Merge remote-tracking branch 'upstream/dev' into pi05

# Conflicts:
#	esphome/components/packet_interface/__init__.py
#	esphome/core/defines.h

CODEOWNERS

@@ -196,6 +196,7 @@ esphome/components/ens160_base/* @latonita @vincentscode
 esphome/components/ens160_i2c/* @latonita
 esphome/components/ens160_spi/* @latonita
 esphome/components/ens210/* @itn3rd77
+esphome/components/entity_transport/* @clydebarrow
 esphome/components/epaper_spi/* @esphome/core
 esphome/components/es7210/* @kahrendt
 esphome/components/es7243/* @Chorty

esphome/components/entity_transport/__init__.py

@@ -0,0 +1,169 @@
+"""Send sensor and binary sensor states to other devices over a packet interface."""
+
+from collections.abc import Callable
+from typing import Any
+
+import esphome.codegen as cg
+from esphome.components import packet_interface
+from esphome.components.binary_sensor import BinarySensor
+from esphome.components.const import CONF_PROVIDER, CONF_REMOTE_ID
+from esphome.components.packet_interface import CONF_PACKET_INTERFACE, interface_layers
+from esphome.components.packet_interface.layers import CONF_SESSION
+from esphome.components.sensor import Sensor
+import esphome.config_validation as cv
+from esphome.const import (
+    CONF_BINARY_SENSORS,
+    CONF_ID,
+    CONF_SENSORS,
+    CONF_TIMEOUT,
+    CONF_UPDATE_INTERVAL,
+)
+from esphome.cpp_generator import MockObj, MockObjClass
+from esphome.types import ConfigType
+
+CODEOWNERS = ["@clydebarrow"]
+DEPENDENCIES = ["packet_interface"]
+DOMAIN = "entity_transport"
+MULTI_CONF = True
+
+CONF_BROADCAST_ID = "broadcast_id"
+CONF_ENTITY_TRANSPORT_ID = "entity_transport_id"
+
+entity_transport_ns = cg.esphome_ns.namespace("entity_transport")
+EntityTransport = entity_transport_ns.class_("EntityTransport", cg.Component)
+
+_request_remote_sensor_slot = cg.slot_counter("ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT")
+_request_remote_binary_sensor_slot = cg.slot_counter(
+    "ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT"
+)
+
+
+def _advertised_id(conf: ConfigType) -> str:
+    return conf.get(CONF_BROADCAST_ID, conf[CONF_ID].id)
+
+
+def _unique_advertised_ids(entities: list[ConfigType]) -> list[ConfigType]:
+    """A receiver tells entities of one type apart by their advertised ids, so each must be different."""
+    seen: set[str] = set()
+    for index, conf in enumerate(entities):
+        if (advertised := _advertised_id(conf)) in seen:
+            raise cv.Invalid(
+                f"The id '{advertised}' is sent for two entities", path=[index]
+            )
+        seen.add(advertised)
+    return entities
+
+
+def _local_entity(cls: MockObjClass) -> Callable[[Any], Any]:
+    return cv.maybe_simple_value(
+        cv.Schema(
+            {
+                cv.Required(CONF_ID): cv.use_id(cls),
+                cv.Optional(CONF_BROADCAST_ID): cv.All(
+                    cv.validate_id_name, cv.Length(max=255)
+                ),
+            }
+        ),
+        key=CONF_ID,
+    )
+
+
+CONFIG_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(): cv.declare_id(EntityTransport),
+        cv.Required(CONF_PACKET_INTERFACE): cv.use_id(packet_interface.PacketInterface),
+        # Defaults to a periodic resend unless the interface has session and reliability layers
+        cv.Optional(CONF_UPDATE_INTERVAL): cv.All(
+            cv.update_interval, cv.Range(min=cv.TimePeriod(seconds=1))
+        ),
+        # Remote entities lose their state when none arrives for this long
+        cv.Optional(CONF_TIMEOUT): cv.All(
+            cv.positive_time_period_milliseconds,
+            cv.Range(min=cv.TimePeriod(seconds=1)),
+        ),
+        # Entities are numbered with one byte
+        cv.Optional(CONF_SENSORS): cv.All(
+            cv.ensure_list(_local_entity(Sensor)),
+            cv.Length(max=255),
+            _unique_advertised_ids,
+        ),
+        cv.Optional(CONF_BINARY_SENSORS): cv.All(
+            cv.ensure_list(_local_entity(BinarySensor)),
+            cv.Length(max=255),
+            _unique_advertised_ids,
+        ),
+    }
+).extend(cv.COMPONENT_SCHEMA)
+
+
+def _final_validate(config: ConfigType) -> ConfigType:
+    """With a session layer, which clears the states when the peer is lost, nothing would refresh an unchanged state."""
+    layers = interface_layers(config[CONF_PACKET_INTERFACE])
+    if CONF_TIMEOUT in config and layers is not None and CONF_SESSION in layers:
+        raise cv.Invalid(
+            f"Not allowed with a '{CONF_SESSION}' layer, which already clears the states when the other device "
+            "cannot be reached; unchanged states may never be sent again, so they would time out on a working link",
+            path=[CONF_TIMEOUT],
+        )
+    return config
+
+
+FINAL_VALIDATE_SCHEMA = _final_validate
+
+
+def _validate_remote_id(config: ConfigType) -> ConfigType:
+    if CONF_REMOTE_ID not in config and not config[CONF_ID].is_manual:
+        raise cv.Invalid(f"'{CONF_REMOTE_ID}' is required unless '{CONF_ID}' is given")
+    return config
+
+
+def remote_entity_schema(base_schema: cv.Schema) -> cv.Schema:
+    """Schema for an entity whose state comes from another device."""
+    return cv.All(
+        base_schema.extend(
+            {
+                cv.GenerateID(CONF_ENTITY_TRANSPORT_ID): cv.use_id(EntityTransport),
+                cv.Required(CONF_PROVIDER): cv.valid_name,
+                cv.Optional(CONF_REMOTE_ID): cv.All(
+                    cv.string_strict, cv.Length(min=1, max=255)
+                ),
+            }
+        ),
+        _validate_remote_id,
+    )
+
+
+def _remote_id(config: ConfigType) -> str:
+    return config.get(CONF_REMOTE_ID) or config[CONF_ID].id
+
+
+async def register_remote_sensor(config: ConfigType, var: MockObj) -> None:
+    hub = await cg.get_variable(config[CONF_ENTITY_TRANSPORT_ID])
+    _request_remote_sensor_slot(str(hub))
+    cg.add(hub.add_remote_sensor(config[CONF_PROVIDER], _remote_id(config), var))
+
+
+async def register_remote_binary_sensor(config: ConfigType, var: MockObj) -> None:
+    hub = await cg.get_variable(config[CONF_ENTITY_TRANSPORT_ID])
+    _request_remote_binary_sensor_slot(str(hub))
+    cg.add(hub.add_remote_binary_sensor(config[CONF_PROVIDER], _remote_id(config), var))
+
+
+async def to_code(config: ConfigType) -> None:
+    interface = await cg.get_variable(config[CONF_PACKET_INTERFACE])
+    var = cg.new_Pvariable(config[CONF_ID], interface)
+    await cg.register_component(var, config)
+    if (interval := config.get(CONF_UPDATE_INTERVAL)) is not None:
+        cg.add(var.set_update_interval(interval))
+    if (timeout := config.get(CONF_TIMEOUT)) is not None:
+        cg.add(var.set_remote_timeout(timeout))
+    sensors = config.get(CONF_SENSORS, [])
+    binary_sensors = config.get(CONF_BINARY_SENSORS, [])
+    if count := len(sensors) + len(binary_sensors):
+        cg.add(var.set_entity_count(count))
+    for conf in sensors:
+        sensor = await cg.get_variable(conf[CONF_ID])
+        cg.add(var.add_sensor(_advertised_id(conf), sensor))
+    for conf in binary_sensors:
+        sensor = await cg.get_variable(conf[CONF_ID])
+        cg.add(var.add_binary_sensor(_advertised_id(conf), sensor))

esphome/components/entity_transport/binary_sensor.py

@@ -0,0 +1,11 @@
+from esphome.components import binary_sensor
+from esphome.types import ConfigType
+
+from . import register_remote_binary_sensor, remote_entity_schema
+
+CONFIG_SCHEMA = remote_entity_schema(binary_sensor.binary_sensor_schema())
+
+
+async def to_code(config: ConfigType) -> None:
+    var = await binary_sensor.new_binary_sensor(config)
+    await register_remote_binary_sensor(config, var)

esphome/components/entity_transport/entity_transport.cpp

@@ -0,0 +1,683 @@
+#include "entity_transport.h"
+#include <algorithm>
+#include <cmath>
+#include <cstring>
+#include "esphome/core/application.h"
+#include "esphome/core/log.h"
+
+namespace esphome::entity_transport {
+
+using packet_interface::PacketBuffer;
+using packet_interface::PacketMetaData;
+
+ESPHOME_LOG_TAG(TAG, "entity_transport");
+
+// The packet type byte holds the format version in its high nibble
+static constexpr uint8_t FORMAT_VERSION = 1;
+static constexpr uint8_t PACKET_DICTIONARY = 0x11;
+static constexpr uint8_t PACKET_DATA = 0x12;
+static constexpr uint8_t PACKET_REQUEST = 0x13;
+// Entity types, in dictionary sections and data records; 0 is not used
+static constexpr uint8_t TYPE_SENSOR = 1;
+static constexpr uint8_t TYPE_BINARY_SENSOR = 2;
+static constexpr uint8_t MAX_TYPE = 31;
+// Packet type and dictionary id
+static constexpr size_t HEADER_SIZE = 5;
+// A dictionary packet's part byte follows the header: the part's index in bits 0-6, and on the last part, bit 7
+static constexpr uint8_t PART_INDEX_MASK = 0x7F;
+static constexpr uint8_t PART_LAST = 0x80;
+static constexpr uint8_t PART_NONE = 0xFF;  // no dictionary is being received in order
+// A data record starts with a 16-bit tag: the entity's number in bits 0-7, its type in bits 8-12, the payload form in
+// bits 13-14, and with PAYLOAD_NONE, the state in bit 15. Any record can be skipped without knowing its type.
+static constexpr size_t TAG_SIZE = 2;
+static constexpr uint8_t TAG_TYPE_SHIFT = 8;
+static constexpr uint8_t TAG_PAYLOAD_SHIFT = 13;
+static constexpr uint16_t TAG_VALUE_BIT = 0x8000;
+static constexpr uint8_t PAYLOAD_NONE = 0;      // the state is in the tag's value bit
+static constexpr uint8_t PAYLOAD_WORD = 1;      // 4 bytes follow
+static constexpr uint8_t PAYLOAD_LENGTH = 2;    // a length byte follows, then that many bytes
+static constexpr uint8_t PAYLOAD_NO_STATE = 3;  // the entity has no state
+static constexpr size_t WORD_SIZE = 4;
+static constexpr size_t MAX_RECORD_SIZE = TAG_SIZE + WORD_SIZE;
+// A dictionary section starts with the entity type, first number and count
+static constexpr size_t SECTION_HEADER_SIZE = 3;
+static constexpr size_t MAX_STRING_LENGTH = 255;
+static constexpr uint32_t DEFAULT_UPDATE_INTERVAL_MS = 15000;
+static constexpr uint32_t REQUEST_INTERVAL_MS = 5000;
+static constexpr uint32_t RETRY_DELAY_MS = 1000;
+static constexpr uint32_t RETRY_TIMER_ID = 0;
+static constexpr uint32_t TIMEOUT_CHECK_INTERVAL_MS = 1000;
+static constexpr uint32_t WARNING_INTERVAL_MS = 60000;
+// How long the interface may refuse every packet before it is reported
+static constexpr uint32_t REFUSED_WARNING_MS = 60000;
+
+static void put_uint32(uint8_t *buf, uint32_t value) {
+  for (size_t i = 0; i != 4; i++)
+    buf[i] = value >> (i * 8);
+}
+
+// True if the NUL terminated str is the same as the len bytes at data
+static bool matches(const char *str, const char *data, size_t len) {
+  return strlen(str) == len && memcmp(str, data, len) == 0;
+}
+
+struct Record {
+  uint8_t number;
+  uint8_t type;
+  uint8_t payload;
+  bool value;
+  const uint8_t *data;  // what follows the tag
+};
+
+static Record parse_record(std::span<const uint8_t> packet, size_t pos) {
+  uint16_t tag = encode_uint16(packet[pos + 1], packet[pos]);
+  return {static_cast<uint8_t>(tag), static_cast<uint8_t>((tag >> TAG_TYPE_SHIFT) & MAX_TYPE),
+          static_cast<uint8_t>((tag >> TAG_PAYLOAD_SHIFT) & 3), (tag & TAG_VALUE_BIT) != 0,
+          packet.data() + pos + TAG_SIZE};
+}
+
+// The size of the data record at pos, or 0 if it does not fit in the packet
+static size_t record_size(std::span<const uint8_t> packet, size_t pos) {
+  if (pos + TAG_SIZE > packet.size())
+    return 0;
+  size_t size = TAG_SIZE;
+  uint8_t payload = parse_record(packet, pos).payload;
+  if (payload == PAYLOAD_WORD) {
+    size += WORD_SIZE;
+  } else if (payload == PAYLOAD_LENGTH) {
+    if (pos + TAG_SIZE == packet.size())
+      return 0;
+    size += 1 + packet[pos + TAG_SIZE];
+  }
+  return pos + size <= packet.size() ? size : 0;
+}
+
+// Checks every record of a data packet, so that a damaged or forged packet changes no state at all
+static bool data_well_formed(std::span<const uint8_t> packet) {
+  for (size_t pos = HEADER_SIZE; pos != packet.size();) {
+    size_t size = record_size(packet, pos);
+    if (size == 0)
+      return false;
+    Record record = parse_record(packet, pos);
+    if (record.type == 0 || (record.value && record.payload != PAYLOAD_NONE))
+      return false;
+    // Types this device knows must use their own payload form; others are skipped
+    if (record.type == TYPE_SENSOR && record.payload != PAYLOAD_WORD && record.payload != PAYLOAD_NO_STATE)
+      return false;
+    if (record.type == TYPE_BINARY_SENSOR && record.payload != PAYLOAD_NONE && record.payload != PAYLOAD_NO_STATE)
+      return false;
+    pos += size;
+  }
+  return true;
+}
+
+// Checks the device name and every section of a dictionary packet before anything is learnt from it
+static bool dictionary_well_formed(std::span<const uint8_t> packet) {
+  size_t pos = HEADER_SIZE + 1;
+  if (pos >= packet.size())
+    return false;
+  pos += 1 + packet[pos];
+  if (pos > packet.size())
+    return false;
+  while (pos != packet.size()) {
+    if (pos + SECTION_HEADER_SIZE > packet.size())
+      return false;
+    uint8_t type = packet[pos];
+    size_t first = packet[pos + 1];
+    size_t count = packet[pos + 2];
+    pos += SECTION_HEADER_SIZE;
+    // Numbers are one byte, and a data record can name types up to MAX_TYPE only
+    if (type == 0 || type > MAX_TYPE || count == 0 || first + count > 256)
+      return false;
+    for (size_t i = 0; i != count; i++) {
+      if (pos == packet.size())
+        return false;
+      pos += 1 + packet[pos];
+      if (pos > packet.size())
+        return false;
+    }
+  }
+  return true;
+}
+
+void EntityTransport::setup() {
+  this->interface_ = this->interface_->top();
+  this->max_packet_size_ = this->interface_->get_max_packet_size();
+  const auto &name = App.get_name();
+  size_t longest_id = 0;
+  for (const auto &entity : this->entities_)
+    longest_id = std::max(longest_id, strlen(entity.id));
+  // A dictionary packet must hold the device name and any one id, which also leaves room for a data record
+  if (name.size() > MAX_STRING_LENGTH || longest_id > MAX_STRING_LENGTH ||
+      HEADER_SIZE + 2 + name.size() + SECTION_HEADER_SIZE + 1 + longest_id > this->max_packet_size_) {
+    this->mark_failed(LOG_STR("max_packet_size too small for the device name and entity ids"));
+    return;
+  }
+  RAMAllocator<uint8_t> allocator;
+  this->buffer_ = allocator.allocate(this->max_packet_size_);
+  if (this->buffer_ == nullptr) {
+    this->mark_failed(LOG_STR("Not enough memory for the packet buffer"));
+    return;
+  }
+  // The dictionary id changes when the device name or any entity changes
+  uint32_t hash = fnv1a_hash(name.c_str());
+  for (const auto &entity : this->entities_) {
+    hash = fnv1a_hash_extend(hash, entity.is_sensor ? TYPE_SENSOR : TYPE_BINARY_SENSOR);
+    hash = fnv1a_hash_extend(hash, entity.id);
+  }
+  this->dictionary_id_ = hash;
+
+  // [&entity] is safe: FixedVector never reallocates, and the parent pointer keeps the capture to one pointer
+  for (auto &entity : this->entities_) {
+#ifdef USE_SENSOR
+    if (entity.is_sensor) {
+      static_cast<sensor::Sensor *>(entity.entity)->add_on_state_callback([&entity](float) {
+        entity.parent->mark_updated_(entity);
+      });
+    }
+#endif
+#ifdef USE_BINARY_SENSOR
+    if (!entity.is_sensor) {
+      // The full state callback also fires for the first state
+      static_cast<binary_sensor::BinarySensor *>(entity.entity)
+          ->add_full_state_callback(
+              [&entity](optional<bool>, optional<bool>) { entity.parent->mark_updated_(entity); });
+    }
+#endif
+  }
+  this->interface_->add_packet_interface_listener(
+      [this](const PacketBuffer &data, const PacketMetaData &) { this->receive_(data); });
+
+  bool session = this->interface_->has_capability(packet_interface::CAPABILITY_SESSION);
+  if (session)
+    this->interface_->add_on_session_state_callback([this](bool connected) { this->on_session_(connected); });
+  // Without a session and reliable delivery, a lost dictionary or state is only replaced by a periodic resend
+  if (this->update_interval_ms_ == 0) {
+    this->update_interval_ms_ = session && this->interface_->has_capability(packet_interface::CAPABILITY_RELIABILITY)
+                                    ? SCHEDULER_DONT_RUN
+                                    : DEFAULT_UPDATE_INTERVAL_MS;
+  }
+  if (this->update_interval_ms_ != SCHEDULER_DONT_RUN)
+    this->set_interval(this->update_interval_ms_, [this] { this->send_all_(); });
+  if (this->remote_timeout_ms_ != 0)
+    this->set_interval(TIMEOUT_CHECK_INTERVAL_MS, [this] { this->clear_remote_states_(true); });
+  this->send_all_();
+}
+
+void EntityTransport::dump_config() {
+  ESP_LOGCONFIG(TAG,
+                "Entity Transport:\n"
+                "  Packet Interface: %s\n"
+                "  Max Packet Size: %zu\n"
+                "  Encryption: %s\n"
+                "  Reliability: %s\n"
+                "  Session: %s\n"
+                "  Dictionary ID: %08" PRIX32,
+                this->interface_->get_name(), this->max_packet_size_,
+                YESNO(this->interface_->has_capability(packet_interface::CAPABILITY_ENCRYPTION)),
+                YESNO(this->interface_->has_capability(packet_interface::CAPABILITY_RELIABILITY)),
+                YESNO(this->interface_->has_capability(packet_interface::CAPABILITY_SESSION)), this->dictionary_id_);
+  if (this->update_interval_ms_ == SCHEDULER_DONT_RUN) {
+    ESP_LOGCONFIG(TAG, "  Update Interval: never");
+  } else {
+    ESP_LOGCONFIG(TAG, "  Update Interval: %" PRIu32 "ms", this->update_interval_ms_);
+  }
+  if (this->remote_timeout_ms_ != 0) {
+    ESP_LOGCONFIG(TAG, "  Remote Timeout: %" PRIu32 "ms", this->remote_timeout_ms_);
+  }
+  for (const auto &entity : this->entities_) {
+    ESP_LOGCONFIG(TAG, "  %s %u: %s", entity.is_sensor ? LOG_STR_LITERAL("Sensor") : LOG_STR_LITERAL("Binary Sensor"),
+                  entity.number, entity.id);
+  }
+#ifdef ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT
+  for (const auto &remote : this->remote_sensors_) {
+    ESP_LOGCONFIG(TAG, "  Remote Sensor: %s from %s", remote.remote_id, remote.provider);
+  }
+#endif
+#ifdef ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT
+  for (const auto &remote : this->remote_binary_sensors_) {
+    ESP_LOGCONFIG(TAG, "  Remote Binary Sensor: %s from %s", remote.remote_id, remote.provider);
+  }
+#endif
+}
+
+void EntityTransport::send_all_() {
+  this->dictionary_pending_ = true;
+  this->dictionary_next_ = 0;
+  this->dictionary_part_ = 0;
+  for (auto &entity : this->entities_)
+    entity.updated = true;
+  this->enable_loop();
+}
+
+void EntityTransport::mark_updated_(LocalEntity &entity) {
+  entity.updated = true;
+  this->enable_loop();
+}
+
+void EntityTransport::loop() {
+  this->disable_loop();
+  // States are only sent once the dictionary that names them has been accepted
+  if ((this->dictionary_pending_ && !this->send_dictionary_()) || !this->send_updated_()) {
+    this->sending_refused_();
+    this->set_timeout(RETRY_TIMER_ID, RETRY_DELAY_MS, [this] { this->enable_loop(); });
+    return;
+  }
+  this->refusing_ = false;
+  if (this->status_has_warning()) {
+    ESP_LOGI(TAG, "Packets are accepted again");
+    this->status_clear_warning();
+  }
+}
+
+// A link that is down or full refuses packets for a while; one that refuses them for long sends nothing at all
+void EntityTransport::sending_refused_() {
+  uint32_t now = App.get_loop_component_start_time();
+  if (!this->refusing_) {
+    this->refusing_ = true;
+    this->refused_since_ms_ = now;
+    return;
+  }
+  if (!this->status_has_warning() && now - this->refused_since_ms_ >= REFUSED_WARNING_MS) {
+    ESP_LOGW(TAG, "The packet interface has accepted no packets for %" PRIu32 "s, so no states are sent",
+             (now - this->refused_since_ms_) / 1000);
+    this->status_set_warning();
+  }
+}
+
+size_t EntityTransport::write_header_(uint8_t type) const {
+  this->buffer_[0] = type;
+  put_uint32(this->buffer_ + 1, this->dictionary_id_);
+  return HEADER_SIZE;
+}
+
+bool EntityTransport::send_dictionary_() {
+  uint8_t *buf = this->buffer_;
+  const auto &name = App.get_name();
+  size_t header = this->write_header_(PACKET_DICTIONARY);
+  size_t part_offset = header++;
+  buf[header++] = name.size();
+  memcpy(buf + header, name.c_str(), name.size());
+  header += name.size();
+  size_t length = header;
+  size_t count_offset = 0;
+  uint8_t section_type = 0;  // none
+  // A dictionary in several packets resumes after the last packet accepted, so a refused one does not resend the rest
+  for (size_t i = this->dictionary_next_; i != this->entities_.size(); i++) {
+    const auto &entity = this->entities_[i];
+    uint8_t type = entity.is_sensor ? TYPE_SENSOR : TYPE_BINARY_SENSOR;
+    size_t id_length = strlen(entity.id);
+    bool new_section = type != section_type;
+    if (length + (new_section ? SECTION_HEADER_SIZE : 0) + 1 + id_length > this->max_packet_size_) {
+      buf[part_offset] = this->dictionary_part_ & PART_INDEX_MASK;
+      if (!this->interface_->send_to_interface(PacketBuffer(buf, length), {})) {
+        ESP_LOGV(TAG, "Dictionary packet of %zu bytes not accepted, will retry", length);
+        return false;
+      }
+      this->dictionary_next_ = i;
+      this->dictionary_part_++;
+      length = header;
+      new_section = true;
+    }
+    if (new_section) {
+      buf[length++] = type;
+      buf[length++] = entity.number;
+      count_offset = length;
+      buf[length++] = 0;
+      section_type = type;
+    }
+    buf[count_offset]++;
+    buf[length++] = id_length;
+    memcpy(buf + length, entity.id, id_length);
+    length += id_length;
+  }
+  if (length != header) {
+    buf[part_offset] = (this->dictionary_part_ & PART_INDEX_MASK) | PART_LAST;
+    if (!this->interface_->send_to_interface(PacketBuffer(buf, length), {})) {
+      ESP_LOGV(TAG, "Dictionary packet of %zu bytes not accepted, will retry", length);
+      return false;
+    }
+  }
+  this->dictionary_pending_ = false;
+  this->dictionary_next_ = 0;
+  this->dictionary_part_ = 0;
+  return true;
+}
+
+// Writes the entity's data record, and returns its size
+size_t EntityTransport::write_record_(const LocalEntity &entity, uint8_t *record) const {
+  uint8_t type = entity.is_sensor ? TYPE_SENSOR : TYPE_BINARY_SENSOR;
+  uint8_t payload = PAYLOAD_NO_STATE;
+  bool value = false;
+  size_t size = TAG_SIZE;
+#ifdef USE_SENSOR
+  if (entity.is_sensor) {
+    float state = static_cast<sensor::Sensor *>(entity.entity)->get_state();
+    if (!std::isnan(state)) {
+      payload = PAYLOAD_WORD;
+      uint32_t bits;
+      memcpy(&bits, &state, sizeof(bits));
+      put_uint32(record + TAG_SIZE, bits);
+      size += WORD_SIZE;
+    }
+  }
+#endif
+#ifdef USE_BINARY_SENSOR
+  if (!entity.is_sensor) {
+    auto *sensor = static_cast<binary_sensor::BinarySensor *>(entity.entity);
+    if (sensor->has_state()) {
+      payload = PAYLOAD_NONE;
+      value = sensor->state;
+    }
+  }
+#endif
+  uint16_t tag =
+      entity.number | (type << TAG_TYPE_SHIFT) | (payload << TAG_PAYLOAD_SHIFT) | (value ? TAG_VALUE_BIT : 0);
+  record[0] = tag;
+  record[1] = tag >> 8;
+  return size;
+}
+
+// Sends every updated entity, in as few packets as fit; returns false if a packet was refused
+bool EntityTransport::send_updated_() {
+  size_t length = this->write_header_(PACKET_DATA);
+  size_t first = 0;
+  for (size_t i = 0; i != this->entities_.size(); i++) {
+    auto &entity = this->entities_[i];
+    if (!entity.updated)
+      continue;
+    uint8_t record[MAX_RECORD_SIZE];
+    size_t size = this->write_record_(entity, record);
+    if (length + size > this->max_packet_size_) {
+      if (!this->flush_(length, first, i))
+        return false;
+      length = HEADER_SIZE;
+      first = i;
+    }
+    memcpy(this->buffer_ + length, record, size);
+    length += size;
+  }
+  return this->flush_(length, first, this->entities_.size());
+}
+
+// Sends the packet holding the updated entities from first up to last, and clears their flags if it was accepted
+bool EntityTransport::flush_(size_t length, size_t first, size_t last) {
+  if (length == HEADER_SIZE)
+    return true;
+  if (!this->interface_->send_to_interface(PacketBuffer(this->buffer_, length), {})) {
+    ESP_LOGV(TAG, "Packet of %zu bytes not accepted, will retry", length);
+    return false;
+  }
+  for (size_t i = first; i != last; i++)
+    this->entities_[i].updated = false;
+  return true;
+}
+
+void EntityTransport::receive_(const PacketBuffer &data) {
+  // Packet interface layers deliver packets in one piece
+  if (!data.is_contiguous()) {
+    ESP_LOGW(TAG, "Packet split across buffers is not supported");
+    return;
+  }
+  auto packet = data.as_span();
+  if (packet.size() < HEADER_SIZE) {
+    if (this->warning_due_()) {
+      ESP_LOGW(TAG, "Packet of %zu bytes is too short", packet.size());
+    }
+    return;
+  }
+  if (packet[0] >> 4 != FORMAT_VERSION) {
+    if (this->warning_due_()) {
+      ESP_LOGW(TAG, "Packet format version %u not supported; the sender may run another ESPHome version",
+               packet[0] >> 4);
+    }
+    return;
+  }
+  uint32_t dictionary_id = encode_uint32(packet[4], packet[3], packet[2], packet[1]);
+  if (dictionary_id == this->dictionary_id_) {
+    // Either a request for our dictionary, or our own packet heard on a broadcast channel
+    if (packet[0] == PACKET_REQUEST && packet.size() == HEADER_SIZE) {
+      ESP_LOGD(TAG, "Dictionary requested");
+      this->send_all_();
+    }
+    return;
+  }
+  if (packet[0] == PACKET_DICTIONARY) {
+    this->receive_dictionary_(packet, dictionary_id);
+  } else if (packet[0] == PACKET_DATA) {
+    this->receive_data_(packet, dictionary_id);
+  } else if (packet[0] != PACKET_REQUEST) {
+    ESP_LOGV(TAG, "Unknown packet type 0x%02X", packet[0]);
+  }
+}
+
+void EntityTransport::receive_dictionary_(std::span<const uint8_t> packet, uint32_t dictionary_id) {
+  if (!dictionary_well_formed(packet)) {
+    if (this->warning_due_()) {
+      ESP_LOGW(TAG, "Dictionary packet of %zu bytes is malformed", packet.size());
+    }
+    return;
+  }
+  size_t pos = HEADER_SIZE;
+  uint8_t part = packet[pos++];
+  uint8_t index = part & PART_INDEX_MASK;
+  // The dictionary is complete, and need not be asked for again, once its last part follows all the others in order
+  bool in_order = index == 0 || (this->rx_dictionary_id_ == dictionary_id && index == this->rx_next_part_);
+  this->rx_dictionary_id_ = dictionary_id;
+  this->rx_next_part_ = in_order ? (index + 1) & PART_INDEX_MASK : PART_NONE;
+  bool completed = in_order && (part & PART_LAST) != 0;
+  bool first_time = completed && !this->dictionary_seen_(dictionary_id);
+  if (first_time) {
+    this->seen_dictionaries_[this->seen_next_] = dictionary_id;
+    this->seen_next_ = (this->seen_next_ + 1) % this->seen_dictionaries_.size();
+  }
+  size_t provider_length = packet[pos++];
+  const auto *provider = reinterpret_cast<const char *>(packet.data() + pos);
+  pos += provider_length;
+  while (pos != packet.size()) {
+    uint8_t type = packet[pos];
+    uint8_t number = packet[pos + 1];
+    uint8_t count = packet[pos + 2];
+    pos += SECTION_HEADER_SIZE;
+    for (uint8_t i = 0; i != count; i++) {
+      size_t id_length = packet[pos++];
+      this->learn_(provider, provider_length, type, number + i, reinterpret_cast<const char *>(packet.data() + pos),
+                   id_length, dictionary_id);
+      pos += id_length;
+    }
+  }
+  // A provider may go back to a dictionary already seen, so every complete one is checked
+  if (completed)
+    this->forget_unnamed_(provider, provider_length, dictionary_id, first_time);
+}
+
+void EntityTransport::forget_unnamed_(const char *provider, size_t provider_length, uint32_t dictionary_id,
+                                      bool first_time) {
+  // A remote entity that its provider's whole dictionary does not name would otherwise keep no or a stale state
+  // silently
+#ifdef ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT
+  for (auto &remote : this->remote_sensors_) {
+    if (!matches(remote.provider, provider, provider_length) || (remote.known && remote.dictionary_id == dictionary_id))
+      continue;
+    if (remote.known || first_time) {
+      ESP_LOGW(TAG, "'%s' does not send '%s'", remote.provider, remote.remote_id);
+    }
+    remote.known = false;
+    if (remote.has_state) {
+      remote.has_state = false;
+      remote.entity->publish_state(NAN);
+    }
+  }
+#endif
+#ifdef ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT
+  for (auto &remote : this->remote_binary_sensors_) {
+    if (!matches(remote.provider, provider, provider_length) || (remote.known && remote.dictionary_id == dictionary_id))
+      continue;
+    if (remote.known || first_time) {
+      ESP_LOGW(TAG, "'%s' does not send '%s'", remote.provider, remote.remote_id);
+    }
+    remote.known = false;
+    if (remote.has_state) {
+      remote.has_state = false;
+      remote.entity->invalidate_state();
+    }
+  }
+#endif
+}
+
+void EntityTransport::learn_(const char *provider, size_t provider_length, uint8_t type, uint8_t number, const char *id,
+                             size_t id_length, uint32_t dictionary_id) {
+#ifdef ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT
+  if (type == TYPE_SENSOR) {
+    for (auto &remote : this->remote_sensors_) {
+      if (matches(remote.provider, provider, provider_length) && matches(remote.remote_id, id, id_length)) {
+        remote.dictionary_id = dictionary_id;
+        remote.number = number;
+        remote.known = true;
+      }
+    }
+  }
+#endif
+#ifdef ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT
+  if (type == TYPE_BINARY_SENSOR) {
+    for (auto &remote : this->remote_binary_sensors_) {
+      if (matches(remote.provider, provider, provider_length) && matches(remote.remote_id, id, id_length)) {
+        remote.dictionary_id = dictionary_id;
+        remote.number = number;
+        remote.known = true;
+      }
+    }
+  }
+#endif
+}
+
+bool EntityTransport::dictionary_seen_(uint32_t dictionary_id) const {
+  return std::find(this->seen_dictionaries_.begin(), this->seen_dictionaries_.end(), dictionary_id) !=
+         this->seen_dictionaries_.end();
+}
+
+void EntityTransport::receive_data_(std::span<const uint8_t> packet, uint32_t dictionary_id) {
+  bool known = false;
+  bool has_remotes = false;
+#ifdef ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT
+  for (const auto &remote : this->remote_sensors_) {
+    known |= remote.known && remote.dictionary_id == dictionary_id;
+    has_remotes = true;
+  }
+#endif
+#ifdef ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT
+  for (const auto &remote : this->remote_binary_sensors_) {
+    known |= remote.known && remote.dictionary_id == dictionary_id;
+    has_remotes = true;
+  }
+#endif
+  if (!has_remotes)
+    return;
+  // Checked first, so a damaged packet does not make the provider resend everything
+  if (!data_well_formed(packet)) {
+    if (this->warning_due_()) {
+      ESP_LOGW(TAG, "Data packet of %zu bytes is malformed", packet.size());
+    }
+    return;
+  }
+  // A dictionary not yet received in full may name our entities, or be a provider's new one; one received in full
+  // named all of them it could, so asking again would only make that provider resend everything
+  if (!this->dictionary_seen_(dictionary_id))
+    this->request_dictionary_(dictionary_id);
+  if (!known)
+    return;
+  [[maybe_unused]] uint32_t now = App.get_loop_component_start_time();
+  for (size_t pos = HEADER_SIZE; pos != packet.size(); pos += record_size(packet, pos)) {
+    [[maybe_unused]] Record record = parse_record(packet, pos);
+#ifdef ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT
+    if (record.type == TYPE_SENSOR) {
+      float value = NAN;
+      if (record.payload == PAYLOAD_WORD) {
+        uint32_t bits = encode_uint32(record.data[3], record.data[2], record.data[1], record.data[0]);
+        memcpy(&value, &bits, sizeof(value));
+      }
+      for (auto &remote : this->remote_sensors_) {
+        if (remote.known && remote.dictionary_id == dictionary_id && remote.number == record.number) {
+          remote.last_state_ms = now;
+          remote.has_state = true;
+          remote.entity->publish_state(value);
+        }
+      }
+    }
+#endif
+#ifdef ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT
+    if (record.type == TYPE_BINARY_SENSOR) {
+      for (auto &remote : this->remote_binary_sensors_) {
+        if (!remote.known || remote.dictionary_id != dictionary_id || remote.number != record.number)
+          continue;
+        remote.last_state_ms = now;
+        remote.has_state = true;
+        if (record.payload == PAYLOAD_NO_STATE) {
+          remote.entity->invalidate_state();
+        } else {
+          remote.entity->publish_state(record.value);
+        }
+      }
+    }
+#endif
+  }
+}
+
+void EntityTransport::request_dictionary_(uint32_t dictionary_id) {
+  uint32_t now = App.get_loop_component_start_time();
+  if (this->request_sent_ && now - this->last_request_ms_ < REQUEST_INTERVAL_MS)
+    return;
+  uint8_t packet[HEADER_SIZE] = {PACKET_REQUEST};
+  put_uint32(packet + 1, dictionary_id);
+  // A refused request is tried again with the next data packet under that dictionary id
+  if (!this->interface_->send_to_interface(PacketBuffer(packet, sizeof(packet)), {})) {
+    ESP_LOGV(TAG, "Request for dictionary %08" PRIX32 " not accepted", dictionary_id);
+    return;
+  }
+  this->request_sent_ = true;
+  this->last_request_ms_ = now;
+  ESP_LOGD(TAG, "Requested dictionary %08" PRIX32, dictionary_id);
+}
+
+// Dropped packets point to a device with other firmware or a fault, and may arrive steadily, so warn only now and then
+bool EntityTransport::warning_due_() {
+  uint32_t now = App.get_loop_component_start_time();
+  if (this->warned_ && now - this->last_warning_ms_ < WARNING_INTERVAL_MS)
+    return false;
+  this->warned_ = true;
+  this->last_warning_ms_ = now;
+  return true;
+}
+
+// A newly connected or restarted peer needs the dictionary and every state; while it is lost, its entities have no
+// state
+void EntityTransport::on_session_(bool connected) {
+  if (connected) {
+    this->send_all_();
+    return;
+  }
+  this->clear_remote_states_(false);
+}
+
+// Clears the state of every remote entity, or with timed_out_only, of those whose state is older than the timeout
+void EntityTransport::clear_remote_states_(bool timed_out_only) {
+  [[maybe_unused]] uint32_t now = App.get_loop_component_start_time();
+#ifdef ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT
+  for (auto &remote : this->remote_sensors_) {
+    if (timed_out_only && (!remote.has_state || now - remote.last_state_ms <= this->remote_timeout_ms_))
+      continue;
+    remote.has_state = false;
+    remote.entity->publish_state(NAN);
+  }
+#endif
+#ifdef ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT
+  for (auto &remote : this->remote_binary_sensors_) {
+    if (timed_out_only && (!remote.has_state || now - remote.last_state_ms <= this->remote_timeout_ms_))
+      continue;
+    remote.has_state = false;
+    remote.entity->invalidate_state();
+  }
+#endif
+}
+
+}  // namespace esphome::entity_transport

esphome/components/entity_transport/entity_transport.h

@@ -0,0 +1,171 @@
+#pragma once
+
+#include <array>
+#include <span>
+
+#include "esphome/components/packet_interface/packet_interface.h"
+#include "esphome/core/component.h"
+#include "esphome/core/defines.h"
+#include "esphome/core/entity_base.h"
+#include "esphome/core/helpers.h"
+#ifdef USE_SENSOR
+#include "esphome/components/sensor/sensor.h"
+#endif
+#ifdef USE_BINARY_SENSOR
+#include "esphome/components/binary_sensor/binary_sensor.h"
+#endif
+
+namespace esphome::entity_transport {
+
+/**
+ * Sends the states of local sensors and binary sensors over a packet interface, and publishes the states received
+ * from other devices to remote entities.
+ *
+ * Each device numbers its sensors and its binary sensors from 0, and sends a dictionary that gives the id for each
+ * number. Data packets then refer to an entity by its number. Every packet carries the sender's dictionary id, a hash
+ * of its device name and dictionary, so that a receiver knows which device sent it and which dictionary applies.
+ * All values are little endian.
+ *
+ * Dictionary packet:
+ * - type: 1 byte, PACKET_DICTIONARY;
+ * - dictionary id: 4 bytes;
+ * - part: 1 byte, the packet's index in the dictionary in bits 0-6, counting from 0 and wrapping, and bit 7 set on
+ *   the last packet;
+ * - sender's device name: length (1 byte), then the name;
+ * - one or more sections, each: entity type (1 byte), first number (1 byte), count (1 byte), then for each entity
+ *   its id: length (1 byte), then the id.
+ *
+ * Data packet:
+ * - type: 1 byte, PACKET_DATA;
+ * - dictionary id: 4 bytes;
+ * - records, to the end, each a 16-bit tag, then its payload. The tag holds the entity's number in bits 0-7, its type
+ *   in bits 8-12, and the payload form in bits 13-14: none, with the state in bit 15 (binary sensors); 4 bytes (a
+ *   sensor's float); a length byte, then that many bytes; or no state, with no payload. The form gives the record's
+ *   size, so a receiver skips records of types it does not know.
+ *
+ * Every packet is checked in full before anything in it is used, so a damaged or forged packet changes no state.
+ *
+ * Dictionary request packet, sent by a device that receives data with a dictionary id it has not received in full,
+ * with every part in order:
+ * - type: 1 byte, PACKET_REQUEST;
+ * - the dictionary id wanted: 4 bytes.
+ *
+ * Changed states are sent from loop(). The dictionary and all states are sent at boot, when a session layer reports
+ * the peer connected, when the dictionary is requested, and every update interval. Unless the interface has both a
+ * session and a reliability layer, the update interval defaults to DEFAULT_UPDATE_INTERVAL_MS. When the interface
+ * refuses a packet, sending is tried again after a delay, and a warning is set once it has refused every packet for a
+ * minute. With a remote timeout, a remote entity loses its state when
+ * none arrives for it within the timeout.
+ */
+class EntityTransport : public Component {
+ public:
+  explicit EntityTransport(packet_interface::PacketInterface *interface) : interface_(interface) {}
+
+  void setup() override;
+  void loop() override;
+  void dump_config() override;
+
+  /// 0 chooses the default for the interface's layers; SCHEDULER_DONT_RUN never resends.
+  void set_update_interval(uint32_t interval_ms) { this->update_interval_ms_ = interval_ms; }
+  /// Remote entities lose their state when none arrives for this long; 0 keeps the last state.
+  void set_remote_timeout(uint32_t timeout_ms) { this->remote_timeout_ms_ = timeout_ms; }
+  void set_entity_count(size_t count) { this->entities_.init(count); }
+  // Sensors must be added before binary sensors, so that data packets hold sensor records first
+#ifdef USE_SENSOR
+  void add_sensor(const char *id, sensor::Sensor *sensor) {
+    this->entities_.push_back({sensor, id, this, this->sensor_count_++, true, true});
+  }
+#endif
+#ifdef USE_BINARY_SENSOR
+  void add_binary_sensor(const char *id, binary_sensor::BinarySensor *sensor) {
+    this->entities_.push_back({sensor, id, this, this->binary_sensor_count_++, false, true});
+  }
+#endif
+#ifdef ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT
+  void add_remote_sensor(const char *provider, const char *remote_id, sensor::Sensor *sensor) {
+    this->remote_sensors_.push_back({provider, remote_id, sensor});
+  }
+#endif
+#ifdef ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT
+  void add_remote_binary_sensor(const char *provider, const char *remote_id, binary_sensor::BinarySensor *sensor) {
+    this->remote_binary_sensors_.push_back({provider, remote_id, sensor});
+  }
+#endif
+
+ protected:
+  struct LocalEntity {
+    EntityBase *entity;  // a sensor::Sensor or a binary_sensor::BinarySensor, as is_sensor says
+    const char *id;
+    EntityTransport *parent;
+    uint8_t number;
+    bool is_sensor;
+    bool updated;
+  };
+  template<typename T> struct RemoteEntity {
+    const char *provider;
+    const char *remote_id;
+    T *entity;
+    uint32_t dictionary_id{0};  // learnt from the provider's dictionary
+    uint32_t last_state_ms{0};
+    uint8_t number{0};
+    bool known{false};
+    bool has_state{false};  // a state arrived, and has not yet timed out
+  };
+
+  void send_all_();
+  void mark_updated_(LocalEntity &entity);
+  bool send_dictionary_();
+  bool send_updated_();
+  size_t write_record_(const LocalEntity &entity, uint8_t *record) const;
+  bool flush_(size_t length, size_t first, size_t last);
+  size_t write_header_(uint8_t type) const;
+  void receive_(const packet_interface::PacketBuffer &data);
+  void receive_dictionary_(std::span<const uint8_t> packet, uint32_t dictionary_id);
+  void receive_data_(std::span<const uint8_t> packet, uint32_t dictionary_id);
+  void learn_(const char *provider, size_t provider_length, uint8_t type, uint8_t number, const char *id,
+              size_t id_length, uint32_t dictionary_id);
+  void request_dictionary_(uint32_t dictionary_id);
+  bool dictionary_seen_(uint32_t dictionary_id) const;
+  /// Clears each remote entity of \p provider that its complete dictionary does not name, warning if the entity was
+  /// known or the dictionary is \p first_time seen, so a resent dictionary does not repeat the warning.
+  void forget_unnamed_(const char *provider, size_t provider_length, uint32_t dictionary_id, bool first_time);
+  bool warning_due_();
+  void sending_refused_();
+  void on_session_(bool connected);
+  void clear_remote_states_(bool timed_out_only);
+
+  packet_interface::PacketInterface *interface_;
+  FixedVector<LocalEntity> entities_;
+#ifdef ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT
+  StaticVector<RemoteEntity<sensor::Sensor>, ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT> remote_sensors_;
+#endif
+#ifdef ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT
+  StaticVector<RemoteEntity<binary_sensor::BinarySensor>, ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT>
+      remote_binary_sensors_;
+#endif
+  // Packets are built here, allocated in setup()
+  uint8_t *buffer_{nullptr};
+  size_t max_packet_size_{0};
+  uint32_t update_interval_ms_{0};
+  uint32_t remote_timeout_ms_{0};
+  uint32_t dictionary_id_{0};
+  uint32_t last_request_ms_{0};
+  uint32_t last_warning_ms_{0};
+  uint32_t refused_since_ms_{0};  // when the interface began refusing every packet
+  uint32_t rx_dictionary_id_{0};  // the dictionary whose parts are arriving
+  // The last dictionary ids received in full, so a provider whose dictionary names none of our entities is not asked
+  // again
+  std::array<uint32_t, 8> seen_dictionaries_{};
+  uint8_t seen_next_{0};
+  uint8_t rx_next_part_{0};     // the index of its next part, or PART_NONE if one was missed
+  uint8_t dictionary_part_{0};  // the index of the next dictionary packet to send
+  uint8_t sensor_count_{0};
+  uint8_t binary_sensor_count_{0};
+  size_t dictionary_next_{0};  // the first entity of the next dictionary packet to send
+  bool dictionary_pending_{true};
+  bool request_sent_{false};
+  bool warned_{false};
+  bool refusing_{false};
+};
+
+}  // namespace esphome::entity_transport

