Merge remote-tracking branch 'origin/dev' into assign-bits-migrate-misc

# Conflicts:
#	esphome/components/rx8111/rx8111.cpp

AGENTS.md

@@ -216,10 +216,9 @@ file does, and it is the authority when they disagree. The most useful starting
         the helper's arguments instead of writing a loop. 7-byte BCD RTC calendars use `time::decode_bcd_calendar<LAYOUT>`
         and `encode_bcd_calendar<LAYOUT>` with `EPSON_BCD_CALENDAR`, `DS_BCD_CALENDAR` or `PCF_BCD_CALENDAR`.
     *   **Bus helpers:** UART drains with `read_available(on_byte)` and `discard_available()`, text lines with
-        `LineReader<N>`, fixed-length binary frames with `FrameReader<N, HEADER...>`; I2C has `update_register`,
-        `write_registers` (PROGMEM `I2CRegValue` table), `check_chip_id` and `read_bytes_16_le`.
-        `if (!this->status_comm_result(ok)) return;` sets or clears the warning. A child sensor that skips repeated
-        values is `SUB_SENSOR_WITH_DEDUP(name, type)`.
+        `LineReader<N>`; I2C has `update_register`, `write_registers` (PROGMEM `I2CRegValue` table), `check_chip_id`
+        and `read_bytes_16_le`. `if (!this->status_comm_result(ok)) return;` sets or clears the warning.
+        A child sensor that skips repeated values is `SUB_SENSOR_WITH_DEDUP(name, type)`.
     *   **An unset float is `NAN`**, not `optional<float>`; the optional costs 4 to 8 bytes of padding per member.
     *   **`PollingComponent` first runs `update()`** after a random share of the interval; call `this->update()` at the
         end of `setup()` when the entities need a state at boot.

esphome/components/dfrobot_sen0465_uart/dfrobot_sen0465_uart.cpp

@@ -1,7 +1,10 @@
 #include "dfrobot_sen0465_uart.h"
 
+#include <algorithm>
+
 namespace esphome::dfrobot_sen0465_uart {
 
+using dfrobot_sen0465_base::FRAME_HEAD;
 using dfrobot_sen0465_base::FRAME_SIZE;
 using dfrobot_sen0465_base::RESPONSE_TIMEOUT_ID;
 
@@ -15,21 +18,30 @@ void DFRobotSEN0465UARTComponent::setup() {
 }
 
 void DFRobotSEN0465UARTComponent::loop() {
-  // The tail of a reading the probe sent on its own may precede the answer; the reader skips to the frame head.
-  const uint8_t *frame = nullptr;
-  this->read_available<FRAME_SIZE>([&](uint8_t c) { return (frame = this->reader_.feed(c)) == nullptr; });
-  if (frame == nullptr)
+  // The tail of a reading the probe sent on its own may precede the answer; start at the next frame head.
+  uint8_t byte;
+  while (this->rx_count_ == 0 && this->peek_byte(&byte) && byte != FRAME_HEAD)
+    this->read_byte(&byte);
+  const size_t available = this->available();
+  if (available == 0)
+    return;
+  const size_t len = std::min<size_t>(available, FRAME_SIZE - this->rx_count_);
+  if (!this->read_array(this->rx_buffer_ + this->rx_count_, len))
+    return;
+  this->rx_count_ += len;
+  if (this->rx_count_ < FRAME_SIZE)
     return;
 
+  this->rx_count_ = 0;
   this->cancel_timeout(RESPONSE_TIMEOUT_ID);
   this->disable_loop();
-  this->handle_response_(frame);
+  this->handle_response_(this->rx_buffer_);
 }
 
 bool DFRobotSEN0465UARTComponent::write_frame(const uint8_t *frame) {
   // Drop anything left over, such as the answer to the mode change or readings sent before passive mode was set
   this->discard_available();
-  this->reader_.reset();
+  this->rx_count_ = 0;
   this->write_array(frame, FRAME_SIZE);
   return true;
 }
@@ -38,7 +50,7 @@ void DFRobotSEN0465UARTComponent::read_response() {
   this->enable_loop();
   this->set_timeout(RESPONSE_TIMEOUT_ID, RESPONSE_TIMEOUT_MS, [this]() {
     this->disable_loop();
-    this->reader_.reset();
+    this->rx_count_ = 0;
     this->handle_no_response_();
   });
 }

esphome/components/dfrobot_sen0465_uart/dfrobot_sen0465_uart.h

@@ -2,7 +2,6 @@
 
 #include "esphome/components/dfrobot_sen0465_base/dfrobot_sen0465_base.h"
 #include "esphome/components/uart/uart.h"
-#include "esphome/core/frame_reader.h"
 
 namespace esphome::dfrobot_sen0465_uart {
 
@@ -18,7 +17,8 @@ class DFRobotSEN0465UARTComponent final : public dfrobot_sen0465_base::DFRobotSE
   bool write_frame(const uint8_t *frame) override;
   void read_response() override;
 
-  FrameReader<dfrobot_sen0465_base::FRAME_SIZE, dfrobot_sen0465_base::FRAME_HEAD> reader_;
+  uint8_t rx_buffer_[dfrobot_sen0465_base::FRAME_SIZE];
+  uint8_t rx_count_{0};
 };
 
 }  // namespace esphome::dfrobot_sen0465_uart

esphome/components/ds1603l/ds1603l.cpp

@@ -1,5 +1,7 @@
 #include "ds1603l.h"
 
