[display] Return the step from the reader again, as a plain enum with END = 0

The bool API cost flash at every call site (about 10 B per mipi_spi
instantiation on ESP8266); the unscoped enum keeps the old code shape and
lets callers loop with while (auto step = reader.next(entry)).

esphome/components/display/__init__.py

@@ -181,6 +181,16 @@ def get_all_display_metadata() -> dict[str, DisplayMetaData]:
     return {id_.id: meta for id_, meta in entries}
 
 
+def config_has_writer(config: ConfigType) -> bool:
+    """Return True when the display config draws anything."""
+    return (
+        config.get(CONF_AUTO_CLEAR_ENABLED) is True
+        or config.get(CONF_PAGES) is not None
+        or config.get(CONF_LAMBDA) is not None
+        or config.get(CONF_SHOW_TEST_CARD) is True
+    )
+
+
 def get_display_metadata(display_id: ID) -> DisplayMetaData:
     """Get display metadata by ID object
 

esphome/components/display/init_sequence.cpp

@@ -0,0 +1,28 @@
+#include "init_sequence.h"
+#include "esphome/core/hal.h"
+
+namespace esphome::display {
+
+InitSequenceStep InitSequenceReader::next(InitSequenceEntry &entry, uint8_t max_count) {
+  const size_t remaining = this->length_ - this->index_;
+  if (remaining == 0)
+    return INIT_SEQUENCE_STEP_END;
+  const uint8_t *pos = this->sequence_ + this->index_;
+  if (remaining >= 2) {
+    entry.value = progmem_read_byte(pos);
+    entry.count = progmem_read_byte(pos + 1);
+    entry.args = pos + 2;
+    if (entry.count == INIT_SEQUENCE_DELAY_FLAG) {
+      this->index_ += 2;
+      return INIT_SEQUENCE_STEP_DELAY;
+    }
+    if (entry.count <= max_count && remaining - 2 >= entry.count) {
+      this->index_ += 2 + entry.count;
+      return INIT_SEQUENCE_STEP_COMMAND;
+    }
+  }
+  this->index_ = this->length_;
+  return INIT_SEQUENCE_STEP_MALFORMED;
+}
+
+}  // namespace esphome::display

esphome/components/display/init_sequence.h

@@ -0,0 +1,40 @@
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+
+namespace esphome::display {
+
+/// Count byte that marks a delay entry: the byte before it is the delay in ms.
+static constexpr uint8_t INIT_SEQUENCE_DELAY_FLAG = 0xFF;
+
+/// Unscoped so `while (auto step = reader.next(entry))` stops at END.
+enum InitSequenceStep : uint8_t {
+  INIT_SEQUENCE_STEP_END = 0,
+  INIT_SEQUENCE_STEP_COMMAND,
+  INIT_SEQUENCE_STEP_DELAY,
+  INIT_SEQUENCE_STEP_MALFORMED,
+};
+
+/// `value` is the command or the delay in ms; `args` points into the (possibly PROGMEM) sequence.
+struct InitSequenceEntry {
+  const uint8_t *args;
+  uint8_t value;
+  uint8_t count;
+};
+
+/// Walks a mipi.flatten_sequence table (`cmd, count, args...` or `ms, 0xFF`) with progmem reads; resumable.
+class InitSequenceReader {
+ public:
+  InitSequenceReader(const uint8_t *sequence, size_t length) : sequence_(sequence), length_(length) {}
+
+  /// Fill `entry` with the next entry; a command with more than `max_count` args is MALFORMED.
+  InitSequenceStep next(InitSequenceEntry &entry, uint8_t max_count = 0xFE);
+
+ protected:
+  const uint8_t *sequence_;
+  size_t length_;
+  size_t index_{0};
+};
+
+}  // namespace esphome::display

esphome/components/ld2410/ld2410.cpp

@@ -163,19 +163,21 @@ static constexpr uint8_t CMD_DURATION_VALUE = 0x02;
 // Bitmasks for target states
 static constexpr uint8_t MOVE_BITMASK = 0x01;
 static constexpr uint8_t STILL_BITMASK = 0x02;
-using ld24xx::CMD_FRAME_FOOTER;
-using ld24xx::CMD_FRAME_HEADER;
-using ld24xx::HEADER_FOOTER_SIZE;
-using ld24xx::validate_header_footer;
-// Smallest frames whose values can all be read: the last value, data footer and check, then the frame footer
-static constexpr uint8_t PERIODIC_FRAME_MIN = DETECT_DISTANCE_HIGH + 3 + HEADER_FOOTER_SIZE;
-static constexpr uint8_t ENGINEERING_FRAME_MIN = OUT_PIN_SENSOR + 3 + HEADER_FOOTER_SIZE;
+// Header & Footer size
+static constexpr uint8_t HEADER_FOOTER_SIZE = 4;
+// Command Header & Footer
+static constexpr uint8_t CMD_FRAME_HEADER[HEADER_FOOTER_SIZE] = {0xFD, 0xFC, 0xFB, 0xFA};
+static constexpr uint8_t CMD_FRAME_FOOTER[HEADER_FOOTER_SIZE] = {0x04, 0x03, 0x02, 0x01};
 // Data Header & Footer
 static constexpr uint8_t DATA_FRAME_HEADER[HEADER_FOOTER_SIZE] = {0xF4, 0xF3, 0xF2, 0xF1};
 static constexpr uint8_t DATA_FRAME_FOOTER[HEADER_FOOTER_SIZE] = {0xF8, 0xF7, 0xF6, 0xF5};
 // MAC address the module uses when Bluetooth is disabled
 static constexpr uint8_t NO_MAC[] = {0x08, 0x05, 0x04, 0x03, 0x02, 0x01};
 
+static inline bool validate_header_footer(const uint8_t *header_footer, const uint8_t *buffer) {
+  return std::memcmp(header_footer, buffer, HEADER_FOOTER_SIZE) == 0;
+}
+
 void LD2410Component::dump_config() {
   char mac_s[MAC_ADDRESS_PRETTY_BUFFER_SIZE];
   char version_s[20];
@@ -326,9 +328,6 @@ void LD2410Component::handle_periodic_data_() {
     0x02: Normal mode
   */
   bool engineering_mode = this->buffer_data_[DATA_TYPES] == 0x01;
-  if (this->buffer_pos_ < (engineering_mode ? ENGINEERING_FRAME_MIN : PERIODIC_FRAME_MIN)) {
-    return;  // too short to hold the values
-  }
 #ifdef USE_SWITCH
   if (this->engineering_mode_switch_ != nullptr) {
     this->engineering_mode_switch_->publish_state(engineering_mode);
@@ -415,15 +414,24 @@ void set_number_value(number::Number *n, float value) {
 }
 #endif
 
-void LD2410Component::handle_ack_data_() {
+bool LD2410Component::handle_ack_data_() {
   ESP_LOGV(TAG, "Handling ACK DATA for COMMAND %02X", this->buffer_data_[COMMAND]);
+  if (this->buffer_pos_ < 10) {
+    ESP_LOGE(TAG, "Invalid length");
+    return true;
+  }
+  if (!ld2410::validate_header_footer(CMD_FRAME_HEADER, this->buffer_data_)) {
+    char hex_buf[format_hex_pretty_size(HEADER_FOOTER_SIZE)];
+    ESP_LOGW(TAG, "Invalid header: %s", format_hex_pretty_to(hex_buf, this->buffer_data_, HEADER_FOOTER_SIZE));
+    return true;
+  }
   if (this->buffer_data_[COMMAND_STATUS] != 0x01) {
     ESP_LOGE(TAG, "Invalid status");
-    return;
+    return true;
   }
   if (this->buffer_data_[8] || this->buffer_data_[9]) {
     ESP_LOGW(TAG, "Invalid command: %02X, %02X", this->buffer_data_[8], this->buffer_data_[9]);
-    return;
+    return true;
   }
 
   switch (this->buffer_data_[COMMAND]) {
@@ -446,9 +454,6 @@ void LD2410Component::handle_ack_data_() {
       break;
 
     case CMD_QUERY_VERSION: {
-      if (this->buffer_pos_ < 12 + sizeof(this->version_) + HEADER_FOOTER_SIZE) {
-        break;  // too short to hold a version
-      }
       std::memcpy(this->version_, &this->buffer_data_[12], sizeof(this->version_));
       char version_s[20];
       ld24xx::format_version_str(this->version_, version_s);
@@ -462,9 +467,6 @@ void LD2410Component::handle_ack_data_() {
     }
 
     case CMD_QUERY_DISTANCE_RESOLUTION: {
-      if (this->buffer_pos_ < 11 + HEADER_FOOTER_SIZE) {
-        break;  // too short to hold the resolution
-      }
       const auto *distance_resolution = find_str(DISTANCE_RESOLUTIONS_BY_UINT, this->buffer_data_[10]);
       ESP_LOGV(TAG, "Distance resolution: %s", distance_resolution);
 #ifdef USE_SELECT
@@ -476,9 +478,6 @@ void LD2410Component::handle_ack_data_() {
     }
 
     case CMD_QUERY_LIGHT_CONTROL: {
-      if (this->buffer_pos_ < 13 + HEADER_FOOTER_SIZE) {
-        break;  // too short to hold the light settings
-      }
       this->light_function_ = this->buffer_data_[10];
       this->light_threshold_ = this->buffer_data_[11];
       this->out_pin_level_ = this->buffer_data_[12];
@@ -503,7 +502,7 @@ void LD2410Component::handle_ack_data_() {
     }
     case CMD_QUERY_MAC_ADDRESS: {
       if (this->buffer_pos_ < 20) {
-        break;  // too short to hold a MAC address
+        return false;
       }
 
       this->bluetooth_on_ = std::memcmp(&this->buffer_data_[10], NO_MAC, sizeof(NO_MAC)) != 0;
@@ -548,11 +547,12 @@ void LD2410Component::handle_ack_data_() {
       break;
 
     case CMD_QUERY: {  // Query parameters response
+      // Config values can contain the footer bytes, so wait for the whole 38 byte frame
       if (this->buffer_pos_ < 38) {
-        break;  // too short to hold the parameters
+        return false;
       }
       if (this->buffer_data_[10] != HEADER)
-        return;  // value head=0xAA
+        return true;  // value head=0xAA
 #ifdef USE_NUMBER
       /*
         Moving distance range: 13th byte
@@ -582,36 +582,39 @@ void LD2410Component::handle_ack_data_() {
     default:
       break;
   }
+
+  return true;
 }
 
 void LD2410Component::readline_(int readch) {
   if (readch < 0) {
     return;  // No data available
   }
-  // Periodic and command frames both carry their count after the header
-  ld24xx::feed_frame_byte(
-      this->buffer_data_, this->buffer_pos_, readch, DATA_FRAME_HEADER, ld24xx::cmd_frame_length,
-      [this](uint8_t frame_len) { return this->handle_frame_(frame_len); }, []() { ESP_LOGW(TAG, "Invalid length"); });
-}
 
-// Dispatches the frame of frame_len bytes at the start of the buffer when its footer is in place
-bool LD2410Component::handle_frame_(uint8_t frame_len) {
-  this->buffer_pos_ = frame_len;  // the handlers take the frame length from here
-  const uint8_t *tail = &this->buffer_data_[frame_len - HEADER_FOOTER_SIZE];
-  if (this->buffer_data_[0] == DATA_FRAME_HEADER[0]) {
-    if (!ld2410::validate_header_footer(DATA_FRAME_FOOTER, tail)) {
-      return false;
-    }
+  if (this->buffer_pos_ < MAX_LINE_LENGTH - 1) {
+    this->buffer_data_[this->buffer_pos_++] = readch;
+    this->buffer_data_[this->buffer_pos_] = 0;
+  } else {
+    // We should never get here, but just in case...
+    ESP_LOGW(TAG, "Max command length exceeded; ignoring");
+    this->buffer_pos_ = 0;
+    return;
+  }
+  if (this->buffer_pos_ < HEADER_FOOTER_SIZE) {
+    return;  // Not enough data to process yet
+  }
+  if (ld2410::validate_header_footer(DATA_FRAME_FOOTER, &this->buffer_data_[this->buffer_pos_ - 4])) {
     ESP_LOGV_HEX(TAG, this->buffer_data_, this->buffer_pos_, MAX_LINE_LENGTH, "Handling Periodic Data: %s");
     this->handle_periodic_data_();
-    return true;
+    this->buffer_pos_ = 0;  // Reset position index for next message
+  } else if (ld2410::validate_header_footer(CMD_FRAME_FOOTER, &this->buffer_data_[this->buffer_pos_ - 4])) {
+    ESP_LOGV_HEX(TAG, this->buffer_data_, this->buffer_pos_, MAX_LINE_LENGTH, "Handling Ack Data: %s");
+    if (this->handle_ack_data_()) {
+      this->buffer_pos_ = 0;  // Reset position index for next message
+    } else {
+      ESP_LOGV(TAG, "Ack Data incomplete");
+    }
   }
-  if (!ld2410::validate_header_footer(CMD_FRAME_FOOTER, tail)) {
-    return false;
-  }
-  ESP_LOGV_HEX(TAG, this->buffer_data_, this->buffer_pos_, MAX_LINE_LENGTH, "Handling Ack Data: %s");
-  this->handle_ack_data_();
-  return true;
 }
 
 void LD2410Component::set_config_mode_(bool enable) {

esphome/components/ld2410/ld2410.h

@@ -106,8 +106,7 @@ class LD2410Component final : public Component, public uart::UARTDevice {
   void send_command_(uint8_t command_str, const uint8_t *command_value, uint8_t command_value_len);
   void set_config_mode_(bool enable);
   void handle_periodic_data_();
-  void handle_ack_data_();
-  bool handle_frame_(uint8_t frame_len);
+  bool handle_ack_data_();
   void readline_(int readch);
   void query_parameters_();
   void get_version_();

esphome/components/ld24xx/ld24xx.h

@@ -3,8 +3,6 @@
 #include "esphome/core/defines.h"
 #include "esphome/core/helpers.h"
 
-#include <algorithm>
-#include <cstring>
 #include <memory>
 #include <span>
 
@@ -55,74 +53,6 @@ inline void format_version_str(const uint8_t *version, std::span<char, 20> buffe
            version[2]);
 }
 
-// Commands and their answers share this header, a little-endian byte count and this footer
-static constexpr uint8_t HEADER_FOOTER_SIZE = 4;
-static constexpr uint8_t CMD_FRAME_HEADER[HEADER_FOOTER_SIZE] = {0xFD, 0xFC, 0xFB, 0xFA};
-static constexpr uint8_t CMD_FRAME_FOOTER[HEADER_FOOTER_SIZE] = {0x04, 0x03, 0x02, 0x01};
-
-inline bool validate_header_footer(const uint8_t *header_footer, const uint8_t *buffer) {
-  return std::memcmp(header_footer, buffer, HEADER_FOOTER_SIZE) == 0;
-}
-
-// Whole command frame length, or 0 until the count arrives; 32 bit so 0xFFFF cannot wrap to 0
-inline uint32_t cmd_frame_length(const uint8_t *buffer, uint8_t count) {
-  if (count < HEADER_FOOTER_SIZE + 2) {
-    return 0;
-  }
-  return HEADER_FOOTER_SIZE + 2 + encode_value_le<uint16_t>(&buffer[HEADER_FOOTER_SIZE]) + HEADER_FOOTER_SIZE;
-}
-
-// Drops the first `from` bytes and anything after them that does not begin a header; returns the bytes kept
-inline uint8_t drop_to_next_header(uint8_t *buffer, uint8_t count, uint8_t from, const uint8_t *data_header) {
-  for (; from < count; from++) {
-    const size_t n = std::min<size_t>(HEADER_FOOTER_SIZE, count - from);
-    if (std::memcmp(&buffer[from], CMD_FRAME_HEADER, n) == 0 || std::memcmp(&buffer[from], data_header, n) == 0) {
-      std::memmove(buffer, &buffer[from], count - from);
-      return count - from;
-    }
-  }
-  return 0;
-}
-
-// Feeds one byte into a frame buffer. A frame opens with the command header or `data_header` and ends at
-// `frame_length(buffer, count)` bytes (0 while unknown); `handle_frame(len)` returns false for a footer out
-// of place. A dropped or oversized frame calls `on_drop()`. The handler may overwrite `pos`; the splitter
-// works from the count it saved.
-template<size_t N, typename LengthFn, typename HandleFn, typename DropFn>
-void feed_frame_byte(uint8_t (&buffer)[N], uint8_t &pos, uint8_t byte, const uint8_t *data_header,
-                     LengthFn frame_length, HandleFn handle_frame, DropFn on_drop) {
-  // The first byte picks the header; bytes that do not continue it are dropped
-  if (pos > 0 && pos < HEADER_FOOTER_SIZE) {
-    const uint8_t *header = buffer[0] == data_header[0] ? data_header : CMD_FRAME_HEADER;
-    if (byte != header[pos]) {
-      pos = 0;
-    }
-  }
-  if (pos == 0 && byte != data_header[0] && byte != CMD_FRAME_HEADER[0]) {
-    return;
-  }
-  buffer[pos++] = byte;
-  // Drain every whole frame left in the buffer
-  while (true) {
-    const uint32_t frame_len = frame_length(buffer, pos);
-    if (frame_len == 0 || (frame_len <= N && pos < frame_len)) {
-      return;
-    }
-    const uint8_t total = pos;
-    // Keep only what begins a header: after the frame, or after its first byte when it is dropped
-    uint8_t from = 1;
-    if (frame_len <= N && handle_frame(static_cast<uint8_t>(frame_len))) {
-      from = frame_len;
-    } else {
-      on_drop();
-    }
-    pos = drop_to_next_header(buffer, total, from, data_header);
-    if (pos == 0) {
-      return;
-    }
-  }
-}
-
 #ifdef USE_SENSOR
 template<typename T> using SensorWithDedup = sensor::SensorWithDedup<T>;
 #endif

tests/components/display/test_init_sequence.cpp

@@ -0,0 +1,82 @@
+#include <gtest/gtest.h>
+
+#include <cstdint>
+
+#include "esphome/components/display/init_sequence.h"
+
+namespace esphome::display::testing {
+
+TEST(InitSequenceReader, WalksCommandsAndDelays) {
+  static const uint8_t SEQ[] = {0x11, 0x00, 120, 0xFF, 0x3A, 0x01, 0x55, 0x2A, 0x02, 0x00, 0xEF};
+  InitSequenceReader reader(SEQ, sizeof(SEQ));
+  InitSequenceEntry entry{};
+
+  ASSERT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_COMMAND);
+  EXPECT_EQ(entry.value, 0x11);
+  EXPECT_EQ(entry.count, 0);
+
+  ASSERT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_DELAY);
+  EXPECT_EQ(entry.value, 120);
+
+  ASSERT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_COMMAND);
+  EXPECT_EQ(entry.value, 0x3A);
+  ASSERT_EQ(entry.count, 1);
+  EXPECT_EQ(entry.args, &SEQ[6]);
+
+  ASSERT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_COMMAND);
+  EXPECT_EQ(entry.value, 0x2A);
+  ASSERT_EQ(entry.count, 2);
+  EXPECT_EQ(entry.args[0], 0x00);
+  EXPECT_EQ(entry.args[1], 0xEF);
+
+  EXPECT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_END);
+  EXPECT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_END);
+}
+
+TEST(InitSequenceReader, EmptySequenceEnds) {
+  InitSequenceReader reader(nullptr, 0);
+  InitSequenceEntry entry{};
+  EXPECT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_END);
+}
+
+// A count above 127 is a real length, not a flag bit.
+TEST(InitSequenceReader, HonoursFullCount) {
+  uint8_t seq[2 + 253 + 2] = {0x32, 253};
+  seq[2 + 253] = 0x29;
+  InitSequenceReader reader(seq, sizeof(seq));
+  InitSequenceEntry entry{};
+
+  ASSERT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_COMMAND);
+  EXPECT_EQ(entry.value, 0x32);
+  EXPECT_EQ(entry.count, 253);
+  ASSERT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_COMMAND);
+  EXPECT_EQ(entry.value, 0x29);
+  EXPECT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_END);
+}
+
+TEST(InitSequenceReader, TruncatedHeaderIsMalformed) {
+  static const uint8_t SEQ[] = {0x11, 0x00, 0x29};
+  InitSequenceReader reader(SEQ, sizeof(SEQ));
+  InitSequenceEntry entry{};
+  ASSERT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_COMMAND);
+  EXPECT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_MALFORMED);
+  EXPECT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_END);
+}
+
+TEST(InitSequenceReader, ShortArgsAreMalformed) {
+  static const uint8_t SEQ[] = {0x2A, 0x04, 0x00, 0x00, 0x00};
+  InitSequenceReader reader(SEQ, sizeof(SEQ));
+  InitSequenceEntry entry{};
+  EXPECT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_MALFORMED);
+  EXPECT_EQ(reader.next(entry), INIT_SEQUENCE_STEP_END);
+}
+
+TEST(InitSequenceReader, CountOverMaxIsMalformed) {
+  static const uint8_t SEQ[] = {10, 0xFF, 0x2A, 0x02, 0x00, 0xEF};
+  InitSequenceReader reader(SEQ, sizeof(SEQ));
+  InitSequenceEntry entry{};
+  ASSERT_EQ(reader.next(entry, 1), INIT_SEQUENCE_STEP_DELAY);
+  EXPECT_EQ(reader.next(entry, 1), INIT_SEQUENCE_STEP_MALFORMED);
+}
+
+}  // namespace esphome::display::testing

tests/components/ld24xx/ld24xx_frame_splitter.cpp

@@ -1,152 +0,0 @@
-#include <cstdint>
-#include <cstring>
-#include <vector>
-#include <gtest/gtest.h>
-#include "esphome/components/ld24xx/ld24xx.h"
-
-namespace esphome::ld24xx::testing {
-
-namespace {
-constexpr uint8_t DATA_FRAME_HEADER[HEADER_FOOTER_SIZE] = {0xF4, 0xF3, 0xF2, 0xF1};
-constexpr uint8_t DATA_FRAME_FOOTER[HEADER_FOOTER_SIZE] = {0xF8, 0xF7, 0xF6, 0xF5};
-constexpr uint8_t DATA_FRAME_SIZE = 12;  // header, four payload bytes, footer
-constexpr size_t BUFFER_SIZE = 40;
-
-// Command frame around the payload
-std::vector<uint8_t> make_cmd_frame(const std::vector<uint8_t> &payload) {
-  std::vector<uint8_t> frame(std::begin(CMD_FRAME_HEADER), std::end(CMD_FRAME_HEADER));
-  frame.push_back(payload.size());
-  frame.push_back(0x00);
-  frame.insert(frame.end(), payload.begin(), payload.end());
-  frame.insert(frame.end(), std::begin(CMD_FRAME_FOOTER), std::end(CMD_FRAME_FOOTER));
-  return frame;
-}
-
-std::vector<uint8_t> make_data_frame(uint8_t fill) {
-  std::vector<uint8_t> frame(std::begin(DATA_FRAME_HEADER), std::end(DATA_FRAME_HEADER));
-  frame.insert(frame.end(), DATA_FRAME_SIZE - 2 * HEADER_FOOTER_SIZE, fill);
-  frame.insert(frame.end(), std::begin(DATA_FRAME_FOOTER), std::end(DATA_FRAME_FOOTER));
-  return frame;
-}
-}  // namespace
-
-// Data frames are fixed size, command frames carry their count, as in the radar components
-class LD24xxFrameSplitterTest : public ::testing::Test {
- protected:
-  void feed_(const std::vector<uint8_t> &bytes) {
-    for (uint8_t byte : bytes) {
-      feed_frame_byte(
-          this->buffer_, this->pos_, byte, DATA_FRAME_HEADER,
-          [](const uint8_t *buffer, uint8_t count) -> uint32_t {
-            if (count < HEADER_FOOTER_SIZE) {
-              return 0;
-            }
-            return buffer[0] == DATA_FRAME_HEADER[0] ? DATA_FRAME_SIZE : cmd_frame_length(buffer, count);
-          },
-          [this](uint8_t frame_len) { return this->handle_frame_(frame_len); }, [this]() { this->drops_++; });
-    }
-  }
-
-  bool handle_frame_(uint8_t frame_len) {
-    const uint8_t *footer = this->buffer_[0] == DATA_FRAME_HEADER[0] ? DATA_FRAME_FOOTER : CMD_FRAME_FOOTER;
-    if (!validate_header_footer(footer, &this->buffer_[frame_len - HEADER_FOOTER_SIZE])) {
-      return false;
-    }
-    this->frames_.emplace_back(this->buffer_, this->buffer_ + frame_len);
-    return true;
-  }
-
-  uint8_t buffer_[BUFFER_SIZE];
-  uint8_t pos_{0};
-  std::vector<std::vector<uint8_t>> frames_;
-  int drops_{0};
-};
-
-TEST_F(LD24xxFrameSplitterTest, FooterBytesInThePayloadDoNotEndTheFrame) {
-  auto frame = make_cmd_frame({0x61, 0x01, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01, 0x55});
-  this->feed_(frame);
-  ASSERT_EQ(this->frames_.size(), 1u);
-  EXPECT_EQ(this->frames_[0], frame);
-  EXPECT_EQ(this->pos_, 0);
-  EXPECT_EQ(this->drops_, 0);
-}
-
-TEST_F(LD24xxFrameSplitterTest, BytesBeforeAHeaderAreNeverBuffered) {
-  this->feed_({0x00, 0x11, 0xFC, 0xFB, 0xFA, 0x04});
-  EXPECT_EQ(this->pos_, 0);
-  this->feed_(make_data_frame(0x22));
-  ASSERT_EQ(this->frames_.size(), 1u);
-  EXPECT_EQ(this->frames_[0], make_data_frame(0x22));
-}
-
-TEST_F(LD24xxFrameSplitterTest, AHeaderRestartsWhenItsBytesStopMatching) {
-  // F3 cannot continue the command header, and FD after FC starts it over
-  this->feed_({0xFD, 0xF3, 0xFD, 0xFC, 0xFD, 0xFC, 0xFB, 0xFA, 0x00, 0x00});
-  EXPECT_EQ(this->pos_, 6) << "only the header that matched through and its count are buffered";
-  this->feed_({0x04, 0x03, 0x02, 0x01});
-  ASSERT_EQ(this->frames_.size(), 1u);
-  EXPECT_EQ(this->frames_[0], make_cmd_frame({}));
-  EXPECT_EQ(this->drops_, 0);
-}
-
-TEST_F(LD24xxFrameSplitterTest, ImpossibleCountsAreDroppedAndTheNextFrameParsed) {
-  // 0xFFF6 totals 65536, which must not wrap to 0
-  this->feed_({0xFD, 0xFC, 0xFB, 0xFA, 0xF6, 0xFF});
-  this->feed_({0xFD, 0xFC, 0xFB, 0xFA, 0xFF, 0xFF});
-  EXPECT_EQ(this->drops_, 2);
-  EXPECT_EQ(this->pos_, 0);
-  this->feed_(make_data_frame(0x33));
-  ASSERT_EQ(this->frames_.size(), 1u);
-  EXPECT_EQ(this->frames_[0], make_data_frame(0x33));
-}
-
-TEST_F(LD24xxFrameSplitterTest, AFrameCaughtInsideADroppedFrameIsReplayed) {
-  // The count reaches 20 bytes in, past the whole frame that follows
-  std::vector<uint8_t> bytes = {0xFD, 0xFC, 0xFB, 0xFA, 0x0A, 0x00, 0x61, 0x01};
-  auto inner = make_data_frame(0x44);
-  bytes.insert(bytes.end(), inner.begin(), inner.end());
-  this->feed_(bytes);
-  EXPECT_EQ(this->drops_, 1);
-  ASSERT_EQ(this->frames_.size(), 1u);
-  EXPECT_EQ(this->frames_[0], inner);
-  EXPECT_EQ(this->pos_, 0);
-}
-
-TEST_F(LD24xxFrameSplitterTest, EveryWholeFrameLeftByADropIsDrained) {
-  // The window holds two whole data frames and the start of a third
-  std::vector<uint8_t> bytes = {0xFD, 0xFC, 0xFB, 0xFA, 0x18, 0x00, 0x61, 0x01};
-  for (uint8_t fill : {0x55, 0x66}) {
-    auto inner = make_data_frame(fill);
-    bytes.insert(bytes.end(), inner.begin(), inner.end());
-  }
-  bytes.insert(bytes.end(), {0xF4, 0xF3});
-  this->feed_(bytes);
-  EXPECT_EQ(this->drops_, 1);
-  ASSERT_EQ(this->frames_.size(), 2u);
-  EXPECT_EQ(this->frames_[0], make_data_frame(0x55));
-  EXPECT_EQ(this->frames_[1], make_data_frame(0x66));
-  EXPECT_EQ(this->pos_, 2) << "the partial header at the end stays buffered";
-}
-
-TEST_F(LD24xxFrameSplitterTest, AWrongFooterDropsOnlyThatFrame) {
-  auto bad = make_cmd_frame({0x01, 0x02});
-  bad[bad.size() - 1] = 0xEE;
-  this->feed_(bad);
-  EXPECT_EQ(this->drops_, 1);
-  EXPECT_TRUE(this->frames_.empty());
-  this->feed_(make_cmd_frame({0x03}));
-  ASSERT_EQ(this->frames_.size(), 1u);
-  EXPECT_EQ(this->frames_[0], make_cmd_frame({0x03}));
-}
-
-TEST_F(LD24xxFrameSplitterTest, BackToBackFramesOfBothKinds) {
-  this->feed_(make_data_frame(0x01));
-  this->feed_(make_cmd_frame({0xA0, 0x01, 0x00, 0x00}));
-  this->feed_(make_data_frame(0x02));
-  ASSERT_EQ(this->frames_.size(), 3u);
-  EXPECT_EQ(this->frames_[1], make_cmd_frame({0xA0, 0x01, 0x00, 0x00}));
-  EXPECT_EQ(this->frames_[2], make_data_frame(0x02));
-  EXPECT_EQ(this->drops_, 0);
-}
-
-}  // namespace esphome::ld24xx::testing

tests/integration/README.md

@@ -58,30 +58,8 @@ 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
 
-#### 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.
+**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.
 
 #### Entity Lookup Helpers (`state_utils.py`)
 
@@ -225,24 +203,46 @@ async def test_my_sensor(
 
 ##### State Subscription Pattern
 
-**Recommended: Using subscribe_after_initial_states**
+**Recommended: Using InitialStateHelper**
 
-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`:
+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:
 
 ```python
-from .state_utils import subscribe_after_initial_states
+from .state_utils import InitialStateHelper
 
+# 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()
-sub = await subscribe_after_initial_states(client, entities)
+initial_state_helper = InitialStateHelper(entities)
 
-# Arm the wait before the action that produces the state
-expected = sub.waiter.expect(
-    lambda s: isinstance(s, SensorState) and s.state == expected_value
-)
+# 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 ...
-result = await expected
+
+# 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
+)
 ```
 
 **Legacy: Manual State Tracking**
@@ -386,33 +386,37 @@ async with run_compiled(yaml_config, line_callback=on_log_line):
     # Test implementation
 ```
 
-To wait for log lines, use `LineWaiter` (`log_utils.py`) instead of a hand written
-callback with one future per pattern:
+Example using futures for specific log patterns:
 ```python
 import re
 
-from .log_utils import LineWaiter
+loop = asyncio.get_running_loop()
+connected_future = loop.create_future()
+service_future = loop.create_future()
 
-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+)"))
+# Patterns to match
+connected_pattern = re.compile(r"Client .* connected from")
+service_pattern = re.compile(r"Service called")
 
-async with run_compiled(yaml_config, line_callback=log.callback):
+
+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 api_client_connected() as client:
-        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
-```
+        # Wait for specific log message
+        await asyncio.wait_for(connected_future, timeout=5.0)
 
-- `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.
+        # Do test actions...
+
+        # Wait for service log
+        await asyncio.wait_for(service_future, timeout=5.0)
+```
 
 **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.
 

tests/integration/fixtures/uart_mock_ld2410.yaml

@@ -30,7 +30,10 @@ uart_mock:
     # Detection distance: 300cm
     # Target state: 0x03 (moving + still)
     #
-    # Frame layout (23 bytes):
+    # Note: LD2410's two_byte_to_int() uses signed char, so low bytes must be
+    # <=127 to produce correct values. Values >127 wrap negative.
+    #
+    # Frame layout (24 bytes):
     #   [0-3]  F4 F3 F2 F1 = data frame header
     #   [4-5]  0D 00       = length 13
     #   [6]    02          = data type (normal)
@@ -41,9 +44,10 @@ uart_mock:
     #   [12-13] 78 00      = still distance 120 (0x0078)
     #   [14]   19          = still energy 25
     #   [15-16] 2C 01      = detection distance 300 (0x012C)
-    #   [17]   55          = data footer marker
-    #   [18]   00          = CRC/check
-    #   [19-22] F8 F7 F6 F5 = data frame footer
+    #   [17]   00          = padding
+    #   [18]   55          = data footer marker
+    #   [19]   00          = CRC/check
+    #   [20-23] F8 F7 F6 F5 = data frame footer
     - delay: 100ms
       inject_rx:
         [
@@ -56,23 +60,28 @@ uart_mock:
           0x78, 0x00,
           0x19,
           0x2C, 0x01,
+          0x00,
           0x55, 0x00,
           0xF8, 0xF7, 0xF6, 0xF5,
         ]
 
-    # Phase 2 (t=200ms): Garbage bytes
-    # None of them can start a frame header, so the parser drops them all.
+    # Phase 2 (t=200ms): Garbage bytes - corrupt the buffer
+    # These random bytes will accumulate in the LD2410's internal buffer.
+    # No footer will be found, so the buffer just grows.
     - delay: 100ms
       inject_rx: [0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x11, 0x22]
 
     # Phase 3 (t=300ms): Truncated frame (header + partial data, no footer)
-    # The header opens the buffer; the parser waits for the 23 bytes its length announces.
+    # More bytes accumulating in the buffer without a footer match.
     - delay: 100ms
       inject_rx: [0xF4, 0xF3, 0xF2, 0xF1, 0x0D, 0x00, 0x02, 0xAA]
 
-    # Phase 4 (t=500ms): 85 bytes of 0xFF
-    # The first 15 complete the truncated frame's 23 bytes without its footer, so it is dropped
-    # with an 'Invalid length' warning; the rest cannot start a header and are dropped too.
+    # Phase 4 (t=500ms): Overflow - inject 85 bytes of 0xFF (MAX_LINE_LENGTH=50)
+    # Buffer has 15 bytes from phases 2+3.
+    # Overflow math: need 35 bytes to trigger first overflow (pos 15->49),
+    # then 50 more to trigger second overflow (pos 0->49). Total = 85 bytes.
+    # After two overflows, buffer_pos_ = 0 with a completely clean buffer.
+    # The LD2410 logs "Max command length exceeded" at each overflow.
     - delay: 200ms
       inject_rx:
         [
@@ -87,8 +96,8 @@ uart_mock:
           0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
         ]
 
-    # Phase 5 (t=600ms): Valid frame after the garbage - recovery test
-    # The buffer is empty again, so this valid frame should parse correctly.
+    # Phase 5 (t=600ms): Valid frame after overflow - recovery test
+    # Buffer was reset by overflow. This valid frame should parse correctly.
     # Moving target: 50cm, energy 100
     # Still target: 75cm, energy 80
     # Detection distance: 127cm
@@ -104,71 +113,7 @@ uart_mock:
           0x4B, 0x00,
           0x50,
           0x7F, 0x00,
-          0x55, 0x00,
-          0xF8, 0xF7, 0xF6, 0xF5,
-        ]
-
-    # Phase 6 (t=3100ms): Values that spell the command footer 04 03 02 01
-    # Moving distance 772 (0x0304), moving energy 2, still distance 1: bytes 9 to 12 are the
-    # command frame footer, which must not end the frame early.
-    # Still energy 5, detection distance 772. The sensors hold and throttle values for a second,
-    # so the gap lets the phase 5 states reach the test before these replace them.
-    - delay: 2500ms
-      inject_rx:
-        [
-          0xF4, 0xF3, 0xF2, 0xF1,
-          0x0D, 0x00,
-          0x02, 0xAA,
-          0x03,
-          0x04, 0x03,
-          0x02,
-          0x01, 0x00,
-          0x05,
-          0x04, 0x03,
-          0x55, 0x00,
-          0xF8, 0xF7, 0xF6, 0xF5,
-        ]
-
-    # Phase 7 (t=5600ms): Truncated frame merged with a complete one
-    # The eight header bytes announce a 23 byte frame, so its window takes in the first 15 bytes of
-    # the valid frame that follows. The merged frame has no footer where the count says, so it is
-    # dropped and the bytes after its header are replayed, which recovers the valid frame.
-    # Moving distance 400, moving energy 7, still distance 200, still energy 9, detection distance 400
-    - delay: 2500ms
-      inject_rx:
-        [
-          0xF4, 0xF3, 0xF2, 0xF1, 0x0D, 0x00, 0x02, 0xAA,
-          0xF4, 0xF3, 0xF2, 0xF1,
-          0x0D, 0x00,
-          0x02, 0xAA,
-          0x03,
-          0x90, 0x01,
-          0x07,
-          0xC8, 0x00,
-          0x09,
-          0x90, 0x01,
-          0x55, 0x00,
-          0xF8, 0xF7, 0xF6, 0xF5,
-        ]
-
-    # Phase 8 (t=8100ms): Impossible counts, then a complete frame
-    # A command header counting 0xFFF6 bytes (65536 in total, which must not wrap to 0) and one counting
-    # 0xFFFF are each dropped as soon as the count arrives, and the valid frame after them is parsed.
-    # Moving distance 600, moving energy 11, still distance 300, still energy 13, detection distance 600
-    - delay: 2500ms
-      inject_rx:
-        [
-          0xFD, 0xFC, 0xFB, 0xFA, 0xF6, 0xFF,
-          0xFD, 0xFC, 0xFB, 0xFA, 0xFF, 0xFF,
-          0xF4, 0xF3, 0xF2, 0xF1,
-          0x0D, 0x00,
-          0x02, 0xAA,
-          0x03,
-          0x58, 0x02,
-          0x0B,
-          0x2C, 0x01,
-          0x0D,
-          0x58, 0x02,
+          0x00,
           0x55, 0x00,
           0xF8, 0xF7, 0xF6, 0xF5,
         ]

tests/integration/fixtures/uart_mock_ld2410_engineering.yaml

@@ -89,6 +89,7 @@ uart_mock:
 
     # Phase 3 (t=300ms): Frame with still target at 291cm (multi-byte distance)
     # This tests the two_byte_to_int function with high byte > 0
+    # Note: low byte 0x2F < 0x80 so it avoids the signed char bug
     - delay: 100ms
       inject_rx:
         [

tests/integration/fixtures/uart_mock_ld2410_query_ack.yaml

@@ -42,6 +42,7 @@ uart_mock:
           0x78, 0x00,
           0x19,
           0x2C, 0x01,
+          0x00,
           0x55, 0x00,
           0xF8, 0xF7, 0xF6, 0xF5,
         ]
@@ -59,24 +60,6 @@ uart_mock:
           0x04, 0x03, 0x02, 0x01,
         ]
 
-    # Version answer cut after two bytes, so the footer sits where the version would be
-    - delay: 100ms
-      inject_rx:
-        [
-          0xFD, 0xFC, 0xFB, 0xFA, 0x08, 0x00, 0xA0, 0x01, 0x00, 0x00,
-          0x00, 0x01, 0x07, 0x02,
-          0x04, 0x03, 0x02, 0x01,
-        ]
-
-    # A whole version answer: firmware type 1, version 2.07.22062416
-    - delay: 100ms
-      inject_rx:
-        [
-          0xFD, 0xFC, 0xFB, 0xFA, 0x0C, 0x00, 0xA0, 0x01, 0x00, 0x00,
-          0x00, 0x01, 0x07, 0x02, 0x16, 0x24, 0x06, 0x22,
-          0x04, 0x03, 0x02, 0x01,
-        ]
-
     # Move thresholds of gates 0 to 3 are the footer bytes 04 03 02 01
     # Max gates 6/5, move 4 3 2 1 60 45 46 47 48, still 20 to 28, timeout 300
     - delay: 100ms
@@ -133,12 +116,6 @@ number:
       still_threshold:
         name: "G8 Still Threshold"
 
-text_sensor:
-  - platform: ld2410
-    ld2410_id: ld2410_dev
-    version:
-      name: "Version"
-
 button:
   - platform: template
     name: "Start Scenario"

tests/integration/log_utils.py

@@ -3,47 +3,23 @@
 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 lines that match.
+    """Collects log lines and lets a test await one containing all needles.
 
     Pass ``callback`` as ``run_compiled``'s ``line_callback``; the callback runs