esphome/components/entity_transport/sensor.py

@@ -0,0 +1,11 @@
+from esphome.components import sensor
+from esphome.types import ConfigType
+
+from . import register_remote_sensor, remote_entity_schema
+
+CONFIG_SCHEMA = remote_entity_schema(sensor.sensor_schema())
+
+
+async def to_code(config: ConfigType) -> None:
+    var = await sensor.new_sensor(config)
+    await register_remote_sensor(config, var)

esphome/components/packet_interface/__init__.py

@@ -6,7 +6,9 @@ import esphome.codegen as cg
 from esphome.components.const import CONF_MAX_PACKET_SIZE
 import esphome.config_validation as cv
 from esphome.const import CONF_ID
+from esphome.core import ID
 from esphome.cpp_generator import MockObj, MockObjClass
+import esphome.final_validate as fv
 from esphome.types import ConfigType
 
 from . import layers
@@ -34,6 +36,7 @@ LINK_LAYER_SCHEMAS = stack_schema()
 __all__ = [
     "CONF_MAX_PACKET_SIZE",
     "CONF_PACKET_INTERFACE",
+    "interface_layers",
     "PacketInterface",
     "new_packet_interface",
     "packet_interface_ns",
@@ -59,6 +62,14 @@ def _validate_layers(config: ConfigType) -> ConfigType:
     return config
 
 
+def interface_layers(interface_id: ID) -> ConfigType | None:
+    """During final validation, the configuration holding the layers of the packet interface with this id."""
+    for interface in fv.full_config.get().get(DOMAIN, []):
+        if interface[CONF_ID].id == interface_id.id:
+            return interface
+    return None
+
+
 def packet_interface_schema(
     class_: MockObjClass,
     default_max_packet_size: int,

esphome/core/defines.h

@@ -76,6 +76,8 @@
 #define USE_DISPLAY
 #define DISPLAY_PAGE_CHANGE_CALLBACK_COUNT 1
 #define DUCO_DEVICE_COUNT 4
+#define ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT 1
+#define ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT 1
 #define USE_EMONTX_VERSION
 #define USE_EMONTX_VERSION_TEXT_SENSOR
 #define USE_ENTITY_DEVICE_CLASS

tests/components/entity_transport/__init__.py

@@ -0,0 +1,19 @@
+import esphome.codegen as cg
+from esphome.core import CORE
+from esphome.types import ConfigType
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    async def to_code_testing(config: ConfigType) -> None:
+        cg.add_define("ENTITY_TRANSPORT_REMOTE_BINARY_SENSOR_COUNT", 2)
+        cg.add_define("ENTITY_TRANSPORT_REMOTE_SENSOR_COUNT", 2)
+        cg.add_define("USE_PACKET_INTERFACE_BASIC_SESSION")
+        # No entity platforms are configured, so count the test's own entities to emit USE_SENSOR and friends
+        CORE.platform_counts["sensor"] += 2
+        CORE.platform_counts["binary_sensor"] += 2
+
+    manifest.to_code = to_code_testing
+    # A multi_conf component gets an empty list in the test build, so to_code would not run
+    manifest.multi_conf = False
+    manifest.dependencies = manifest.dependencies + ["sensor", "binary_sensor"]

tests/components/entity_transport/common.yaml

@@ -0,0 +1,33 @@
+packet_interface:
+  - platform: uart
+    id: entity_transport_link
+    reliability:
+
+sensor:
+  - platform: template
+    id: entity_transport_sensor
+    lambda: return 1.0;
+  - platform: entity_transport
+    provider: remote-device
+    remote_id: entity_transport_sensor
+    name: Remote sensor
+
+binary_sensor:
+  - platform: template
+    id: entity_transport_binary_sensor
+    lambda: return true;
+  - platform: entity_transport
+    id: remote_binary_sensor
+    provider: remote-device
+    remote_id: door
+    name: Remote binary sensor
+
+entity_transport:
+  - packet_interface: entity_transport_link
+    update_interval: 60s
+    timeout: 3min
+    sensors:
+      - entity_transport_sensor
+    binary_sensors:
+      - id: entity_transport_binary_sensor
+        broadcast_id: door

tests/components/entity_transport/entity_transport.cpp

@@ -0,0 +1,552 @@
+#include <gtest/gtest.h>
+#include <cmath>
+#include <cstdint>
+#include <cstring>
+#include <initializer_list>
+#include <memory>
+#include <vector>
+#include "esphome/components/entity_transport/entity_transport.h"
+#include "esphome/core/application.h"
+
+namespace esphome::entity_transport::testing {
+
+// Copies a packet into a vector, for comparing in tests
+static std::vector<uint8_t> to_vector(const esphome::packet_interface::PacketBuffer &data) {
+  std::vector<uint8_t> result(data.size());
+  data.copy_to(result.data(), result.size());
+  return result;
+}
+
+using packet_interface::PacketBuffer;
+using packet_interface::PacketMetaData;
+using Packet = std::vector<uint8_t>;
+
+static constexpr const char *DEVICE_NAME = "this-device";
+static constexpr uint8_t PACKET_DICTIONARY = 0x11;
+static constexpr uint8_t PACKET_DATA = 0x12;
+static constexpr uint8_t PACKET_REQUEST = 0x13;
+
+// A packet interface that records what is sent, can refuse to send, and can act as a session layer
+class MockInterface : public packet_interface::PacketInterface {
+ public:
+  explicit MockInterface(size_t max_packet_size = 64) : packet_interface::PacketInterface(max_packet_size) {}
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &) override {
+    if (!this->accept || this->budget == 0)
+      return false;
+    this->budget--;
+    this->sent.push_back(to_vector(data));
+    return true;
+  }
+  void inject(const Packet &packet) { this->on_receive_from_interface_(PacketBuffer(packet), {}); }
+  void set_capabilities(uint8_t capabilities) { this->capabilities_ = capabilities; }
+  void report_session(bool connected) { this->session_callback_.call(connected); }
+  // Deliver everything sent so far to another interface
+  void deliver_to(MockInterface &other) {
+    auto packets = std::move(this->sent);
+    this->sent.clear();
+    for (auto &packet : packets)
+      other.inject(packet);
+  }
+
+  std::vector<Packet> sent;
+  bool accept{true};
+  size_t budget{SIZE_MAX};  // packets accepted before refusing, like a full queue
+};
+
+class TestTransport : public EntityTransport {
+ public:
+  using EntityTransport::EntityTransport;
+  uint32_t get_dictionary_id() const { return this->dictionary_id_; }
+  uint32_t get_update_interval() const { return this->update_interval_ms_; }
+  using EntityTransport::clear_remote_states_;
+};
+
+static Packet header(uint8_t packet_type, uint32_t id) {
+  return {packet_type, static_cast<uint8_t>(id), static_cast<uint8_t>(id >> 8), static_cast<uint8_t>(id >> 16),
+          static_cast<uint8_t>(id >> 24)};
+}
+
+// The start of a dictionary packet from this device's name: the header, the part byte, then the name
+static Packet dictionary_start(uint32_t id, uint8_t part = 0x80) {
+  Packet packet = header(PACKET_DICTIONARY, id);
+  packet.push_back(part);
+  packet.push_back(strlen(DEVICE_NAME));
+  packet.insert(packet.end(), DEVICE_NAME, DEVICE_NAME + strlen(DEVICE_NAME));
+  return packet;
+}
+
+class EntityTransportTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    App.~Application();
+    new (&App) Application();
+    static char name[] = "this-device";
+    static char friendly_name[] = "";
+    App.pre_setup(name, sizeof(name) - 1, friendly_name, 0);
+  }
+  void TearDown() override {
+    App.~Application();
+    new (&App) Application();  // later tests and the test listener use App
+  }
+
+  // The sender, with a sensor and a binary sensor
+  std::unique_ptr<TestTransport> make_sender_(MockInterface *link = nullptr) {
+    auto transport = std::make_unique<TestTransport>(link != nullptr ? link : &this->link_a_);
+    transport->set_entity_count(2);
+    transport->add_sensor("temp", &this->temp_);
+    transport->add_binary_sensor("door", &this->door_);
+    transport->setup();
+    return transport;
+  }
+  // The receiver, showing the sender's entities; both run on this device, so they share its name
+  std::unique_ptr<TestTransport> make_receiver_(const char *provider = DEVICE_NAME, uint32_t timeout_ms = 0) {
+    auto transport = std::make_unique<TestTransport>(&this->link_b_);
+    transport->set_remote_timeout(timeout_ms);
+    transport->add_remote_sensor(provider, "temp", &this->remote_temp_);
+    transport->add_remote_binary_sensor(provider, "door", &this->remote_door_);
+    transport->setup();
+    return transport;
+  }
+
+  MockInterface link_a_, link_b_;
+  sensor::Sensor temp_, remote_temp_;
+  binary_sensor::BinarySensor door_, remote_door_;
+};
+
+TEST_F(EntityTransportTest, SendsDictionaryThenStates) {
+  this->temp_.publish_state(21.5f);
+  this->door_.publish_state(true);
+  auto sender = this->make_sender_();
+  sender->loop();
+  uint32_t id = sender->get_dictionary_id();
+  Packet dictionary = dictionary_start(id);
+  dictionary.insert(dictionary.end(), {1, 0, 1, 4, 't', 'e', 'm', 'p', 2, 0, 1, 4, 'd', 'o', 'o', 'r'});
+  Packet data = header(PACKET_DATA, id);
+  // Tags: sensor 0 with a 4 byte state, then binary sensor 0 with its state in the top bit
+  data.insert(data.end(), {0x00, 0x21, 0x00, 0x00, 0xAC, 0x41, 0x00, 0x82});
+  ASSERT_EQ(this->link_a_.sent.size(), 2);
+  EXPECT_EQ(this->link_a_.sent[0], dictionary);
+  EXPECT_EQ(this->link_a_.sent[1], data);
+}
+
+TEST_F(EntityTransportTest, SendsOnlyChangedStates) {
+  auto sender = this->make_sender_();
+  sender->loop();
+  this->link_a_.sent.clear();
+  this->temp_.publish_state(3.0f);
+  sender->loop();
+  Packet data = header(PACKET_DATA, sender->get_dictionary_id());
+  data.insert(data.end(), {0x00, 0x21, 0x00, 0x00, 0x40, 0x40});
+  ASSERT_EQ(this->link_a_.sent.size(), 1);
+  EXPECT_EQ(this->link_a_.sent[0], data);
+}
+
+TEST_F(EntityTransportTest, StatesReachRemoteEntities) {
+  this->temp_.publish_state(12.25f);
+  this->door_.publish_state(true);
+  auto sender = this->make_sender_();
+  auto receiver = this->make_receiver_();
+  sender->loop();
+  this->link_a_.deliver_to(this->link_b_);
+  EXPECT_EQ(this->remote_temp_.get_state(), 12.25f);
+  EXPECT_TRUE(this->remote_door_.state);
+  this->door_.invalidate_state();
+  this->temp_.publish_state(NAN);
+  sender->loop();
+  ASSERT_EQ(this->link_a_.sent.size(), 1);
+  // Both records have no state, and so no payload
+  EXPECT_EQ(Packet(this->link_a_.sent[0].begin() + 5, this->link_a_.sent[0].end()), (Packet{0x00, 0x61, 0x00, 0x62}));
+  this->link_a_.deliver_to(this->link_b_);
+  EXPECT_FALSE(this->remote_door_.has_state());
+  EXPECT_TRUE(std::isnan(this->remote_temp_.get_state()));
+}
+
+TEST_F(EntityTransportTest, SplitsAcrossPackets) {
+  MockInterface link(5 + 2 + strlen(DEVICE_NAME) + 3 + 5);
+  this->temp_.publish_state(1.0f);
+  this->door_.publish_state(true);
+  auto sender = this->make_sender_(&link);
+  auto receiver = this->make_receiver_();
+  sender->loop();
+  EXPECT_EQ(link.sent.size(), 3);  // one dictionary packet for each entity, then one data packet
+  link.deliver_to(this->link_b_);
+  EXPECT_EQ(this->remote_temp_.get_state(), 1.0f);
+  EXPECT_TRUE(this->remote_door_.state);
+}
+
+TEST_F(EntityTransportTest, RefusedPacketSentAgain) {
+  this->link_a_.accept = false;
+  auto sender = this->make_sender_();
+  sender->loop();
+  EXPECT_TRUE(this->link_a_.sent.empty());
+  this->link_a_.accept = true;
+  sender->loop();
+  EXPECT_EQ(this->link_a_.sent.size(), 2);
+}
+
+TEST_F(EntityTransportTest, LongRefusalWarns) {
+  this->link_a_.accept = false;
+  auto sender = this->make_sender_();
+  {
+    LoopBlockingGuard guard{nullptr, nullptr, 1000};
+  }
+  sender->loop();
+  {
+    LoopBlockingGuard guard{nullptr, nullptr, 60999};
+  }
+  sender->loop();
+  EXPECT_FALSE(sender->status_has_warning());
+  {
+    LoopBlockingGuard guard{nullptr, nullptr, 61000};
+  }
+  sender->loop();
+  EXPECT_TRUE(sender->status_has_warning());
+  this->link_a_.accept = true;
+  sender->loop();
+  EXPECT_FALSE(sender->status_has_warning());
+  EXPECT_EQ(this->link_a_.sent.size(), 2);
+}
+
+TEST_F(EntityTransportTest, UnknownDictionaryRequested) {
+  this->temp_.publish_state(5.0f);
+  auto sender = this->make_sender_();
+  auto receiver = this->make_receiver_();
+  sender->loop();
+  this->link_a_.sent.erase(this->link_a_.sent.begin());  // the dictionary is lost
+  this->link_a_.deliver_to(this->link_b_);
+  EXPECT_FALSE(this->remote_temp_.has_state());
+  ASSERT_EQ(this->link_b_.sent.size(), 1);
+  EXPECT_EQ(this->link_b_.sent[0], header(PACKET_REQUEST, sender->get_dictionary_id()));
+  this->link_b_.deliver_to(this->link_a_);
+  sender->loop();
+  this->link_a_.deliver_to(this->link_b_);
+  EXPECT_EQ(this->remote_temp_.get_state(), 5.0f);
+}
+
+TEST_F(EntityTransportTest, ChangedDictionaryRequested) {
+  this->temp_.publish_state(5.0f);
+  auto sender = this->make_sender_();
+  auto receiver = this->make_receiver_();
+  sender->loop();
+  this->link_a_.deliver_to(this->link_b_);
+  EXPECT_EQ(this->remote_temp_.get_state(), 5.0f);
+  // The provider restarts with other entities, and its new dictionary is lost
+  sensor::Sensor temp;
+  MockInterface link;
+  TestTransport changed(&link);
+  changed.set_entity_count(1);
+  changed.add_sensor("temp", &temp);
+  changed.setup();
+  temp.publish_state(6.0f);
+  changed.loop();
+  link.sent.erase(link.sent.begin());
+  link.deliver_to(this->link_b_);
+  ASSERT_EQ(this->link_b_.sent.size(), 1);
+  EXPECT_EQ(this->link_b_.sent[0], header(PACKET_REQUEST, changed.get_dictionary_id()));
+  this->link_b_.deliver_to(link);
+  changed.loop();
+  link.deliver_to(this->link_b_);
+  EXPECT_EQ(this->remote_temp_.get_state(), 6.0f);
+}
+
+TEST_F(EntityTransportTest, MalformedDataNotRequested) {
+  auto receiver = this->make_receiver_();
+  // A data packet under an unknown dictionary, cut off inside its first record
+  Packet data = header(PACKET_DATA, 0x12345678);
+  data.insert(data.end(), {0x00, 0x21, 0x00});
+  this->link_b_.inject(data);
+  EXPECT_TRUE(this->link_b_.sent.empty());
+}
+
+TEST_F(EntityTransportTest, EntityLeftOutOfNewDictionaryCleared) {
+  this->temp_.publish_state(5.0f);
+  this->door_.publish_state(true);
+  auto sender = this->make_sender_();
+  auto receiver = this->make_receiver_();
+  sender->loop();
+  this->link_a_.deliver_to(this->link_b_);
+  EXPECT_TRUE(this->remote_door_.state);
+  // The provider restarts without the door
+  sensor::Sensor temp;
+  MockInterface link;
+  TestTransport changed(&link);
+  changed.set_entity_count(1);
+  changed.add_sensor("temp", &temp);
+  changed.setup();
+  temp.publish_state(6.0f);
+  changed.loop();
+  link.deliver_to(this->link_b_);
+  EXPECT_EQ(this->remote_temp_.get_state(), 6.0f);
+  EXPECT_FALSE(this->remote_door_.has_state());
+}
+
+TEST_F(EntityTransportTest, EntityLeftOutOfSeenDictionaryCleared) {
+  // The provider names only the sensor, then both entities, then goes back to naming only the sensor
+  sensor::Sensor temp;
+  MockInterface link;
+  auto make_temp_only = [&] {
+    auto transport = std::make_unique<TestTransport>(&link);
+    transport->set_entity_count(1);
+    transport->add_sensor("temp", &temp);
+    transport->setup();
+    return transport;
+  };
+  temp.publish_state(4.0f);
+  auto receiver = this->make_receiver_();
+  make_temp_only()->loop();
+  link.deliver_to(this->link_b_);
+  this->door_.publish_state(true);
+  this->make_sender_()->loop();
+  this->link_a_.deliver_to(this->link_b_);
+  EXPECT_TRUE(this->remote_door_.state);
+  make_temp_only()->loop();
+  link.deliver_to(this->link_b_);
+  EXPECT_EQ(this->remote_temp_.get_state(), 4.0f);
+  EXPECT_FALSE(this->remote_door_.has_state());
+}
+
+TEST_F(EntityTransportTest, LostDictionaryPartRequested) {
+  // One entity in each dictionary packet
+  MockInterface link(5 + 2 + strlen(DEVICE_NAME) + 3 + 5);
+  this->temp_.publish_state(2.0f);
+  this->door_.publish_state(true);
+  auto sender = this->make_sender_(&link);
+  auto receiver = this->make_receiver_();
+  sender->loop();
+  ASSERT_EQ(link.sent.size(), 3);
+  EXPECT_EQ(link.sent[0][5], 0x00);
+  EXPECT_EQ(link.sent[1][5], 0x81);
+  link.sent.erase(link.sent.begin() + 1);  // the part naming the binary sensor is lost
+  link.deliver_to(this->link_b_);
+  EXPECT_EQ(this->remote_temp_.get_state(), 2.0f);
+  EXPECT_FALSE(this->remote_door_.has_state());
+  // The sensor is known, but the dictionary is not complete, so it is asked for
+  ASSERT_EQ(this->link_b_.sent.size(), 1);
+  EXPECT_EQ(this->link_b_.sent[0], header(PACKET_REQUEST, sender->get_dictionary_id()));
+  this->link_b_.deliver_to(link);
+  sender->loop();
+  link.deliver_to(this->link_b_);
+  EXPECT_TRUE(this->remote_door_.state);
+  // Now complete, it is not asked for again
+  {
+    LoopBlockingGuard guard{nullptr, nullptr, 10000};
+  }
+  this->door_.publish_state(false);
+  sender->loop();
+  link.deliver_to(this->link_b_);
+  EXPECT_FALSE(this->remote_door_.state);
+  EXPECT_TRUE(this->link_b_.sent.empty());
+}
+
+TEST_F(EntityTransportTest, RefusedRequestSentAgain) {
+  auto receiver = this->make_receiver_();
+  Packet data = header(PACKET_DATA, 0x12345678);
+  this->link_b_.accept = false;
+  this->link_b_.inject(data);
+  this->link_b_.accept = true;
+  this->link_b_.inject(data);
+  ASSERT_EQ(this->link_b_.sent.size(), 1);
+  EXPECT_EQ(this->link_b_.sent[0], header(PACKET_REQUEST, 0x12345678));
+}
+
+TEST_F(EntityTransportTest, RequestsLimited) {
+  auto receiver = this->make_receiver_();
+  Packet data = header(PACKET_DATA, 0x12345678);
+  data.insert(data.end(), {0x00, 0x61});  // sensor 0 with no state
+  this->link_b_.inject(data);
+  this->link_b_.inject(data);
+  EXPECT_EQ(this->link_b_.sent.size(), 1);
+}
+
+TEST_F(EntityTransportTest, OtherProviderAndOwnPacketsIgnored) {
+  this->temp_.publish_state(5.0f);
+  auto sender = this->make_sender_();
+  auto receiver = this->make_receiver_("other-device");
+  sender->loop();
+  auto packets = this->link_a_.sent;
+  this->link_a_.deliver_to(this->link_b_);
+  EXPECT_FALSE(this->remote_temp_.has_state());
+  // The dictionary named none of the receiver's entities, so it is not asked for again
+  EXPECT_TRUE(this->link_b_.sent.empty());
+  for (auto &packet : packets)
+    this->link_a_.inject(packet);
+  this->link_a_.inject(header(PACKET_REQUEST, 0x12345678));
+  sender->loop();
+  EXPECT_TRUE(this->link_a_.sent.empty());
+}
+
+TEST_F(EntityTransportTest, MalformedPacketsIgnored) {
+  auto sender = this->make_sender_();
+  auto receiver = this->make_receiver_();
+  sender->loop();
+  Packet dictionary = this->link_a_.sent[0];
+  Packet data = this->link_a_.sent[1];
+  this->link_b_.inject(Packet(dictionary.begin(), dictionary.end() - 1));
+  this->link_b_.inject(dictionary);
+  this->link_b_.inject(Packet(data.begin(), data.end() - 1));
+  this->link_b_.inject({PACKET_DATA, 1, 2});
+  this->link_b_.inject({0x21, 0, 0, 0, 0});
+  EXPECT_FALSE(this->remote_temp_.has_state());
+}
+
+// Data packets from the sender's dictionary, holding the given records
+class EntityTransportRecordTest : public EntityTransportTest {
+ protected:
+  void SetUp() override {
+    EntityTransportTest::SetUp();
+    this->sender_ = this->make_sender_();
+    this->receiver_ = this->make_receiver_();
+    this->sender_->loop();
+    this->link_a_.sent.resize(1);  // the dictionary only
+    this->link_a_.deliver_to(this->link_b_);
+  }
+  void inject_data_(std::initializer_list<uint8_t> records) {
+    Packet data = header(PACKET_DATA, this->sender_->get_dictionary_id());
+    data.insert(data.end(), records);
+    this->link_b_.inject(data);
+  }
+
+  std::unique_ptr<TestTransport> sender_, receiver_;
+};
+
+TEST_F(EntityTransportRecordTest, WholePacketCheckedBeforeUse) {
+  // A good record for each entity, then a record cut short
+  this->inject_data_({0x00, 0x21, 0x00, 0x00, 0x10, 0x41, 0x00, 0x82, 0x00, 0x21, 0x00});
+  // A good record, then a sensor record without its state, then a binary sensor record with a payload
+  this->inject_data_({0x00, 0x21, 0x00, 0x00, 0x10, 0x41, 0x00, 0x01});
+  this->inject_data_({0x00, 0x82, 0x00, 0x22, 0x00, 0x00, 0x00, 0x00});
+  // The value bit is only for records without a payload, and type 0 is not used
+  this->inject_data_({0x00, 0xA1, 0x00, 0x00, 0x10, 0x41});
+  this->inject_data_({0x00, 0x82, 0x00, 0x00});
+  // A length byte past the end
+  this->inject_data_({0x00, 0x82, 0x00, 0x45, 0x03, 0x01});
+  EXPECT_FALSE(this->remote_temp_.has_state());
+  EXPECT_FALSE(this->remote_door_.has_state());
+  this->inject_data_({0x00, 0x21, 0x00, 0x00, 0x10, 0x41, 0x00, 0x82});
+  EXPECT_EQ(this->remote_temp_.get_state(), 9.0f);
+  EXPECT_TRUE(this->remote_door_.state);
+}
+
+TEST_F(EntityTransportRecordTest, UnknownTypesSkipped) {
+  // Type 5 with 3 bytes after a length byte, type 6 with a 4 byte payload, type 7 without a state, then the sensor
+  this->inject_data_({0x00, 0x45, 0x03, 0x01, 0x02, 0x03, 0x01, 0x26, 0x00, 0x00,
+                      0x00, 0x00, 0x02, 0x67, 0x00, 0x21, 0x00, 0x00, 0x10, 0x41});
+  EXPECT_EQ(this->remote_temp_.get_state(), 9.0f);
+}
+
+TEST_F(EntityTransportTest, PartlyValidDictionaryNotLearnt) {
+  auto receiver = this->make_receiver_();
+  static constexpr uint32_t OTHER_ID = 0x0BADF00D;
+  Packet dictionary = dictionary_start(OTHER_ID);
+  // A good section naming the sensor, then sections cut short, of type 0, and of more than 255 numbers
+  dictionary.insert(dictionary.end(), {1, 0, 1, 4, 't', 'e', 'm', 'p'});
+  for (Packet bad : {Packet{2, 0, 2, 4, 'd', 'o', 'o', 'r'}, Packet{0, 0, 1, 1, 'x'}, Packet{2, 255, 2, 1, 'a', 1, 'b'},
+                     Packet{2, 0, 1}}) {
+    Packet packet = dictionary;
+    packet.insert(packet.end(), bad.begin(), bad.end());
+    this->link_b_.inject(packet);
+  }
+  Packet data = header(PACKET_DATA, OTHER_ID);
+  data.insert(data.end(), {0x00, 0x21, 0x00, 0x00, 0x80, 0x3F});
+  this->link_b_.inject(data);
+  EXPECT_FALSE(this->remote_temp_.has_state());
+  this->link_b_.inject(dictionary);
+  this->link_b_.inject(data);
+  EXPECT_EQ(this->remote_temp_.get_state(), 1.0f);
+}
+
+TEST_F(EntityTransportTest, UpdateIntervalFollowsCapabilities) {
+  EXPECT_EQ(this->make_sender_()->get_update_interval(), 15000);
+  this->link_a_.set_capabilities(packet_interface::CAPABILITY_SESSION);
+  EXPECT_EQ(this->make_sender_()->get_update_interval(), 15000);
+  this->link_a_.set_capabilities(packet_interface::CAPABILITY_SESSION | packet_interface::CAPABILITY_RELIABILITY);
+  EXPECT_EQ(this->make_sender_()->get_update_interval(), SCHEDULER_DONT_RUN);
+  auto transport = std::make_unique<TestTransport>(&this->link_a_);
+  transport->set_update_interval(5000);
+  transport->setup();
+  EXPECT_EQ(transport->get_update_interval(), 5000);
+}
+
+TEST_F(EntityTransportTest, SessionConnectSendsAllAndLossClearsRemote) {
+  this->link_a_.set_capabilities(packet_interface::CAPABILITY_SESSION);
+  this->link_b_.set_capabilities(packet_interface::CAPABILITY_SESSION);
+  this->temp_.publish_state(7.0f);
+  auto sender = this->make_sender_();
+  auto receiver = this->make_receiver_();
+  sender->loop();
+  this->link_a_.deliver_to(this->link_b_);
+  this->link_a_.report_session(true);
+  sender->loop();
+  EXPECT_EQ(this->link_a_.sent.size(), 2);
+  this->link_b_.report_session(false);
+  EXPECT_TRUE(std::isnan(this->remote_temp_.get_state()));
+  EXPECT_FALSE(this->remote_door_.has_state());
+}
+
+TEST_F(EntityTransportTest, RefusedDictionaryPartResumes) {
+  // One entity in each dictionary packet
+  MockInterface link(5 + 2 + strlen(DEVICE_NAME) + 3 + 5);
+  this->temp_.publish_state(4.0f);
+  this->door_.publish_state(true);
+  auto sender = this->make_sender_(&link);
+  auto receiver = this->make_receiver_();
+  link.budget = 1;
+  sender->loop();
+  ASSERT_EQ(link.sent.size(), 1);
+  link.budget = SIZE_MAX;
+  sender->loop();
+  int dictionaries = 0;
+  for (auto &packet : link.sent)
+    dictionaries += packet[0] == PACKET_DICTIONARY;
+  EXPECT_EQ(dictionaries, 2);  // the first part was not sent again
+  link.deliver_to(this->link_b_);
+  EXPECT_EQ(this->remote_temp_.get_state(), 4.0f);
+  EXPECT_TRUE(this->remote_door_.state);
+}
+
+TEST_F(EntityTransportTest, RemoteStatesTimeOut) {
+  this->temp_.publish_state(6.0f);
+  this->door_.publish_state(true);
+  auto sender = this->make_sender_();
+  auto receiver = this->make_receiver_(DEVICE_NAME, 5000);
+  {
+    LoopBlockingGuard guard{nullptr, nullptr, 1000};
+  }
+  sender->loop();
+  this->link_a_.deliver_to(this->link_b_);
+  {
+    LoopBlockingGuard guard{nullptr, nullptr, 6000};
+  }
+  receiver->clear_remote_states_(true);
+  EXPECT_EQ(this->remote_temp_.get_state(), 6.0f);
+  EXPECT_TRUE(this->remote_door_.has_state());
+  {
+    LoopBlockingGuard guard{nullptr, nullptr, 6001};
+  }
+  receiver->clear_remote_states_(true);
+  EXPECT_TRUE(std::isnan(this->remote_temp_.get_state()));
+  EXPECT_FALSE(this->remote_door_.has_state());
+}
+
+TEST_F(EntityTransportTest, IdMatchingOnlyUpToNulNotLearnt) {
+  auto receiver = this->make_receiver_();
+  static constexpr uint32_t OTHER_ID = 0x0BADF00D;
+  Packet dictionary = dictionary_start(OTHER_ID);
+  // The id "temp" followed by a NUL and more bytes, nine in all
+  dictionary.insert(dictionary.end(), {1, 0, 1, 9, 't', 'e', 'm', 'p', 0, 'a', 'b', 'c', 'd'});
+  this->link_b_.inject(dictionary);
+  Packet data = header(PACKET_DATA, OTHER_ID);
+  data.insert(data.end(), {0x00, 0x21, 0x00, 0x00, 0x80, 0x3F});
+  this->link_b_.inject(data);
+  EXPECT_FALSE(this->remote_temp_.has_state());
+}
+
+TEST_F(EntityTransportTest, PacketTooSmallForIdsFailsSetup) {
+  // One byte short of a dictionary packet holding the device name and a four character id
+  MockInterface small_link(5 + 2 + strlen(DEVICE_NAME) + 3 + 1 + 4 - 1);
+  auto sender = this->make_sender_(&small_link);
+  EXPECT_TRUE(sender->is_failed());
+  EXPECT_FALSE(this->make_sender_()->is_failed());
+}
+
+}  // namespace esphome::entity_transport::testing

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

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

