[packet_interface] Add addressing and fragmentation layers

The addressing layer lets several devices share a channel such as an
RS485 bus or a radio, with 16-bit addresses. A link can carry several
peers, each with its own stack of layers and its own id. Layers that work
between two devices only are rejected with broadcast addressing. The
fragmentation layer splits packets larger than the layers below allow,
and joins them again.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

esphome/components/entity_transport/__init__.py

@@ -7,7 +7,11 @@ 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 import (
+    CONF_PACKET_INTERFACE,
+    final_validate_interface,
+    interface_layers,
+)
 from esphome.components.packet_interface.layers import CONF_SESSION
 from esphome.components.sensor import Sensor
 import esphome.config_validation as cv
@@ -108,7 +112,9 @@ def _final_validate(config: ConfigType) -> ConfigType:
     return config
 
 
-FINAL_VALIDATE_SCHEMA = _final_validate
+FINAL_VALIDATE_SCHEMA = cv.All(
+    final_validate_interface(CONF_PACKET_INTERFACE), _final_validate
+)
 
 
 def _validate_remote_id(config: ConfigType) -> ConfigType:

esphome/components/packet_interface/__init__.py

@@ -1,3 +1,4 @@
+from collections.abc import Callable
 import importlib
 import pkgutil
 from typing import Any