-    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``.
+    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).
     """
 
     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()
@@ -70,38 +46,3 @@ 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,7 +4,6 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import Awaitable, Callable
-from dataclasses import dataclass
 import logging
 from typing import TypeVar
 
@@ -101,22 +100,6 @@ 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]],
@@ -419,58 +402,6 @@ 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/integration/test_uart_mock_ld2410.py

@@ -5,12 +5,9 @@ test_uart_mock_ld2410 (normal mode):
   1. Happy path - valid data frame publishes correct sensor values
   2. Garbage resilience - random bytes don't crash the component
   3. Truncated frame handling - partial frame doesn't corrupt state
-  4. Truncated frame recovery - the frame is dropped once its length is reached
-  5. Post-garbage parsing - the next valid frame is parsed correctly
-  6. Footer bytes in the values - a frame whose values spell the command footer is parsed whole
-  7. Replay after a drop - a valid frame caught inside a dropped frame's window is still parsed
-  8. Impossible counts - a length field of 0xFFF6 or 0xFFFF is dropped and the next frame is parsed
-  9. TX logging - verifies LD2410 sends expected setup commands
+  4. Buffer overflow recovery - overflow resets the parser
+  5. Post-overflow parsing - next valid frame after overflow is parsed correctly
+  6. TX logging - verifies LD2410 sends expected setup commands
 
 test_uart_mock_ld2410_engineering (engineering mode):
   1. Engineering mode frames with per-gate energy data and light sensor
@@ -36,7 +33,7 @@ async def test_uart_mock_ld2410(
     run_compiled: RunCompiledFunction,
     api_client_connected: APIClientConnectedFactory,
 ) -> None:
-    """Test LD2410 data parsing with happy path, garbage, a truncated frame, and recovery."""
+    """Test LD2410 data parsing with happy path, garbage, overflow, and recovery."""
     waiter = LineWaiter()
 
     collector = SensorStateCollector(
@@ -63,22 +60,6 @@ async def test_uart_mock_ld2410(
             and pytest.approx(100.0) in collector.sensor_states["moving_energy"]
             and pytest.approx(80.0) in collector.sensor_states["still_energy"]
             and pytest.approx(127.0) in collector.sensor_states["detection_distance"]
-            # Every value asserted below, so no state can still be in flight when the waiter fires
-            and pytest.approx(772.0) in collector.sensor_states["moving_distance"]
-            and pytest.approx(2.0) in collector.sensor_states["moving_energy"]
-            and pytest.approx(1.0) in collector.sensor_states["still_distance"]
-            and pytest.approx(5.0) in collector.sensor_states["still_energy"]
-            and pytest.approx(772.0) in collector.sensor_states["detection_distance"]
-            and pytest.approx(400.0) in collector.sensor_states["moving_distance"]
-            and pytest.approx(7.0) in collector.sensor_states["moving_energy"]
-            and pytest.approx(200.0) in collector.sensor_states["still_distance"]
-            and pytest.approx(9.0) in collector.sensor_states["still_energy"]
-            and pytest.approx(400.0) in collector.sensor_states["detection_distance"]
-            and pytest.approx(600.0) in collector.sensor_states["moving_distance"]
-            and pytest.approx(11.0) in collector.sensor_states["moving_energy"]
-            and pytest.approx(300.0) in collector.sensor_states["still_distance"]
-            and pytest.approx(13.0) in collector.sensor_states["still_energy"]
-            and pytest.approx(600.0) in collector.sensor_states["detection_distance"]
         )
     )
 
@@ -112,7 +93,7 @@ async def test_uart_mock_ld2410(
         assert collector.sensor_states["detection_distance"][0] == pytest.approx(300.0)
 
         # Wait for the recovery frame (Phase 5) to be parsed
-        # This proves the component survived garbage + truncated frame + filler bytes
+        # This proves the component survived garbage + truncated + overflow
         await wait_or_fail(
             recovery_received,
             15.0,
@@ -120,9 +101,9 @@ async def test_uart_mock_ld2410(
             sensor_states=collector.sensor_states,
         )
 
-        # Verify the truncated frame was dropped with a warning
-        assert any("Invalid length" in line for line in waiter.lines), (
-            "Expected 'Invalid length' warning in logs"
+        # Verify overflow warning was logged
+        assert any("Max command length exceeded" in line for line in waiter.lines), (
+            "Expected 'Max command length exceeded' warning in logs"
         )
 
         # Verify LD2410 sent setup commands (TX logging)
@@ -159,27 +140,6 @@ async def test_uart_mock_ld2410(
             recovery_idx
         ] == pytest.approx(127.0)
 
-        # Phase 6: the values 04 03 02 01 did not end the frame early
-        assert pytest.approx(772.0) in collector.sensor_states["moving_distance"]
-        assert pytest.approx(2.0) in collector.sensor_states["moving_energy"]
-        assert pytest.approx(1.0) in collector.sensor_states["still_distance"]
-        assert pytest.approx(5.0) in collector.sensor_states["still_energy"]
-        assert pytest.approx(772.0) in collector.sensor_states["detection_distance"]
-
-        # Phase 7: the valid frame inside the dropped frame's window was replayed and parsed
-        assert pytest.approx(400.0) in collector.sensor_states["moving_distance"]
-        assert pytest.approx(7.0) in collector.sensor_states["moving_energy"]
-        assert pytest.approx(200.0) in collector.sensor_states["still_distance"]
-        assert pytest.approx(9.0) in collector.sensor_states["still_energy"]
-        assert pytest.approx(400.0) in collector.sensor_states["detection_distance"]
-
-        # Phase 8: both impossible counts were dropped and the frame after them was parsed
-        assert pytest.approx(600.0) in collector.sensor_states["moving_distance"]
-        assert pytest.approx(11.0) in collector.sensor_states["moving_energy"]
-        assert pytest.approx(300.0) in collector.sensor_states["still_distance"]
-        assert pytest.approx(13.0) in collector.sensor_states["still_energy"]
-        assert pytest.approx(600.0) in collector.sensor_states["detection_distance"]
-
         # Verify binary sensors detected targets (from Phase 1 frame)
         assert collector.binary_states["has_target"][0] is True
         assert collector.binary_states["has_moving_target"][0] is True

tests/integration/test_uart_mock_ld2410_query_ack.py

@@ -2,8 +2,7 @@
 
 An ordinary data frame comes first, then two query ACKs. Move thresholds 4, 3, 2, 1
 put the command footer inside the ACK values. Every value must come from the whole
-38 byte frame, not from bytes left by the previous frame. A version answer cut short between
-the ACKs publishes nothing; the whole one after it does.
+38 byte frame, not from bytes left by the previous frame.
 The ACKs are synthetic, built from the layout the component parses, not device captures.
 """
 
