[display] Loop on the reader's step in display drivers and cap mipi_spi args in the reader

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
 
@@ -210,10 +220,7 @@ def get_display_metadata(display_id: ID) -> DisplayMetaData:
         height=dimensions[1],
         has_hardware_rotation=False,
         byte_order=disp_config.get(CONF_BYTE_ORDER, cv.UNDEFINED),
-        has_writer=disp_config.get(CONF_AUTO_CLEAR_ENABLED) is True
-        or disp_config.get(CONF_PAGES) is not None
-        or disp_config.get(CONF_LAMBDA) is not None
-        or disp_config.get(CONF_SHOW_TEST_CARD) is True,
+        has_writer=config_has_writer(disp_config),
         rotation=disp_config.get(CONF_ROTATION, 0),
         draw_rounding=disp_config.get(CONF_DRAW_ROUNDING, 0),
     )

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/ds248x/ds248x_one_wire_bus.cpp

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

esphome/components/epaper_spi/display.py

@@ -15,7 +15,6 @@ from esphome.components.mipi import (
 import esphome.config_validation as cv
 from esphome.config_validation import update_interval
 from esphome.const import (
-    CONF_AUTO_CLEAR_ENABLED,
     CONF_BUSY_PIN,
     CONF_CS_PIN,
     CONF_DATA_RATE,
@@ -189,10 +188,7 @@ def customise_schema(config):
         height,
         has_hardware_rotation=True,
         byte_order=cv.UNDEFINED,
-        has_writer=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,
+        has_writer=display.config_has_writer(config),
         rotation=config.get(CONF_ROTATION, 0),
         draw_rounding=0,
     )

esphome/components/epaper_spi/epaper_spi.cpp

@@ -1,4 +1,5 @@
 #include "epaper_spi.h"
+#include "esphome/components/display/init_sequence.h"
 #include <cinttypes>
 #include "esphome/core/application.h"
 #include "esphome/core/helpers.h"
@@ -315,26 +316,18 @@ void EPaperBase::start_data_() {
 void EPaperBase::on_safe_shutdown() { this->deep_sleep(); }
 
 void EPaperBase::send_init_sequence_(const uint8_t *sequence, size_t length) {
-  size_t index = 0;
-
-  while (index != length) {
-    if (length - index < 2) {
+  display::InitSequenceReader reader(sequence, length);
+  display::InitSequenceEntry entry;
+  while (auto step = reader.next(entry)) {
+    if (step == display::INIT_SEQUENCE_STEP_MALFORMED) {
       this->mark_failed(LOG_STR("Malformed init sequence"));
       return;
     }
-    const uint8_t cmd = sequence[index++];
-    if (const uint8_t x = sequence[index++]; x == DELAY_FLAG) {
-      ESP_LOGV(TAG, "Delay %dms", cmd);
-      delay(cmd);
+    if (step == display::INIT_SEQUENCE_STEP_DELAY) {
+      ESP_LOGV(TAG, "Delay %dms", entry.value);
+      delay(entry.value);
     } else {
-      const uint8_t num_args = x & 0x7F;
-      if (length - index < num_args) {
-        ESP_LOGE(TAG, "Malformed init sequence, cmd = %X, num_args = %u", cmd, num_args);
-        this->mark_failed();
-        return;
-      }
-      this->cmd_data(cmd, sequence + index, num_args);
-      index += num_args;
+      this->cmd_data(entry.value, entry.args, entry.count);
     }
   }
 }

esphome/components/epaper_spi/epaper_spi.h

@@ -34,7 +34,6 @@ static constexpr uint32_t MAX_TRANSFER_TIME = 10;  // Transfer in 10ms blocks to
 static constexpr size_t MAX_TRANSFER_SIZE = 128;
 // Longer busy waits are the waveform or a stuck pin, and are polled at the normal loop pace
 static constexpr uint32_t FAST_BUSY_POLL_MS = 50;
-static constexpr uint8_t DELAY_FLAG = 0xFF;
 
 class EPaperBase : public Display,
                    public spi::SPIDevice<spi::BIT_ORDER_MSB_FIRST, spi::CLOCK_POLARITY_LOW, spi::CLOCK_PHASE_LEADING,

esphome/components/htu21d/htu21d.cpp

@@ -1,7 +1,6 @@
 #include "htu21d.h"
-#include "esphome/core/hal.h"
-#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
+#include "esphome/core/hal.h"
 
 namespace esphome::htu21d {
 
@@ -113,7 +112,11 @@ void HTU21DComponent::set_heater(bool status) {
     this->status_set_warning();
     return;
   }
-  assign_bits(raw_heater, 1 << HTU21D_REG_HTRE_BIT, status);
+  if (status) {
+    raw_heater |= (1 << HTU21D_REG_HTRE_BIT);
+  } else {
+    raw_heater &= ~(1 << HTU21D_REG_HTRE_BIT);
+  }
   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,7 +1,6 @@
 #pragma once
 
 #include "light_color_values.h"
-#include "esphome/core/helpers.h"
 #include "esphome/core/string_ref.h"
 
 namespace esphome {
@@ -230,7 +229,13 @@ 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 { assign_bits(this->flags_, flag, value); }
+  void set_flag_(FieldFlags flag, bool value = true) ESPHOME_ALWAYS_INLINE {
+    if (value) {
+      this->flags_ |= flag;
+    } else {
+      this->flags_ &= ~flag;
+    }
+  }
 
   // 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,7 +1,6 @@
 #include "media_player.h"
-#include "esphome/core/controller_registry.h"
 #include "esphome/core/defines.h"
-#include "esphome/core/helpers.h"
+#include "esphome/core/controller_registry.h"
 #include "esphome/core/log.h"
 #include "esphome/core/progmem.h"
 
@@ -78,12 +77,19 @@ const char *media_player_command_to_string(MediaPlayerCommand command) {
 }
 
 void MediaPlayerTraits::set_supports_pause(bool supports_pause) {
-  assign_bits(this->feature_flags_, MediaPlayerEntityFeature::PAUSE | MediaPlayerEntityFeature::PLAY, supports_pause);
+  if (supports_pause) {
+    this->feature_flags_ |= MediaPlayerEntityFeature::PAUSE | MediaPlayerEntityFeature::PLAY;
+  } else {
+    this->feature_flags_ &= ~(MediaPlayerEntityFeature::PAUSE | MediaPlayerEntityFeature::PLAY);
+  }
 }
 
 void MediaPlayerTraits::set_supports_turn_off_on(bool supports_turn_off_on) {
-  assign_bits(this->feature_flags_, MediaPlayerEntityFeature::TURN_OFF | MediaPlayerEntityFeature::TURN_ON,
-              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);
+  }
 }
 
 void MediaPlayerCall::validate_() {

esphome/components/mipi_dsi/display.py

@@ -38,7 +38,6 @@ from esphome.components.mipi import (
 )
 import esphome.config_validation as cv
 from esphome.const import (
-    CONF_AUTO_CLEAR_ENABLED,
     CONF_COLOR_ORDER,
     CONF_DIMENSIONS,
     CONF_ENABLE_PIN,
@@ -167,8 +166,7 @@ def _config_schema(config: ConfigType) -> ConfigType:
         height,
         has_hardware_rotation=False,
         byte_order=config[CONF_BYTE_ORDER],
-        has_writer=requires_buffer(config)
-        or config.get(CONF_AUTO_CLEAR_ENABLED) is True,
+        has_writer=display.config_has_writer(config),
         rotation=config.get(CONF_ROTATION, 0),
         draw_rounding=config.get(CONF_DRAW_ROUNDING, 0),
     )

esphome/components/mipi_dsi/mipi_dsi.cpp

@@ -1,6 +1,7 @@
 #ifdef USE_ESP32_VARIANT_ESP32P4
 #include <utility>
 #include "mipi_dsi.h"
+#include "esphome/components/display/init_sequence.h"
 #include "esphome/core/helpers.h"
 
 #include <algorithm>
@@ -123,47 +124,38 @@ void MipiDsi::setup() {
     this->smark_failed(LOG_STR("esp_lcd_init failed"), err);
     return;
   }
-  size_t index = 0;
-  const uint8_t *seq = this->init_sequence_;
-  const size_t len = this->init_sequence_len_;
-  while (index != len) {
-    if (len - index < 2) {
+  display::InitSequenceReader reader(this->init_sequence_, this->init_sequence_len_);
+  display::InitSequenceEntry entry;
+  while (auto step = reader.next(entry)) {
+    if (step == display::INIT_SEQUENCE_STEP_MALFORMED) {
       this->mark_failed(LOG_STR("Malformed init sequence"));
       return;
     }
-    uint8_t cmd = seq[index++];
-    uint8_t x = seq[index++];
-    if (x == DELAY_FLAG) {
-      ESP_LOGD(TAG, "Delay %dms", cmd);
-      delay(cmd);
-    } else {
-      uint8_t num_args = x & 0x7F;
-      if (len - index < num_args) {
-        this->mark_failed(LOG_STR("Malformed init sequence"));
-        return;
-      }
-      if (cmd == SLEEP_OUT) {
-        // are we ready, boots?
-        int duration = when - millis();
-        if (duration > 0) {
-          delay(duration);
-        }
-      }
-      const auto *ptr = seq + index;
-#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERY_VERBOSE
-      char hex_buf[format_hex_pretty_size(MIPI_DSI_MAX_CMD_LOG_BYTES)];
-#endif
-      ESP_LOGVV(TAG, "Command %02X, length %d, byte(s) %s", cmd, num_args,
-                format_hex_pretty_to(hex_buf, ptr, num_args, '.'));
-      err = esp_lcd_panel_io_tx_param(this->io_handle_, cmd, ptr, num_args);
-      if (err != ESP_OK) {
-        this->smark_failed(LOG_STR("lcd_panel_io_tx_param failed"), err);
-        return;
-      }
-      index += num_args;
-      if (cmd == SLEEP_OUT)
-        delay(10);
+    if (step == display::INIT_SEQUENCE_STEP_DELAY) {
+      ESP_LOGD(TAG, "Delay %dms", entry.value);
+      delay(entry.value);
+      continue;
     }
+    const uint8_t cmd = entry.value;
+    if (cmd == SLEEP_OUT) {
+      // are we ready, boots?
+      int duration = when - millis();
+      if (duration > 0) {
+        delay(duration);
+      }
+    }
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERY_VERBOSE
+    char hex_buf[format_hex_pretty_size(MIPI_DSI_MAX_CMD_LOG_BYTES)];
+#endif
+    ESP_LOGVV(TAG, "Command %02X, length %d, byte(s) %s", cmd, entry.count,
+              format_hex_pretty_to(hex_buf, entry.args, entry.count, '.'));
+    err = esp_lcd_panel_io_tx_param(this->io_handle_, cmd, entry.args, entry.count);
+    if (err != ESP_OK) {
+      this->smark_failed(LOG_STR("lcd_panel_io_tx_param failed"), err);
+      return;
+    }
+    if (cmd == SLEEP_OUT)
+      delay(10);
   }
   this->io_lock_ = xSemaphoreCreateBinary();
   esp_lcd_dpi_panel_event_callbacks_t cbs = {

esphome/components/mipi_dsi/mipi_dsi.h

@@ -31,7 +31,6 @@ const uint8_t INVERT_OFF = 0x20;
 const uint8_t INVERT_ON = 0x21;
 const uint8_t DISPLAY_ON = 0x29;
 const uint8_t CMD2_BKSEL = 0xFF;
-const uint8_t DELAY_FLAG = 0xFF;
 const uint8_t MADCTL_BGR = 0x08;
 const uint8_t MADCTL_MX = 0x40;
 const uint8_t MADCTL_MY = 0x80;

esphome/components/mipi_rgb/display.py

@@ -50,7 +50,6 @@ from esphome.components.spi import (
 )
 import esphome.config_validation as cv
 from esphome.const import (
-    CONF_AUTO_CLEAR_ENABLED,
     CONF_BLUE,
     CONF_COLOR_ORDER,
     CONF_CS_PIN,
@@ -243,8 +242,7 @@ def _config_schema(config: ConfigType) -> ConfigType:
         height,
         has_hardware_rotation=True,
         byte_order=config[CONF_BYTE_ORDER],
-        has_writer=requires_buffer(config)
-        or config.get(CONF_AUTO_CLEAR_ENABLED) is True,
+        has_writer=display.config_has_writer(config),
         rotation=config.get(CONF_ROTATION, 0),
         draw_rounding=config.get(CONF_DRAW_ROUNDING, 0),
     )

esphome/components/mipi_rgb/mipi_rgb.cpp

@@ -1,5 +1,6 @@
 #if defined(USE_ESP32_VARIANT_ESP32S3) || defined(USE_ESP32_VARIANT_ESP32P4) || defined(USE_ESP32_VARIANT_ESP32S31)
 #include "mipi_rgb.h"
+#include "esphome/components/display/init_sequence.h"
 #include "esphome/core/gpio.h"
 #include "esphome/core/hal.h"
 #include "esphome/core/helpers.h"
@@ -13,8 +14,6 @@
 
 namespace esphome::mipi_rgb {
 
-static const uint8_t DELAY_FLAG = 0xFF;
-
 // Maximum bytes to log for init commands (truncated if larger)
 static constexpr size_t MIPI_RGB_MAX_CMD_LOG_BYTES = 64;
 static constexpr uint8_t MADCTL_MY = 0x80;     // Bit 7 Bottom to top
@@ -75,44 +74,30 @@ void MipiRgbSpi::write_data_(uint8_t value) {
   this->disable();
 }
 
-/**
- * this relies upon the init sequence being well-formed, which is guaranteed by the Python init code.
- */
-
 void MipiRgbSpi::write_init_sequence_() {
-  size_t index = 0;
-  const uint8_t *seq = this->init_sequence_;
-  const size_t len = this->init_sequence_len_;
-  while (index != len) {
-    if (len - index < 2) {
+  display::InitSequenceReader reader(this->init_sequence_, this->init_sequence_len_);
+  display::InitSequenceEntry entry;
+  while (auto step = reader.next(entry)) {
+    if (step == display::INIT_SEQUENCE_STEP_MALFORMED) {
       this->mark_failed(LOG_STR("Malformed init sequence"));
       return;
     }
-    uint8_t cmd = seq[index++];
-    uint8_t x = seq[index++];
-    if (x == DELAY_FLAG) {
-      ESP_LOGD(TAG, "Delay %dms", cmd);
-      delay(cmd);
-    } else {
-      uint8_t num_args = x & 0x7F;
-      if (len - index < num_args) {
-        this->mark_failed(LOG_STR("Malformed init sequence"));
-        return;
-      }
-      if (cmd == SLEEP_OUT) {
-        delay(120);  // NOLINT
-      }
-      const auto *ptr = seq + index;
-      char hex_buf[format_hex_pretty_size(MIPI_RGB_MAX_CMD_LOG_BYTES)];
-      ESP_LOGD(TAG, "Write command %02X, length %d, byte(s) %s", cmd, num_args,
-               format_hex_pretty_to(hex_buf, ptr, num_args, '.'));
-      index += num_args;
-      this->write_command_(cmd);
-      while (num_args-- != 0)
-        this->write_data_(*ptr++);
-      if (cmd == SLEEP_OUT)
-        delay(10);
+    if (step == display::INIT_SEQUENCE_STEP_DELAY) {
+      ESP_LOGD(TAG, "Delay %dms", entry.value);
+      delay(entry.value);
+      continue;
     }
+    const uint8_t cmd = entry.value;
+    if (cmd == SLEEP_OUT)
+      delay(120);  // NOLINT
+    char hex_buf[format_hex_pretty_size(MIPI_RGB_MAX_CMD_LOG_BYTES)];
+    ESP_LOGD(TAG, "Write command %02X, length %d, byte(s) %s", cmd, entry.count,
+             format_hex_pretty_to(hex_buf, entry.args, entry.count, '.'));
+    this->write_command_(cmd);
+    for (uint8_t i = 0; i < entry.count; i++)
+      this->write_data_(entry.args[i]);
+    if (cmd == SLEEP_OUT)
+      delay(10);
   }
   // this->spi_teardown();  // SPI not needed after this
   delay(10);

esphome/components/mipi_spi/display.py

@@ -33,7 +33,6 @@ from esphome.components.spi import CONF_SPI_MODE, TYPE_OCTAL, TYPE_QUAD, TYPE_SI
 import esphome.config_validation as cv
 from esphome.config_validation import ALLOW_EXTRA
 from esphome.const import (
-    CONF_AUTO_CLEAR_ENABLED,
     CONF_BRIGHTNESS,
     CONF_BUFFER_SIZE,
     CONF_COLOR_ORDER,
@@ -73,6 +72,10 @@ Model = mipi_spi_ns.enum("Model")
 PixelMode = mipi_spi_ns.enum("PixelMode")
 BusType = mipi_spi_ns.enum("BusType")
 
+# Matches INIT_SEQUENCE_MAX_ARGS, the flash copy buffer in mipi_spi.h
+INIT_SEQUENCE_MAX_ARGS = 128
+
+
 COLOR_ORDERS = {
     MODE_RGB: ColorOrder.COLOR_ORDER_RGB,
     MODE_BGR: ColorOrder.COLOR_ORDER_BGR,
@@ -216,7 +219,15 @@ def model_schema(config: ConfigType) -> cv.All | cv.Schema:
                     bus_mode, lower=True
                 ),
                 cv.Required(CONF_MODEL): cv.one_of(model.name, upper=True),
-                iseqconf: cv.ensure_list(map_sequence),
+                iseqconf: cv.ensure_list(
+                    cv.All(
+                        map_sequence,
+                        cv.Length(
+                            max=INIT_SEQUENCE_MAX_ARGS + 1,
+                            msg=f"An init sequence command can have at most {INIT_SEQUENCE_MAX_ARGS} data bytes",
+                        ),
+                    )
+                ),
                 cv.Optional(CONF_BUFFER_SIZE): cv.All(
                     cv.percentage, cv.Range(0.12, 1.0)
                 ),
@@ -290,8 +301,7 @@ def customise_schema(config: ConfigType) -> ConfigType:
         height,
         has_hardware_transform,
         byte_order=config[CONF_BYTE_ORDER],
-        has_writer=requires_buffer(config)
-        or config.get(CONF_AUTO_CLEAR_ENABLED) is True,
+        has_writer=display.config_has_writer(config),
         rotation=config.get(CONF_ROTATION, 0),
         draw_rounding=config.get(CONF_DRAW_ROUNDING, 0),
     )

esphome/components/mipi_spi/mipi_spi.h

@@ -7,12 +7,16 @@
 #include "esphome/components/spi/spi.h"
 #include "esphome/components/display/display.h"
 #include "esphome/components/display/display_color_utils.h"
+#include "esphome/components/display/init_sequence.h"
 #include "esphome/core/hal.h"
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
 namespace esphome::mipi_spi {
 
+/// Largest argument count in one init command; display.py rejects longer ones.
+static constexpr size_t INIT_SEQUENCE_MAX_ARGS = 128;
+
 ESPHOME_LOG_TAG(TAG, "display.mipi_spi");
 
 // Maximum bytes to log for commands (truncated if larger)
@@ -48,7 +52,6 @@ static constexpr uint8_t MADCTL_XFLIP = 0x02;        // Mirror the display horiz
 static constexpr uint8_t MADCTL_YFLIP = 0x01;        // Mirror the display vertically
 static constexpr uint16_t MADCTL_FLIP_FLAG = 0x100;  // controller uses axis flip bits
 
-static constexpr uint8_t DELAY_FLAG = 0xFF;
 // store a 16 bit value in a buffer, big endian.
 static inline void put16_be(uint8_t *buf, uint16_t value) {
   buf[0] = value >> 8;
@@ -163,36 +166,23 @@ class MipiSpi : public display::Display,
 
     // need to know when the display is ready for SLPOUT command - will be 120ms after reset
     auto when = millis() + 120;
-    size_t index = 0;
-    const uint8_t *seq = this->init_sequence_;
-    const size_t len = this->init_sequence_len_;
-    while (index != len) {
-      if (len - index < 2) {
-        esph_log_e(TAG, "Malformed init sequence");
-        this->mark_failed();
+    display::InitSequenceReader reader(this->init_sequence_, this->init_sequence_len_);
+    display::InitSequenceEntry entry;
+    // The sequence is in flash, which SPI DMA (and ESP8266 byte loads) cannot read
+    uint8_t args[INIT_SEQUENCE_MAX_ARGS];
+    while (auto step = reader.next(entry, sizeof(args))) {
+      if (step == display::INIT_SEQUENCE_STEP_MALFORMED) {
+        this->mark_failed(LOG_STR("Malformed init sequence"));
         return;
       }
-      uint8_t cmd = progmem_read_byte(seq + index++);
-      uint8_t x = progmem_read_byte(seq + index++);
-      if (x == DELAY_FLAG) {
-        if (cmd == 0) {
-          cmd = clamp_at_least((int) (when - millis()), 0);
-        }
-        esph_log_d(TAG, "Delay %dms", cmd);
-        delay(cmd);
-      } else {
-        uint8_t num_args = x & 0x7F;
-        if (len - index < num_args) {
-          esph_log_e(TAG, "Malformed init sequence");
-          this->mark_failed();
-          return;
-        }
-        // The sequence is in flash, which SPI DMA (and ESP8266 byte loads) cannot read
-        uint8_t args[0x80];
-        progmem_memcpy(args, seq + index, num_args);
-        this->write_command_(cmd, args, num_args);
-        index += num_args;
+      if (step == display::INIT_SEQUENCE_STEP_DELAY) {
+        const uint8_t ms = entry.value != 0 ? entry.value : clamp_at_least((int) (when - millis()), 0);
+        esph_log_d(TAG, "Delay %dms", ms);
+        delay(ms);
+        continue;
       }
+      progmem_memcpy(args, entry.args, entry.count);
+      this->write_command_(entry.value, args, entry.count);
     }
     this->reset_params_();
     // Marks init as done, so later commands log at verbose level instead of debug

esphome/components/mpr121/mpr121.cpp

@@ -3,7 +3,6 @@
 #include <cstdint>
 
 #include "esphome/core/hal.h"
-#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
 namespace esphome::mpr121 {
@@ -85,7 +84,11 @@ 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;
-  assign_bits(this->gpio_output_, 1 << ionum, value);
+  if (value) {
+    this->gpio_output_ |= (1 << ionum);
+  } else {
+    this->gpio_output_ &= ~(1 << ionum);
+  }
   this->flush_gpio_();
 }
 

esphome/components/provisioning/provisioning.h

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

esphome/components/socket/bsd_sockets_impl.cpp

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

esphome/components/socket/lwip_sockets_impl.cpp

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

esphome/components/st7305/display.py

@@ -4,7 +4,6 @@ import pkgutil
 from esphome import pins
 import esphome.codegen as cg
 from esphome.components import display, spi
-from esphome.components.display import CONF_SHOW_TEST_CARD
 from esphome.components.mipi import (
     DriverChip,
     flatten_sequence,
@@ -13,16 +12,13 @@ from esphome.components.mipi import (
 )
 import esphome.config_validation as cv
 from esphome.const import (
-    CONF_AUTO_CLEAR_ENABLED,
     CONF_CS_PIN,
     CONF_DATA_RATE,
     CONF_DC_PIN,
     CONF_HEIGHT,
     CONF_ID,
     CONF_INIT_SEQUENCE,
-    CONF_LAMBDA,
     CONF_MODEL,
-    CONF_PAGES,
     CONF_RESET_PIN,
     CONF_ROTATION,
     CONF_WIDTH,
@@ -182,10 +178,7 @@ def customise_schema(config: ConfigType) -> ConfigType:
         config[CONF_HEIGHT],
         has_hardware_rotation=False,
         byte_order=cv.UNDEFINED,
-        has_writer=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,
+        has_writer=display.config_has_writer(config),
         rotation=config.get(CONF_ROTATION, 0),
     )
     return config

esphome/components/st7305/st7305.cpp

@@ -56,33 +56,26 @@ void ST7305::reset_() {
 }
 
 void ST7305::send_init_sequence_() {
-  while (this->init_index_ < this->init_sequence_length_) {
-    if (this->init_sequence_length_ - this->init_index_ < 2) {
-      this->mark_failed(LOG_STR("Malformed init sequence"));
-      return;
-    }
-    const uint8_t command = progmem_read_byte(&this->init_sequence_[this->init_index_++]);
-    const uint8_t count = progmem_read_byte(&this->init_sequence_[this->init_index_++]);
-    if (count == 0xFF) {
-      // Resume the sequence after the delay instead of blocking the main loop.
-      this->set_timeout(STARTUP_TIMEOUT_ID, command, [this]() { this->send_init_sequence_(); });
-      return;
-    }
-    if (this->init_sequence_length_ - this->init_index_ < count) {
-      this->mark_failed(LOG_STR("Malformed init sequence"));
-      return;
-    }
+  display::InitSequenceEntry entry;
+  display::InitSequenceStep step;
+  while ((step = this->init_reader_.next(entry)) == display::INIT_SEQUENCE_STEP_COMMAND) {
     this->enable();
-    this->send_command_(command);
-    if (count != 0) {
+    this->send_command_(entry.value);
+    if (entry.count != 0) {
       this->dc_pin_->digital_write(true);
-      for (uint8_t i = 0; i < count; i++)
-        this->write_byte(progmem_read_byte(&this->init_sequence_[this->init_index_ + i]));
+      for (uint8_t i = 0; i < entry.count; i++)
+        this->write_byte(progmem_read_byte(entry.args + i));
     }
     this->disable();
-    this->init_index_ += count;
   }
-  this->finish_init_();
+  if (step == display::INIT_SEQUENCE_STEP_DELAY) {
+    // Resume the sequence after the delay instead of blocking the main loop.
+    this->set_timeout(STARTUP_TIMEOUT_ID, entry.value, [this]() { this->send_init_sequence_(); });
+  } else if (step == display::INIT_SEQUENCE_STEP_MALFORMED) {
+    this->mark_failed(LOG_STR("Malformed init sequence"));
+  } else {
+    this->finish_init_();
+  }
 }
 
 void ST7305::finish_init_() {

esphome/components/st7305/st7305.h

@@ -4,6 +4,7 @@
 #include <cstdint>
 
 #include "esphome/components/display/display_buffer.h"
+#include "esphome/components/display/init_sequence.h"
 #include "esphome/components/spi/spi.h"
 #include "esphome/core/component.h"
 #include "esphome/core/hal.h"
@@ -35,8 +36,7 @@ class ST7305 : public display::DisplayBuffer,
         row_start_(row_start),
         row_end_(row_end),
         buffer_length_(static_cast<size_t>(width) * static_cast<size_t>(height) / 8),
-        init_sequence_(init_sequence),
-        init_sequence_length_(init_sequence_length) {}
+        init_reader_(init_sequence, init_sequence_length) {}
 
   void setup() override;
   void update() override;
@@ -113,9 +113,7 @@ class ST7305 : public display::DisplayBuffer,
   uint8_t row_start_;
   uint8_t row_end_;
   size_t buffer_length_;
-  const uint8_t *init_sequence_;
-  size_t init_sequence_length_;
-  size_t init_index_{0};
+  display::InitSequenceReader init_reader_;
   bool ready_{false};
 
   int x_low_{0};

esphome/components/uart/uart_component_host.cpp

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

esphome/components/ufire_ec/ufire_ec.cpp

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

esphome/components/ufire_ise/ufire_ise.cpp

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

esphome/components/water_heater/water_heater.cpp

@@ -5,7 +5,6 @@
 #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>
@@ -60,13 +59,21 @@ WaterHeaterCall &WaterHeaterCall::set_target_temperature_high(float temperature)
 }
 
 WaterHeaterCall &WaterHeaterCall::set_away(bool away) {
-  assign_bits(this->state_, WATER_HEATER_STATE_AWAY, away);
+  if (away) {
+    this->state_ |= WATER_HEATER_STATE_AWAY;
+  } else {
+    this->state_ &= ~WATER_HEATER_STATE_AWAY;
+  }
   this->state_mask_ |= WATER_HEATER_STATE_AWAY;
   return *this;
 }
 
 WaterHeaterCall &WaterHeaterCall::set_on(bool on) {
-  assign_bits(this->state_, WATER_HEATER_STATE_ON, on);
+  if (on) {
+    this->state_ |= WATER_HEATER_STATE_ON;
+  } else {
+    this->state_ &= ~WATER_HEATER_STATE_ON;
+  }
   this->state_mask_ |= WATER_HEATER_STATE_ON;
   return *this;
 }

esphome/components/water_heater/water_heater.h

@@ -252,7 +252,13 @@ 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) { assign_bits(this->state_, flag, value); }
+  void set_state_flag_(uint32_t flag, bool value) {
+    if (value) {
+      this->state_ |= flag;
+    } else {
+      this->state_ &= ~flag;
+    }
+  }
 
   WaterHeaterMode mode_{WATER_HEATER_MODE_OFF};
   float current_temperature_{NAN};

tests/component_tests/mipi_spi/test_init.py

@@ -93,6 +93,16 @@ def run_schema_validation(config: ConfigType) -> None:
             r"value must be a power of two for dictionary value @ data\['draw_rounding'\]",
             id="invalid_draw_rounding",
         ),
+        pytest.param(
+            {
+                "model": "custom",
+                "dc_pin": 18,
+                "dimensions": {"width": 320, "height": 240},
+                "init_sequence": [[0xE0] + [0x00] * 129],
+            },
+            r"at most 128 data bytes",
+            id="init_command_too_long",
+        ),
     ],
 )
 def test_basic_configuration_errors(
@@ -606,3 +616,19 @@ def test_identical_init_sequences_share_one_table(
     assert first is not None and second is not None
     assert first.group(1) == second.group(1)
     assert main_cpp.count("mipi_spi_init_sequence[] PROGMEM") == 1
+
+
+def test_init_command_at_buffer_limit(set_core_config: SetCoreConfigCallable) -> None:
+    """A command with exactly 128 data bytes fits the C++ copy buffer."""
+    set_core_config(
+        PlatformFramework.ESP32_IDF,
+        platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
+    )
+    run_schema_validation(
+        {
+            "model": "custom",
+            "dc_pin": 18,
+            "dimensions": {"width": 320, "height": 240},
+            "init_sequence": [[0xE0] + [0x00] * 128],
+        }
+    )

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/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/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/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