@@ -13,9 +14,16 @@ from esphome.types import ConfigType
 
 from . import layers
 from .layers import (
+    ADDRESSING,
+    BROADCAST_ADDRESS,
+    CONF_ADDRESSING,
+    CONF_DESTINATION,
+    CONF_PEERS,
+    STACK_ORDER,
     PacketInterface,
     build_stack,
     check_session,
+    check_single_peer,
     packet_interface_ns,
     stack_auto_load,
     stack_schema,
@@ -31,13 +39,17 @@ CONF_PACKET_INTERFACE = "packet_interface"
 for module_info in pkgutil.iter_modules(layers.__path__):
     importlib.import_module(f".layers.{module_info.name}", package=__package__)
 
-LINK_LAYER_SCHEMAS = stack_schema()
+LINK_LAYER_SCHEMAS = {
+    cv.Optional(CONF_ADDRESSING): ADDRESSING.schema(),
+    **stack_schema(),
+}
 
 __all__ = [
     "CONF_MAX_PACKET_SIZE",
     "CONF_PACKET_INTERFACE",
     "interface_layers",
     "PacketInterface",
+    "final_validate_interface",
     "new_packet_interface",
     "packet_interface_ns",
     "packet_interface_schema",
@@ -45,7 +57,7 @@ __all__ = [
 
 
 def AUTO_LOAD(config: list[ConfigType] | None = None) -> list[str]:
-    """Load only what the configured layers use.
+    """Load only what the configured layers use, on the link and in each addressing peer.
 
     Without a config, as tooling that lists dependencies calls it, everything a layer could load.
     """
@@ -54,22 +66,61 @@ def AUTO_LOAD(config: list[ConfigType] | None = None) -> list[str]:
     loads: set[str] = set()
     for interface in config:
         loads |= stack_auto_load(interface)
+        for peer in interface.get(CONF_ADDRESSING, {}).get(CONF_PEERS, []):
+            loads |= stack_auto_load(peer)
     return sorted(loads)
 
 
 def _validate_layers(config: ConfigType) -> ConfigType:
+    """With peers, the layers go in each peer; otherwise check them against the one destination."""
     check_session(config)
+    if (addressing := config.get(CONF_ADDRESSING)) is None:
+        return config
+    if CONF_PEERS in addressing:
+        for kind in STACK_ORDER:
+            if kind.key in config:
+                raise cv.Invalid(
+                    f"With '{CONF_PEERS}', '{kind.key}' goes in each peer",
+                    path=[kind.key],
+                )
+        return config
+    addressing.setdefault(CONF_DESTINATION, BROADCAST_ADDRESS)
+    check_single_peer(config, addressing[CONF_DESTINATION])
     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."""
+    """During final validation, the configuration holding the layers of the packet interface with this id.
+
+    The id is a link's, or an addressing peer's.
+    """
     for interface in fv.full_config.get().get(DOMAIN, []):
         if interface[CONF_ID].id == interface_id.id:
             return interface
+        for peer in interface.get(CONF_ADDRESSING, {}).get(CONF_PEERS, []):
+            if peer[CONF_ID].id == interface_id.id:
+                return peer
     return None
 
 
+def final_validate_interface(key: str) -> Callable[[ConfigType], ConfigType]:
+    """Final validation for an application: a link with several peers must be used through one of the peers."""
+
+    def validator(config: ConfigType) -> ConfigType:
+        interface_id: ID = config[key]
+        for interface in fv.full_config.get().get(DOMAIN, []):
+            if interface[CONF_ID].id == interface_id.id and CONF_PEERS in interface.get(
+                CONF_ADDRESSING, {}
+            ):
+                raise cv.Invalid(
+                    f"'{interface_id.id}' has several peers, so use the id of one of them",
+                    path=[key],
+                )
+        return config
+
+    return validator
+
+
 def packet_interface_schema(
     class_: MockObjClass,
     default_max_packet_size: int,
@@ -97,10 +148,17 @@ async def new_packet_interface(config: ConfigType, *args: Any) -> MockObj:
     The link's constructor takes its maximum packet size, usually as a std::span over a buffer that code generation
     sized from the configuration, so the two cannot differ.
 
-    Returns the link layer; applications reach the top layer with PacketInterface::top().
+    Returns the link layer; applications reach the top layer with PacketInterface::top(). With several addressing
+    peers, applications use the peers' ids instead.
     """
     var = cg.new_Pvariable(config[CONF_ID], *args)
     cg.add(var.set_name(config[CONF_ID].id))
     await cg.register_component(var, config)
-    await build_stack(config, var)
+    if (addressing := config.get(CONF_ADDRESSING)) is None:
+        await build_stack(config, var)
+        return var
+    for port, layers_config in await ADDRESSING.get_type(addressing).build(
+        addressing, var
+    ):
+        await build_stack(config if layers_config is None else layers_config, port)
     return var

esphome/components/packet_interface/basic_addressing.cpp

@@ -0,0 +1,87 @@
+#include "basic_addressing.h"
+#ifdef USE_PACKET_INTERFACE_BASIC_ADDRESSING
+#include "esphome/core/log.h"
+
+namespace esphome::packet_interface {
+
+ESPHOME_LOG_TAG(TAG, "packet_interface.addressing");
+
+void BasicAddressing::setup() {
+  this->link_->add_packet_interface_listener(
+      [this](const PacketBuffer &data, const PacketMetaData &meta_data) { this->receive_(data, meta_data); });
+}
+
+void BasicAddressing::dump_config() {
+  ESP_LOGCONFIG(TAG,
+                "Basic Addressing on '%s':\n"
+                "  Address: 0x%04X",
+                this->link_->get_name(), this->address_);
+}
+
+bool BasicAddressing::send(const PacketBuffer &data, const PacketMetaData &meta_data, uint16_t destination) {
+  if (meta_data.destination_address != 0)
+    destination = meta_data.destination_address;
+  uint8_t header[HEADER_SIZE] = {static_cast<uint8_t>(destination), static_cast<uint8_t>(destination >> 8),
+                                 static_cast<uint8_t>(this->address_), static_cast<uint8_t>(this->address_ >> 8)};
+  PacketBuffer packet(header, HEADER_SIZE, &data, nullptr, 0);
+  return this->link_->send_to_interface(packet, meta_data);
+}
+
+void BasicAddressing::receive_(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  // Lower layers deliver packets in one piece; the payload is passed up without copying
+  if (!data.is_contiguous()) {
+    ESP_LOGW(TAG, "%s: Packet split across buffers is not supported", this->link_->get_name());
+    return;
+  }
+  auto packet = data.as_span();
+  if (packet.size() < HEADER_SIZE)
+    return;
+  uint16_t destination = encode_uint16(packet[1], packet[0]);
+  uint16_t source = encode_uint16(packet[3], packet[2]);
+  if (source == this->address_ || (destination != this->address_ && destination != BROADCAST_ADDRESS))
+    return;
+  // A device with a port of its own is heard only there; the broadcast port hears the rest
+  AddressingPort *target = nullptr;
+  for (auto *port : this->ports_) {
+    if (port->destination_ == source) {
+      target = port;
+      break;
+    }
+    if (port->destination_ == BROADCAST_ADDRESS)
+      target = port;
+  }
+  if (target == nullptr) {
+    ESP_LOGV(TAG, "%s: Packet from 0x%04X dropped", this->link_->get_name(), source);
+    return;
+  }
+  PacketMetaData received = meta_data;
+  received.source_address = source;
+  received.destination_address = destination;
+  target->on_receive_from_interface_(PacketBuffer(packet.data() + HEADER_SIZE, packet.size() - HEADER_SIZE), received);
+}
+
+AddressingPort::AddressingPort(BasicAddressing *addressing, uint16_t destination, bool only_port)
+    : addressing_(addressing), destination_(destination) {
+  if (only_port) {
+    this->stack_on_(addressing->get_link(), CAPABILITY_ADDRESSING);
+  } else {
+    this->inherit_from_(addressing->get_link(), CAPABILITY_ADDRESSING);
+  }
+  this->max_packet_size_ = addressing->get_link()->get_max_packet_size() - BasicAddressing::HEADER_SIZE;
+  addressing->add_port(this);
+}
+
+void AddressingPort::dump_config() {
+  ESP_LOGCONFIG(TAG,
+                "Addressing Port '%s':\n"
+                "  Max Packet Size: %zu",
+                this->name_, this->max_packet_size_);
+  if (this->destination_ == BasicAddressing::BROADCAST_ADDRESS) {
+    ESP_LOGCONFIG(TAG, "  Destination: broadcast");
+  } else {
+    ESP_LOGCONFIG(TAG, "  Destination: 0x%04X", this->destination_);
+  }
+}
+
+}  // namespace esphome::packet_interface
+#endif

esphome/components/packet_interface/basic_addressing.h

@@ -0,0 +1,68 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+#ifdef USE_PACKET_INTERFACE_BASIC_ADDRESSING
+#include "esphome/core/helpers.h"
+#include "packet_interface.h"
+
+namespace esphome::packet_interface {
+
+class AddressingPort;
+
+/**
+ * An addressing layer for channels shared by several devices, such as an RS485 bus or a LoRa radio.
+ *
+ * Each packet starts with a 16-bit destination address and a 16-bit source address, little endian.
+ * BROADCAST_ADDRESS sends to every device. A packet is dropped unless it is addressed to this device or broadcast, and
+ * a packet from this device's own address, such as an echo on a half-duplex bus, is dropped too.
+ *
+ * Above this layer are one or more ports, each the bottom of its own stack of layers. A port with a destination
+ * address is a point-to-point link to that device: it sends to it and receives only its packets, so the reliability
+ * and session layers and the rolling code, which work between two devices only, can be used above it. A port with
+ * the broadcast destination sends to every device, and receives packets from devices that have no port of their own.
+ */
+class BasicAddressing : public Component {
+ public:
+  static constexpr uint16_t BROADCAST_ADDRESS = 0xFFFF;
+  /// Destination and source addresses.
+  static constexpr size_t HEADER_SIZE = 4;
+
+  BasicAddressing(PacketInterface *link, uint16_t address) : link_(link), address_(address) {}
+
+  void setup() override;
+  void dump_config() override;
+
+  PacketInterface *get_link() const { return this->link_; }
+  void add_port(AddressingPort *port) { this->ports_.push_back(port); }
+  /// Sends \p data to \p destination, or to the destination in \p meta_data if set.
+  bool send(const PacketBuffer &data, const PacketMetaData &meta_data, uint16_t destination);
+
+ protected:
+  void receive_(const PacketBuffer &data, const PacketMetaData &meta_data);
+
+  PacketInterface *link_;
+  const uint16_t address_;
+  StaticVector<AddressingPort *, PACKET_INTERFACE_ADDRESSING_PORT_COUNT> ports_;
+};
+
+/// The bottom of the stack of layers for one destination, or for broadcast, on an addressed channel.
+class AddressingPort : public PacketInterface {
+ public:
+  /// With \p only_port, the port is the top of the link, so top() on the link reaches this port's stack.
+  AddressingPort(BasicAddressing *addressing, uint16_t destination, bool only_port);
+
+  void dump_config() override;
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) override {
+    return this->addressing_->send(data, meta_data, this->destination_);
+  }
+  uint16_t get_destination() const { return this->destination_; }
+
+ protected:
+  friend class BasicAddressing;
+
+  BasicAddressing *addressing_;
+  const uint16_t destination_;
+};
+
+}  // namespace esphome::packet_interface
+#endif

esphome/components/packet_interface/basic_fragmentation.cpp

@@ -0,0 +1,138 @@
+#include "basic_fragmentation.h"
+#ifdef USE_PACKET_INTERFACE_BASIC_FRAGMENTATION
+#include <algorithm>
+#include <cstring>
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
+
+namespace esphome::packet_interface {
+
+ESPHOME_LOG_TAG(TAG, "packet_interface.fragmentation");
+
+static constexpr uint8_t LAST_FRAGMENT = 0x80;
+static constexpr uint8_t INDEX_MASK = 0x7F;
+
+BasicFragmentation::BasicFragmentation(PacketInterface *lower, size_t max_packet_size) : lower_(lower) {
+  this->stack_on_(lower, CAPABILITY_FRAGMENTATION);
+  this->max_packet_size_ = max_packet_size;
+}
+
+size_t BasicFragmentation::fragments_for_(size_t size) const {
+  size_t fragment_size = this->fragment_size_();
+  return std::max<size_t>(1, (size + fragment_size - 1) / fragment_size);
+}
+
+void BasicFragmentation::setup() {
+  if (this->fragments_for_(this->max_packet_size_) > MAX_FRAGMENTS) {
+    this->mark_failed(LOG_STR("max_packet_size needs more than 128 fragments"));
+    return;
+  }
+  // Checked now, while the layer below holds nothing
+  this->queue_too_short_ = this->fragments_for_(this->max_packet_size_) > this->lower_->get_send_capacity();
+  this->tx_buffer_ = std::make_unique<uint8_t[]>(this->fragment_size_());
+  this->rx_buffer_ = std::make_unique<uint8_t[]>(this->max_packet_size_);
+  // Numbering from a random start makes it unlikely that a packet joins fragments from before a restart
+  this->tx_number_ = random_uint32();
+  this->lower_->add_packet_interface_listener(
+      [this](const PacketBuffer &data, const PacketMetaData &meta_data) { this->receive_(data, meta_data); });
+}
+
+void BasicFragmentation::dump_config() {
+  size_t fragments = this->fragments_for_(this->max_packet_size_);
+  ESP_LOGCONFIG(TAG,
+                "Basic Fragmentation on '%s':\n"
+                "  Max Packet Size: %zu\n"
+                "  Fragment Size: %zu\n"
+                "  Fragments in the Largest Packet: %zu",
+                this->name_, this->max_packet_size_, this->fragment_size_(), fragments);
+  if (this->queue_too_short_) {
+    ESP_LOGW(TAG, "  The reliability queue is too short for the largest packets, which cannot be sent");
+  }
+}
+
+bool BasicFragmentation::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  if (this->is_failed())
+    return false;
+  size_t size = data.size();
+  if (size > this->max_packet_size_) {
+    ESP_LOGW(TAG, "%s: Packet of %zu bytes is larger than %zu", this->name_, size, this->max_packet_size_);
+    return false;
+  }
+  size_t fragments = this->fragments_for_(size);
+  if (fragments > this->lower_->get_send_capacity()) {
+    ESP_LOGV(TAG, "%s: No room for %zu fragments", this->name_, fragments);
+    return false;
+  }
+  uint8_t header[HEADER_SIZE] = {this->tx_number_++, 0};
+  if (fragments == 1) {
+    header[1] = LAST_FRAGMENT;
+    return this->lower_->send_to_interface(PacketBuffer(header, HEADER_SIZE, &data, nullptr, 0), meta_data);
+  }
+  size_t fragment_size = this->fragment_size_();
+  for (size_t index = 0; index != fragments; index++) {
+    size_t length = data.copy_to(this->tx_buffer_.get(), fragment_size, index * fragment_size);
+    header[1] = index | (index + 1 == fragments ? LAST_FRAGMENT : 0);
+    PacketBuffer body(this->tx_buffer_.get(), length);
+    if (!this->lower_->send_to_interface(PacketBuffer(header, HEADER_SIZE, &body, nullptr, 0), meta_data)) {
+      // The peer drops the part already sent when the next packet starts
+      ESP_LOGV(TAG, "%s: Fragment %zu of %zu not accepted", this->name_, index, fragments);
+      return false;
+    }
+  }
+  return true;
+}
+
+void BasicFragmentation::receive_(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  // Lower layers deliver packets in one piece; a packet in one fragment is passed up without copying
+  if (!data.is_contiguous()) {
+    ESP_LOGW(TAG, "%s: Packet split across buffers is not supported", this->name_);
+    return;
+  }
+  auto fragment = data.as_span();
+  if (fragment.size() < HEADER_SIZE)
+    return;
+  uint8_t number = fragment[0];
+  uint8_t index = fragment[1] & INDEX_MASK;
+  bool last = (fragment[1] & LAST_FRAGMENT) != 0;
+  const uint8_t *payload = fragment.data() + HEADER_SIZE;
+  size_t length = fragment.size() - HEADER_SIZE;
+  if (index == 0) {
+    if (this->rx_joining_) {
+      ESP_LOGV(TAG, "%s: Incomplete packet %u dropped", this->name_, this->rx_number_);
+    }
+    this->rx_joining_ = false;
+    if (last) {
+      if (length > this->max_packet_size_) {
+        ESP_LOGV(TAG, "%s: Packet %u is larger than %zu, dropped", this->name_, number, this->max_packet_size_);
+        return;
+      }
+      this->on_receive_from_interface_(PacketBuffer(payload, length), meta_data);
+      return;
+    }
+    this->rx_joining_ = true;
+    this->rx_number_ = number;
+    this->rx_source_ = meta_data.source_address;
+    this->rx_next_index_ = 0;
+    this->rx_length_ = 0;
+  } else if (!this->rx_joining_ || number != this->rx_number_ || meta_data.source_address != this->rx_source_ ||
+             index != this->rx_next_index_) {
+    ESP_LOGV(TAG, "%s: Fragment %u of packet %u dropped", this->name_, index, number);
+    this->rx_joining_ = false;
+    return;
+  }
+  if (this->rx_length_ + length > this->max_packet_size_) {
+    ESP_LOGV(TAG, "%s: Packet %u is larger than %zu, dropped", this->name_, number, this->max_packet_size_);
+    this->rx_joining_ = false;
+    return;
+  }
+  memcpy(this->rx_buffer_.get() + this->rx_length_, payload, length);
+  this->rx_length_ += length;
+  this->rx_next_index_++;
+  if (last) {
+    this->rx_joining_ = false;
+    this->on_receive_from_interface_(PacketBuffer(this->rx_buffer_.get(), this->rx_length_), meta_data);
+  }
+}
+
+}  // namespace esphome::packet_interface
+#endif

esphome/components/packet_interface/basic_fragmentation.h

@@ -0,0 +1,56 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+#ifdef USE_PACKET_INTERFACE_BASIC_FRAGMENTATION
+#include <memory>
+
+#include "packet_interface.h"
+
+namespace esphome::packet_interface {
+
+/**
+ * A fragmentation layer: packets up to this layer's max_packet_size are split into fragments that fit the layer
+ * below, and joined again at the other end.
+ *
+ * Each fragment starts with a 2 byte header: an 8-bit packet number, then the fragment's index in bits 0-6, with bit
+ * 7 set on the last fragment. A packet that fits in one fragment still has the header, and is sent and passed up
+ * without being copied.
+ *
+ * A packet is started only if the layer below can take all of its fragments now (get_send_capacity()), so a full
+ * reliability queue does not leave the peer with part of a packet. The receiver joins one packet at a time, and drops
+ * it when a fragment is missing, or one arrives with another packet number or from another source address. Above a
+ * reliability layer, fragments arrive in order and none are missing.
+ */
+class BasicFragmentation : public PacketInterface {
+ public:
+  /// Packet number, then index and last flag.
+  static constexpr size_t HEADER_SIZE = 2;
+  /// The index has 7 bits.
+  static constexpr size_t MAX_FRAGMENTS = 128;
+
+  BasicFragmentation(PacketInterface *lower, size_t max_packet_size);
+
+  void setup() override;
+  void dump_config() override;
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) override;
+
+ protected:
+  void receive_(const PacketBuffer &data, const PacketMetaData &meta_data);
+  size_t fragment_size_() const { return this->lower_->get_max_packet_size() - HEADER_SIZE; }
+  size_t fragments_for_(size_t size) const;
+
+  PacketInterface *lower_;
+  // One fragment being sent, and the packet being joined; both allocated in setup()
+  std::unique_ptr<uint8_t[]> tx_buffer_;
+  std::unique_ptr<uint8_t[]> rx_buffer_;
+  size_t rx_length_{0};
+  uint16_t rx_source_{0};
+  uint8_t tx_number_{0};
+  uint8_t rx_number_{0};
+  uint8_t rx_next_index_{0};
+  bool rx_joining_{false};
+  bool queue_too_short_{false};
+};
+
+}  // namespace esphome::packet_interface
+#endif

esphome/components/packet_interface/layers/__init__.py

@@ -2,7 +2,7 @@
 
 Each module in this package implements one layer type and registers it with its kind, for example:
 
-    @RELIABILITY.register("hdlc", schema=HDLC_SCHEMA)
+    @RELIABILITY.register("hdlc", schema=HDLC_SCHEMA, single_peer=lambda config: True)
     async def build_hdlc(config: ConfigType, lower: MockObj) -> MockObj:
         ...
 
@@ -16,17 +16,23 @@ from typing import Any
 
 import esphome.codegen as cg
 import esphome.config_validation as cv
-from esphome.const import CONF_ENCRYPTION, CONF_KEY, CONF_TYPE
+from esphome.const import CONF_ENCRYPTION, CONF_FRAGMENTATION, CONF_KEY, CONF_TYPE
 from esphome.cpp_generator import MockObj
 from esphome.types import ConfigType
 
 packet_interface_ns = cg.esphome_ns.namespace("packet_interface")
 PacketInterface = packet_interface_ns.class_("PacketInterface", cg.Component)
 
+CONF_ADDRESSING = "addressing"
+CONF_DESTINATION = "destination"
+CONF_PEERS = "peers"
 CONF_RELIABILITY = "reliability"
 CONF_SESSION = "session"
 
-# Builds a layer above lower and returns it
+BROADCAST_ADDRESS = 0xFFFF
+
+# Builds a layer above lower and returns it. An addressing layer instead returns a list of (port, config) pairs, one
+# for each stack it carries, where config holds the layers for that stack, or is None for the layers on the link.
 LayerBuilder = Callable[[ConfigType, MockObj], Awaitable[Any]]
 
 
@@ -42,8 +48,11 @@ def _nothing(config: ConfigType) -> list[str]:
 class LayerType:
     """One implementation of a kind of layer."""
 
-    schema: Any
+    # Returns the schema; called once every layer type is registered, as some schemas contain other layers
+    schema_factory: Callable[[], Any]
     build: LayerBuilder
+    # True if this configuration keeps state for one peer, so cannot be used with broadcast addressing
+    single_peer: Callable[[ConfigType], bool]
     # True if this configuration works only when the peer sends packets back, which a session layer makes sure of
     needs_session: Callable[[ConfigType], bool]
     # The components this configuration needs loaded
@@ -68,11 +77,16 @@ class LayerKind:
         self,
         name: str,
         *,
-        schema: Any,
+        schema: Any = None,
+        schema_factory: Callable[[], Any] | None = None,
+        single_peer: Callable[[ConfigType], bool] = _never,
         needs_session: Callable[[ConfigType], bool] = _never,
         auto_load: Callable[[ConfigType], list[str]] = _nothing,
     ) -> Callable[[LayerBuilder], LayerBuilder]:
-        """Register a type with its schema."""
+        """Register a type with its schema, or with a function returning a schema that contains other layers."""
+
+        def fixed_schema() -> Any:
+            return schema
 
         def decorator(build: LayerBuilder) -> LayerBuilder:
             if self.frozen:
@@ -84,7 +98,13 @@ class LayerKind:
                 raise ValueError(
                     f"{self.key} layer type '{name}' is already registered"
                 )
-            self.types[name] = LayerType(schema, build, needs_session, auto_load)
+            self.types[name] = LayerType(
+                schema_factory or fixed_schema,
+                build,
+                single_peer,
+                needs_session,
+                auto_load,
+            )
             return build
 
         return decorator
@@ -103,7 +123,7 @@ class LayerKind:
         return cv.All(
             self._expand,
             cv.typed_schema(
-                {name: layer.schema for name, layer in self.types.items()},
+                {name: layer.schema_factory() for name, layer in self.types.items()},
                 default_type=self.default_type,
             ),
         )
@@ -112,11 +132,15 @@ class LayerKind:
         return self.types[config[CONF_TYPE]]
 
 
+# An addressing type's configuration has either a destination or a list of peers, each with an id, a destination
+# and its own layers.
+ADDRESSING = LayerKind(CONF_ADDRESSING, "basic")
 ENCRYPTION = LayerKind(CONF_ENCRYPTION, "xxtea", shorthand=CONF_KEY)
 RELIABILITY = LayerKind(CONF_RELIABILITY, "hdlc")
+FRAGMENTATION = LayerKind(CONF_FRAGMENTATION, "basic")
 SESSION = LayerKind(CONF_SESSION, "basic")
-# The kinds stacked above the link, from the bottom up
-STACK_ORDER = (ENCRYPTION, RELIABILITY, SESSION)
+# The kinds stacked above the link, or above each port of an addressing layer, from the bottom up
+STACK_ORDER = (ENCRYPTION, RELIABILITY, FRAGMENTATION, SESSION)
 
 
 def stack_schema() -> dict:
@@ -139,6 +163,21 @@ def check_session(config: ConfigType) -> None:
             )
 
 
+def check_single_peer(config: ConfigType, destination: int) -> None:
+    """Layers that keep state for one peer need a destination other than broadcast."""
+    if destination != BROADCAST_ADDRESS:
+        return
+    for kind in STACK_ORDER:
+        if (layer := config.get(kind.key)) is not None and kind.get_type(
+            layer
+        ).single_peer(layer):
+            raise cv.Invalid(
+                f"This '{kind.key}' works between two devices only, so it needs a '{CONF_DESTINATION}' other than "
+                "broadcast",
+                path=[kind.key],
+            )
+
+
 def stack_auto_load(config: ConfigType) -> set[str]:
     """The components that the layers configured in config need loaded."""
     loads: set[str] = set()

esphome/components/packet_interface/layers/basic_addressing.py

@@ -0,0 +1,107 @@
+"""Addressing layer with 16-bit addresses, carrying one stack of layers or one for each of several peers."""
+
+from typing import Any
+
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.const import CONF_ADDRESS, CONF_ID
+from esphome.core import ID
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
+
+from . import (
+    ADDRESSING,
+    BROADCAST_ADDRESS,
+    CONF_DESTINATION,
+    CONF_PEERS,
+    PacketInterface,
+    check_session,
+    check_single_peer,
+    packet_interface_ns,
+    stack_schema,
+)
+
+CONF_PORT_ID = "port_id"
+
+BasicAddressing = packet_interface_ns.class_("BasicAddressing", cg.Component)
+AddressingPort = packet_interface_ns.class_("AddressingPort", PacketInterface)
+
+_request_port_slot = cg.slot_counter("PACKET_INTERFACE_ADDRESSING_PORT_COUNT")
+
+
+def _destination(value: Any) -> int:
+    """A device address, or broadcast."""
+    if isinstance(value, str) and value.lower() == "broadcast":
+        return BROADCAST_ADDRESS
+    return cv.hex_int_range(1, BROADCAST_ADDRESS)(value)
+
+
+def _validate_peer(config: ConfigType) -> ConfigType:
+    check_session(config)
+    check_single_peer(config, config[CONF_DESTINATION])
+    return config
+
+
+def _unique_destinations(peers: list[ConfigType]) -> list[ConfigType]:
+    destinations = [peer[CONF_DESTINATION] for peer in peers]
+    if len(set(destinations)) != len(destinations):
+        raise cv.Invalid("Each peer must have a different destination")
+    return peers
+
+
+# A function, as each peer holds the other layers, whose schemas exist only once every layer type is registered
+def _basic_addressing_schema() -> cv.Schema:
+    peer_schema = cv.All(
+        cv.Schema(
+            {
+                cv.Required(CONF_ID): cv.declare_id(AddressingPort),
+                cv.Required(CONF_DESTINATION): _destination,
+                **stack_schema(),
+            }
+        ),
+        _validate_peer,
+    )
+    return cv.All(
+        cv.Schema(
+            {
+                cv.GenerateID(): cv.declare_id(BasicAddressing),
+                cv.GenerateID(CONF_PORT_ID): cv.declare_id(AddressingPort),
+                cv.Required(CONF_ADDRESS): cv.hex_int_range(1, BROADCAST_ADDRESS - 1),
+                cv.Optional(CONF_DESTINATION): _destination,
+                cv.Optional(CONF_PEERS): cv.All(
+                    cv.ensure_list(peer_schema), cv.Length(min=1), _unique_destinations
+                ),
+            }
+        ).extend(cv.COMPONENT_SCHEMA),
+        cv.has_at_most_one_key(CONF_DESTINATION, CONF_PEERS),
+    )
+
+
+async def _new_port(
+    port_id: ID, addressing: MockObj, destination: int, only_port: bool
+) -> MockObj:
+    _request_port_slot(str(addressing))
+    port = cg.new_Pvariable(port_id, addressing, destination, only_port)
+    await cg.register_component(port, {})
+    return port
+
+
+@ADDRESSING.register("basic", schema_factory=_basic_addressing_schema)
+async def build_basic_addressing(
+    config: ConfigType, link: MockObj
+) -> list[tuple[MockObj, ConfigType | None]]:
+    cg.add_define("USE_PACKET_INTERFACE_BASIC_ADDRESSING")
+    hub = cg.new_Pvariable(config[CONF_ID], link, config[CONF_ADDRESS])
+    await cg.register_component(hub, config)
+    if (peers := config.get(CONF_PEERS)) is None:
+        port = await _new_port(
+            config[CONF_PORT_ID], hub, config[CONF_DESTINATION], True
+        )
+        return [(port, None)]
+    stacks = []
+    # Each peer has its own stack, named after the peer for logs and saved preferences
+    for peer in peers:
+        port = await _new_port(peer[CONF_ID], hub, peer[CONF_DESTINATION], False)
+        cg.add(port.set_name(peer[CONF_ID].id))
+        stacks.append((port, peer))
+    return stacks

esphome/components/packet_interface/layers/basic_fragmentation.py

@@ -0,0 +1,31 @@
+"""Fragmentation layer that splits packets larger than the layers below allow, and joins them again."""
+
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.const import CONF_ID
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
+
+from . import FRAGMENTATION, PacketInterface, packet_interface_ns
+
+CONF_MAX_PACKET_SIZE = "max_packet_size"
+
+BasicFragmentation = packet_interface_ns.class_("BasicFragmentation", PacketInterface)
+
+BASIC_FRAGMENTATION_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(): cv.declare_id(BasicFragmentation),
+        # The packet being joined is held in a buffer of this size
+        cv.Optional(CONF_MAX_PACKET_SIZE, default=1024): cv.int_range(16, 16384),
+    }
+).extend(cv.COMPONENT_SCHEMA)
+
+
+# Fragments from different devices are told apart by the addressing layer's source address, so several devices can
+# share a channel; without addressing, two devices sending fragmented packets at once can make the receiver drop both
+@FRAGMENTATION.register("basic", schema=BASIC_FRAGMENTATION_SCHEMA)
+async def build_basic_fragmentation(config: ConfigType, lower: MockObj) -> MockObj:
+    cg.add_define("USE_PACKET_INTERFACE_BASIC_FRAGMENTATION")
+    var = cg.new_Pvariable(config[CONF_ID], lower, config[CONF_MAX_PACKET_SIZE])
+    await cg.register_component(var, config)
+    return var

esphome/components/packet_interface/layers/basic_session.py

@@ -57,6 +57,7 @@ BASIC_SESSION_SCHEMA = cv.All(
 @SESSION.register(
     "basic",
     schema=BASIC_SESSION_SCHEMA,
+    single_peer=lambda config: True,
     auto_load=lambda config: ["binary_sensor"] if CONF_BINARY_SENSOR in config else [],
 )
 async def build_basic_session(config: ConfigType, lower: MockObj) -> MockObj:

esphome/components/packet_interface/layers/hdlc_reliability.py

@@ -47,7 +47,7 @@ HDLC_SCHEMA = cv.All(
 )
 
 
-@RELIABILITY.register("hdlc", schema=HDLC_SCHEMA)
+@RELIABILITY.register("hdlc", schema=HDLC_SCHEMA, single_peer=lambda config: True)
 async def build_hdlc(config: ConfigType, lower: MockObj) -> MockObj:
     cg.add_define("USE_PACKET_INTERFACE_HDLC")
     var = cg.new_Pvariable(config[CONF_ID], lower)

esphome/components/packet_interface/layers/xxtea_encryption.py

@@ -23,10 +23,12 @@ XXTEA_SCHEMA = cv.Schema(
 ).extend(cv.COMPONENT_SCHEMA)
 
 
+# The rolling code keeps the last counter of one peer, and accepts a packet only once the peer has sent back this
+# device's boot number
 @ENCRYPTION.register(
     "xxtea",
     schema=XXTEA_SCHEMA,
-    # The rolling code accepts a packet only once the peer has sent back this device's boot number
+    single_peer=lambda config: config[CONF_ROLLING_CODE_ENABLE],
     needs_session=lambda config: config[CONF_ROLLING_CODE_ENABLE],
     auto_load=lambda config: ["xxtea"],
 )

esphome/components/packet_interface/packet_transport/__init__.py

@@ -7,7 +7,12 @@ from esphome.components.packet_transport import (
 import esphome.config_validation as cv
 from esphome.types import ConfigType
 
-from .. import CONF_PACKET_INTERFACE, PacketInterface, packet_interface_ns
+from .. import (
+    CONF_PACKET_INTERFACE,
+    PacketInterface,
+    final_validate_interface,
+    packet_interface_ns,
+)
 
 CODEOWNERS = ["@clydebarrow"]
 DEPENDENCIES = ["packet_interface"]
@@ -22,6 +27,8 @@ CONFIG_SCHEMA = transport_schema(PacketInterfaceTransport).extend(
     }
 )
 
+FINAL_VALIDATE_SCHEMA = final_validate_interface(CONF_PACKET_INTERFACE)
+
 
 async def to_code(config: ConfigType) -> None:
     interface = await cg.get_variable(config[CONF_PACKET_INTERFACE])

esphome/core/defines.h

@@ -259,6 +259,9 @@
 #define USE_OTA_STATE_LISTENER
 #define USE_OUTPUT
 #define USE_OUTPUT_FLOAT_POWER_SCALING
+#define PACKET_INTERFACE_ADDRESSING_PORT_COUNT 2
+#define USE_PACKET_INTERFACE_BASIC_ADDRESSING
+#define USE_PACKET_INTERFACE_BASIC_FRAGMENTATION
 #define USE_PACKET_INTERFACE_BASIC_SESSION
 #define USE_PACKET_INTERFACE_HDLC
 #define USE_PACKET_INTERFACE_XXTEA

tests/components/packet_interface/__init__.py

@@ -4,6 +4,9 @@ from tests.testing_helpers import ComponentManifestOverride
 
 def override_manifest(manifest: ComponentManifestOverride) -> None:
     async def to_code_testing(config: object) -> None:
+        cg.add_define("PACKET_INTERFACE_ADDRESSING_PORT_COUNT", 4)
+        cg.add_define("USE_PACKET_INTERFACE_BASIC_ADDRESSING")
+        cg.add_define("USE_PACKET_INTERFACE_BASIC_FRAGMENTATION")
         cg.add_define("USE_PACKET_INTERFACE_BASIC_SESSION")
         cg.add_define("USE_PACKET_INTERFACE_HDLC")
         cg.add_define("USE_PACKET_INTERFACE_XXTEA")

tests/components/packet_interface/basic_addressing.cpp

@@ -0,0 +1,188 @@
+#include <gtest/gtest.h>
+#include <cstdint>
+#include <initializer_list>
+#include <memory>
+#include <vector>
+#include "esphome/components/packet_interface/basic_addressing.h"
+#include "esphome/components/packet_interface/hdlc_reliability.h"
+#include "esphome/core/application.h"
+
+namespace esphome::packet_interface::addressing_testing {
+
+using Packet = std::vector<uint8_t>;
+
+static constexpr uint16_t BROADCAST = BasicAddressing::BROADCAST_ADDRESS;
+
+// A link layer that records what is sent, so a test can deliver it to other devices
+class MockLink : public PacketInterface {
+ public:
+  MockLink() { this->set_max_packet_size(64); }
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &) override {
+    this->sent.push_back(data.to_vector());
+    return true;
+  }
+  void inject(const Packet &packet) { this->on_receive_from_interface_(PacketBuffer(packet), {}); }
+
+  std::vector<Packet> sent;
+};
+
+// What one port receives
+struct Port {
+  Port(BasicAddressing *addressing, uint16_t destination, bool only_port) {
+    this->port = std::make_unique<AddressingPort>(addressing, destination, only_port);
+    this->port->add_packet_interface_listener([this](const PacketBuffer &data, const PacketMetaData &meta_data) {
+      this->received.push_back(data.to_vector());
+      this->sources.push_back(meta_data.source_address);
+    });
+  }
+  void send(const Packet &data, uint16_t destination = 0) {
+    PacketMetaData meta_data;
+    meta_data.destination_address = destination;
+    this->port->send_to_interface(PacketBuffer(data), meta_data);
+  }
+
+  std::unique_ptr<AddressingPort> port;
+  std::vector<Packet> received;
+  std::vector<uint16_t> sources;
+};
+
+// A device on the bus with one port per destination given
+struct Device {
+  Device(uint16_t address, std::initializer_list<uint16_t> destinations) : addressing(&this->link, address) {
+    this->link.set_name("bus");
+    for (uint16_t destination : destinations)
+      this->ports.push_back(std::make_unique<Port>(&this->addressing, destination, destinations.size() == 1));
+    this->addressing.setup();
+  }
+  explicit Device(uint16_t address, uint16_t destination = BROADCAST) : Device(address, {destination}) {}
+  Port &port(size_t index = 0) { return *this->ports[index]; }
+  // Every device on the bus hears every packet, including its own
+  void deliver(std::initializer_list<Device *> bus) {
+    for (auto &packet : this->link.sent) {
+      for (auto *device : bus)
+        device->link.inject(packet);
+    }
+    this->link.sent.clear();
+  }
+
+  MockLink link;
+  BasicAddressing addressing;
+  std::vector<std::unique_ptr<Port>> ports;
+};
+
+TEST(BasicAddressingTest, HeaderAndCapability) {
+  Device a(0x1234, 0x5678);
+  EXPECT_EQ(a.port().port->get_max_packet_size(), 64 - BasicAddressing::HEADER_SIZE);
+  EXPECT_TRUE(a.port().port->has_capability(CAPABILITY_ADDRESSING));
+  EXPECT_EQ(a.link.top(), a.port().port.get());
+  a.port().send({0xAA});
+  a.port().send({0xBB}, 0x0042);
+  EXPECT_EQ(a.link.sent, (std::vector<Packet>{{0x78, 0x56, 0x34, 0x12, 0xAA}, {0x42, 0x00, 0x34, 0x12, 0xBB}}));
+}
+
+TEST(BasicAddressingTest, BroadcastReachesEveryoneButSender) {
+  Device a(1), b(2), c(3);
+  a.port().send({0x10});
+  a.deliver({&a, &b, &c});
+  EXPECT_TRUE(a.port().received.empty());
+  EXPECT_EQ(b.port().received, std::vector<Packet>{{0x10}});
+  EXPECT_EQ(c.port().received, std::vector<Packet>{{0x10}});
+  EXPECT_EQ(b.port().sources, std::vector<uint16_t>{1});
+}
+
+TEST(BasicAddressingTest, UnicastReachesOnlyDestination) {
+  Device a(1, 2), b(2), c(3);
+  a.port().send({0x20});
+  a.deliver({&b, &c});
+  EXPECT_EQ(b.port().received, std::vector<Packet>{{0x20}});
+  EXPECT_TRUE(c.port().received.empty());
+}
+
+TEST(BasicAddressingTest, PointToPointAcceptsOnlyPeer) {
+  Device a(1, 2), b(2, 1), c(3);
+  c.port().send({0x30});     // broadcast from a third device
+  c.port().send({0x31}, 2);  // addressed to b, from the wrong device
+  a.port().send({0x32});     // from the peer
+  c.deliver({&b});
+  a.deliver({&b});
+  EXPECT_EQ(b.port().received, std::vector<Packet>{{0x32}});
+  EXPECT_EQ(b.port().sources, std::vector<uint16_t>{1});
+}
+
+TEST(BasicAddressingTest, PeersHaveSeparateStacks) {
+  Device hub(1, {2, 3, BROADCAST}), b(2, 1), c(3, 1), d(4);
+  EXPECT_EQ(hub.link.top(), &hub.link);  // with several ports, applications use the ports
+  EXPECT_EQ(hub.port(0).port->top(), hub.port(0).port.get());
+  b.port().send({0x02});
+  c.port().send({0x03});
+  d.port().send({0x04});
+  b.deliver({&hub});
+  c.deliver({&hub});
+  d.deliver({&hub});
+  EXPECT_EQ(hub.port(0).received, std::vector<Packet>{{0x02}});
+  EXPECT_EQ(hub.port(1).received, std::vector<Packet>{{0x03}});
+  EXPECT_EQ(hub.port(2).received, std::vector<Packet>{{0x04}});  // only devices without a port of their own
+  hub.port(1).send({0x13});
+  hub.deliver({&b, &c, &d});
+  EXPECT_EQ(c.port().received, std::vector<Packet>{{0x13}});
+  EXPECT_TRUE(b.port().received.empty());
+  EXPECT_TRUE(d.port().received.empty());
+}
+
+TEST(BasicAddressingTest, WithoutBroadcastPortUnknownDevicesDropped) {
+  Device hub(1, {2, 3}), d(4);
+  d.port().send({0x04});
+  d.deliver({&hub});
+  EXPECT_TRUE(hub.port(0).received.empty());
+  EXPECT_TRUE(hub.port(1).received.empty());
+}
+
+TEST(BasicAddressingTest, ShortPacketDropped) {
+  Device a(1);
+  a.link.inject({0xFF, 0xFF, 0x02});
+  EXPECT_TRUE(a.port().received.empty());
+}
+
+// The test main does not construct App; the reliability layer's timers need its scheduler
+class AddressingReliabilityTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    new (&App) Application;
+    App.pre_setup("addressing-tests", 16, "", 0);
+  }
+  void TearDown() override { App.~Application(); }
+};
+
+// A reliability layer on each port, recording what it delivers
+struct ReliablePort {
+  explicit ReliablePort(AddressingPort *port) : hdlc(port) {
+    this->hdlc.add_packet_interface_listener(
+        [this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(data.to_vector()); });
+    this->hdlc.setup();
+  }
+  void send(const Packet &data) { this->hdlc.send_to_interface(PacketBuffer(data), {}); }
+  HdlcReliability hdlc;
+  std::vector<Packet> received;
+};
+
+TEST_F(AddressingReliabilityTest, SeveralReliablePeersOnOneBus) {
+  Device hub(1, {2, 3}), b(2, 1), c(3, 1);
+  ReliablePort hub_b(hub.port(0).port.get()), hub_c(hub.port(1).port.get());
+  ReliablePort b_hub(b.port().port.get()), c_hub(c.port().port.get());
+  hub_b.send({0x12});
+  hub_c.send({0x13});
+  b_hub.send({0x21});
+  c_hub.send({0x31});
+  for (int i = 0; i != 20; i++) {
+    for (auto *port : {&hub_b, &hub_c, &b_hub, &c_hub})
+      port->hdlc.loop();
+    for (auto *device : {&hub, &b, &c})
+      device->deliver({&hub, &b, &c});
+  }
+  EXPECT_EQ(b_hub.received, std::vector<Packet>{{0x12}});
+  EXPECT_EQ(c_hub.received, std::vector<Packet>{{0x13}});
+  EXPECT_EQ(hub_b.received, std::vector<Packet>{{0x21}});
+  EXPECT_EQ(hub_c.received, std::vector<Packet>{{0x31}});
+}
+
+}  // namespace esphome::packet_interface::addressing_testing

tests/components/packet_interface/basic_fragmentation.cpp

@@ -0,0 +1,194 @@
+#include <gtest/gtest.h>
+#include <cstdint>
+#include <memory>
+#include <vector>
+#include "esphome/components/packet_interface/basic_fragmentation.h"
+#include "esphome/components/packet_interface/hdlc_reliability.h"
+#include "esphome/core/application.h"
+
+namespace esphome::packet_interface::fragmentation_testing {
+
+using Packet = std::vector<uint8_t>;
+
+// A link layer that records what is sent, with a limited send capacity
+class MockLink : public PacketInterface {
+ public:
+  MockLink() { this->set_max_packet_size(12); }
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &) override {
+    this->sent.push_back(data.to_vector());
+    return true;
+  }
+  size_t get_send_capacity() const override { return this->capacity; }
+  void inject(const Packet &packet, uint16_t source = 0) {
+    PacketMetaData meta_data;
+    meta_data.source_address = source;
+    this->on_receive_from_interface_(PacketBuffer(packet), meta_data);
+  }
+
+  std::vector<Packet> sent;
+  size_t capacity{SIZE_MAX};
+};
+
+struct Peer {
+  explicit Peer(PacketInterface *lower) : fragmentation(lower, 100) {
+    this->fragmentation.add_packet_interface_listener(
+        [this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(data.to_vector()); });
+    this->fragmentation.setup();
+  }
+  bool send(const Packet &data) { return this->fragmentation.send_to_interface(PacketBuffer(data), {}); }
+
+  BasicFragmentation fragmentation;
+  std::vector<Packet> received;
+};
+
+static Packet make_packet(size_t size) {
+  Packet packet(size);
+  for (size_t i = 0; i != size; i++)
+    packet[i] = i * 7 + 1;
+  return packet;
+}
+
+class TestHdlc : public HdlcReliability {
+ public:
+  using HdlcReliability::HdlcReliability;
+  using HdlcReliability::on_timeout_;
+};
+
+// The test main does not construct App; the reliability layer's timers need its scheduler
+class BasicFragmentationTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    new (&App) Application;
+    App.pre_setup("fragmentation-tests", 19, "", 0);
+  }
+  void TearDown() override { App.~Application(); }
+
+  // Deliver what a has sent to b
+  static void deliver(MockLink &from, MockLink &to) {
+    for (auto &packet : from.sent)
+      to.inject(packet);
+    from.sent.clear();
+  }
+
+  MockLink link_a, link_b;
+};
+
+TEST_F(BasicFragmentationTest, SizesAndCapability) {
+  Peer a(&this->link_a);
+  EXPECT_EQ(a.fragmentation.get_max_packet_size(), 100);
+  EXPECT_TRUE(a.fragmentation.has_capability(CAPABILITY_FRAGMENTATION));
+  EXPECT_EQ(this->link_a.top(), &a.fragmentation);
+}
+
+TEST_F(BasicFragmentationTest, SmallPacketIsOneFragment) {
+  Peer a(&this->link_a), b(&this->link_b);
+  a.send({1, 2, 3});
+  ASSERT_EQ(this->link_a.sent.size(), 1);
+  EXPECT_EQ(this->link_a.sent[0].size(), 3 + BasicFragmentation::HEADER_SIZE);
+  EXPECT_EQ(this->link_a.sent[0][1], 0x80);  // index 0, last
+  deliver(this->link_a, this->link_b);
+  EXPECT_EQ(b.received, (std::vector<Packet>{{1, 2, 3}}));
+}
+
+TEST_F(BasicFragmentationTest, LargePacketSplitAndJoined) {
+  Peer a(&this->link_a), b(&this->link_b);
+  for (size_t size : {10, 11, 50, 100}) {
+    Packet packet = make_packet(size);
+    ASSERT_TRUE(a.send(packet));
+    EXPECT_EQ(this->link_a.sent.size(), (size + 9) / 10);  // 10 bytes of payload in each 12 byte fragment
+    EXPECT_EQ(this->link_a.sent.back()[1] & 0x80, 0x80);
+    deliver(this->link_a, this->link_b);
+    EXPECT_EQ(b.received.back(), packet);
+  }
+  EXPECT_EQ(b.received.size(), 4);
+}
+
+TEST_F(BasicFragmentationTest, CompositePacketSplitAcrossFragments) {
+  Peer a(&this->link_a), b(&this->link_b);
+  Packet pre = make_packet(7), mid(13, 0xAA), post(9, 0x55);
+  PacketBuffer body(mid.data(), mid.size());
+  ASSERT_TRUE(
+      a.fragmentation.send_to_interface(PacketBuffer(pre.data(), pre.size(), &body, post.data(), post.size()), {}));
+  deliver(this->link_a, this->link_b);
+  Packet expected = pre;
+  expected.insert(expected.end(), mid.begin(), mid.end());
+  expected.insert(expected.end(), post.begin(), post.end());
+  EXPECT_EQ(b.received, std::vector<Packet>{expected});
+}
+
+TEST_F(BasicFragmentationTest, MissingFragmentDropsOnlyThatPacket) {
+  Peer a(&this->link_a), b(&this->link_b);
+  a.send(make_packet(30));
+  this->link_a.sent.erase(this->link_a.sent.begin() + 1);
+  deliver(this->link_a, this->link_b);
+  EXPECT_TRUE(b.received.empty());
+  a.send(make_packet(25));
+  deliver(this->link_a, this->link_b);
+  EXPECT_EQ(b.received, std::vector<Packet>{make_packet(25)});
+}
+
+TEST_F(BasicFragmentationTest, NotStartedWithoutRoomForAllFragments) {
+  Peer a(&this->link_a);
+  this->link_a.capacity = 2;
+  EXPECT_FALSE(a.send(make_packet(30)));  // needs 3
+  EXPECT_TRUE(this->link_a.sent.empty());
+  EXPECT_TRUE(a.send(make_packet(20)));
+  EXPECT_FALSE(a.send(make_packet(101)));
+}
+
+TEST_F(BasicFragmentationTest, InterleavedSourcesNotMixed) {
+  Peer a(&this->link_a), c(&this->link_a), b(&this->link_b);
+  a.send(make_packet(20));
+  auto from_a = std::move(this->link_a.sent);
+  this->link_a.sent.clear();
+  c.send(make_packet(20));
+  auto from_c = std::move(this->link_a.sent);
+  // Fragments from devices 1 and 2 arrive interleaved, perhaps with the same packet number
+  this->link_b.inject(from_a[0], 1);
+  this->link_b.inject(from_c[0], 2);
+  this->link_b.inject(from_a[1], 1);
+  this->link_b.inject(from_c[1], 2);
+  for (auto &packet : b.received)
+    EXPECT_EQ(packet, make_packet(20));  // nothing mixed from the two
+}
+
+TEST_F(BasicFragmentationTest, OverReliabilityWithLosses) {
+  // Fragmentation above HDLC on each side, with every fourth frame from a dropped on the way; not every third, as
+  // each resend is a burst of three, so the same fragment would be lost every time
+  TestHdlc hdlc_a(&this->link_a), hdlc_b(&this->link_b);
+  hdlc_a.setup();
+  hdlc_b.setup();
+  // HDLC adds 3 bytes to the 12 byte link, leaving 7 bytes of payload in each fragment
+  Peer a(&hdlc_a), b(&hdlc_b);
+  Packet packet = make_packet(20);  // 3 fragments, within the queue of 4
+  ASSERT_TRUE(a.send(packet));
+  int count = 0;
+  for (int i = 0; i != 30; i++) {
+    hdlc_a.loop();
+    hdlc_b.loop();
+    if (i % 5 == 4)
+      hdlc_a.on_timeout_();
+    for (auto &frame : this->link_a.sent) {
+      if (++count % 4 != 0)
+        this->link_b.inject(frame);
+    }
+    this->link_a.sent.clear();
+    deliver(this->link_b, this->link_a);
+  }
+  EXPECT_EQ(b.received, std::vector<Packet>{packet});
+}
+
+TEST_F(BasicFragmentationTest, OversizedSingleFragmentDropped) {
+  // A link packet holds more than this layer's max_packet_size
+  BasicFragmentation fragmentation(&this->link_a, 4);
+  std::vector<Packet> received;
+  fragmentation.add_packet_interface_listener(
+      [&received](const PacketBuffer &data, const PacketMetaData &) { received.push_back(data.to_vector()); });
+  fragmentation.setup();
+  this->link_a.inject({0x01, 0x80, 1, 2, 3, 4, 5});
+  EXPECT_TRUE(received.empty());
+  this->link_a.inject({0x02, 0x80, 1, 2, 3, 4});
+  EXPECT_EQ(received, std::vector<Packet>{Packet({1, 2, 3, 4})});
+}
+
+}  // namespace esphome::packet_interface::fragmentation_testing

tests/components/packet_interface/common.yaml

@@ -9,24 +9,40 @@ packet_interface:
   - platform: uart
     id: packet_interface_uart
     max_packet_size: 256
-    reliability:
-      timeout: 300ms
-      ack_delay: 50ms
-      max_queue_length: 3
-      max_retries: 4
-    encryption:
-      key: Johnny Cash
-      rolling_code_enable: true
-    session:
-      keep_alive: 5s
-      timeout: 12s
-      binary_sensor:
-        name: UART session
+    addressing:
+      address: 0x0001
+      peers:
+        - id: packet_interface_peer_a
+          destination: 0x0002
+          encryption:
+            key: Johnny Cash
+            rolling_code_enable: true
+          reliability:
+            timeout: 300ms
+            max_queue_length: 3
+            max_retries: 4
+          session:
+            keep_alive: 5s
+            timeout: 12s
+            binary_sensor:
+              name: Peer A session
+        - id: packet_interface_peer_b
+          destination: 0x0003
+          reliability:
+          fragmentation:
+            max_packet_size: 2048
+          session:
+        - id: packet_interface_bus
+          destination: broadcast
   - platform: udp
     id: packet_interface_network
     udp_id: packet_interface_udp
+    addressing:
+      address: 0x0001
+      destination: 0x0002
     encryption: Johnny Cash
     reliability:
+    fragmentation:
     session:
 
 sensor:
@@ -36,7 +52,7 @@ sensor:
 
 packet_transport:
   - platform: packet_interface
-    packet_interface: packet_interface_uart
+    packet_interface: packet_interface_peer_a
     sensors:
       - packet_interface_sensor
   - platform: packet_interface

tests/unit_tests/test_packet_interface.py

@@ -1,4 +1,4 @@
-"""Unit tests for the packet_interface component."""
+"""Unit tests for the packet_interface component and its platforms."""
 
 from typing import Any
 
@@ -8,6 +8,8 @@ from esphome import config_validation as cv
 from esphome.components import packet_interface
 from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE
 from esphome.components.packet_interface.layers import (
+    CONF_ADDRESSING,
+    CONF_FRAGMENTATION,
     CONF_RELIABILITY,
     CONF_SESSION,
     RELIABILITY,
@@ -68,61 +70,6 @@ def test_reliability_queue_length_limit(setup_core) -> None:
         )
 
 
-def test_reliability_timeout_minimum(setup_core) -> None:
-    """A timeout shorter than a few loop passes cannot work."""
-    uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "50ms"}})
-    with pytest.raises(cv.Invalid):
-        uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "40ms"}})
-
-
-def test_ack_delay_shorter_than_timeout(setup_core) -> None:
-    """The acknowledgement must not wait longer than the peer's timeout."""
-    reliability = uart_packet_interface.CONFIG_SCHEMA(
-        {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "100ms"}}
-    )[CONF_RELIABILITY]
-    assert reliability["ack_delay"].total_milliseconds == 100
-    with pytest.raises(cv.Invalid, match="shorter"):
-        uart_packet_interface.CONFIG_SCHEMA(
-            {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "500ms"}}
-        )
-
-
-def test_layer_types_registered() -> None:
-    """Importing packet_interface registers the built-in layer types."""
-    assert {kind.key: set(kind.types) for kind in STACK_ORDER} == {
-        CONF_ENCRYPTION: {"xxtea"},
-        CONF_RELIABILITY: {"hdlc"},
-        CONF_SESSION: {"basic"},
-    }
-
-
-async def _build(config: dict, lower: Any) -> Any:
-    return lower
-
-
-def test_registered_type_selectable(setup_core) -> None:
-    """A newly registered layer type can be selected with type."""
-    kind = LayerKind("test_kind", "test")
-    kind.register("test", schema=cv.Schema({}))(_build)
-    config = kind.schema()({"type": "test"})
-    assert config["type"] == "test"
-    assert kind.get_type(config).build is _build
-
-
-def test_duplicate_layer_type_rejected(setup_core) -> None:
-    """Registering a type under a name already in use is an error, not a silent replacement."""
-    kind = LayerKind("test_kind", "test")
-    kind.register("test", schema=cv.Schema({}))(_build)
-    with pytest.raises(ValueError, match="already registered"):
-        kind.register("test", schema=cv.Schema({}))(_build)
-
-
-def test_late_registration_rejected(setup_core) -> None:
-    """A type registered after the schemas were built would be missing from them, so it is an error."""
-    with pytest.raises(ValueError, match="after the schemas were built"):
-        RELIABILITY.register("late", schema=cv.Schema({}))(_build)
-
-
 def test_session_defaults(setup_core) -> None:
     """An empty session key selects the basic session, timing out after three keep-alives."""
     session = uart_packet_interface.CONFIG_SCHEMA({CONF_SESSION: None})[CONF_SESSION]
@@ -176,13 +123,162 @@ def test_rolling_code_requires_session(setup_core) -> None:
     uart_packet_interface.CONFIG_SCHEMA({CONF_ENCRYPTION: "secret"})
 
 
+def test_addressing_defaults_to_broadcast(setup_core) -> None:
+    """Without a destination, packets are broadcast; broadcast can also be named."""
+    for addressing in ({"address": 1}, {"address": 1, "destination": "broadcast"}):
+        config = uart_packet_interface.CONFIG_SCHEMA({CONF_ADDRESSING: addressing})
+        assert config[CONF_ADDRESSING]["destination"] == 0xFFFF
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA({CONF_ADDRESSING: {"address": 0xFFFF}})
+
+
+@pytest.mark.parametrize(
+    "layers",
+    [
+        {CONF_RELIABILITY: None},
+        {CONF_SESSION: None},
+        {
+            CONF_ENCRYPTION: {"key": "secret", "rolling_code_enable": True},
+            CONF_SESSION: None,
+        },
+    ],
+)
+def test_two_device_layers_need_destination(setup_core, layers: dict) -> None:
+    """Layers that work between two devices only cannot be used with broadcast addressing."""
+    with pytest.raises(cv.Invalid, match="destination"):
+        uart_packet_interface.CONFIG_SCHEMA({CONF_ADDRESSING: {"address": 1}, **layers})
+    uart_packet_interface.CONFIG_SCHEMA(
+        {CONF_ADDRESSING: {"address": 1, "destination": 2}, **layers}
+    )
+
+
+def _peer(peer_id: str, destination: Any, **layers: Any) -> dict:
+    return {"id": peer_id, "destination": destination, **layers}
+
+
+def test_addressing_peers(setup_core) -> None:
+    """Each peer has its own destination and layers."""
+    config = uart_packet_interface.CONFIG_SCHEMA(
+        {
+            CONF_ADDRESSING: {
+                "address": 1,
+                "peers": [
+                    _peer("peer_a", 2, reliability=None, session=None),
+                    _peer("peer_b", 3, encryption="secret"),
+                    _peer("everyone", "broadcast"),
+                ],
+            }
+        }
+    )
+    peers = config[CONF_ADDRESSING]["peers"]
+    assert [peer["destination"] for peer in peers] == [2, 3, 0xFFFF]
+    assert CONF_RELIABILITY in peers[0]
+
+
+@pytest.mark.parametrize(
+    ("addressing", "layers", "match"),
+    [
+        ({"peers": [_peer("a", 2), _peer("b", 2)]}, {}, "different destination"),
+        ({"peers": [_peer("a", "broadcast", reliability=None)]}, {}, "broadcast"),
+        ({"peers": [_peer("a", 2)]}, {CONF_RELIABILITY: None}, "goes in each peer"),
+        ({"destination": 2, "peers": [_peer("a", 3)]}, {}, "destination"),
+    ],
+)
+def test_addressing_peers_invalid(
+    setup_core, addressing: dict, layers: dict, match: str
+) -> None:
+    """Peers need different destinations, two-device layers need a destination, and layers go in the peers."""
+    with pytest.raises(cv.Invalid, match=match):
+        uart_packet_interface.CONFIG_SCHEMA(
+            {CONF_ADDRESSING: {"address": 1, **addressing}, **layers}
+        )
+
+
+def test_layer_types_registered() -> None:
+    """Importing packet_interface registers the built-in layer types."""
+    assert {kind.key: set(kind.types) for kind in STACK_ORDER} == {
+        CONF_ENCRYPTION: {"xxtea"},
+        CONF_RELIABILITY: {"hdlc"},
+        CONF_FRAGMENTATION: {"basic"},
+        CONF_SESSION: {"basic"},
+    }
+
+
+async def _build(config: dict, lower: Any) -> Any:
+    return lower
+
+
+def test_registered_type_selectable(setup_core) -> None:
+    """A newly registered layer type can be selected with type."""
+    kind = LayerKind("test_kind", "test")
+    kind.register("test", schema=cv.Schema({}))(_build)
+    config = kind.schema()({"type": "test"})
+    assert config["type"] == "test"
+    assert kind.get_type(config).build is _build
+
+
+def test_duplicate_layer_type_rejected(setup_core) -> None:
+    """Registering a type under a name already in use is an error, not a silent replacement."""
+    kind = LayerKind("test_kind", "test")
+    kind.register("test", schema=cv.Schema({}))(_build)
+    with pytest.raises(ValueError, match="already registered"):
+        kind.register("test", schema=cv.Schema({}))(_build)
+
+
+def test_late_registration_rejected(setup_core) -> None:
+    """A type registered after the schemas were built would be missing from them, so it is an error."""
+    with pytest.raises(ValueError, match="after the schemas were built"):
+        RELIABILITY.register("late", schema=cv.Schema({}))(_build)
+
+
+def test_reliability_timeout_minimum(setup_core) -> None:
+    """A timeout shorter than a few loop passes cannot work."""
+    uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "50ms"}})
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "40ms"}})
+
+
+def test_ack_delay_shorter_than_timeout(setup_core) -> None:
+    """The acknowledgement must not wait longer than the peer's timeout."""
+    reliability = uart_packet_interface.CONFIG_SCHEMA(
+        {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "100ms"}}
+    )[CONF_RELIABILITY]
+    assert reliability["ack_delay"].total_milliseconds == 100
+    with pytest.raises(cv.Invalid, match="shorter"):
+        uart_packet_interface.CONFIG_SCHEMA(
+            {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "500ms"}}
+        )
+
+
+def test_fragmentation_defaults(setup_core) -> None:
+    """An empty fragmentation key joins packets of up to 1024 bytes."""
+    fragmentation = uart_packet_interface.CONFIG_SCHEMA({CONF_FRAGMENTATION: None})[
+        CONF_FRAGMENTATION
+    ]
+    assert fragmentation["type"] == "basic"
+    assert fragmentation["max_packet_size"] == 1024
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA(
+            {CONF_FRAGMENTATION: {"max_packet_size": 8}}
+        )
+
+
 @pytest.mark.parametrize(
     ("config", "loads"),
     [
         ({}, []),
-        ({CONF_RELIABILITY: None, CONF_SESSION: None}, []),
-        ({CONF_SESSION: {"binary_sensor": {"name": "Link"}}}, ["binary_sensor"]),
+        ({CONF_RELIABILITY: None, "session": None}, []),
+        ({"session": {"binary_sensor": {"name": "Link"}}}, ["binary_sensor"]),
         ({CONF_ENCRYPTION: "secret"}, ["xxtea"]),
+        (
+            {
+                CONF_ADDRESSING: {
+                    "address": 1,
+                    "peers": [_peer("a", 2, encryption="secret"), _peer("b", 3)],
+                }
+            },
+            ["xxtea"],
+        ),
     ],
 )
 def test_auto_load_follows_layers(