@@ -18,7 +17,6 @@ from aioesphomeapi import (
     NumberState,
     SensorInfo,
     SensorState,
-    TextSensorState,
 )
 import pytest
 
@@ -40,9 +38,6 @@ EXPECTED: dict[str, tuple[float, float]] = {
     "g8_still_threshold": (15.0, 28.0),
 }
 
-# The whole version answer formats as this; the short one must publish nothing
-VERSION_EXPECTED = "2.07.22062416"
-
 # Sensor object_id: value from the data frame
 SENSOR_EXPECTED: dict[str, float] = {
     "moving_distance": 100.0,
@@ -75,12 +70,7 @@ async def test_uart_mock_ld2410_query_ack(
     data_received = loop.create_future()
     final_received = loop.create_future()
 
-    version_states: list[str] = []
-
     def on_state(state: EntityState) -> None:
-        if isinstance(state, TextSensorState) and not state.missing_state:
-            version_states.append(state.state)
-            return
         if isinstance(state, SensorState) and not state.missing_state:
             if (name := sensor_key_to_name.get(state.key)) is None:
                 return
@@ -152,7 +142,7 @@ async def test_uart_mock_ld2410_query_ack(
         # Each number is published once per ACK: the seed value, then the value under test
         for name, (seed, value) in EXPECTED.items():
             assert number_states[name] == pytest.approx([seed, value]), name
-        # Only the whole version answer reached the text sensor
-        assert version_states == [VERSION_EXPECTED]
-        # The footer bytes inside the values never ended the frame early, so nothing was rejected
+        # The parser waits past the footer bytes in the values; the log line
+        # comes over stdout, which can trail the API states
+        await waiter.wait_for("[ld2410", "Ack Data incomplete", timeout=2.0)
         assert not ack_errors(), f"Unexpected ACK errors: {ack_errors()}"

tests/unit_tests/components/test_display.py

@@ -1,4 +1,4 @@
-"""Tests for display.add_writer."""
+"""Tests for display.add_writer and display.config_has_writer."""
 
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -6,7 +6,8 @@ import pytest
 
 from esphome import codegen as cg
 from esphome.components import display
-from esphome.const import CONF_LAMBDA
+from esphome.components.display import CONF_SHOW_TEST_CARD
+from esphome.const import CONF_AUTO_CLEAR_ENABLED, CONF_LAMBDA, CONF_PAGES
 
 
 @pytest.mark.parametrize(
@@ -39,3 +40,20 @@ async def test_add_writer_without_lambda_does_nothing() -> None:
 
     process.assert_not_awaited()
     add.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("config", "expected"),
+    [
+        ({}, False),
+        ({CONF_AUTO_CLEAR_ENABLED: True}, True),
+        ({CONF_AUTO_CLEAR_ENABLED: False}, False),
+        ({CONF_AUTO_CLEAR_ENABLED: "unspecified"}, False),
+        ({CONF_PAGES: []}, True),
+        ({CONF_LAMBDA: "x"}, True),
+        ({CONF_SHOW_TEST_CARD: True}, True),
+        ({CONF_SHOW_TEST_CARD: False}, False),
+    ],
+)
+def test_config_has_writer(config: dict, expected: bool) -> None:
+    assert display.config_has_writer(config) is expected

tests/unit_tests/test_integration_log_utils.py

@@ -1,81 +0,0 @@
-"""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

@@ -1,108 +0,0 @@
-"""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