[ld2410s] Add HLK-LD2410S presence radar component (#10452)

Co-authored-by: NovakI <novaki@debian.ka11>
Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>

CODEOWNERS

@@ -306,6 +306,7 @@ esphome/components/kuntze/* @ssieb
 esphome/components/lc709203f/* @ilikecake
 esphome/components/lcd_menu/* @numo68
 esphome/components/ld2410/* @regevbr @sebcaps
+esphome/components/ld2410s/* @NovakIrs
 esphome/components/ld2412/* @Rihan9
 esphome/components/ld2420/* @descipher
 esphome/components/ld2450/* @hareeshmu

esphome/components/ld2410s/__init__.py

@@ -0,0 +1,40 @@
+import esphome.codegen as cg
+from esphome.components import uart
+import esphome.config_validation as cv
+from esphome.const import CONF_ID
+from esphome.types import ConfigType
+
+CODEOWNERS = ["@NovakIrs"]
+DEPENDENCIES = ["uart"]
+DOMAIN = "ld2410s"
+MULTI_CONF = True
+
+CONF_LD2410S_ID = "ld2410s_id"
+
+ld2410s_ns = cg.esphome_ns.namespace(DOMAIN)
+LD2410SComponent = ld2410s_ns.class_("LD2410SComponent", cg.Component, uart.UARTDevice)
+
+CONFIG_SCHEMA = (
+    cv.Schema(
+        {
+            cv.GenerateID(): cv.declare_id(LD2410SComponent),
+        }
+    )
+    .extend(uart.UART_DEVICE_SCHEMA)
+    .extend(cv.COMPONENT_SCHEMA)
+)
+
+FINAL_VALIDATE_SCHEMA = uart.final_validate_device_schema(
+    DOMAIN,
+    require_tx=True,
+    require_rx=True,
+    baud_rate=115200,
+    parity="NONE",
+    stop_bits=1,
+)
+
+
+async def to_code(config: ConfigType) -> None:
+    var = cg.new_Pvariable(config[CONF_ID])
+    await cg.register_component(var, config)
+    await uart.register_uart_device(var, config)

esphome/components/ld2410s/binary_sensor.py

@@ -0,0 +1,23 @@
+import esphome.codegen as cg
+from esphome.components import binary_sensor
+import esphome.config_validation as cv
+from esphome.const import CONF_HAS_TARGET, DEVICE_CLASS_OCCUPANCY, ICON_ACCOUNT
+from esphome.types import ConfigType
+
+from . import CONF_LD2410S_ID, LD2410SComponent
+
+DEPENDENCIES = ["ld2410s"]
+
+CONFIG_SCHEMA = {
+    cv.GenerateID(CONF_LD2410S_ID): cv.use_id(LD2410SComponent),
+    cv.Optional(CONF_HAS_TARGET): binary_sensor.binary_sensor_schema(
+        device_class=DEVICE_CLASS_OCCUPANCY,
+        icon=ICON_ACCOUNT,
+    ),
+}
+
+
+async def to_code(config: ConfigType) -> None:
+    hub = await cg.get_variable(config[CONF_LD2410S_ID])
+    binary_sensors = binary_sensor.sub_binary_sensors(config)
+    await binary_sensors(CONF_HAS_TARGET, hub.set_target_binary_sensor)

esphome/components/ld2410s/ld2410s.cpp

@@ -0,0 +1,314 @@
+#include "ld2410s.h"
+#include <algorithm>
+#include <cstring>
+#include "esphome/core/application.h"
+#include "esphome/core/hal.h"
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
+
+namespace esphome::ld2410s {
+
+ESPHOME_LOG_TAG(TAG, "ld2410s");
+
+static constexpr uint8_t SHORT_DATA_FRAME_HEADER = 0x6E;
+static constexpr uint8_t SHORT_DATA_FRAME_FOOTER[] PROGMEM = {0x62};
+static constexpr uint8_t SHORT_DATA_FRAME_SIZE = 5;  // header, state, distance low, distance high, footer
+static constexpr uint8_t STD_DATA_FRAME_HEADER[] PROGMEM = {0xF4, 0xF3, 0xF2, 0xF1};
+static constexpr uint8_t STD_DATA_FRAME_FOOTER[] PROGMEM = {0xF8, 0xF7, 0xF6, 0xF5};
+static constexpr uint8_t CMD_FRAME_HEADER[] PROGMEM = {0xFD, 0xFC, 0xFB, 0xFA};
+static constexpr uint8_t CMD_FRAME_FOOTER[] PROGMEM = {0x04, 0x03, 0x02, 0x01};
+static constexpr uint8_t LONG_HEADER_SIZE = 4;
+static constexpr uint8_t LENGTH_FIELD_SIZE = 2;
+static constexpr uint8_t LONG_PAYLOAD_POS = LONG_HEADER_SIZE + LENGTH_FIELD_SIZE;
+
+static constexpr uint16_t CMD_CONFIRMATION = 0x0100;  // set in the command word of an acknowledgement
+static constexpr uint16_t CONFIG_MODE_START_CMD = 0x00FF;
+static constexpr uint16_t CONFIG_MODE_START_VALUE = 0x0001;
+static constexpr uint16_t CONFIG_MODE_END_CMD = 0x00FE;
+static constexpr uint16_t OUTPUT_MODE_SWITCH_CMD = 0x007A;
+// Minimal output mode (short frames carrying presence and distance only) is selected by six zero bytes
+static constexpr uint8_t OUTPUT_MODE_MINIMAL_SIZE = 6;
+static constexpr uint8_t CMD_FRAME_MAX_SIZE =
+    LONG_PAYLOAD_POS + 2 + OUTPUT_MODE_MINIMAL_SIZE + sizeof(CMD_FRAME_FOOTER);
+
+// Everything the module needs at start: enter configuration mode, pick the output mode, leave it
+static constexpr uint16_t INIT_SEQUENCE[] PROGMEM = {CONFIG_MODE_START_CMD, OUTPUT_MODE_SWITCH_CMD,
+                                                     CONFIG_MODE_END_CMD};
+static constexpr uint8_t INIT_SEQUENCE_LENGTH = sizeof(INIT_SEQUENCE) / sizeof(INIT_SEQUENCE[0]);
+static uint16_t init_command(uint8_t step) { return progmem_read_uint16(&INIT_SEQUENCE[step]); }
+static constexpr uint8_t INIT_MAX_TIMEOUTS = 2;
+static constexpr uint32_t ACK_TIMEOUT_MS = 300;
+static constexpr uint32_t COMMAND_GAP_MS = 300;     // pause after an acknowledgement before the next command
+static constexpr uint32_t REINIT_PAUSE_MS = 15000;  // pause before starting over after the module stopped answering
+static constexpr uint32_t LINK_TIMEOUT_MS = 5000;   // the module streams frames continuously; silence means it is gone
+
+static constexpr uint8_t DATA_TYPE_STANDARD = 0x01;
+
+void LD2410SComponent::dump_config() {
+  ESP_LOGCONFIG(TAG, "LD2410S:");
+#ifdef USE_BINARY_SENSOR
+  LOG_BINARY_SENSOR("  ", "Has Target", this->target_binary_sensor_);
+#endif
+}
+
+void LD2410SComponent::loop() {
+  uint8_t chunk[16];
+  // Take the count once so one pass handles a bounded number of bytes
+  size_t avail = this->available();
+  while (avail > 0) {
+    const size_t len = std::min(avail, sizeof(chunk));
+    if (!this->read_array(chunk, len)) {
+      ESP_LOGV(TAG, "UART read of %u bytes failed", static_cast<unsigned>(len));
+      break;
+    }
+    for (size_t i = 0; i < len; i++) {
+      this->receive_byte_(chunk[i]);
+    }
+    avail -= len;
+  }
+  const uint32_t now = App.get_loop_component_start_time();
+  this->run_init_sequence_(now);
+  if (this->init_step_ >= INIT_SEQUENCE_LENGTH && now - this->last_frame_at_ > LINK_TIMEOUT_MS) {
+    // The count tells real silence apart from frames that never parse (wrong baud rate or firmware)
+    ESP_LOGW(TAG, "No data for %" PRIu32 " s (%u frames dropped), starting over", LINK_TIMEOUT_MS / 1000,
+             this->dropped_frames_);
+    this->dropped_frames_ = 0;
+    this->status_set_warning();
+    this->init_step_ = 0;
+    this->init_timeouts_ = 0;
+    this->awaiting_ack_ = false;
+    this->next_send_at_ = now;
+  }
+}
+
+void LD2410SComponent::run_init_sequence_(uint32_t now) {
+  if (this->init_step_ >= INIT_SEQUENCE_LENGTH || static_cast<int32_t>(now - this->next_send_at_) < 0) {
+    return;
+  }
+  const uint16_t command = init_command(this->init_step_);
+  if (this->awaiting_ack_) {
+    if (++this->init_timeouts_ > INIT_MAX_TIMEOUTS) {
+      ESP_LOGW(TAG, "No acknowledgement for command %04X, starting over in %" PRIu32 " s", command,
+               REINIT_PAUSE_MS / 1000);
+      // Leave configuration mode whatever state the module is in, then begin again
+      this->send_command_(CONFIG_MODE_END_CMD);
+      this->init_step_ = 0;
+      this->init_timeouts_ = 0;
+      this->awaiting_ack_ = false;
+      this->next_send_at_ = now + REINIT_PAUSE_MS;
+      return;
+    }
+    ESP_LOGD(TAG, "No acknowledgement for command %04X, resending", command);
+  }
+  this->send_command_(command);
+  this->awaiting_ack_ = true;
+  this->next_send_at_ = now + ACK_TIMEOUT_MS;
+}
+
+void LD2410SComponent::send_command_(uint16_t command) {
+  uint8_t frame[CMD_FRAME_MAX_SIZE];
+  progmem_memcpy(frame, CMD_FRAME_HEADER, LONG_HEADER_SIZE);
+  uint8_t pos = LONG_PAYLOAD_POS;
+  frame[pos++] = command & 0xFF;
+  frame[pos++] = command >> 8;
+  switch (command) {
+    case OUTPUT_MODE_SWITCH_CMD:
+      memset(&frame[pos], 0, OUTPUT_MODE_MINIMAL_SIZE);
+      pos += OUTPUT_MODE_MINIMAL_SIZE;
+      break;
+    case CONFIG_MODE_START_CMD:
+      frame[pos++] = CONFIG_MODE_START_VALUE & 0xFF;
+      frame[pos++] = CONFIG_MODE_START_VALUE >> 8;
+      break;
+    default:
+      break;
+  }
+  frame[LONG_HEADER_SIZE] = pos - LONG_PAYLOAD_POS;  // payload length, little-endian
+  frame[LONG_HEADER_SIZE + 1] = 0;
+  progmem_memcpy(&frame[pos], CMD_FRAME_FOOTER, sizeof(CMD_FRAME_FOOTER));
+  pos += sizeof(CMD_FRAME_FOOTER);
+  this->write_array(frame, pos);
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERY_VERBOSE
+  char hex_buf[format_hex_pretty_size(CMD_FRAME_MAX_SIZE)];
+  ESP_LOGVV(TAG, "> %s", format_hex_pretty_to(hex_buf, frame, pos, ' '));
+#endif
+}
+
+void LD2410SComponent::receive_byte_(uint8_t byte) {
+  if (this->rx_len_ >= RX_BUFFER_SIZE) {
+    this->reset_frame_();
+  }
+  this->rx_buffer_[this->rx_len_++] = byte;
+
+  if (this->frame_type_ == FrameType::NONE) {
+    if (this->rx_len_ == 1 && byte == SHORT_DATA_FRAME_HEADER) {
+      this->frame_type_ = FrameType::SHORT_DATA;
+      this->expected_len_ = SHORT_DATA_FRAME_SIZE;
+      return;
+    }
+    // The two long headers are told apart once all four bytes are in
+    uint8_t std_header[LONG_HEADER_SIZE];
+    uint8_t cmd_header[LONG_HEADER_SIZE];
+    progmem_memcpy(std_header, STD_DATA_FRAME_HEADER, LONG_HEADER_SIZE);
+    progmem_memcpy(cmd_header, CMD_FRAME_HEADER, LONG_HEADER_SIZE);
+    const bool std_prefix = memcmp(this->rx_buffer_, std_header, this->rx_len_) == 0;
+    const bool cmd_prefix = memcmp(this->rx_buffer_, cmd_header, this->rx_len_) == 0;
+    if (std_prefix || cmd_prefix) {
+      if (this->rx_len_ == LONG_HEADER_SIZE) {
+        this->frame_type_ = std_prefix ? FrameType::STD_DATA : FrameType::COMMAND;
+      }
+      return;
+    }
+    // Not a header: drop what was collected, and try this byte as the start of the next frame
+    const bool retry = this->rx_len_ > 1;
+    this->reset_frame_();
+    if (retry) {
+      this->receive_byte_(byte);
+    }
+    return;
+  }
+
+  if (this->expected_len_ == 0) {
+    if (this->rx_len_ < LONG_PAYLOAD_POS) {
+      return;
+    }
+    // Standard data and command frames carry a little-endian payload length after the header
+    const uint16_t payload_len =
+        encode_uint16(this->rx_buffer_[LONG_HEADER_SIZE + 1], this->rx_buffer_[LONG_HEADER_SIZE]);
+    if (payload_len > RX_BUFFER_SIZE - LONG_PAYLOAD_POS - sizeof(CMD_FRAME_FOOTER)) {
+      ESP_LOGV(TAG, "Dropping frame with a %u byte payload, larger than the receive buffer", payload_len);
+      this->count_dropped_frame_();
+      this->reset_frame_();
+      return;
+    }
+    this->expected_len_ = LONG_PAYLOAD_POS + payload_len + sizeof(CMD_FRAME_FOOTER);
+    return;
+  }
+
+  if (this->rx_len_ < this->expected_len_) {
+    return;
+  }
+  if (this->handle_frame_() || this->rescanning_) {
+    this->reset_frame_();
+    return;
+  }
+  // A truncated long frame swallowed the bytes that followed it; try them again as new frames, once
+  uint8_t rest[RX_BUFFER_SIZE];
+  const uint8_t rest_len = this->rx_len_ - 1;
+  memcpy(rest, &this->rx_buffer_[1], rest_len);
+  this->reset_frame_();
+  this->rescanning_ = true;
+  for (uint8_t i = 0; i < rest_len; i++) {
+    this->receive_byte_(rest[i]);
+  }
+  this->rescanning_ = false;
+}
+
+bool LD2410SComponent::handle_frame_() {
+  const uint8_t *footer_src;
+  uint8_t footer_size = sizeof(CMD_FRAME_FOOTER);
+  uint8_t payload_pos = LONG_PAYLOAD_POS;
+  switch (this->frame_type_) {
+    case FrameType::SHORT_DATA:
+      footer_src = SHORT_DATA_FRAME_FOOTER;
+      footer_size = sizeof(SHORT_DATA_FRAME_FOOTER);
+      payload_pos = 1;
+      break;
+    case FrameType::STD_DATA:
+      footer_src = STD_DATA_FRAME_FOOTER;
+      break;
+    default:
+      footer_src = CMD_FRAME_FOOTER;
+      break;
+  }
+  uint8_t footer[sizeof(CMD_FRAME_FOOTER)];
+  progmem_memcpy(footer, footer_src, footer_size);
+  if (memcmp(&this->rx_buffer_[this->rx_len_ - footer_size], footer, footer_size) != 0) {
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
+    char hex_buf[format_hex_pretty_size(RX_BUFFER_SIZE)];
+    ESP_LOGV(TAG, "Footer does not match header: %s",
+             format_hex_pretty_to(hex_buf, this->rx_buffer_, this->rx_len_, ' '));
+#endif
+    this->count_dropped_frame_();
+    return false;
+  }
+  const uint8_t *payload = &this->rx_buffer_[payload_pos];
+  const uint16_t payload_len = this->rx_len_ - payload_pos - footer_size;
+  switch (this->frame_type_) {
+    case FrameType::SHORT_DATA:
+    case FrameType::STD_DATA:
+      if (!this->handle_data_frame_(payload, payload_len)) {
+        this->count_dropped_frame_();
+        break;
+      }
+      this->last_frame_at_ = App.get_loop_component_start_time();
+      this->dropped_frames_ = 0;
+      break;
+    default:
+      this->handle_command_ack_(payload, payload_len);
+      break;
+  }
+  return true;
+}
+
+bool LD2410SComponent::handle_data_frame_(const uint8_t *payload, uint16_t len) {
+  uint8_t state;
+  if (this->frame_type_ == FrameType::SHORT_DATA) {
+    // [state][distance low][distance high]
+    state = payload[0];
+  } else if (len >= 4 && payload[0] == DATA_TYPE_STANDARD) {
+    // [type][state][distance low][distance high]...; the module only sends these before the init
+    // sequence switches it to minimal output
+    state = payload[1];
+  } else {
+    if (len >= 1) {
+      ESP_LOGV(TAG, "Ignoring data frame type %02X", payload[0]);
+    }
+    return true;
+  }
+  // States 0 and 1 mean no target, 2 and 3 a target; anything else is line noise that happened to
+  // carry the header and footer bytes
+  if (state > 3) {
+    ESP_LOGV(TAG, "Dropping frame with impossible state %u", state);
+    return false;
+  }
+  this->publish_target_(state > 1);
+  return true;
+}
+
+void LD2410SComponent::handle_command_ack_(const uint8_t *payload, uint16_t len) {
+  if (len < 4) {
+    ESP_LOGV(TAG, "Acknowledgement too short (%u bytes)", len);
+    return;
+  }
+  const uint16_t command_word = encode_uint16(payload[1], payload[0]);
+  const uint16_t ack = encode_uint16(payload[3], payload[2]);
+  if (!this->awaiting_ack_ || command_word != (init_command(this->init_step_) | CMD_CONFIRMATION)) {
+    ESP_LOGV(TAG, "Unexpected acknowledgement %04X", command_word);
+    return;
+  }
+  if (ack != 0) {
+    // Leave the command pending so the timeout resends it
+    ESP_LOGW(TAG, "Module rejected command %04X, ack %04X", init_command(this->init_step_), ack);
+    return;
+  }
+  this->awaiting_ack_ = false;
+  this->init_timeouts_ = 0;
+  this->init_step_++;
+  this->next_send_at_ = App.get_loop_component_start_time() + COMMAND_GAP_MS;
+  if (this->init_step_ >= INIT_SEQUENCE_LENGTH) {
+    ESP_LOGD(TAG, "Setup done");
+    this->last_frame_at_ = App.get_loop_component_start_time();
+    this->status_clear_warning();
+  }
+}
+
+void LD2410SComponent::publish_target_(bool target) {
+#ifdef USE_BINARY_SENSOR
+  if (this->target_binary_sensor_ != nullptr) {
+    this->target_binary_sensor_->publish_state(target);
+  }
+#endif
+}
+
+}  // namespace esphome::ld2410s

esphome/components/ld2410s/ld2410s.h

@@ -0,0 +1,62 @@
+#pragma once
+
+#include "esphome/core/component.h"
+#include "esphome/core/defines.h"
+#include "esphome/components/uart/uart.h"
+#ifdef USE_BINARY_SENSOR
+#include "esphome/components/binary_sensor/binary_sensor.h"
+#endif
+
+namespace esphome::ld2410s {
+
+// Largest frame read here is the standard data frame: 4 header + 2 length + payload + 4 footer
+static constexpr size_t RX_BUFFER_SIZE = 96;
+static_assert(RX_BUFFER_SIZE <= 255, "frame lengths are kept in a byte");
+
+class LD2410SComponent final : public Component, public uart::UARTDevice {
+#ifdef USE_BINARY_SENSOR
+  SUB_BINARY_SENSOR(target)
+#endif
+
+ public:
+  void setup() override { this->status_set_warning(); }  // cleared once the module acknowledged the init sequence
+  void loop() override;
+  void dump_config() override;
+
+ protected:
+  enum class FrameType : uint8_t { NONE, SHORT_DATA, STD_DATA, COMMAND };
+
+  void receive_byte_(uint8_t byte);
+  /// Returns false when the footer does not match, so the bytes can be scanned again.
+  bool handle_frame_();
+  /// Returns false for a data frame whose state byte is impossible, so it counts as dropped.
+  bool handle_data_frame_(const uint8_t *payload, uint16_t len);
+  void handle_command_ack_(const uint8_t *payload, uint16_t len);
+  void count_dropped_frame_() {
+    if (this->dropped_frames_ < UINT8_MAX) {
+      this->dropped_frames_++;
+    }
+  }
+  void reset_frame_() {
+    this->rx_len_ = 0;
+    this->expected_len_ = 0;
+    this->frame_type_ = FrameType::NONE;
+  }
+  void run_init_sequence_(uint32_t now);
+  void send_command_(uint16_t command);
+  void publish_target_(bool target);
+
+  uint32_t next_send_at_{0};
+  uint32_t last_frame_at_{0};
+  uint8_t rx_buffer_[RX_BUFFER_SIZE];
+  uint8_t rx_len_{0};
+  uint8_t expected_len_{0};
+  uint8_t init_step_{0};
+  uint8_t init_timeouts_{0};
+  uint8_t dropped_frames_{0};  // since the last accepted data frame, saturates at 255
+  FrameType frame_type_{FrameType::NONE};
+  bool awaiting_ack_{false};
+  bool rescanning_{false};
+};
+
+}  // namespace esphome::ld2410s

tests/components/ld2410s/common.yaml

@@ -0,0 +1,7 @@
+ld2410s:
+  uart_id: uart_bus
+
+binary_sensor:
+  - platform: ld2410s
+    has_target:
+      name: Presence

tests/components/ld2410s/test.esp32-idf.yaml

@@ -0,0 +1,3 @@
+packages:
+  uart_115200: !include ../../test_build_components/common/uart_115200/esp32-idf.yaml
+  ld2410s: !include common.yaml

tests/components/ld2410s/test.esp8266-ard.yaml

@@ -0,0 +1,3 @@
+packages:
+  uart_115200: !include ../../test_build_components/common/uart_115200/esp8266-ard.yaml
+  ld2410s: !include common.yaml

tests/components/ld2410s/test.rp2040-ard.yaml

@@ -0,0 +1,3 @@
+packages:
+  uart_115200: !include ../../test_build_components/common/uart_115200/rp2040-ard.yaml
+  ld2410s: !include common.yaml

tests/integration/fixtures/uart_mock_ld2410s.yaml

@@ -0,0 +1,84 @@
+esphome:
+  name: uart-mock-ld2410s-test
+
+host:
+api:
+  batch_delay: 0ms  # Disable batching to receive all state updates
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy ld2410s's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  id: mock_uart
+  baud_rate: 115200
+  auto_start: false
+  responses:
+    # Config-mode start, acknowledged with protocol version 2 and buffer size 0x20
+    - expect_tx: [0xFD, 0xFC, 0xFB, 0xFA, 0x04, 0x00, 0xFF, 0x00, 0x01, 0x00, 0x04, 0x03, 0x02, 0x01]
+      inject_rx: [0xFD, 0xFC, 0xFB, 0xFA, 0x08, 0x00, 0xFF, 0x01, 0x00, 0x00, 0x02, 0x00, 0x20, 0x00, 0x04, 0x03, 0x02, 0x01]
+    # Output-mode switch to minimal, acknowledged
+    - expect_tx: [0xFD, 0xFC, 0xFB, 0xFA, 0x08, 0x00, 0x7A, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01]
+      inject_rx: [0xFD, 0xFC, 0xFB, 0xFA, 0x04, 0x00, 0x7A, 0x01, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01]
+    # Config-mode end, acknowledged
+    - expect_tx: [0xFD, 0xFC, 0xFB, 0xFA, 0x02, 0x00, 0xFE, 0x00, 0x04, 0x03, 0x02, 0x01]
+      inject_rx: [0xFD, 0xFC, 0xFB, 0xFA, 0x04, 0x00, 0xFE, 0x01, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01]
+  injections:
+    # Short frame, state 2 (target), distance 300 -> presence on
+    - delay: 100ms
+      inject_rx: [0x6E, 0x02, 0x2C, 0x01, 0x62]
+    # Short frame, state 0 (no target) -> presence off
+    - delay: 100ms
+      inject_rx: [0x6E, 0x00, 0x00, 0x00, 0x62]
+    # Standard data frame, type 1, state 2, distance 300 -> presence on
+    - delay: 100ms
+      inject_rx: [0xF4, 0xF3, 0xF2, 0xF1, 0x05, 0x00, 0x01, 0x02, 0x2C, 0x01, 0x00, 0xF8, 0xF7, 0xF6, 0xF5]
+    # Short frame, state 1 (no target) -> presence off
+    - delay: 100ms
+      inject_rx: [0x6E, 0x01, 0x00, 0x00, 0x62]
+    # Noise byte directly followed by a short frame, state 2 -> presence on
+    - delay: 100ms
+      inject_rx: [0xAA, 0x6E, 0x02, 0x2C, 0x01, 0x62]
+    # Short frame, state 0 -> presence off
+    - delay: 100ms
+      inject_rx: [0x6E, 0x00, 0x00, 0x00, 0x62]
+    # Partial standard header byte directly followed by a short frame, state 3 -> presence on
+    - delay: 100ms
+      inject_rx: [0xF4, 0x6E, 0x03, 0x2C, 0x01, 0x62]
+    # Short frame with the impossible state 7, dropped, then a short frame, state 0 -> presence off
+    - delay: 100ms
+      inject_rx: [0x6E, 0x07, 0x00, 0x00, 0x62, 0x6E, 0x00, 0x00, 0x00, 0x62]
+    # Standard data frame, type 1, with the impossible state 7, dropped, then a short frame, state 2
+    # -> presence on
+    - delay: 100ms
+      inject_rx:
+        [
+          0xF4, 0xF3, 0xF2, 0xF1, 0x05, 0x00, 0x01, 0x07, 0x2C, 0x01, 0x00, 0xF8, 0xF7, 0xF6, 0xF5, 0x6E,
+          0x02, 0x2C, 0x01, 0x62,
+        ]
+
+ld2410s:
+  id: ld2410s_dev
+  uart_id: mock_uart
+
+binary_sensor:
+  - platform: ld2410s
+    ld2410s_id: ld2410s_dev
+    has_target:
+      name: "Has Target"
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: 'id(mock_uart).start_scenario();'

tests/integration/fixtures/uart_mock_ld2410s_link_lost.yaml

@@ -0,0 +1,75 @@
+esphome:
+  name: uart-mock-ld2410s-link-lost
+
+host:
+api:
+  batch_delay: 0ms  # Disable batching to receive all state updates
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy ld2410s's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  id: mock_uart
+  baud_rate: 115200
+  auto_start: false
+  responses:
+    # Config-mode start, acknowledged with protocol version 2 and buffer size 0x20
+    - expect_tx: [0xFD, 0xFC, 0xFB, 0xFA, 0x04, 0x00, 0xFF, 0x00, 0x01, 0x00, 0x04, 0x03, 0x02, 0x01]
+      inject_rx: [0xFD, 0xFC, 0xFB, 0xFA, 0x08, 0x00, 0xFF, 0x01, 0x00, 0x00, 0x02, 0x00, 0x20, 0x00, 0x04, 0x03, 0x02, 0x01]
+    # Output-mode switch to minimal, acknowledged
+    - expect_tx: [0xFD, 0xFC, 0xFB, 0xFA, 0x08, 0x00, 0x7A, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01]
+      inject_rx: [0xFD, 0xFC, 0xFB, 0xFA, 0x04, 0x00, 0x7A, 0x01, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01]
+    # Config-mode end, acknowledged
+    - expect_tx: [0xFD, 0xFC, 0xFB, 0xFA, 0x02, 0x00, 0xFE, 0x00, 0x04, 0x03, 0x02, 0x01]
+      inject_rx: [0xFD, 0xFC, 0xFB, 0xFA, 0x04, 0x00, 0xFE, 0x01, 0x00, 0x00, 0x04, 0x03, 0x02, 0x01]
+  injections:
+    # Short frames every second for 7 s, longer than the 5 s link timeout, so they must keep the link alive
+    # t=1 s: short frame, state 2 -> presence on
+    - delay: 1000ms
+      inject_rx: [0x6E, 0x02, 0x2C, 0x01, 0x62]
+    # t=2 s: short frame, state 0 -> presence off
+    - delay: 1000ms
+      inject_rx: [0x6E, 0x00, 0x00, 0x00, 0x62]
+    # t=3 s: short frame, state 2 -> presence on
+    - delay: 1000ms
+      inject_rx: [0x6E, 0x02, 0x2C, 0x01, 0x62]
+    # t=4 s: short frame, state 0 -> presence off
+    - delay: 1000ms
+      inject_rx: [0x6E, 0x00, 0x00, 0x00, 0x62]
+    # t=5 s: short frame, state 2 -> presence on
+    - delay: 1000ms
+      inject_rx: [0x6E, 0x02, 0x2C, 0x01, 0x62]
+    # t=6 s: short frame, state 0 -> presence off
+    - delay: 1000ms
+      inject_rx: [0x6E, 0x00, 0x00, 0x00, 0x62]
+    # t=7 s: short frame, state 2 -> presence on
+    - delay: 1000ms
+      inject_rx: [0x6E, 0x02, 0x2C, 0x01, 0x62]
+    # Then silence: the link watchdog must start the init sequence over
+
+ld2410s:
+  id: ld2410s_dev
+  uart_id: mock_uart
+
+binary_sensor:
+  - platform: ld2410s
+    ld2410s_id: ld2410s_dev
+    has_target:
+      name: "Has Target"
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: 'id(mock_uart).start_scenario();'

tests/integration/fixtures/uart_mock_ld2410s_recovery.yaml

@@ -0,0 +1,100 @@
+esphome:
+  name: uart-mock-ld2410s-recovery
+
+host:
+api:
+  batch_delay: 0ms  # Disable batching to receive all state updates
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy ld2410s's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+# No responses: the init commands go unanswered, so the test also covers the resend and give-up path;
+# this does not affect receiving data frames
+uart_mock:
+  id: mock_uart
+  baud_rate: 115200
+  auto_start: false
+  injections:
+    # Short frame, state 2 -> presence on (baseline)
+    - delay: 100ms
+      inject_rx: [0x6E, 0x02, 0x2C, 0x01, 0x62]
+    # Standard header with a 255 byte length field (over the 96 byte buffer), three payload bytes,
+    # then a short frame, state 0 -> frame dropped, presence off
+    - delay: 100ms
+      inject_rx: [0xF4, 0xF3, 0xF2, 0xF1, 0xFF, 0x00, 0x01, 0x02, 0x03, 0x6E, 0x00, 0x00, 0x00, 0x62]
+    # Standard frame declaring 5 payload bytes but carrying 3, then a short frame, state 2: the short
+    # frame is read as the rest of the declared length, so the frame stays one byte short
+    - delay: 100ms
+      inject_rx: [0xF4, 0xF3, 0xF2, 0xF1, 0x05, 0x00, 0x01, 0x02, 0x2C, 0x6E, 0x02, 0x2C, 0x01, 0x62]
+    # Short frame, state 0: its first byte completes the truncated frame (footer mismatch); the
+    # rescan of the swallowed bytes rescues the state 2 short frame -> presence on, and leaves this
+    # frame's header pending, so the rest of it completes -> presence off
+    - delay: 100ms
+      inject_rx: [0x6E, 0x00, 0x00, 0x00, 0x62]
+    # Short frame, state 2 -> presence on
+    - delay: 100ms
+      inject_rx: [0x6E, 0x02, 0x2C, 0x01, 0x62]
+    # Command frame with zeros where the footer belongs, then a short frame, state 0 -> presence off
+    - delay: 100ms
+      inject_rx:
+        [
+          0xFD, 0xFC, 0xFB, 0xFA, 0x04, 0x00, 0xFE, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x6E, 0x00,
+          0x00, 0x00, 0x62,
+        ]
+    # 100 bytes that never start a header, then a short frame, state 2 -> presence on
+    - delay: 100ms
+      inject_rx:
+        [
+          0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55,
+          0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55,
+          0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55,
+          0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55,
+          0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55,
+          0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55,
+          0x55, 0x55, 0x55, 0x55, 0x6E, 0x02, 0x2C, 0x01, 0x62,
+        ]
+    # Short frame, state 0 -> presence off
+    - delay: 100ms
+      inject_rx: [0x6E, 0x00, 0x00, 0x00, 0x62]
+    # Standard frame of the largest size that fits (86 byte payload: type 1, state 2, distance 300,
+    # 82 padding bytes) -> presence on
+    - delay: 100ms
+      inject_rx:
+        [
+          0xF4, 0xF3, 0xF2, 0xF1, 0x56, 0x00, 0x01, 0x02, 0x2C, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+          0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+          0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+          0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+          0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+          0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xF8, 0xF7, 0xF6, 0xF5,
+        ]
+    # Short frame, state 0 -> presence off
+    - delay: 100ms
+      inject_rx: [0x6E, 0x00, 0x00, 0x00, 0x62]
+
+ld2410s:
+  id: ld2410s_dev
+  uart_id: mock_uart
+
+binary_sensor:
+  - platform: ld2410s
+    ld2410s_id: ld2410s_dev
+    has_target:
+      name: "Has Target"
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: 'id(mock_uart).start_scenario();'

tests/integration/test_uart_mock_ld2410s.py

@@ -0,0 +1,329 @@
+"""Integration tests for the LD2410S component with mock UART.
+
+test_uart_mock_ld2410s:
+  1. Init sequence - config-mode start, output-mode switch and config-mode end go
+     out in order, each after the previous one is acknowledged, then the
+     component logs "Setup done" and clears its warning
+  2. Presence from short frames (states 0/1 off, 2/3 on) and from a standard
+     data frame
+  3. Resync - a noise byte or a partial header directly before a short frame
+     does not hide that frame
+  4. A short frame and a standard frame whose state byte is above 3 are
+     dropped without publishing, and the short frame after each still publishes
+
+test_uart_mock_ld2410s_recovery (the mock never answers the init commands):
+  1. Config-mode start is resent after the acknowledgement timeout
+  2. Once the retries run out, config-mode end is sent and the component warns
+     that it is starting over
+  3. A standard frame whose length field exceeds the receive buffer is dropped
+  4. A truncated standard frame swallows the bytes after it up to its declared
+     length and fails the footer check; the swallowed bytes are rescanned, so
+     the short frames hidden in them still publish
+  5. A command frame with a bad footer is dropped
+  6. 100 bytes that never start a header are skipped
+  7. A standard frame of the largest size that fits (86 byte payload) is accepted
+
+test_uart_mock_ld2410s_link_lost:
+  1. Short frames arriving for longer than the 5 s link timeout keep the link up
+  2. 5 s of silence logs "No data for 5 s (0 frames dropped), starting over", sets the warning flag
+     and sends a new config-mode start
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+
+from aioesphomeapi import APIClient, ButtonInfo
+import pytest
+
+from .state_utils import InitialStateHelper, SensorStateCollector, require_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+CONFIG_START_TX = "FD:FC:FB:FA:04:00:FF:00:01:00:04:03:02:01"
+OUTPUT_MODE_TX = "FD:FC:FB:FA:08:00:7A:00:00:00:00:00:00:00:04:03:02:01"
+CONFIG_END_TX = "FD:FC:FB:FA:02:00:FE:00:04:03:02:01"
+
+ANSI_ESCAPE = re.compile(r"\x1b\[[0-9;]*m")
+
+# has_target after each injection: short 2, short 0, standard frame state 2,
+# short 1, noise + short 2, short 0, partial header + short 3,
+# short 7 (dropped) + short 0, standard frame state 7 (dropped) + short 2
+EXPECTED_PRESENCE = [True, False, True, False, True, False, True, False, True]
+# The short frame and the standard frame with state 7 are logged as dropped
+EXPECTED_DROPPED_STATES = ["7", "7"]
+
+# has_target after each recovery phase (see uart_mock_ld2410s_recovery.yaml).
+# The truncated frame swallows the state 2 short frame after it plus the header
+# byte of the following state 0 short frame; the footer mismatch rescans those
+# bytes, which publishes the swallowed frame and resumes the next one.
+EXPECTED_RECOVERY_PRESENCE = [
+    True,  # baseline short frame
+    False,  # short frame after the dropped oversized length field
+    True,  # rescued: short frame swallowed by the truncated standard frame
+    False,  # short frame whose header byte completed the truncated frame
+    True,  # short frame
+    False,  # short frame after the command frame with a bad footer
+    True,  # short frame after 100 junk bytes
+    False,  # short frame
+    True,  # maximum-size standard frame
+    False,  # short frame
+]
+# The truncated frame takes the next 6 bytes to reach its declared length
+TRUNCATED_FRAME_HEX = "F4 F3 F2 F1 05 00 01 02 2C 6E 02 2C 01 62 6E"
+BAD_COMMAND_FRAME_HEX = "FD FC FB FA 04 00 FE 01 00 00 00 00 00 00"
+
+
+def _tx_frame(line: str) -> str | None:
+    """Return the hex payload of a uart_mock TX log line, or None."""
+    if "uart_mock" not in line or "TX " not in line or "bytes: " not in line:
+        return None
+    return ANSI_ESCAPE.sub("", line.split("bytes: ", 1)[1]).strip()
+
+
+async def _start_scenario(client: APIClient, collector: SensorStateCollector) -> None:
+    """Subscribe the collector past the initial states, then start the mock scenario."""
+    entities, _ = await client.list_entities_services()
+    collector.build_key_mapping(entities)
+    initial_state_helper = InitialStateHelper(entities)
+    client.subscribe_states(initial_state_helper.on_state_wrapper(collector.on_state))
+    try:
+        await initial_state_helper.wait_for_initial_states()
+    except TimeoutError:
+        pytest.fail("Timeout waiting for initial states")
+    start_btn = require_entity(entities, "start_scenario", ButtonInfo)
+    client.button_command(start_btn.key)
+
+
+async def _wait_for_presence(
+    collector: SensorStateCollector, waiter: asyncio.Future[bool]
+) -> None:
+    try:
+        await asyncio.wait_for(waiter, timeout=5.0)
+    except TimeoutError:
+        pytest.fail(
+            f"Timeout waiting for presence updates. Received: "
+            f"{collector.binary_states['has_target']}"
+        )
+
+
+@pytest.mark.asyncio
+async def test_uart_mock_ld2410s(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test the LD2410S init sequence, presence parsing and receiver resync."""
+    loop = asyncio.get_running_loop()
+    tx_frames: list[str] = []
+    setup_done = loop.create_future()
+    warning_cleared = loop.create_future()
+    dropped_states: list[str] = []
+
+    def line_callback(line: str) -> None:
+        if (frame := _tx_frame(line)) is not None:
+            tx_frames.append(frame)
+        if "[ld2410s" in line and "Setup done" in line and not setup_done.done():
+            setup_done.set_result(True)
+        if "ld2410s cleared Warning flag" in line and not warning_cleared.done():
+            warning_cleared.set_result(True)
+        if "Dropping frame with impossible state " in line:
+            dropped_states.append(
+                ANSI_ESCAPE.sub("", line.split("state ", 1)[1]).strip()
+            )
+
+    collector = SensorStateCollector(
+        sensor_names=[], binary_sensor_names=["has_target"]
+    )
+    all_presence = collector.add_waiter(
+        lambda: len(collector.binary_states["has_target"]) >= len(EXPECTED_PRESENCE)
+    )
+
+    async with (
+        run_compiled(yaml_config, line_callback=line_callback),
+        api_client_connected() as client,
+    ):
+        try:
+            await asyncio.wait_for(asyncio.gather(setup_done, warning_cleared), 5.0)
+        except TimeoutError:
+            pytest.fail(f"Init sequence did not finish. TX frames: {tx_frames}")
+
+        assert tx_frames[:3] == [CONFIG_START_TX, OUTPUT_MODE_TX, CONFIG_END_TX]
+
+        await _start_scenario(client, collector)
+        await _wait_for_presence(collector, all_presence)
+        # Let any unexpected extra update arrive before comparing
+        await asyncio.sleep(0.3)
+
+        assert collector.binary_states["has_target"] == EXPECTED_PRESENCE
+        assert dropped_states == EXPECTED_DROPPED_STATES
+        # Data frames must not restart the init sequence
+        assert len(tx_frames) == 3, tx_frames
+
+
+@pytest.mark.asyncio
+async def test_uart_mock_ld2410s_recovery(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test the unanswered-command retry, and that bad frames do not stop later ones."""
+    loop = asyncio.get_running_loop()
+    tx_frames: list[tuple[float, str]] = []
+    resent = loop.create_future()
+    gave_up = loop.create_future()
+    oversized_dropped = loop.create_future()
+    footer_mismatches: list[str] = []
+
+    def line_callback(line: str) -> None:
+        if (frame := _tx_frame(line)) is not None:
+            tx_frames.append((loop.time(), frame))
+        if "No acknowledgement for command 00FF, resending" in line and not (
+            resent.done()
+        ):
+            resent.set_result(True)
+        if (
+            "No acknowledgement for command 00FF, starting over in 15 s" in line
+            and not gave_up.done()
+        ):
+            gave_up.set_result(True)
+        if (
+            "Dropping frame with a 255 byte payload, larger than the receive buffer"
+            in line
+            and not oversized_dropped.done()
+        ):
+            oversized_dropped.set_result(True)
+        if "Footer does not match header: " in line:
+            footer_mismatches.append(
+                ANSI_ESCAPE.sub("", line.split("header: ", 1)[1]).strip()
+            )
+
+    collector = SensorStateCollector(
+        sensor_names=[], binary_sensor_names=["has_target"]
+    )
+    all_presence = collector.add_waiter(
+        lambda: (
+            len(collector.binary_states["has_target"])
+            >= len(EXPECTED_RECOVERY_PRESENCE)
+        )
+    )
+
+    async with (
+        run_compiled(yaml_config, line_callback=line_callback),
+        api_client_connected() as client,
+    ):
+        await _start_scenario(client, collector)
+        await _wait_for_presence(collector, all_presence)
+        try:
+            await asyncio.wait_for(asyncio.gather(resent, gave_up), 5.0)
+        except TimeoutError:
+            pytest.fail(f"No resend or give-up warning. TX frames: {tx_frames}")
+        # Let any unexpected extra update arrive before comparing
+        await asyncio.sleep(0.3)
+
+        # Recovery shows in the published states alone; the footer log below is
+        # only a cross-check that needs the fixture's VERBOSE logger level
+        assert collector.binary_states["has_target"] == EXPECTED_RECOVERY_PRESENCE
+        assert oversized_dropped.done(), "Oversized length field was not dropped"
+        assert footer_mismatches == [TRUNCATED_FRAME_HEX, BAD_COMMAND_FRAME_HEX]
+
+        frames = [frame for _, frame in tx_frames]
+        # First send plus INIT_MAX_TIMEOUTS (2) resends, then config-mode end;
+        # the restart after the 15 s pause is out of the test's time window
+        assert frames == [CONFIG_START_TX] * 3 + [CONFIG_END_TX]
+        # Resends follow the 300 ms acknowledgement timeout
+        gaps = [b[0] - a[0] for a, b in zip(tx_frames, tx_frames[1:], strict=False)]
+        assert all(0.2 <= gap <= 1.0 for gap in gaps), gaps
+
+
+# has_target after each short frame in uart_mock_ld2410s_link_lost.yaml
+EXPECTED_LINK_PRESENCE = [True, False, True, False, True, False, True]
+LINK_TIMEOUT_S = 5.0
+
+
+@pytest.mark.asyncio
+async def test_uart_mock_ld2410s_link_lost(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that short frames keep the link alive and silence restarts the init."""
+    loop = asyncio.get_running_loop()
+    tx_frames: list[str] = []
+    rx_injected_at: list[float] = []
+    setup_done = loop.create_future()
+    link_lost = loop.create_future()
+    warning_set_after_loss = loop.create_future()
+    restarted = loop.create_future()
+    config_start_before_loss = 0
+
+    def line_callback(line: str) -> None:
+        nonlocal config_start_before_loss
+        if (frame := _tx_frame(line)) is not None:
+            tx_frames.append(frame)
+            if frame == CONFIG_START_TX and link_lost.done() and not restarted.done():
+                restarted.set_result(True)
+        if "uart_mock" in line and "Injecting " in line and "(injection " in line:
+            rx_injected_at.append(loop.time())
+        if "[ld2410s" in line and "Setup done" in line and not setup_done.done():
+            setup_done.set_result(True)
+        if (
+            "No data for 5 s (0 frames dropped), starting over" in line
+            and not link_lost.done()
+        ):
+            config_start_before_loss = tx_frames.count(CONFIG_START_TX)
+            link_lost.set_result(loop.time())
+        if (
+            link_lost.done()
+            and "ld2410s set Warning flag" in line
+            and not warning_set_after_loss.done()
+        ):
+            warning_set_after_loss.set_result(True)
+
+    collector = SensorStateCollector(
+        sensor_names=[], binary_sensor_names=["has_target"]
+    )
+    all_presence = collector.add_waiter(
+        lambda: (
+            len(collector.binary_states["has_target"]) >= len(EXPECTED_LINK_PRESENCE)
+        )
+    )
+
+    async with (
+        run_compiled(yaml_config, line_callback=line_callback),
+        api_client_connected() as client,
+    ):
+        try:
+            await asyncio.wait_for(setup_done, 5.0)
+        except TimeoutError:
+            pytest.fail(f"Init sequence did not finish. TX frames: {tx_frames}")
+
+        await _start_scenario(client, collector)
+        try:
+            await asyncio.wait_for(all_presence, timeout=10.0)
+        except TimeoutError:
+            pytest.fail(
+                f"Timeout waiting for presence updates. Received: "
+                f"{collector.binary_states['has_target']}"
+            )
+        assert collector.binary_states["has_target"] == EXPECTED_LINK_PRESENCE
+
+        try:
+            lost_at = await asyncio.wait_for(link_lost, timeout=7.0)
+            await asyncio.wait_for(
+                asyncio.gather(warning_set_after_loss, restarted), timeout=1.0
+            )
+        except TimeoutError:
+            pytest.fail(
+                f"No link-lost warning, warning flag or restart. TX frames: {tx_frames}"
+            )
+
+        # The stream ran past the link timeout, so only the silence after it counts
+        silence = lost_at - rx_injected_at[-1]
+        assert silence >= LINK_TIMEOUT_S - 0.5, (
+            f"Link declared lost {silence:.2f} s after the last frame"
+        )
+
+        # The init sequence starts over with a new config-mode start
+        assert config_start_before_loss == 1, tx_frames
+        assert tx_frames.count(CONFIG_START_TX) == 2, tx_frames