tests/unit_tests/test_entity_transport.py

@@ -0,0 +1,89 @@
+"""Unit tests for the entity_transport component."""
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.const import CONF_PROVIDER, CONF_REMOTE_ID
+from esphome.components.entity_transport import (
+    CONFIG_SCHEMA,
+    FINAL_VALIDATE_SCHEMA,
+    remote_entity_schema,
+)
+from esphome.components.sensor import sensor_schema
+from esphome.const import (
+    CONF_ID,
+    CONF_NAME,
+    CONF_TIMEOUT,
+    CONF_UPDATE_INTERVAL,
+    SCHEDULER_DONT_RUN,
+)
+from esphome.core import ID
+import esphome.final_validate as fv
+
+
+def test_remote_id_required_without_manual_id(setup_core) -> None:
+    """A remote entity needs remote_id unless its own id is given."""
+    schema = remote_entity_schema(sensor_schema())
+    with pytest.raises(cv.Invalid, match="remote_id"):
+        schema({CONF_NAME: "Temp", CONF_PROVIDER: "remote-device"})
+    assert schema(
+        {CONF_NAME: "Temp 2", CONF_PROVIDER: "remote-device", CONF_REMOTE_ID: "temp"}
+    )
+    config = schema(
+        {CONF_PROVIDER: "remote-device", CONF_ID: ID("temp", is_manual=True)}
+    )
+    assert config[CONF_ID].id == "temp"
+
+
+def test_update_interval_never(setup_core) -> None:
+    """update_interval accepts never, and is left out when not given."""
+    base = {"packet_interface": "link"}
+    assert CONF_UPDATE_INTERVAL not in CONFIG_SCHEMA(base)
+    config = CONFIG_SCHEMA({**base, CONF_UPDATE_INTERVAL: "never"})
+    assert config[CONF_UPDATE_INTERVAL].total_milliseconds == SCHEDULER_DONT_RUN
+    with pytest.raises(cv.Invalid):
+        CONFIG_SCHEMA({**base, CONF_UPDATE_INTERVAL: "500ms"})
+
+
+def test_timeout(setup_core) -> None:
+    """The timeout is optional, and at least a second."""
+    base = {"packet_interface": "link"}
+    assert CONF_TIMEOUT not in CONFIG_SCHEMA(base)
+    assert (
+        CONFIG_SCHEMA({**base, CONF_TIMEOUT: "1min"})[CONF_TIMEOUT].total_seconds == 60
+    )
+    with pytest.raises(cv.Invalid):
+        CONFIG_SCHEMA({**base, CONF_TIMEOUT: "500ms"})
+
+
+@pytest.mark.parametrize(
+    ("layers", "allowed"),
+    [({}, True), ({"reliability": None}, True), ({"session": {}}, False)],
+)
+def test_timeout_not_with_session(setup_core, layers: dict, allowed: bool) -> None:
+    """With a session layer, unchanged states are never sent again, so a timeout would clear them on a working link."""
+    config = CONFIG_SCHEMA({"packet_interface": "link", CONF_TIMEOUT: "1min"})
+    token = fv.full_config.set(
+        {"packet_interface": [{CONF_ID: ID("link", is_manual=True), **layers}]}
+    )
+    try:
+        if allowed:
+            FINAL_VALIDATE_SCHEMA(config)
+        else:
+            with pytest.raises(cv.Invalid, match="session") as error:
+                FINAL_VALIDATE_SCHEMA(config)
+            assert error.value.path == [CONF_TIMEOUT]
+        # Without a timeout, any layers are fine
+        FINAL_VALIDATE_SCHEMA(CONFIG_SCHEMA({"packet_interface": "link"}))
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_advertised_ids_unique(setup_core) -> None:
+    """Two entities of one type cannot be sent with the same id."""
+    base = {"packet_interface": "link"}
+    assert CONFIG_SCHEMA({**base, "sensors": ["a", {"id": "b", "broadcast_id": "c"}]})
+    with pytest.raises(cv.Invalid, match="'a' is sent for two entities"):
+        CONFIG_SCHEMA({**base, "sensors": ["a", {"id": "b", "broadcast_id": "a"}]})
+    # Sensors and binary sensors are numbered separately, so may share an id
+    assert CONFIG_SCHEMA({**base, "sensors": ["a"], "binary_sensors": ["a"]})