+#include <cstring>
+
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
@@ -8,28 +10,35 @@ namespace esphome::ds1603l {
 ESPHOME_LOG_TAG(TAG, "ds1603l.sensor");
 
 void DS1603L::loop() {
-  this->read_available([this](uint8_t byte) {
-    const uint8_t *frame = this->reader_.feed(byte);
-    if (frame == nullptr) {
-      if (this->reader_.length() == 0) {
-        ESP_LOGV(TAG, "Skipping byte 0x%02X while looking for header", byte);
-      }
-      return;
+  // Assemble frames one byte at a time so a stream that starts mid-frame can realign
+  uint8_t byte;
+  while (this->available() > 0 && this->read_byte(&byte)) {
+    if (this->rx_count_ == 0 && byte != HEADER_BYTE) {
+      ESP_LOGV(TAG, "Skipping byte 0x%02X while looking for header", byte);
+      continue;
     }
-    if (!this->parse_data_(frame)) {
+
+    this->rx_buffer_[this->rx_count_++] = byte;
+    if (this->rx_count_ < FRAME_SIZE) {
+      continue;
+    }
+
+    if (this->parse_data_()) {
+      this->rx_count_ = 0;
+    } else {
       // The header byte was part of the payload of a misaligned frame, so realign instead of dropping everything
-      this->reader_.resync();
+      this->resync_();
     }
-  });
+  }
 }
 
 void DS1603L::dump_config() { LOG_SENSOR("", "DS1603L", this); }
 
-bool DS1603L::parse_data_(const uint8_t *frame) {
-  uint8_t header = frame[0];
-  uint8_t data_h = frame[1];
-  uint8_t data_l = frame[2];
-  uint8_t checksum = frame[3];
+bool DS1603L::parse_data_() {
+  uint8_t header = this->rx_buffer_[0];
+  uint8_t data_h = this->rx_buffer_[1];
+  uint8_t data_l = this->rx_buffer_[2];
+  uint8_t checksum = this->rx_buffer_[3];
 
   uint8_t computed_checksum = (header + data_h + data_l) & 0xFF;
 
@@ -44,4 +53,16 @@ bool DS1603L::parse_data_(const uint8_t *frame) {
   return true;
 }
 
+void DS1603L::resync_() {
+  // Drop the byte that was treated as the header, then look for the next candidate header in what is left
+  size_t start = 1;
+  while (start < this->rx_count_ && this->rx_buffer_[start] != HEADER_BYTE) {
+    start++;
+  }
+  this->rx_count_ -= start;
+  if (this->rx_count_ > 0) {
+    memmove(this->rx_buffer_, this->rx_buffer_ + start, this->rx_count_);
+  }
+}
+
 }  // namespace esphome::ds1603l

esphome/components/ds1603l/ds1603l.h

@@ -6,7 +6,6 @@
 #include "esphome/components/sensor/sensor.h"
 #include "esphome/components/uart/uart.h"
 #include "esphome/core/component.h"
-#include "esphome/core/frame_reader.h"
 
 namespace esphome::ds1603l {
 
@@ -19,10 +18,13 @@ class DS1603L final : public sensor::Sensor, public Component, public uart::UART
   static constexpr uint8_t HEADER_BYTE = 0xFF;
   static constexpr size_t FRAME_SIZE = 4;
 
-  // Validates the checksum of the frame and publishes it. Returns false if the frame is invalid.
-  bool parse_data_(const uint8_t *frame);
+  // Validates the checksum of the frame in rx_buffer_ and publishes it. Returns false if the frame is invalid.
+  bool parse_data_();
+  // Drops the first buffered byte and realigns the buffer on the next possible header byte.
+  void resync_();
 
-  FrameReader<FRAME_SIZE, HEADER_BYTE> reader_;
+  uint8_t rx_buffer_[FRAME_SIZE];  // Buffer for the frame being assembled
+  size_t rx_count_{0};             // Number of bytes currently in rx_buffer_
 };
 
 }  // namespace esphome::ds1603l

esphome/components/ds248x/ds248x_one_wire_bus.cpp

@@ -1,5 +1,6 @@
 #include "ds248x_one_wire_bus.h"
 #include "ds248x.h"
+#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
 namespace esphome::ds248x {
@@ -151,11 +152,7 @@ uint64_t DS248xOneWireBus::search_int() {
     }
 
     // Update address based on direction taken
-    if (dir_taken) {
-      address |= bit_mask;
-    } else {
-      address &= ~bit_mask;
-    }
+    assign_bits(address, bit_mask, dir_taken);
   }
 
   // Search successful

esphome/components/htu21d/htu21d.cpp

@@ -1,6 +1,7 @@
 #include "htu21d.h"
-#include "esphome/core/log.h"
 #include "esphome/core/hal.h"
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
 
 namespace esphome::htu21d {
 
@@ -112,11 +113,7 @@ void HTU21DComponent::set_heater(bool status) {
     this->status_set_warning();
     return;
   }
-  if (status) {
-    raw_heater |= (1 << HTU21D_REG_HTRE_BIT);
-  } else {
-    raw_heater &= ~(1 << HTU21D_REG_HTRE_BIT);
-  }
+  assign_bits(raw_heater, 1 << HTU21D_REG_HTRE_BIT, status);
   if (this->write_register(HTU21D_WRITERHT_REG_CMD, &raw_heater, 1) != i2c::ERROR_OK) {
     this->status_set_warning();
     return;

esphome/components/light/light_call.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "light_color_values.h"
+#include "esphome/core/helpers.h"
 #include "esphome/core/string_ref.h"
 
 namespace esphome {
@@ -229,13 +230,7 @@ class LightCall {
   inline bool get_save_() { return (this->flags_ & FLAG_SAVE) != 0; }
 
   // Helper to set flag - defaults to true for common case
-  void set_flag_(FieldFlags flag, bool value = true) ESPHOME_ALWAYS_INLINE {
-    if (value) {
-      this->flags_ |= flag;
-    } else {
-      this->flags_ &= ~flag;
-    }
-  }
+  void set_flag_(FieldFlags flag, bool value = true) ESPHOME_ALWAYS_INLINE { assign_bits(this->flags_, flag, value); }
 
   // Helper to clear flag - reduces code size for common case
   void clear_flag_(FieldFlags flag) ESPHOME_ALWAYS_INLINE { this->flags_ &= ~flag; }

esphome/components/media_player/media_player.cpp

@@ -1,6 +1,7 @@
 #include "media_player.h"
-#include "esphome/core/defines.h"
 #include "esphome/core/controller_registry.h"
+#include "esphome/core/defines.h"
+#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 #include "esphome/core/progmem.h"
 
@@ -77,19 +78,12 @@ const char *media_player_command_to_string(MediaPlayerCommand command) {
 }
 
 void MediaPlayerTraits::set_supports_pause(bool supports_pause) {
-  if (supports_pause) {
-    this->feature_flags_ |= MediaPlayerEntityFeature::PAUSE | MediaPlayerEntityFeature::PLAY;
-  } else {
-    this->feature_flags_ &= ~(MediaPlayerEntityFeature::PAUSE | MediaPlayerEntityFeature::PLAY);
-  }
+  assign_bits(this->feature_flags_, MediaPlayerEntityFeature::PAUSE | MediaPlayerEntityFeature::PLAY, supports_pause);
 }
 
 void MediaPlayerTraits::set_supports_turn_off_on(bool supports_turn_off_on) {
-  if (supports_turn_off_on) {
-    this->feature_flags_ |= MediaPlayerEntityFeature::TURN_OFF | MediaPlayerEntityFeature::TURN_ON;
-  } else {
-    this->feature_flags_ &= ~(MediaPlayerEntityFeature::TURN_OFF | MediaPlayerEntityFeature::TURN_ON);
-  }
+  assign_bits(this->feature_flags_, MediaPlayerEntityFeature::TURN_OFF | MediaPlayerEntityFeature::TURN_ON,
+              supports_turn_off_on);
 }
 
 void MediaPlayerCall::validate_() {

esphome/components/mpr121/mpr121.cpp

@@ -3,6 +3,7 @@
 #include <cstdint>
 
 #include "esphome/core/hal.h"
+#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
 namespace esphome::mpr121 {
@@ -84,11 +85,7 @@ bool MPR121Component::digital_read(uint8_t pin) { return (this->gpio_input_ & (1
 
 void MPR121Component::digital_write(uint8_t pin, bool value) {
   const uint8_t ionum = pin - 4;
-  if (value) {
-    this->gpio_output_ |= (1 << ionum);
-  } else {
-    this->gpio_output_ &= ~(1 << ionum);
-  }
+  assign_bits(this->gpio_output_, 1 << ionum, value);
   this->flush_gpio_();
 }
 

esphome/components/provisioning/provisioning.h

@@ -46,11 +46,7 @@ class ProvisioningManager : public Component {
   void set_source_provisioned(uint8_t source, bool provisioned) {
     if (source >= MAX_SOURCES)
       return;
-    if (provisioned) {
-      this->provisioned_mask_ |= (1UL << source);
-    } else {
-      this->provisioned_mask_ &= ~(1UL << source);
-    }
+    assign_bits(this->provisioned_mask_, 1UL << source, provisioned);
   }
 
   // True once every registered source is provisioned. Config validation guarantees

esphome/components/sm300d2/sm300d2.cpp

@@ -1,27 +1,33 @@
 #include "sm300d2.h"
-#include "esphome/core/frame_reader.h"
 #include "esphome/core/log.h"
 
 namespace esphome::sm300d2 {
 
 ESPHOME_LOG_TAG(TAG, "sm300d2");
-static constexpr uint8_t SM300D2_RESPONSE_LENGTH = 17;
+static const uint8_t SM300D2_RESPONSE_LENGTH = 17;
 
 void SM300D2Sensor::update() {
-  // Takes the first frame already received; only one cut off at the end of the buffer is waited for
-  FrameReader<SM300D2_RESPONSE_LENGTH, 0x3C, 0x02> reader;
-  const uint8_t *response = nullptr;
-  this->read_available<1>([&](uint8_t c) { return (response = reader.feed(c)) == nullptr; });
-  for (uint8_t c; response == nullptr && reader.length() > 0 && this->read_byte(&c);)
-    response = reader.feed(c);
+  uint8_t response[SM300D2_RESPONSE_LENGTH];
+  uint8_t peeked;
 
-  if (response == nullptr) {
+  while (this->available() > 0 && this->peek_byte(&peeked) && peeked != 0x3C)
+    this->read();
+
+  bool read_success = read_array(response, SM300D2_RESPONSE_LENGTH);
+
+  if (!read_success) {
     ESP_LOGW(TAG, "Reading data from SM300D2 failed!");
+    status_set_warning();
+    return;
+  }
+
+  if (response[0] != 0x3C || response[1] != 0x02) {
+    ESP_LOGW(TAG, "Invalid preamble for SM300D2 response!");
     this->status_set_warning();
     return;
   }
 
-  uint16_t calculated_checksum = sum_bytes(response, SM300D2_RESPONSE_LENGTH - 1);
+  uint16_t calculated_checksum = this->sm300d2_checksum_(response);
   // Occasionally the checksum has a +/- 0x80 offset. Negative temperatures are
   // responsible for some of these. The rest are unknown/undocumented.
   if ((calculated_checksum != response[SM300D2_RESPONSE_LENGTH - 1]) &&
@@ -77,6 +83,8 @@ void SM300D2Sensor::update() {
     this->humidity_sensor_->publish_state(humidity);
 }
 
+uint16_t SM300D2Sensor::sm300d2_checksum_(uint8_t *ptr) { return sum_bytes(ptr, SM300D2_RESPONSE_LENGTH - 1); }
+
 void SM300D2Sensor::dump_config() {
   ESP_LOGCONFIG(TAG, "SM300D2:");
   LOG_SENSOR("  ", "CO2", this->co2_sensor_);

esphome/components/sm300d2/sm300d2.h

@@ -20,6 +20,8 @@ class SM300D2Sensor final : public PollingComponent, public uart::UARTDevice {
   void dump_config() override;
 
  protected:
+  uint16_t sm300d2_checksum_(uint8_t *ptr);
+
   sensor::Sensor *co2_sensor_{nullptr};
   sensor::Sensor *formaldehyde_sensor_{nullptr};
   sensor::Sensor *tvoc_sensor_{nullptr};

esphome/components/socket/bsd_sockets_impl.cpp

@@ -62,11 +62,7 @@ int BSDSocketImpl::setblocking(bool blocking) {
   if (fl < 0) {
     return fl;
   }
-  if (blocking) {
-    fl &= ~O_NONBLOCK;
-  } else {
-    fl |= O_NONBLOCK;
-  }
+  assign_bits(fl, O_NONBLOCK, !blocking);
   return ::fcntl(this->fd_, F_SETFL, fl);
 }
 

esphome/components/socket/lwip_sockets_impl.cpp

@@ -52,11 +52,7 @@ int LwIPSocketImpl::setblocking(bool blocking) {
   if (fl < 0) {
     return fl;
   }
-  if (blocking) {
-    fl &= ~O_NONBLOCK;
-  } else {
-    fl |= O_NONBLOCK;
-  }
+  assign_bits(fl, O_NONBLOCK, !blocking);
   return lwip_fcntl(this->fd_, F_SETFL, fl);
 }
 

esphome/components/uart/uart_component_host.cpp

@@ -162,11 +162,7 @@ void HostUartComponent::setup() {
       break;
   };
   // Set stop bits
-  if (this->stop_bits_ == 2) {
-    options.c_cflag |= CSTOPB;
-  } else {
-    options.c_cflag &= ~CSTOPB;
-  }
+  assign_bits(options.c_cflag, CSTOPB, this->stop_bits_ == 2);
   cfsetispeed(&options, baud);
   cfsetospeed(&options, baud);
   tcsetattr(this->file_descriptor_, TCSANOW, &options);

esphome/components/ufire_ec/ufire_ec.cpp

@@ -1,3 +1,4 @@
+#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 #include "ufire_ec.h"
 #include <cmath>
@@ -19,11 +20,8 @@ void UFireECComponent::setup() {
   // Write option for temperature adjustments
   uint8_t config;
   this->read_byte(REGISTER_CONFIG, &config);
-  if (this->temperature_sensor_ == nullptr && this->temperature_sensor_external_ == nullptr) {
-    config &= ~CONFIG_TEMP_COMPENSATION;
-  } else {
-    config |= CONFIG_TEMP_COMPENSATION;
-  }
+  assign_bits(config, CONFIG_TEMP_COMPENSATION,
+              this->temperature_sensor_ != nullptr || this->temperature_sensor_external_ != nullptr);
   this->write_byte(REGISTER_CONFIG, config);
 
   // Update temperature compensation

esphome/components/ufire_ise/ufire_ise.cpp

@@ -1,3 +1,4 @@
+#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 #include "ufire_ise.h"
 
@@ -20,11 +21,8 @@ void UFireISEComponent::setup() {
   // Write option for temperature adjustments
   uint8_t config;
   this->read_byte(REGISTER_CONFIG, &config);
-  if (this->temperature_sensor_ == nullptr && this->temperature_sensor_external_ == nullptr) {
-    config &= ~CONFIG_TEMP_COMPENSATION;
-  } else {
-    config |= CONFIG_TEMP_COMPENSATION;
-  }
+  assign_bits(config, CONFIG_TEMP_COMPENSATION,
+              this->temperature_sensor_ != nullptr || this->temperature_sensor_external_ != nullptr);
   this->write_byte(REGISTER_CONFIG, config);
 }
 

esphome/components/ufm01/ufm01.cpp

@@ -44,6 +44,8 @@ static constexpr std::array<uint8_t, 7> GET_SOFTWARE_VERSION = {0xFE, 0xFE, 0x11
 static constexpr size_t FRAME_DEVICE_ID_INDEX = 2;
 static constexpr size_t FRAME_CHECKSUM_INDEX = 30;
 static constexpr size_t FRAME_STOP_INDEX = 31;
+static constexpr uint8_t FRAME_START_BYTE_1 = 0x3C;
+static constexpr uint8_t FRAME_START_BYTE_2 = 0x32;
 static constexpr uint8_t PASSIVE_START_BYTE_2 = 0x64;
 static constexpr uint8_t PASSIVE_START_BYTE_2_WITH_ID = 0x96;
 static constexpr uint8_t FRAME_STOP_BYTE = 0x16;
@@ -118,8 +120,9 @@ static bool check_byte(const uint8_t data[FRAME_SIZE], size_t index, uint8_t exp
 
 static bool validate_active_frame(const uint8_t data[FRAME_SIZE]) {
   const uint8_t sum = sum_bytes(data, FRAME_CHECKSUM_INDEX);
-  // FrameReader already matched the start bytes
-  return check_byte(data, FRAME_INDEX_INSTANT_FLOW_FLAG, FRAME_FLAG_INSTANT_FLOW, "instant flow flag") &&
+  return check_byte(data, 0, FRAME_START_BYTE_1, "start byte 1") &&
+         check_byte(data, 1, FRAME_START_BYTE_2, "start byte 2") &&
+         check_byte(data, FRAME_INDEX_INSTANT_FLOW_FLAG, FRAME_FLAG_INSTANT_FLOW, "instant flow flag") &&
          check_byte(data, FRAME_INDEX_RESERVED_SECTION, FRAME_FLAG_RESERVED_SECTION, "reserved section flag") &&
          check_byte(data, FRAME_INDEX_TEMP_FLAG, FRAME_FLAG_TEMP, "temperature flag") &&
          check_byte(data, FRAME_CHECKSUM_INDEX, sum, "checksum") &&
@@ -212,7 +215,7 @@ static bool read_flow_rate_out_of_range(const uint8_t data[FRAME_SIZE]) {
 
 void UFM01Component::flush_rx_() {
   this->discard_available();
-  this->active_reader_.reset();
+  this->read_index_ = 0;
 }
 
 void UFM01Component::send_command_no_wait_(const std::array<uint8_t, 7> &command) {
@@ -420,7 +423,7 @@ bool UFM01Component::service_software_version_(uint32_t now) {
 }
 #endif  // USE_UFM01_SOFTWARE_VERSION
 
-void UFM01Component::on_active_frame_(const uint8_t data[FRAME_SIZE], uint32_t now) {
+void UFM01Component::on_active_frame_(uint8_t data[FRAME_SIZE], uint32_t now) {
 #ifdef USE_UFM01_METER_ID
   this->publish_meter_id_from_frame_(data);
 #endif
@@ -458,26 +461,40 @@ void UFM01Component::on_active_frame_(const uint8_t data[FRAME_SIZE], uint32_t n
 bool UFM01Component::process_active_stream_(uint32_t now) {
   bool got_valid_frame = false;
 
-  const bool read_ok = this->read_available([this, now, &got_valid_frame](uint8_t byte) {
-    const uint8_t *frame = this->active_reader_.feed(byte);
-    if (frame == nullptr) {
-      if (this->active_reader_.length() == 0) {
-        ESP_LOGD(TAG, "not start of data (is 0x%02X)", byte);
-      }
-      return;
+  while (this->available()) {
+    if (!this->read_byte(&this->data_[this->read_index_])) {
+      ESP_LOGW(TAG, "unable to read byte");
+      this->read_index_ = 0;
+      continue;
     }
-    if (validate_active_frame(frame)) {
-      this->on_active_frame_(frame, now);
+    if ((this->read_index_ == 0 && this->data_[0] != FRAME_START_BYTE_1) ||
+        (this->read_index_ == 1 && this->data_[1] != FRAME_START_BYTE_2)) {
+      ESP_LOGD(TAG, "not start of data at %" PRId32 " (is 0x%02X)", this->read_index_, this->data_[this->read_index_]);
+      this->read_index_ = 0;
+      continue;
+    }
+    if (++this->read_index_ < static_cast<int32_t>(FRAME_SIZE))
+      continue;
+
+    if (validate_active_frame(this->data_)) {
+      this->on_active_frame_(this->data_, now);
+      this->read_index_ = 0;
       got_valid_frame = true;
-      return;
+      continue;
     }
-    log_hex(frame, FRAME_SIZE);
+
+    log_hex(this->data_, sizeof(this->data_));
     ESP_LOGW(TAG, "unable to read data");
-    this->active_reader_.resync();
-  });
-  if (!read_ok) {
-    ESP_LOGW(TAG, "unable to read byte");
-    this->active_reader_.reset();
+    for (int32_t i = 2;
+         i < static_cast<int32_t>(FRAME_STOP_INDEX) && this->read_index_ == static_cast<int32_t>(FRAME_SIZE); ++i) {
+      if ((this->data_[i] == FRAME_START_BYTE_1) && (this->data_[i + 1] == FRAME_START_BYTE_2)) {
+        for (int32_t j = i; j < static_cast<int32_t>(FRAME_SIZE); ++j)
+          this->data_[j - i] = this->data_[j];
+        this->read_index_ = static_cast<int32_t>(FRAME_SIZE) - i;
+      }
+    }
+    if (this->read_index_ == static_cast<int32_t>(FRAME_SIZE))
+      this->read_index_ = 0;
   }
 
   return got_valid_frame;

esphome/components/ufm01/ufm01.h

@@ -2,7 +2,6 @@
 
 #include "esphome/core/component.h"
 #include "esphome/core/defines.h"
-#include "esphome/core/frame_reader.h"
 #ifdef USE_BINARY_SENSOR
 #include "esphome/components/binary_sensor/binary_sensor.h"
 #endif
@@ -26,8 +25,6 @@ class TestableUFM01;
 }  // namespace testing
 
 static constexpr size_t FRAME_SIZE = 32;
-static constexpr uint8_t FRAME_START_BYTE_1 = 0x3C;
-static constexpr uint8_t FRAME_START_BYTE_2 = 0x32;
 static constexpr size_t PASSIVE_FRAME_SIZE = 23;
 static constexpr size_t PASSIVE_FRAME_WITH_ID_SIZE = 39;
 #ifdef USE_UFM01_METER_ID
@@ -111,7 +108,7 @@ class UFM01Component : public uart::UARTDevice, public Component {
   bool consume_ack_();
   void flush_rx_();
   bool process_active_stream_(uint32_t now);
-  void on_active_frame_(const uint8_t data[FRAME_SIZE], uint32_t now);
+  void on_active_frame_(uint8_t data[FRAME_SIZE], uint32_t now);
 #ifdef USE_UFM01_METER_ID
   void publish_meter_id_from_frame_(const uint8_t data[FRAME_SIZE]);
 #endif
@@ -148,6 +145,7 @@ class UFM01Component : public uart::UARTDevice, public Component {
 #ifdef USE_UFM01_CLEAR_ACCUMULATED_FLOW_ACTION
   uint32_t pending_clear_start_ms_{0};
 #endif
+  int32_t read_index_{0};
 
   OperatingMode operating_mode_{OperatingMode::STARTUP};
   StartupPhase startup_phase_{StartupPhase::WAIT};
@@ -172,7 +170,7 @@ class UFM01Component : public uart::UARTDevice, public Component {
 #endif
 
   uint8_t passive_frame_[PASSIVE_FRAME_MAX_SIZE];
-  FrameReader<FRAME_SIZE, FRAME_START_BYTE_1, FRAME_START_BYTE_2> active_reader_;
+  uint8_t data_[FRAME_SIZE];
 
   friend class testing::TestableUFM01;
 };

esphome/components/water_heater/water_heater.cpp

@@ -5,6 +5,7 @@
 #include "esphome/core/application.h"
 #include "esphome/core/controller_registry.h"
 #include "esphome/core/progmem.h"
+#include "esphome/core/helpers.h"
 
 #include <cmath>
 #include <cstring>
@@ -59,21 +60,13 @@ WaterHeaterCall &WaterHeaterCall::set_target_temperature_high(float temperature)
 }
 
 WaterHeaterCall &WaterHeaterCall::set_away(bool away) {
-  if (away) {
-    this->state_ |= WATER_HEATER_STATE_AWAY;
-  } else {
-    this->state_ &= ~WATER_HEATER_STATE_AWAY;
-  }
+  assign_bits(this->state_, WATER_HEATER_STATE_AWAY, away);
   this->state_mask_ |= WATER_HEATER_STATE_AWAY;
   return *this;
 }
 
 WaterHeaterCall &WaterHeaterCall::set_on(bool on) {
-  if (on) {
-    this->state_ |= WATER_HEATER_STATE_ON;
-  } else {
-    this->state_ &= ~WATER_HEATER_STATE_ON;
-  }
+  assign_bits(this->state_, WATER_HEATER_STATE_ON, on);
   this->state_mask_ |= WATER_HEATER_STATE_ON;
   return *this;
 }

esphome/components/water_heater/water_heater.h

@@ -252,13 +252,7 @@ class WaterHeater : public EntityBase {
   /// Set the state flags. Should only be called from control().
   void set_state_(uint32_t state) { this->state_ = state; }
   /// Set or clear a state flag. Should only be called from control().
-  void set_state_flag_(uint32_t flag, bool value) {
-    if (value) {
-      this->state_ |= flag;
-    } else {
-      this->state_ &= ~flag;
-    }
-  }
+  void set_state_flag_(uint32_t flag, bool value) { assign_bits(this->state_, flag, value); }
 
   WaterHeaterMode mode_{WATER_HEATER_MODE_OFF};
   float current_temperature_{NAN};

esphome/components/wts01/wts01.cpp

@@ -4,38 +4,70 @@
 
 namespace esphome::wts01 {
 
+constexpr uint8_t HEADER_1 = 0x55;
+constexpr uint8_t HEADER_2 = 0x01;
+constexpr uint8_t HEADER_3 = 0x01;
+constexpr uint8_t HEADER_4 = 0x04;
+
 ESPHOME_LOG_TAG(TAG, "wts01");
 
 void WTS01Sensor::loop() {
-  this->read_available([this](uint8_t c) {
-    if (const uint8_t *packet = this->reader_.feed(c)) {
-      this->process_packet_(packet);
-    }
-  });
+  this->read_available([this](uint8_t c) { this->handle_char_(c); });
 }
 
 void WTS01Sensor::dump_config() { LOG_SENSOR("", "WTS01 Sensor", this); }
 
-void WTS01Sensor::process_packet_(const uint8_t *packet) {
+void WTS01Sensor::handle_char_(uint8_t c) {
+  // State machine for processing the header. Reset if something doesn't match.
+  if (this->buffer_pos_ == 0 && c != HEADER_1) {
+    return;
+  }
+
+  if (this->buffer_pos_ == 1 && c != HEADER_2) {
+    this->buffer_pos_ = 0;
+    return;
+  }
+
+  if (this->buffer_pos_ == 2 && c != HEADER_3) {
+    this->buffer_pos_ = 0;
+    return;
+  }
+
+  if (this->buffer_pos_ == 3 && c != HEADER_4) {
+    this->buffer_pos_ = 0;
+    return;
+  }
+
+  // Add byte to buffer
+  this->buffer_[this->buffer_pos_++] = c;
+
+  // Process complete packet
+  if (this->buffer_pos_ >= PACKET_SIZE) {
+    this->process_packet_();
+    this->buffer_pos_ = 0;
+  }
+}
+
+void WTS01Sensor::process_packet_() {
   // Based on Tasmota implementation
   // Format: 55 01 01 04 01 11 16 12 95
   // header            T  Td Ck  - T = Temperature, Td = Temperature decimal, Ck = Checksum
-  const uint8_t calculated_checksum = sum_bytes(packet, PACKET_SIZE - 1);
+  const uint8_t calculated_checksum = sum_bytes(this->buffer_, PACKET_SIZE - 1);
 
-  uint8_t received_checksum = packet[PACKET_SIZE - 1];
+  uint8_t received_checksum = this->buffer_[PACKET_SIZE - 1];
   if (calculated_checksum != received_checksum) {
     ESP_LOGW(TAG, "WTS01 Checksum doesn't match: 0x%02X != 0x%02X", received_checksum, calculated_checksum);
     return;
   }
 
   // Extract temperature value
-  const uint8_t raw = packet[6];
+  const uint8_t raw = this->buffer_[6];
 
   // WTS01 encodes sign in bit 7, magnitude in bits 0-6
   const bool negative = (raw & 0x80) != 0;
   const uint8_t magnitude = raw & 0x7F;
 
-  const float decimal = static_cast<float>(packet[7]) / 100.0f;
+  const float decimal = static_cast<float>(this->buffer_[7]) / 100.0f;
 
   float temperature = static_cast<float>(magnitude) + decimal;
 

esphome/components/wts01/wts01.h

@@ -1,7 +1,6 @@
 #pragma once
 
 #include "esphome/core/component.h"
-#include "esphome/core/frame_reader.h"
 #include "esphome/components/sensor/sensor.h"
 #include "esphome/components/uart/uart.h"
 
@@ -15,9 +14,11 @@ class WTS01Sensor final : public sensor::Sensor, public uart::UARTDevice, public
   void dump_config() override;
 
  protected:
-  void process_packet_(const uint8_t *packet);
+  uint8_t buffer_[PACKET_SIZE];
+  uint8_t buffer_pos_{0};
 
-  FrameReader<PACKET_SIZE, 0x55, 0x01, 0x01, 0x04> reader_;
+  void handle_char_(uint8_t c);
+  void process_packet_();
 };
 
 }  // namespace esphome::wts01

esphome/core/frame_reader.h

@@ -1,67 +0,0 @@
-#pragma once
-
-#include <cstddef>
-#include <cstdint>
-#include <type_traits>
-
-namespace esphome {
-
-/// Collects fixed length binary frames of N bytes that start with the header bytes HEADER...; the binary twin of
-/// LineReader. Bytes before a header are skipped; checksums are left to the caller.
-template<size_t N, uint8_t... HEADER> class FrameReader {
-  using pos_t = std::conditional_t<(N <= 0xFF), uint8_t, uint16_t>;
-  static constexpr size_t HEADER_LEN = sizeof...(HEADER);
-  static_assert(HEADER_LEN >= 1 && HEADER_LEN <= 8, "FrameReader takes 1 to 8 header bytes");  // bounds replay_() depth
-  static_assert(HEADER_LEN < N && N <= 0xFFFF, "FrameReader size out of range");
-
- public:
-  /// Feed one byte; returns the completed frame, header included, valid until the next call, or nullptr.
-  const uint8_t *feed(uint8_t c) {
-    pos_t pos = this->pos_;
-    if (pos < HEADER_LEN && c != header_byte(pos)) {
-      // A later byte of the broken header, or c, may start the real one
-      this->buf_[pos] = c;
-      this->replay_(pos);
-      return nullptr;
-    }
-    this->buf_[pos++] = c;
-    if (pos < N) {
-      this->pos_ = pos;
-      return nullptr;
-    }
-    this->pos_ = 0;
-    return this->buf_;
-  }
-
-  /// Number of bytes of the partial frame; 0 after a frame is returned and while looking for a header.
-  size_t length() const { return this->pos_; }
-
-  /// Forget the partial frame and look for a header from the next byte.
-  void reset() { this->pos_ = 0; }
-
-  /// For a frame the caller rejected (bad checksum): drop its first byte and realign on a header in the rest.
-  /// Call only right after feed() returned that frame; use reset() otherwise.
-  void resync() { this->replay_(N - 1); }
-
- protected:
-  // A fold, not an array, so no header table lands in RAM on ESP8266
-  static constexpr uint8_t header_byte(size_t i) {
-    size_t j = 0;
-    uint8_t r = 0;
-    ((r = (j++ == i) ? HEADER : r), ...);
-    return r;
-  }
-
-  // Start over and feed buf_[1..last] again; never completes a frame as last < N
-  void replay_(size_t last) {
-    this->pos_ = 0;
-    for (size_t i = 1; i <= last; i++) {
-      this->feed(this->buf_[i]);
-    }
-  }
-
-  uint8_t buf_[N];
-  pos_t pos_{0};
-};
-
-}  // namespace esphome

tests/components/core/frame_reader_test.cpp

@@ -1,143 +0,0 @@
-#include <gtest/gtest.h>
-#include <vector>
-#include "esphome/core/frame_reader.h"
-
-namespace esphome::core::testing {
-
-using Bytes = std::vector<uint8_t>;
-using Frames = std::vector<Bytes>;
-
-// Feeds bytes and collects every completed frame.
-template<size_t N, uint8_t... H> static Frames feed_all(FrameReader<N, H...> &reader, const Bytes &input) {
-  Frames frames;
-  for (uint8_t c : input) {
-    const uint8_t *frame = reader.feed(c);
-    if (frame != nullptr) {
-      EXPECT_EQ(reader.length(), 0u);
-      frames.emplace_back(frame, frame + N);
-    }
-  }
-  return frames;
-}
-
-static_assert(sizeof(FrameReader<4, 0xFF>) == 5);
-static_assert(sizeof(FrameReader<9, 0x55, 0x01, 0x01, 0x04>) == 10);
-static_assert(sizeof(FrameReader<0xFF, 0xAA>) == 0x100);
-static_assert(sizeof(FrameReader<0x100, 0xAA>) == 0x102);
-
-TEST(FrameReaderTest, SingleByteHeader) {
-  FrameReader<4, 0xFF> reader;
-  EXPECT_EQ(feed_all(reader, {0xFF, 0x01, 0x02, 0x02, 0xFF, 0x03, 0x04, 0x06}),
-            (Frames{{0xFF, 0x01, 0x02, 0x02}, {0xFF, 0x03, 0x04, 0x06}}));
-}
-
-TEST(FrameReaderTest, SkipsBytesBeforeHeader) {
-  FrameReader<3, 0xFF> reader;
-  EXPECT_EQ(feed_all(reader, {0x01, 0x02, 0xFF, 0x10, 0x20}), (Frames{{0xFF, 0x10, 0x20}}));
-}
-
-TEST(FrameReaderTest, HeaderByteInsidePayloadIsData) {
-  FrameReader<4, 0xFF> reader;
-  EXPECT_EQ(feed_all(reader, {0xFF, 0xFF, 0xFF, 0x00}), (Frames{{0xFF, 0xFF, 0xFF, 0x00}}));
-}
-
-TEST(FrameReaderTest, LengthTracksPartialFrame) {
-  FrameReader<4, 0xAA, 0xBB> reader;
-  EXPECT_EQ(reader.length(), 0u);
-  reader.feed(0x00);
-  EXPECT_EQ(reader.length(), 0u);
-  reader.feed(0xAA);
-  EXPECT_EQ(reader.length(), 1u);
-  reader.feed(0xBB);
-  reader.feed(0x01);
-  EXPECT_EQ(reader.length(), 3u);
-  reader.reset();
-  EXPECT_EQ(reader.length(), 0u);
-  EXPECT_EQ(feed_all(reader, {0x02, 0xAA, 0xBB, 0x03, 0x04}), (Frames{{0xAA, 0xBB, 0x03, 0x04}}));
-}
-
-TEST(FrameReaderTest, MultiByteHeaderMismatchRestarts) {
-  FrameReader<6, 0x55, 0x01, 0x01, 0x04> reader;
-  EXPECT_EQ(feed_all(reader, {0x55, 0x01, 0x02, 0x55, 0x01, 0x01, 0x04, 0x11, 0x22}),
-            (Frames{{0x55, 0x01, 0x01, 0x04, 0x11, 0x22}}));
-}
-
-TEST(FrameReaderTest, MismatchedByteCanStartHeader) {
-  FrameReader<6, 0x55, 0x01, 0x01, 0x04> reader;
-  // The second 0x55 breaks the first header and starts the real one
-  EXPECT_EQ(feed_all(reader, {0x55, 0x01, 0x55, 0x01, 0x01, 0x04, 0x11, 0x22}),
-            (Frames{{0x55, 0x01, 0x01, 0x04, 0x11, 0x22}}));
-}
-
-TEST(FrameReaderTest, RepeatedHeaderPrefixIsKept) {
-  // AA AA AA BB: the mismatch at the third byte still leaves AA AA matched
-  FrameReader<4, 0xAA, 0xAA, 0xBB> reader;
-  EXPECT_EQ(feed_all(reader, {0xAA, 0xAA, 0xAA, 0xBB, 0x01}), (Frames{{0xAA, 0xAA, 0xBB, 0x01}}));
-}
-
-TEST(FrameReaderTest, OverlappingHeaderPrefix) {
-  // A1 B2 A1 B2: the mismatch on the last B2 keeps the trailing A1 B2 as the start of the header
-  FrameReader<5, 0xA1, 0xB2, 0xA1, 0xC3> reader;
-  EXPECT_EQ(feed_all(reader, {0xA1, 0xB2, 0xA1, 0xB2, 0xA1, 0xC3, 0x07}), (Frames{{0xA1, 0xB2, 0xA1, 0xC3, 0x07}}));
-}
-
-TEST(FrameReaderTest, ResyncFindsHeaderInRejectedFrame) {
-  FrameReader<4, 0xFF> reader;
-  // Started mid-frame: the 0xFF taken as a header was payload; the real frame starts at the second 0xFF
-  const uint8_t *frame = nullptr;
-  for (uint8_t c : {0xFF, 0x12, 0xFF, 0x01}) {
-    frame = reader.feed(c);
-  }
-  ASSERT_NE(frame, nullptr);
-  reader.resync();
-  EXPECT_EQ(reader.length(), 2u);
-  EXPECT_EQ(feed_all(reader, {0x02, 0x02}), (Frames{{0xFF, 0x01, 0x02, 0x02}}));
-}
-
-TEST(FrameReaderTest, ResyncWithoutHeaderStartsOver) {
-  FrameReader<4, 0xFF> reader;
-  const uint8_t *frame = nullptr;
-  for (uint8_t c : {0xFF, 0x01, 0x02, 0x03}) {
-    frame = reader.feed(c);
-  }
-  ASSERT_NE(frame, nullptr);
-  reader.resync();
-  EXPECT_EQ(reader.length(), 0u);
-}
-
-TEST(FrameReaderTest, ResyncKeepsPartialHeaderAtEnd) {
-  FrameReader<5, 0x3C, 0x32> reader;
-  const uint8_t *frame = nullptr;
-  for (uint8_t c : {0x3C, 0x32, 0x00, 0x00, 0x3C}) {
-    frame = reader.feed(c);
-  }
-  ASSERT_NE(frame, nullptr);
-  reader.resync();
-  EXPECT_EQ(reader.length(), 1u);
-  EXPECT_EQ(feed_all(reader, {0x32, 0x01, 0x02, 0x03}), (Frames{{0x3C, 0x32, 0x01, 0x02, 0x03}}));
-}
-
-TEST(FrameReaderTest, ResyncMultiByteHeader) {
-  FrameReader<6, 0x3C, 0x32> reader;
-  const uint8_t *frame = nullptr;
-  for (uint8_t c : {0x3C, 0x32, 0x3C, 0x3C, 0x32, 0x09}) {
-    frame = reader.feed(c);
-  }
-  ASSERT_NE(frame, nullptr);
-  reader.resync();
-  EXPECT_EQ(reader.length(), 3u);
-  EXPECT_EQ(feed_all(reader, {0x0A, 0x0B, 0x0C}), (Frames{{0x3C, 0x32, 0x09, 0x0A, 0x0B, 0x0C}}));
-}
-
-TEST(FrameReaderTest, SixteenBitIndex) {
-  FrameReader<300, 0xAA, 0x55> reader;
-  Bytes input{0x00, 0xAA, 0x55};
-  input.resize(input.size() + 298, 0x42);
-  Frames frames = feed_all(reader, input);
-  ASSERT_EQ(frames.size(), 1u);
-  EXPECT_EQ(frames[0][0], 0xAA);
-  EXPECT_EQ(frames[0][1], 0x55);
-  EXPECT_EQ(frames[0][299], 0x42);
-}
-
-}  // namespace esphome::core::testing

tests/components/ufm01/common.h

@@ -20,6 +20,8 @@
 
 namespace esphome::ufm01::testing {
 
+static constexpr uint8_t FRAME_START_BYTE_1 = 0x3C;
+static constexpr uint8_t FRAME_START_BYTE_2 = 0x32;
 static constexpr uint8_t PASSIVE_START_BYTE_2 = 0x64;
 static constexpr uint8_t PASSIVE_START_BYTE_2_WITH_ID = 0x96;
 static constexpr uint8_t FRAME_STOP_BYTE = 0x16;
@@ -129,7 +131,7 @@ class TestableUFM01 : public UFM01Component {
 
   StartupPhase startup_phase() const { return this->startup_phase_; }
 
-  int32_t read_index() const { return this->active_reader_.length(); }
+  int32_t read_index() const { return this->read_index_; }
 
   uint8_t passive_index() const { return this->passive_index_; }
 
@@ -163,7 +165,7 @@ class TestableUFM01 : public UFM01Component {
     this->operating_mode_ = OperatingMode::ACTIVE_STREAM;
     this->last_valid_frame_ms_ = now();
     this->passive_read_pending_ = false;
-    this->active_reader_.reset();
+    this->read_index_ = 0;
   }
 
   void begin_active_wait_frame() {
@@ -180,7 +182,7 @@ class TestableUFM01 : public UFM01Component {
   }
 
   void reset_state() {
-    this->active_reader_.reset();
+    this->read_index_ = 0;
     this->last_valid_frame_ms_ = 0;
     this->passive_index_ = 0;
     this->passive_read_pending_ = false;

tests/components/ufm01/ufm01_frame_test.cpp

@@ -30,15 +30,6 @@ TEST_F(UFM01Test, InvalidActiveFrameChecksumRejected) {
   EXPECT_EQ(this->ufm01_.last_valid_frame_ms(), 0u);
 }
 
-TEST_F(UFM01Test, RejectedActiveFrameResyncsOnStartBytesInside) {
-  // A cut off frame start makes the first 32 bytes invalid; the real frame starts inside them
-  this->mock_uart_.enqueue({FRAME_START_BYTE_1, FRAME_START_BYTE_2, 0x00, 0x00});
-  this->mock_uart_.enqueue(make_active_frame());
-
-  EXPECT_TRUE(this->ufm01_.process_active_stream());
-  EXPECT_EQ(this->ufm01_.read_index(), 0);
-}
-
 TEST_F(UFM01Test, ValidPassiveFrameReadSuccess) {
   auto frame = make_passive_frame();
   this->mock_uart_.enqueue(frame);

tests/integration/README.md

@@ -58,8 +58,30 @@ The `InitialStateHelper` class solves a common problem in integration tests: whe
 - Provides debug logging to track state reception (use `--log-cli-level=DEBUG`)
 - Safe for concurrent use with multiple entity types
 
-**Future work:**
-Consider converting existing integration tests to use `InitialStateHelper` for more reliable state tracking and to eliminate race conditions related to initial state broadcasts.
+#### Subscribing after the initial states (`state_utils.py`)
+
+`subscribe_after_initial_states()` does the `InitialStateHelper` setup in one call: it subscribes, waits
+for every entity's connect time state, and fails the test with the entities still missing on timeout.
+It returns a `StateSubscription` with `initial_states` (by key) and a `waiter`, a `StateWaiter` that sees
+every later state. `waiter.next_state(key=..., state_type=...)` arms a wait for the next matching state;
+call it before sending the command. `press_and_wait()` presses a button and returns the next state of a
+type, from one entity when `key=` is given:
+
+```python
+from .state_utils import press_and_wait, subscribe_after_initial_states
+
+entities, _ = await client.list_entities_services()
+sub = await subscribe_after_initial_states(client, entities)
+
+state = await press_and_wait(client, entities, sub.waiter, "set_position", CoverState)
+assert state.position == pytest.approx(0.5, abs=0.01)
+
+next_light = sub.waiter.next_state(key=light.key, state_type=LightState)
+client.light_command(key=light.key, state=True)
+assert (await next_light).state is True
+```
+
+Pass `on_state=` as well when the test also needs to collect every state.
 
 #### Entity Lookup Helpers (`state_utils.py`)
 
@@ -203,46 +225,24 @@ async def test_my_sensor(
 
 ##### State Subscription Pattern
 
-**Recommended: Using InitialStateHelper**
+**Recommended: Using subscribe_after_initial_states**
 
-When an API client connects, ESPHome automatically sends the current state of all entities. The `InitialStateHelper` (from `state_utils.py`) handles this by swallowing these initial states and only forwarding subsequent state changes to your test callback:
+When an API client connects, ESPHome automatically sends the current state of all entities.
+`subscribe_after_initial_states()` (from `state_utils.py`) swallows these initial states and only forwards
+subsequent state changes, to your callback and to the returned `waiter`:
 
 ```python
-from .state_utils import InitialStateHelper
+from .state_utils import subscribe_after_initial_states
 
-# Track state changes with futures
-loop = asyncio.get_running_loop()
-states: dict[int, EntityState] = {}
-state_future: asyncio.Future[EntityState] = loop.create_future()
-
-
-def on_state(state: EntityState) -> None:
-    """This callback only receives NEW state changes, not initial states."""
-    states[state.key] = state
-    # Check for specific condition using isinstance
-    if isinstance(state, SensorState) and state.state == expected_value:
-        if not state_future.done():
-            state_future.set_result(state)
-
-
-# Get entities and set up state synchronization
 entities, services = await client.list_entities_services()
-initial_state_helper = InitialStateHelper(entities)
+sub = await subscribe_after_initial_states(client, entities)
 
-# Subscribe with the wrapper that filters initial states
-client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
-
-# Wait for all initial states to be broadcast; on timeout this fails the test
-# and lists the entities that never sent one
-await initial_state_helper.wait_for_initial_states()
-
-# Now perform your test actions - on_state will only receive new changes
-# ... trigger state changes ...
-
-# Wait for expected state; on timeout the test fails and shows the states received
-result = await wait_or_fail(
-    state_future, 5.0, "Expected state not received", states=states
+# Arm the wait before the action that produces the state
+expected = sub.waiter.expect(
+    lambda s: isinstance(s, SensorState) and s.state == expected_value
 )
+# ... trigger state changes ...
+result = await expected
 ```
 
 **Legacy: Manual State Tracking**
@@ -386,38 +386,34 @@ async with run_compiled(yaml_config, line_callback=on_log_line):
     # Test implementation
 ```
 
-Example using futures for specific log patterns:
+To wait for log lines, use `LineWaiter` (`log_utils.py`) instead of a hand written
+callback with one future per pattern:
 ```python
 import re
 
-loop = asyncio.get_running_loop()
-connected_future = loop.create_future()
-service_future = loop.create_future()
+from .log_utils import LineWaiter
 
-# Patterns to match
-connected_pattern = re.compile(r"Client .* connected from")
-service_pattern = re.compile(r"Service called")
+log = LineWaiter()
+# Arm expectations before the action that logs them; any number may be pending.
+connected = log.expect(re.compile(r"Client .* connected from"))
+value = log.expect(re.compile(r"value=(\d+)"))
 
-
-def check_output(line: str) -> None:
-    """Check log output for expected messages."""
-    if not connected_future.done() and connected_pattern.search(line):
-        connected_future.set_result(True)
-    elif not service_future.done() and service_pattern.search(line):
-        service_future.set_result(True)
-
-
-async with run_compiled(yaml_config, line_callback=check_output):
+async with run_compiled(yaml_config, line_callback=log.callback):
     async with api_client_connected() as client:
-        # Wait for specific log message
-        await asyncio.wait_for(connected_future, timeout=5.0)
-
-        # Do test actions...
-
-        # Wait for service log
-        await asyncio.wait_for(service_future, timeout=5.0)
+        await log.wait(connected)
+        match = await log.wait(value, timeout=5.0, msg="no value logged")
+        assert match.group(1) == "42"
+        assert log.count("Service called") == 1
 ```
 
+- `expect(pattern)` only matches lines that arrive after the call. A `str` matches as a
+  substring and resolves to the line; a compiled regex is searched and resolves to its match.
+  Each expectation resolves on the next matching line, so two armed for the same text both
+  resolve on its first occurrence: arm the second after the first resolves, or use `count`.
+- `wait(future)` fails the test on timeout with the pattern and the last 20 log lines.
+- `wait_for(*needles)` also matches lines seen before the call, but only one may be pending.
+- `count(pattern)` counts the lines seen so far.
+
 **Note**: Tests that monitor log messages typically have fewer race conditions compared to state-based testing, making them more reliable. However, be aware that the host platform currently does not have a thread-safe logger, so logging from threads will not work correctly.
 
 ##### Timeout Handling

tests/integration/log_utils.py

@@ -3,23 +3,47 @@
 from __future__ import annotations
 
 import asyncio
+import re
+from typing import Any, overload
+
+from .wait_utils import wait_or_fail
+
+type LinePattern = str | re.Pattern[str]
+type LineMatch = str | re.Match[str]
+
+
+def _match(pattern: LinePattern, line: str) -> LineMatch | None:
+    """Return ``line`` if a str pattern is in it, or the search match of a regex."""
+    if isinstance(pattern, str):
+        return line if pattern in line else None
+    return pattern.search(line)
 
 
 class LineWaiter:
-    """Collects log lines and lets a test await one containing all needles.
+    """Collects log lines and lets a test await lines that match.
 
     Pass ``callback`` as ``run_compiled``'s ``line_callback``; the callback runs
-    on the test's own event loop, so futures are resolved directly. Only one
-    ``wait_for`` may be outstanding at a time (tests await sequentially).
+    on the test's own event loop, so futures are resolved directly. ``wait_for``
+    also matches lines seen before the call, and only one may be outstanding at a
+    time. ``expect`` matches only later lines, and any number may be armed at once;
+    each resolves on the next matching line, so two armed for the same text both
+    resolve on its first occurrence. Arm the second after the first, or use ``count``.
     """
 
     def __init__(self) -> None:
         self.lines: list[str] = []
         self._needles: tuple[str, ...] = ()
         self._future: asyncio.Future | None = None
+        self._expectations: dict[asyncio.Future[Any], LinePattern] = {}
 
     def callback(self, line: str) -> None:
         self.lines.append(line)
+        for future, pattern in list(self._expectations.items()):
+            if future.done():
+                del self._expectations[future]
+            elif (match := _match(pattern, line)) is not None:
+                future.set_result(match)
+                del self._expectations[future]
         if (
             self._future is not None
             and not self._future.done()
@@ -46,3 +70,38 @@ class LineWaiter:
         finally:
             self._future = None
             self._needles = ()
+
+    @overload
+    def expect(self, pattern: str) -> asyncio.Future[str]: ...  # noqa: E704
+
+    @overload
+    def expect(self, pattern: re.Pattern[str]) -> asyncio.Future[re.Match[str]]: ...  # noqa: E704
+
+    def expect(self, pattern: LinePattern) -> asyncio.Future[LineMatch]:
+        """Arm a wait for the next line matching ``pattern``; earlier lines never match.
+
+        A str matches as a substring and resolves to the line; a compiled regex is
+        searched and resolves to its match. Arm it before the action that logs the line.
+        """
+        future: asyncio.Future[LineMatch] = asyncio.get_running_loop().create_future()
+        self._expectations[future] = pattern
+        return future
+
+    async def wait[T: str | re.Match[str]](
+        self,
+        future: asyncio.Future[T],
+        timeout: float = 5.0,
+        msg: str = "log line not seen",
+    ) -> T:
+        """Await a future from ``expect``; on timeout fail with its pattern and the last lines."""
+        return await wait_or_fail(
+            future,
+            timeout,
+            msg,
+            pattern=self._expectations.get(future),
+            lines=self.lines,
+        )
+
+    def count(self, pattern: LinePattern) -> int:
+        """Return how many lines seen so far match ``pattern``."""
+        return sum(_match(pattern, line) is not None for line in self.lines)

tests/integration/state_utils.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
 import logging
 from typing import TypeVar
 
@@ -100,6 +101,22 @@ class StateWaiter:
         self._waiters.append(entry)
         return self._wait(entry, timeout, label)
 
+    def next_state[T: EntityState](
+        self,
+        key: int | None = None,
+        state_type: type[T] = EntityState,
+        timeout: float = 5.0,
+    ) -> Awaitable[T]:
+        """Arm a wait for the next ``state_type`` state, for entity ``key`` if given.
+
+        Like :meth:`expect`, call it before sending the command and await the result afterwards.
+        """
+        return self.expect(
+            lambda s: isinstance(s, state_type) and (key is None or s.key == key),
+            timeout,
+            state_type.__name__ if key is None else f"{state_type.__name__} key={key}",
+        )
+
     async def _wait(
         self,
         entry: tuple[Callable[[EntityState], bool], asyncio.Future[EntityState]],
@@ -402,6 +419,58 @@ class InitialStateHelper:
         )
 
 
+@dataclass(frozen=True, slots=True)
+class StateSubscription:
+    """A state subscription that is past the connect time states."""
+
+    initial_states: dict[int, EntityState]
+    waiter: StateWaiter
+
+
+async def subscribe_after_initial_states(
+    client: APIClient,
+    entities: list[EntityInfo],
+    on_state: Callable[[EntityState], None] | None = None,
+    timeout: float = 5.0,
+) -> StateSubscription:
+    """Subscribe to states and wait until every entity has sent its connect time state.
+
+    Later states go to ``on_state`` (if given) and to the returned ``waiter``. The connect time
+    states are kept in ``initial_states`` by key. Fails the test listing the entities still missing.
+    """
+    helper = InitialStateHelper(entities)
+    waiter = StateWaiter()
+
+    def forward(state: EntityState) -> None:
+        waiter.on_state(state)
+        if on_state is not None:
+            on_state(state)
+
+    client.subscribe_states(helper.on_state_wrapper(forward))
+    await helper.wait_for_initial_states(timeout)
+    return StateSubscription(helper.initial_states, waiter)
+
+
+async def press_and_wait[T: EntityState](
+    client: APIClient,
+    entities: list[EntityInfo],
+    waiter: StateWaiter,
+    button: str,
+    state_type: type[T],
+    *,
+    key: int | None = None,
+    timeout: float = 5.0,
+) -> T:
+    """Press the button whose object_id contains ``button`` and return the next ``state_type`` state.
+
+    Pass the entity's ``key`` when the fixture has more than one entity of that type.
+    """
+    btn = require_entity(entities, button, ButtonInfo)
+    next_state = waiter.next_state(key=key, state_type=state_type, timeout=timeout)
+    client.button_command(btn.key)
+    return await next_state
+
+
 class SensorStateCollector:
     """Collects sensor, binary sensor, and text sensor state updates with wait helpers.
 

tests/unit_tests/test_integration_log_utils.py

@@ -0,0 +1,81 @@
+"""Tests for the integration test LineWaiter."""
+
+from __future__ import annotations
+
+import asyncio
+import re
+
+import pytest
+
+from tests.integration.log_utils import LineWaiter
+
+
+@pytest.mark.asyncio
+async def test_expect_matches_only_later_lines() -> None:
+    waiter = LineWaiter()
+    waiter.callback("[I][app]: ready")
+    ready = waiter.expect("ready")
+    value = waiter.expect(re.compile(r"value=(\d+)"))
+    assert not ready.done()
+
+    waiter.callback("[D][x]: value=42")
+    waiter.callback("[I][app]: ready again")
+    assert await waiter.wait(ready) == "[I][app]: ready again"
+    match = await waiter.wait(value)
+    assert isinstance(match, re.Match)
+    assert match.group(1) == "42"
+    # wait_for still matches lines seen before the call.
+    assert await waiter.wait_for("app", "ready") == "[I][app]: ready"
+
+
+@pytest.mark.asyncio
+async def test_wait_timeout_reports_pattern_and_lines() -> None:
+    waiter = LineWaiter()
+    for i in range(30):
+        waiter.callback(f"line {i}")
+    future = waiter.expect("never")
+    with pytest.raises(pytest.fail.Exception) as err:
+        await waiter.wait(future, 0.01, "boot message missing")
+    message = str(err.value)
+    assert message.startswith("boot message missing (after 0.01s)")
+    assert "pattern: 'never'" in message
+    assert "lines (last 20 of 30):" in message
+    assert message.endswith("line 29")
+    assert future.cancelled()
+    # The cancelled expectation is dropped on the next line.
+    waiter.callback("never")
+    assert not waiter._expectations
+
+
+def test_count() -> None:
+    waiter = LineWaiter()
+    for line in ("tick 1", "tock", "tick 2"):
+        waiter.callback(line)
+    assert waiter.count("tick") == 2
+    assert waiter.count(re.compile(r"t\w+k$")) == 1
+    assert waiter.count("missing") == 0
+
+
+@pytest.mark.asyncio
+async def test_expect_resolves_from_callback_while_waiting() -> None:
+    waiter = LineWaiter()
+    first = waiter.expect("first")
+    second = waiter.expect("second")
+    loop = asyncio.get_running_loop()
+    loop.call_soon(waiter.callback, "second")
+    loop.call_soon(waiter.callback, "first")
+    assert await waiter.wait(first) == "first"
+    assert await waiter.wait(second) == "second"
+
+
+@pytest.mark.asyncio
+async def test_line_waiter_expectations_share_a_matching_line() -> None:
+    waiter = LineWaiter()
+    first = waiter.expect("tick")
+    second = waiter.expect("tick")
+    waiter.callback("tick 1")
+    assert first.result() == second.result() == "tick 1"
+    third = waiter.expect("tick")
+    waiter.callback("tick 2")
+    assert third.result() == "tick 2"
+    assert waiter.count("tick") == 2

tests/unit_tests/test_integration_state_subscribe.py

@@ -0,0 +1,108 @@
+"""Tests for the state subscription helpers in tests/integration/state_utils.py."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+
+from aioesphomeapi import ButtonInfo, EntityState, SensorInfo, SensorState, SwitchState
+import pytest
+
+from tests.integration.state_utils import (
+    StateSubscription,
+    StateWaiter,
+    press_and_wait,
+    subscribe_after_initial_states,
+)
+
+ENTITIES = [
+    SensorInfo(object_id="temp", key=1),
+    SensorInfo(object_id="humidity", key=2),
+    ButtonInfo(object_id="set_temp", key=3),
+]
+
+
+class FakeClient:
+    """Records the state callback and button presses."""
+
+    def __init__(self) -> None:
+        self.callback: Callable[[EntityState], None] | None = None
+        self.pressed: list[int] = []
+
+    def subscribe_states(self, callback: Callable[[EntityState], None]) -> None:
+        self.callback = callback
+
+    def button_command(self, key: int) -> None:
+        self.pressed.append(key)
+        self.callback(SensorState(key=1, state=42.0))
+
+
+async def _subscribe(client: FakeClient, **kwargs: object) -> StateSubscription:
+    task = asyncio.create_task(
+        subscribe_after_initial_states(client, ENTITIES, **kwargs)
+    )
+    await asyncio.sleep(0)
+    client.callback(SensorState(key=1, state=1.0))
+    client.callback(SensorState(key=2, state=2.0))
+    return await task
+
+
+@pytest.mark.asyncio
+async def test_subscribe_swallows_initial_states_and_forwards_later_ones() -> None:
+    client = FakeClient()
+    received: list[EntityState] = []
+    sub = await _subscribe(client, on_state=received.append)
+    assert sub.initial_states[1].state == 1.0
+    assert sub.initial_states[2].state == 2.0
+    assert received == []
+
+    next_state = sub.waiter.next_state(key=2, state_type=SensorState)
+    client.callback(SensorState(key=1, state=3.0))
+    client.callback(SensorState(key=2, state=4.0))
+    assert (await next_state).state == 4.0
+    assert [s.state for s in received] == [3.0, 4.0]
+
+
+@pytest.mark.asyncio
+async def test_subscribe_timeout_lists_missing_entities() -> None:
+    client = FakeClient()
+    task = asyncio.create_task(
+        subscribe_after_initial_states(client, ENTITIES, timeout=0.01)
+    )
+    await asyncio.sleep(0)
+    client.callback(SensorState(key=1, state=1.0))
+    with pytest.raises(pytest.fail.Exception, match=r"still_waiting[\s\S]*humidity"):
+        await task
+
+
+@pytest.mark.asyncio
+async def test_next_state_filters_by_type_and_times_out() -> None:
+    waiter = StateWaiter()
+    next_switch = waiter.next_state(state_type=SwitchState)
+    waiter.on_state(SensorState(key=1, state=1.0))
+    waiter.on_state(SwitchState(key=5, state=True))
+    assert (await next_switch).key == 5
+    with pytest.raises(TimeoutError, match="SwitchState key=9"):
+        await waiter.next_state(key=9, state_type=SwitchState, timeout=0.01)
+
+
+@pytest.mark.asyncio
+async def test_press_and_wait_presses_button_and_returns_state() -> None:
+    client = FakeClient()
+    sub = await _subscribe(client)
+    state = await press_and_wait(client, ENTITIES, sub.waiter, "set_temp", SensorState)
+    assert client.pressed == [3]
+    assert state.state == 42.0
+
+
+@pytest.mark.asyncio
+async def test_press_and_wait_waits_for_the_given_key() -> None:
+    client = FakeClient()
+    sub = await _subscribe(client)
+    pending = asyncio.ensure_future(
+        press_and_wait(client, ENTITIES, sub.waiter, "set_temp", SensorState, key=2)
+    )
+    await asyncio.sleep(0)
+    assert not pending.done()  # the press published key 1
+    client.callback(SensorState(key=2, state=55.0))
+    assert (await pending).state == 55.0