[espnow] Warn now and then when a packet interface link's sends keep failing

A failure that repeats on every send, such as a peer that could not be
added, was logged only at verbose level. The error name helper is now
shared so the link can name the error.

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

CODEOWNERS

@@ -213,6 +213,7 @@ esphome/components/esp_ldo/* @clydebarrow
 esphome/components/espectre/* @francescopace
 esphome/components/esphome/ota/* @esphome/core
 esphome/components/espnow/* @jesserockz
+esphome/components/espnow/packet_interface/* @clydebarrow
 esphome/components/espnow/packet_transport/* @EasilyBoredEngineer
 esphome/components/ethernet_info/* @gtjadsonsantos
 esphome/components/event/* @nohat
@@ -630,7 +631,9 @@ esphome/components/sun_gtil2/* @Mat931
 esphome/components/switch/* @esphome/core
 esphome/components/switch/binary_sensor/* @ssieb
 esphome/components/sx126x/* @swoboda1337
+esphome/components/sx126x/packet_interface/* @clydebarrow
 esphome/components/sx127x/* @swoboda1337
+esphome/components/sx127x/packet_interface/* @clydebarrow
 esphome/components/sy6970/* @linkedupbits
 esphome/components/syslog/* @clydebarrow
 esphome/components/systa_bus/* @Mat931
@@ -642,6 +645,8 @@ esphome/components/tca8418/* @zebble
 esphome/components/tca9548a/* @andreashergert1984
 esphome/components/tca9555/* @mobrembski
 esphome/components/tcl112/* @glmnet
+esphome/components/tcp/* @clydebarrow
+esphome/components/tcp/packet_interface/* @clydebarrow
 esphome/components/tcp_uart/* @Bascht74
 esphome/components/tee501/* @Stock-M
 esphome/components/teleinfo/* @0hax @FredM67

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_has_layer,
+)
 from esphome.components.packet_interface.layers import CONF_SESSION
 from esphome.components.sensor import Sensor
 import esphome.config_validation as cv
@@ -98,17 +102,21 @@ CONFIG_SCHEMA = cv.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:
+    if CONF_TIMEOUT in config and interface_has_layer(
+        config[CONF_PACKET_INTERFACE], CONF_SESSION
+    ):
         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",
+            f"Not allowed with a '{CONF_SESSION}' layer, or a link that does its job, 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
+FINAL_VALIDATE_SCHEMA = cv.All(
+    final_validate_interface(CONF_PACKET_INTERFACE), _final_validate
+)
 
 
 def _validate_remote_id(config: ConfigType) -> ConfigType:

esphome/components/espnow/espnow_component.cpp

@@ -24,7 +24,7 @@ ESPHOME_LOG_TAG(TAG, "espnow");
 
 ESPNowComponent *global_esp_now = nullptr;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
 
-static const LogString *espnow_error_to_str(esp_err_t error) {
+const LogString *espnow_error_to_str(esp_err_t error) {
   switch (error) {
     case ESP_ERR_ESPNOW_FAILED:
       return LOG_STR("ESPNow is in fail mode");

esphome/components/espnow/espnow_component.h

@@ -25,6 +25,9 @@ namespace esphome::espnow {
 
 using peer_address_t = std::array<uint8_t, ESP_NOW_ETH_ALEN>;
 
+/// A short description of an ESP-NOW error code, for logging.
+const LogString *espnow_error_to_str(esp_err_t error);
+
 enum class ESPNowTriggers : uint8_t {
   TRIGGER_NONE = 0,
   ON_NEW_PEER = 1,
@@ -123,6 +126,7 @@ class ESPNowComponent final : public Component {
   uint8_t get_wifi_channel();
 
   void set_auto_add_peer(bool value) { this->auto_add_peer_ = value; }
+  bool get_auto_add_peer() const { return this->auto_add_peer_; }
 
 #if defined(USE_WIFI) && defined(USE_WIFI_CONNECT_STATE_LISTENERS)
   // WiFiConnectStateListener interface: refresh the cached channel after each (re)connect

esphome/components/espnow/packet_interface/__init__.py

@@ -0,0 +1,90 @@
+import esphome.codegen as cg
+from esphome.components import packet_interface
+from esphome.components.packet_interface import (
+    CONF_MAX_PACKET_SIZE,
+    new_packet_interface,
+    packet_interface_schema,
+)
+from esphome.components.packet_interface.layers import (
+    CONF_ADDRESSING,
+    check_broadcast_link,
+)
+import esphome.config_validation as cv
+from esphome.const import CONF_ID
+from esphome.core import HexInt
+import esphome.final_validate as fv
+from esphome.types import ConfigType
+
+from .. import (
+    CONF_MAX_PAYLOAD_SIZE,
+    DOMAIN as ESPNOW_DOMAIN,
+    ESPNowComponent,
+    espnow_ns,
+)
+
+CODEOWNERS = ["@clydebarrow"]
+DEPENDENCIES = ["espnow"]
+
+CONF_ESPNOW_ID = "espnow_id"
+CONF_PEER_ADDRESS = "peer_address"
+
+# The ESP-NOW v1 payload; a larger max_payload_size on the espnow component allows a larger max_packet_size
+MAX_PACKET_SIZE = 250
+BROADCAST_MAC = "FF:FF:FF:FF:FF:FF"
+
+ESPNowPacketInterface = espnow_ns.class_(
+    "ESPNowPacketInterface", packet_interface.PacketInterface
+)
+
+
+def _is_broadcast(config: ConfigType) -> bool:
+    return str(config[CONF_PEER_ADDRESS]) == BROADCAST_MAC
+
+
+def _validate_addressing(config: ConfigType) -> ConfigType:
+    if _is_broadcast(config):
+        check_broadcast_link(config)
+    elif CONF_ADDRESSING in config:
+        raise cv.Invalid(
+            f"Packets to one '{CONF_PEER_ADDRESS}' are already addressed, so '{CONF_ADDRESSING}' is only "
+            "for broadcast",
+            path=[CONF_ADDRESSING],
+        )
+    return config
+
+
+CONFIG_SCHEMA = cv.All(
+    packet_interface_schema(ESPNowPacketInterface, MAX_PACKET_SIZE, 1470).extend(
+        {
+            cv.GenerateID(CONF_ESPNOW_ID): cv.use_id(ESPNowComponent),
+            cv.Optional(CONF_PEER_ADDRESS, default=BROADCAST_MAC): cv.mac_address,
+        }
+    ),
+    _validate_addressing,
+)
+
+
+def _validate_packet_size(config: ConfigType) -> ConfigType:
+    payload = fv.full_config.get()[ESPNOW_DOMAIN][CONF_MAX_PAYLOAD_SIZE]
+    if config[CONF_MAX_PACKET_SIZE] > payload:
+        raise cv.Invalid(
+            f"Must not be larger than the espnow {CONF_MAX_PAYLOAD_SIZE} of {payload}",
+            path=[CONF_MAX_PACKET_SIZE],
+        )
+    return config
+
+
+FINAL_VALIDATE_SCHEMA = _validate_packet_size
+
+
+async def to_code(config: ConfigType) -> None:
+    parent = await cg.get_variable(config[CONF_ESPNOW_ID])
+    peer = [HexInt(x) for x in config[CONF_PEER_ADDRESS].parts]
+    if not _is_broadcast(config):
+        cg.add(parent.add_peer(peer))
+    # The copy buffer's size is known here, so it is a static array rather than allocated at run time
+    tx_buffer = f"{config[CONF_ID].id}__tx_buffer"
+    cg.add_global(
+        cg.RawStatement(f"static uint8_t {tx_buffer}[{config[CONF_MAX_PACKET_SIZE]}];")
+    )
+    await new_packet_interface(config, parent, cg.RawExpression(tx_buffer), peer)

esphome/components/espnow/packet_interface/espnow_packet_interface.cpp

@@ -0,0 +1,125 @@
+#include "espnow_packet_interface.h"
+
+#if defined(USE_ESP32) || defined(USE_ESP8266)
+
+#include <cstring>
+
+#include "esphome/core/application.h"
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
+
+namespace esphome::espnow {
+
+using packet_interface::PacketBuffer;
+using packet_interface::PacketMetaData;
+
+ESPHOME_LOG_TAG(TAG, "espnow.packet_interface");
+
+ESPNowPacketInterface::ESPNowPacketInterface(ESPNowComponent *parent, std::span<uint8_t> tx_buffer,
+                                             peer_address_t peer_address)
+    : packet_interface::PacketInterface(tx_buffer.size()),
+      parent_(parent),
+      peer_address_(peer_address),
+      tx_buffer_(tx_buffer) {
+  if (!this->is_broadcast_())
+    this->capabilities_ = packet_interface::CAPABILITY_ADDRESSING;
+}
+
+void ESPNowPacketInterface::setup() {
+  // Each mode hears only its own kind of packet, so a peer link and a broadcast link can share the component
+  if (!this->is_broadcast_() || !ESPNOW_REPORTS_DESTINATION)
+    this->parent_->register_receive_handler(this);
+  if (this->is_broadcast_()) {
+    this->parent_->register_broadcast_handler(this);
+    this->parent_->register_unknown_peer_handler(this);
+  }
+}
+
+void ESPNowPacketInterface::dump_config() {
+  char peer_buf[MAC_ADDRESS_PRETTY_BUFFER_SIZE];
+  format_mac_addr_upper(this->peer_address_.data(), peer_buf);
+  ESP_LOGCONFIG(TAG,
+                "ESP-NOW Packet Interface '%s':\n"
+                "  Peer: %s\n"
+                "  Max Packet Size: %zu",
+                this->name_, peer_buf, this->max_packet_size_);
+}
+
+bool ESPNowPacketInterface::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  // Layers above are set up first, and may send while setup waits for another component
+  if (!this->is_ready())
+    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;
+  }
+  const uint8_t *payload;
+  if (data.is_contiguous()) {
+    payload = data.as_span().data();
+  } else {
+    data.copy_to(this->tx_buffer_.data(), size);
+    payload = this->tx_buffer_.data();
+  }
+  esp_err_t err = this->parent_->send(this->peer_address_.data(), payload, size);
+  if (err != ESP_OK) {
+    this->send_failed_(err);
+    return false;
+  }
+  return true;
+}
+
+// A full send queue is logged by the espnow component, and ESP-NOW is not running before setup or while disabled.
+// Other errors, such as a peer that could not be added, repeat on every send, so warn now and then
+void ESPNowPacketInterface::send_failed_(esp_err_t err) {
+  static constexpr uint32_t WARNING_INTERVAL_MS = 60000;
+  uint32_t now = App.get_loop_component_start_time();
+  if (err == ESP_ERR_ESPNOW_NO_MEM || err == ESP_ERR_ESPNOW_NOT_INIT ||
+      (this->send_warned_ && now - this->last_send_warning_ms_ < WARNING_INTERVAL_MS)) {
+    ESP_LOGV(TAG, "%s: Send failed: %s", this->name_, LOG_STR_ARG(espnow_error_to_str(err)));
+    return;
+  }
+  this->send_warned_ = true;
+  this->last_send_warning_ms_ = now;
+  ESP_LOGW(TAG, "%s: Send failed: %s", this->name_, LOG_STR_ARG(espnow_error_to_str(err)));
+}
+
+bool ESPNowPacketInterface::on_receive(const ESPNowRecvInfo &info, const uint8_t *data, uint16_t size) {
+  this->receive_(info, data, size);
+  return false;
+}
+
+bool ESPNowPacketInterface::on_broadcast(const ESPNowRecvInfo &info, const uint8_t *data, uint16_t size) {
+  this->receive_(info, data, size);
+  return false;
+}
+
+bool ESPNowPacketInterface::on_unknown_peer(const ESPNowRecvInfo &info, const uint8_t *data, uint16_t size) {
+  // Only known peers reach the other handlers; with auto_add_peer this sender is added and arrives there
+  if (this->parent_->get_auto_add_peer())
+    return false;
+  if (!ESPNOW_REPORTS_DESTINATION || memcmp(info.des_addr, ESPNOW_BROADCAST_ADDR, ESP_NOW_ETH_ALEN) == 0)
+    this->receive_(info, data, size);
+  return false;
+}
+
+void ESPNowPacketInterface::receive_(const ESPNowRecvInfo &info, const uint8_t *data, uint16_t size) {
+  if (size == 0)
+    return;
+  if (size > this->max_packet_size_) {
+    ESP_LOGV(TAG, "%s: Packet of %u bytes is larger than %zu, dropped", this->name_, size, this->max_packet_size_);
+    return;
+  }
+  if (!this->is_broadcast_() && memcmp(info.src_addr, this->peer_address_.data(), ESP_NOW_ETH_ALEN) != 0)
+    return;
+  PacketMetaData meta_data;
+  memcpy(meta_data.mac_address, info.src_addr, ESP_NOW_ETH_ALEN);
+#ifdef USE_ESP32
+  meta_data.rssi = info.rx_ctrl->rssi;
+#endif
+  this->on_receive_from_interface_(PacketBuffer(data, size), meta_data);
+}
+
+}  // namespace esphome::espnow
+
+#endif  // USE_ESP32 || USE_ESP8266

esphome/components/espnow/packet_interface/espnow_packet_interface.h

@@ -0,0 +1,59 @@
+#pragma once
+
+#include "../espnow_component.h"
+
+#if defined(USE_ESP32) || defined(USE_ESP8266)
+
+#include <cstring>
+#include <span>
+
+#include "esphome/components/packet_interface/packet_interface.h"
+
+namespace esphome::espnow {
+
+/**
+ * A packet interface link over ESP-NOW.
+ *
+ * With a peer address other than broadcast, packets go to that device only, and only packets sent to this device by
+ * it are received, so the link provides the addressing capability itself. With the broadcast address, packets go to
+ * every device on the channel and every broadcast packet is received, from known and unknown senders; an addressing
+ * layer then tells the devices apart. ESP8266 does not report a packet's destination, so there a broadcast link also
+ * receives packets sent to this device, and a peer link also receives the peer's broadcasts.
+ *
+ * ESP-NOW queues each packet and reports later whether it was delivered; a reliability layer above resends what is
+ * lost.
+ */
+class ESPNowPacketInterface : public packet_interface::PacketInterface,
+                              public ESPNowReceivedPacketHandler,
+                              public ESPNowBroadcastHandler,
+                              public ESPNowUnknownPeerHandler {
+ public:
+  /// @param tx_buffer Joins packets split across buffers; its size is the largest packet, and it must outlive the link
+  ESPNowPacketInterface(ESPNowComponent *parent, std::span<uint8_t> tx_buffer, peer_address_t peer_address);
+
+  void setup() override;
+  void dump_config() override;
+  bool send_to_interface(const packet_interface::PacketBuffer &data,
+                         const packet_interface::PacketMetaData &meta_data) override;
+
+  bool on_receive(const ESPNowRecvInfo &info, const uint8_t *data, uint16_t size) override;
+  bool on_broadcast(const ESPNowRecvInfo &info, const uint8_t *data, uint16_t size) override;
+  bool on_unknown_peer(const ESPNowRecvInfo &info, const uint8_t *data, uint16_t size) override;
+
+ protected:
+  bool is_broadcast_() const {
+    return memcmp(this->peer_address_.data(), ESPNOW_BROADCAST_ADDR, ESP_NOW_ETH_ALEN) == 0;
+  }
+  void receive_(const ESPNowRecvInfo &info, const uint8_t *data, uint16_t size);
+  void send_failed_(esp_err_t err);
+
+  ESPNowComponent *parent_;
+  peer_address_t peer_address_;
+  const std::span<uint8_t> tx_buffer_;
+  uint32_t last_send_warning_ms_{0};
+  bool send_warned_{false};
+};
+
+}  // namespace esphome::espnow
+
+#endif  // USE_ESP32 || USE_ESP8266

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,19 @@ from esphome.types import ConfigType
 
 from . import layers
 from .layers import (
+    ADDRESSING,
+    BROADCAST_ADDRESS,
+    CONF_ADDRESSING,
+    CONF_DESTINATION,
+    CONF_PEERS,
+    CONF_SESSION,
+    STACK_ORDER,
     PacketInterface,
     build_stack,
+    check_packet_sizes,
     check_session,
+    check_single_peer,
+    layer_needing_session,
     packet_interface_ns,
     stack_auto_load,
     stack_schema,
@@ -31,13 +42,21 @@ 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()
+# The layer keys that each link class does itself, by class name
+_PROVIDED_LAYERS: dict[str, tuple[str, ...]] = {}
+
+LINK_LAYER_SCHEMAS = {
+    cv.Optional(CONF_ADDRESSING): ADDRESSING.schema(),
+    **stack_schema(),
+}
 
 __all__ = [
     "CONF_MAX_PACKET_SIZE",
     "CONF_PACKET_INTERFACE",
+    "interface_has_layer",
     "interface_layers",
     "PacketInterface",
+    "final_validate_interface",
     "new_packet_interface",
     "packet_interface_ns",
     "packet_interface_schema",
@@ -45,7 +64,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,41 +73,121 @@ 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)
+    size = config[CONF_MAX_PACKET_SIZE]
+    if (addressing := config.get(CONF_ADDRESSING)) is None:
+        check_packet_sizes(config, size)
+        return config
+    size = ADDRESSING.get_type(addressing).packet_size(addressing, size)
+    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],
+                )
+        for index, peer in enumerate(addressing[CONF_PEERS]):
+            try:
+                check_packet_sizes(peer, size)
+            except cv.Invalid as err:
+                err.prepend([CONF_ADDRESSING, CONF_PEERS, index])
+                raise
+        return config
+    addressing.setdefault(CONF_DESTINATION, BROADCAST_ADDRESS)
+    check_single_peer(config, addressing[CONF_DESTINATION])
+    check_packet_sizes(config, size)
     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 interface_has_layer(interface_id: ID, key: str) -> bool:
+    """During final validation, whether the packet interface with this id has the layer, configured or done by its link."""
+    interface = interface_layers(interface_id)
+    if interface is None:
+        return False
+    return key in interface or key in _PROVIDED_LAYERS.get(
+        str(interface[CONF_ID].type), ()
+    )
+
+
+def final_validate_interface(key: str) -> Callable[[ConfigType], ConfigType]:
+    """Final validation for an application: a link with peers must be used through one of the peers."""
+
+    def validator(config: ConfigType) -> ConfigType:
+        interface_id: ID = config[key]
+        interface = interface_layers(interface_id)
+        if interface is not None and (
+            peers := interface.get(CONF_ADDRESSING, {}).get(CONF_PEERS)
+        ):
+            raise cv.Invalid(
+                f"'{interface_id.id}' has peers, so use the id of one of them, such as '{peers[0][CONF_ID].id}'",
+                path=[key],
+            )
+        return config
+
+    return validator
+
+
+def _check_provided_session(config: ConfigType) -> ConfigType:
+    """A link's own session does not send packets through the layers, so none can need them from the peer."""
+    if key := layer_needing_session(config):
+        raise cv.Invalid(
+            f"This '{key}' needs packets from the peer, which this link does not send through its layers, "
+            "so it cannot be used on this link",
+            path=[key],
+        )
+    return config
+
+
 def packet_interface_schema(
     class_: MockObjClass,
     default_max_packet_size: int,
     max_packet_size_limit: int = 4096,
+    provides: tuple[str, ...] = (),
 ) -> cv.Schema:
-    """Schema for a link platform, with the layers that can be stacked on it."""
-    return (
-        cv.Schema(
-            {
-                cv.GenerateID(): cv.declare_id(class_),
-                cv.Optional(
-                    CONF_MAX_PACKET_SIZE, default=default_max_packet_size
-                ): cv.int_range(16, max_packet_size_limit),
-                **LINK_LAYER_SCHEMAS,
-            }
+    """Schema for a link platform; provides names the layer keys, such as reliability, that the link does itself."""
+    if provides:
+        _PROVIDED_LAYERS[str(class_)] = provides
+    link_layers = {
+        key: (
+            cv.invalid(f"This link provides '{key}' itself, so it cannot be configured")
+            if key in provides
+            else schema
         )
-        .extend(cv.COMPONENT_SCHEMA)
-        .add_extra(_validate_layers)
-    )
+        for key, schema in LINK_LAYER_SCHEMAS.items()
+    }
+    schema = cv.Schema(
+        {
+            cv.GenerateID(): cv.declare_id(class_),
+            cv.Optional(
+                CONF_MAX_PACKET_SIZE, default=default_max_packet_size
+            ): cv.int_range(16, max_packet_size_limit),
+            **link_layers,
+        }
+    ).extend(cv.COMPONENT_SCHEMA)
+    if CONF_SESSION in provides:
+        schema = schema.add_extra(_check_provided_session)
+    return schema.add_extra(_validate_layers)
 
 
 async def new_packet_interface(config: ConfigType, *args: Any) -> MockObj:
@@ -97,10 +196,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,97 @@
+#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) {
+    ESP_LOGV(TAG, "%s: Packet of %zu bytes is shorter than the address header, dropped", this->link_->get_name(),
+             packet.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;
+  AddressingPort *source_port = nullptr;
+  AddressingPort *broadcast_port = nullptr;
+  for (auto *port : this->ports_) {
+    if (port->destination_ == source) {
+      source_port = port;
+    } else if (port->destination_ == BROADCAST_ADDRESS) {
+      broadcast_port = port;
+    }
+  }
+  // A broadcast goes to the broadcast port, even from a device with a port of its own, as that port's layers expect
+  // only its point-to-point packets; anything else goes to the source's port, or to the broadcast port if it has none
+  AddressingPort *target = source_port;
+  if (broadcast_port != nullptr && (destination == BROADCAST_ADDRESS || source_port == nullptr))
+    target = broadcast_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)
+    : PacketInterface(
+          remaining_packet_size(addressing->get_link()->get_max_packet_size(), BasicAddressing::HEADER_SIZE)),
+      addressing_(addressing),
+      destination_(destination) {
+  if (only_port) {
+    this->stack_on_(addressing->get_link(), CAPABILITY_ADDRESSING);
+  } else {
+    this->inherit_from_(addressing->get_link(), CAPABILITY_ADDRESSING);
+  }
+  // A link with no room leaves the layers above none either, so they fail their setup
+  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,69 @@
+#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 every broadcast, as well as packets from devices
+ * that have no port of their own. Without a broadcast port, a device's broadcasts go to its own port.
+ */
+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,160 @@
+#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)
+    : PacketInterface(max_packet_size), lower_(lower) {
+  this->stack_on_(lower, CAPABILITY_FRAGMENTATION);
+}
+
+size_t BasicFragmentation::fragments_for_(size_t size) const {
+  size_t fragment_size = this->fragment_size_();
+  if (fragment_size == 0)
+    return SIZE_MAX;
+  return std::max<size_t>(1, (size + fragment_size - 1) / fragment_size);
+}
+
+void BasicFragmentation::setup() {
+  if (this->fragment_size_() == 0) {
+    this->mark_failed(LOG_STR("max_packet_size too small for the configured layers"));
+    return;
+  }
+  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; a failed layer below reports its own failure instead
+  this->queue_too_short_ =
+      !this->lower_->is_failed() && this->fragments_for_(this->max_packet_size_) > this->lower_->get_send_capacity();
+  RAMAllocator<uint8_t> allocator;
+  this->tx_buffer_ = allocator.allocate(this->fragment_size_() + this->max_packet_size_);
+  if (this->tx_buffer_ == nullptr) {
+    this->mark_failed(LOG_STR("Not enough memory for the packet buffers"));
+    return;
+  }
+  this->rx_buffer_ = this->tx_buffer_ + this->fragment_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() {
+  // A link with no room for fragments fails setup, which core reports below
+  size_t fragments = this->fragment_size_() == 0 ? 0 : 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->tx_buffer_ == nullptr)
+    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_, fragment_size, index * fragment_size);
+    header[1] = index | (index + 1 == fragments ? LAST_FRAGMENT : 0);
+    PacketBuffer body(this->tx_buffer_, 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) {
+    ESP_LOGV(TAG, "%s: Fragment of %zu bytes is shorter than its header, dropped", this->name_, fragment.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 && last) {
+    // A packet in one fragment leaves a packet being joined from another device intact
+    if (this->rx_joining_ && meta_data.source_address == this->rx_source_) {
+      ESP_LOGV(TAG, "%s: Incomplete packet %u dropped", this->name_, this->rx_number_);
+      this->rx_joining_ = false;
+    }
+    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;
+  }
+  if (index == 0) {
+    if (this->rx_joining_) {
+      ESP_LOGV(TAG, "%s: Incomplete packet %u dropped", this->name_, this->rx_number_);
+    }
+    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_ && meta_data.source_address != this->rx_source_) {
+    // Only this fragment is lost, so of two devices sending at once, the one that started last still gets through
+    ESP_LOGV(TAG, "%s: Fragment %u from 0x%04X dropped", this->name_, index, meta_data.source_address);
+    return;
+  } else if (!this->rx_joining_ || number != this->rx_number_ || 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_ + 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_, this->rx_length_), meta_data);
+  }
+}
+
+}  // namespace esphome::packet_interface
+#endif

esphome/components/packet_interface/basic_fragmentation.h

@@ -0,0 +1,60 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+#ifdef USE_PACKET_INTERFACE_BASIC_FRAGMENTATION
+
+#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 arrives with another packet number, or when the same source, or another one,
+ * starts a new packet of several fragments. Other fragments from another source address are dropped, leaving the
+ * packet being joined intact. 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 {
+    size_t lower_size = this->lower_->get_max_packet_size();
+    return lower_size > HEADER_SIZE ? lower_size - HEADER_SIZE : 0;
+  }
+  size_t fragments_for_(size_t size) const;
+
+  PacketInterface *lower_;
+  // One fragment being sent, and the packet being joined; one block allocated in setup(), and tx_buffer_ stays null
+  // if that fails
+  uint8_t *tx_buffer_{nullptr};
+  uint8_t *rx_buffer_{nullptr};
+  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]]
 
 
@@ -38,16 +44,37 @@ def _nothing(config: ConfigType) -> list[str]:
     return []
 
 
+def _same_size(config: ConfigType, size: int) -> int:
+    return size
+
+
+def _no_limit(config: ConfigType) -> int | None:
+    return None
+
+
+def _fits(config: ConfigType, size: int, capacity: int | None) -> None:
+    """Accept any layer below."""
+
+
 @dataclass(frozen=True)
 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
     auto_load: Callable[[ConfigType], list[str]]
+    # The largest packet this configuration takes, from the largest that the layer below takes, as the C++ works it out
+    packet_size: Callable[[ConfigType, int], int]
+    # How many packets this configuration takes at once, or None for no limit
+    send_capacity: Callable[[ConfigType], int | None]
+    # Raises cv.Invalid if this configuration cannot work above a layer with this packet size and send capacity
+    check_lower: Callable[[ConfigType, int, int | None], None]
 
 
 class LayerKind:
@@ -68,11 +95,19 @@ 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,
+        packet_size: Callable[[ConfigType, int], int] = _same_size,
+        send_capacity: Callable[[ConfigType], int | None] = _no_limit,
+        check_lower: Callable[[ConfigType, int, int | None], None] = _fits,
     ) -> 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 +119,16 @@ 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,
+                packet_size,
+                send_capacity,
+                check_lower,
+            )
             return build
 
         return decorator
@@ -103,7 +147,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 +156,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:
@@ -124,17 +172,72 @@ def stack_schema() -> dict:
     return {cv.Optional(kind.key): kind.schema() for kind in STACK_ORDER}
 
 
-def check_session(config: ConfigType) -> None:
-    """Layers that need packets back from the peer need a session layer, whose keep-alives send them."""
-    if CONF_SESSION in config:
-        return
+def layer_needing_session(config: ConfigType) -> str | None:
+    """The key of the first layer in this stack that needs packets back from the peer, if any."""
     for kind in STACK_ORDER:
         if (layer := config.get(kind.key)) is not None and kind.get_type(
             layer
         ).needs_session(layer):
+            return kind.key
+    return None
+
+
+def check_session(config: ConfigType) -> None:
+    """Layers that need packets back from the peer need a session layer, whose keep-alives send them."""
+    if CONF_SESSION not in config and (key := layer_needing_session(config)):
+        raise cv.Invalid(
+            f"This '{key}' needs packets from the peer, so it needs a '{CONF_SESSION}' layer, whose "
+            "keep-alives make sure they are sent",
+            path=[key],
+        )
+
+
+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}' needs packets from the peer, so it needs a '{CONF_SESSION}' layer, whose "
-                "keep-alives make sure they are sent",
+                f"This '{kind.key}' works between two devices only, so it needs a '{CONF_DESTINATION}' other than "
+                "broadcast",
+                path=[kind.key],
+            )
+
+
+def remaining_packet_size(size: int, used: int) -> int:
+    return max(size - used, 0)
+
+
+def check_packet_sizes(config: ConfigType, size: int) -> None:
+    """Check each layer in config against the layer below it; size is the largest packet that the stack sits on takes."""
+    capacity = None
+    for kind in STACK_ORDER:
+        if (layer := config.get(kind.key)) is None:
+            continue
+        layer_type = kind.get_type(layer)
+        try:
+            layer_type.check_lower(layer, size, capacity)
+        except cv.Invalid as err:
+            err.prepend([kind.key])
+            raise
+        size = layer_type.packet_size(layer, size)
+        capacity = layer_type.send_capacity(layer)
+
+
+def check_broadcast_link(config: ConfigType) -> None:
+    """On a link that every device hears, layers that keep state for one peer need an addressing layer."""
+    if CONF_ADDRESSING in config:
+        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, but every device hears this link, so it needs "
+                f"'{CONF_ADDRESSING}' with a '{CONF_DESTINATION}'",
                 path=[kind.key],
             )
 

esphome/components/packet_interface/layers/basic_addressing.py

@@ -0,0 +1,132 @@
+"""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,
+    remaining_packet_size,
+    stack_schema,
+)
+
+CONF_PORT_ID = "port_id"
+
+# The destination and source addresses
+HEADER_SIZE = 4
+
+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
+
+
+def _not_own_address(config: ConfigType) -> ConfigType:
+    """Packets from this device's own address are dropped as echoes, so it cannot be a destination."""
+    address = config[CONF_ADDRESS]
+    if config.get(CONF_DESTINATION) == address:
+        raise cv.Invalid(
+            "Must not be this device's own address", path=[CONF_DESTINATION]
+        )
+    for index, peer in enumerate(config.get(CONF_PEERS, [])):
+        if peer[CONF_DESTINATION] == address:
+            raise cv.Invalid(
+                "Must not be this device's own address",
+                path=[CONF_PEERS, index, CONF_DESTINATION],
+            )
+    return config
+
+
+# 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),
+        _not_own_address,
+    )
+
+
+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,
+    packet_size=lambda config, size: remaining_packet_size(size, HEADER_SIZE),
+)
+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,56 @@
+"""Fragmentation layer that splits packets larger than the layers below allow, and joins them again."""
+
+import math
+
+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.cpp_generator import MockObj
+from esphome.types import ConfigType
+
+from . import FRAGMENTATION, PacketInterface, packet_interface_ns
+
+# The packet number, and the fragment's index with the flag on the last fragment
+HEADER_SIZE = 2
+MAX_FRAGMENTS = 128
+
+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)
+
+
+def _check_lower(config: ConfigType, size: int, capacity: int | None) -> None:
+    """The largest packet must fit in the fragments that the layer below can take at once."""
+    fragment_size = size - HEADER_SIZE
+    if fragment_size <= 0:
+        raise cv.Invalid("The layers below leave no room for fragments")
+    fragments = max(1, math.ceil(config[CONF_MAX_PACKET_SIZE] / fragment_size))
+    limit = MAX_FRAGMENTS if capacity is None else min(MAX_FRAGMENTS, capacity)
+    if fragments > limit:
+        raise cv.Invalid(
+            f"A packet of {config[CONF_MAX_PACKET_SIZE]} bytes needs {fragments} fragments of {fragment_size} bytes, "
+            f"but at most {limit} can be sent at once",
+            path=[CONF_MAX_PACKET_SIZE],
+        )
+
+
+# Fragments from different devices are told apart by the addressing layer's source address, so of two devices sending
+# fragmented packets at once, one still gets through; without addressing, the receiver can drop both
+@FRAGMENTATION.register(
+    "basic",
+    schema=BASIC_FRAGMENTATION_SCHEMA,
+    packet_size=lambda config, size: config[CONF_MAX_PACKET_SIZE],
+    check_lower=_check_lower,
+)
+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

@@ -6,12 +6,15 @@ from esphome.const import CONF_ID, CONF_TIMEOUT
 from esphome.cpp_generator import MockObj
 from esphome.types import ConfigType
 
-from . import RELIABILITY, PacketInterface, packet_interface_ns
+from . import RELIABILITY, PacketInterface, packet_interface_ns, remaining_packet_size
 
 CONF_ACK_DELAY = "ack_delay"
 CONF_MAX_QUEUE_LENGTH = "max_queue_length"
 CONF_MAX_RETRIES = "max_retries"
 
+# The control byte and the CRC-16
+FRAME_OVERHEAD = 3
+
 HdlcReliability = packet_interface_ns.class_("HdlcReliability", PacketInterface)
 
 
@@ -47,7 +50,13 @@ HDLC_SCHEMA = cv.All(
 )
 
 
-@RELIABILITY.register("hdlc", schema=HDLC_SCHEMA)
+@RELIABILITY.register(
+    "hdlc",
+    schema=HDLC_SCHEMA,
+    single_peer=lambda config: True,
+    packet_size=lambda config, size: remaining_packet_size(size, FRAME_OVERHEAD),
+    send_capacity=lambda config: config[CONF_MAX_QUEUE_LENGTH],
+)
 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

@@ -8,7 +8,7 @@ from esphome.const import CONF_ID, CONF_KEY
 from esphome.cpp_generator import MockObj
 from esphome.types import ConfigType
 
-from . import ENCRYPTION, PacketInterface, packet_interface_ns
+from . import ENCRYPTION, PacketInterface, packet_interface_ns, remaining_packet_size
 
 CONF_ROLLING_CODE_ENABLE = "rolling_code_enable"
 
@@ -18,6 +18,10 @@ CONF_ROLLING_CODE_ENABLE = "rolling_code_enable"
 KEY_SALT = b"esphome packet_interface xxtea"
 KEY_ITERATIONS = 600_000
 
+# The check word, and with the rolling code the boot numbers and counter
+CHECK_SIZE = 4
+ROLLING_SIZE = 12
+
 XxteaEncryption = packet_interface_ns.class_("XxteaEncryption", PacketInterface)
 
 XXTEA_SCHEMA = cv.Schema(
@@ -35,12 +39,19 @@ def derive_key(text: str) -> list[int]:
     return [int.from_bytes(digest[i : i + 4], "little") for i in range(0, 32, 4)]
 
 
+# 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"],
+    # XXTEA works on whole 32-bit words
+    packet_size=lambda config, size: remaining_packet_size(
+        size & ~3,
+        CHECK_SIZE + (ROLLING_SIZE if config[CONF_ROLLING_CODE_ENABLE] else 0),
+    ),
 )
 async def build_xxtea(config: ConfigType, lower: MockObj) -> MockObj:
     cg.add_define("USE_PACKET_INTERFACE_XXTEA")

esphome/components/packet_interface/listen_before_talk.h

@@ -0,0 +1,45 @@
+#pragma once
+
+#include <cstdint>
+
+namespace esphome::packet_interface {
+
+/**
+ * Decides whether a half-duplex radio link may send while its radio reports that it is receiving a packet.
+ *
+ * Sending is refused while the radio is busy, so the packet being received is not lost. A busy period starts when a
+ * send first finds the radio busy, and ends when a send finds it free or end_busy() is called. Once a busy period has
+ * lasted max_busy_ms, sends go ahead, so a busy flag that never clears cannot stop sending for good, however often
+ * sends are tried. Code generation sets max_busy_ms from the time the radio takes to receive the largest packet.
+ */
+class ListenBeforeTalk {
+ public:
+  static constexpr uint32_t DEFAULT_MAX_BUSY_MS = 2000;
+
+  void set_max_busy(uint32_t max_busy_ms) { this->max_busy_ms_ = max_busy_ms; }
+  uint32_t get_max_busy() const { return this->max_busy_ms_; }
+
+  /// True if a packet may be sent at \p now; \p busy says whether the radio is receiving.
+  bool may_send(bool busy, uint32_t now) {
+    if (!busy) {
+      this->busy_ = false;
+      return true;
+    }
+    if (!this->busy_) {
+      this->busy_ = true;
+      this->busy_since_ms_ = now;
+    }
+    return now - this->busy_since_ms_ >= this->max_busy_ms_;
+  }
+  /// Ends the busy period, when the radio is seen to be free other than by a send.
+  void end_busy() { this->busy_ = false; }
+  /// How long the current busy period has lasted at \p now.
+  uint32_t busy_for(uint32_t now) const { return now - this->busy_since_ms_; }
+
+ protected:
+  uint32_t busy_since_ms_{0};
+  uint32_t max_busy_ms_{DEFAULT_MAX_BUSY_MS};
+  bool busy_{false};
+};
+
+}  // namespace esphome::packet_interface

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/components/packet_interface/radio_link.cpp

@@ -0,0 +1,54 @@
+#include "radio_link.h"
+#include <cinttypes>
+#include "esphome/core/application.h"
+#include "esphome/core/log.h"
+
+namespace esphome::packet_interface {
+
+ESPHOME_LOG_TAG(TAG, "packet_interface.radio");
+
+bool RadioLinkBase::check_size_(size_t size) const {
+  if (size <= this->max_packet_size_)
+    return true;
+  ESP_LOGW(TAG, "Packet of %zu bytes is larger than %zu", size, this->max_packet_size_);
+  return false;
+}
+
+bool RadioLinkBase::may_send_(bool busy) {
+  uint32_t now = App.get_loop_component_start_time();
+  if (busy)
+    this->enable_loop();
+  if (!this->listen_before_talk_.may_send(busy, now))
+    return false;
+  if (busy) {
+    // The limit is longer than any packet takes, so the radio's busy indication is stuck or the channel is jammed
+    ESP_LOGW(TAG, "Radio busy for %" PRIu32 "ms, sending anyway", this->listen_before_talk_.busy_for(now));
+  }
+  return true;
+}
+
+bool RadioLinkBase::skip_empty_() const {
+  ESP_LOGV(TAG, "Empty packet not sent");
+  return true;
+}
+
+void RadioLinkBase::radio_not_ready_(bool failed) {
+  if (!failed || this->failed_warned_)
+    return;
+  this->failed_warned_ = true;
+  ESP_LOGW(TAG, "%s: The radio failed to set up, so nothing can be sent", this->name_);
+}
+
+void RadioLinkBase::radio_free_() {
+  this->listen_before_talk_.end_busy();
+  this->disable_loop();
+}
+
+void RadioLinkBase::receive_(const std::vector<uint8_t> &packet, float rssi, float snr) {
+  PacketMetaData meta_data;
+  meta_data.rssi = rssi;
+  meta_data.snr = snr;
+  this->on_receive_from_interface_(PacketBuffer(packet.data(), packet.size()), meta_data);
+}
+
+}  // namespace esphome::packet_interface

esphome/components/packet_interface/radio_link.h

@@ -0,0 +1,87 @@
+#pragma once
+
+#include <vector>
+
+#include "esphome/components/packet_interface/listen_before_talk.h"
+#include "esphome/components/packet_interface/packet_interface.h"
+
+namespace esphome::packet_interface {
+
+/// The parts of RadioLink that do not depend on the radio.
+class RadioLinkBase : public PacketInterface {
+ public:
+  void set_max_busy(uint32_t max_busy_ms) { this->listen_before_talk_.set_max_busy(max_busy_ms); }
+
+ protected:
+  explicit RadioLinkBase(size_t max_packet_size) : PacketInterface(max_packet_size) {}
+  /// False if a packet of this size is too large.
+  bool check_size_(size_t size) const;
+  /// False if sending must wait, as the radio is busy; starts loop() to watch the radio.
+  bool may_send_(bool busy);
+  /// Returns true for an empty packet, which the radios cannot send, so there is nothing to retry.
+  bool skip_empty_() const;
+  /// Warns once when a send finds the radio failed; until its setup has run, sends are refused quietly.
+  void radio_not_ready_(bool failed);
+  /// Ends the busy period when loop() finds the radio free.
+  void radio_free_();
+  void receive_(const std::vector<uint8_t> &packet, float rssi, float snr);
+
+  ListenBeforeTalk listen_before_talk_;
+  bool failed_warned_{false};
+};
+
+/**
+ * A packet interface link over a packet radio: each packet is one radio packet, with the RSSI and SNR of received
+ * packets in the metadata. Sending waits for the radio to finish transmitting. A radio cannot hear while it sends, so
+ * sending is refused while the radio is receiving a packet. A reliability layer sends the packet again shortly, and
+ * entity_transport after a delay; other callers drop it, logging at debug level at most. While a busy period lasts,
+ * loop() watches the radio and ends the period as soon as the radio is free, so a later reception starts a new one.
+ *
+ * Radio provides is_receiving(), register_listener(Listener *) and transmit_packet(const uint8_t *, size_t), which
+ * returns Error::NONE on success. Listener declares on_packet(packet, rssi, snr). MaxPacketSize is the radio's limit.
+ */
+template<typename Radio, typename Listener, typename Error, size_t MaxPacketSize>
+class RadioLink : public RadioLinkBase, public Listener {
+ public:
+  RadioLink(Radio *radio, size_t max_packet_size) : RadioLinkBase(max_packet_size), radio_(radio) {}
+
+  void setup() override {
+    this->radio_->register_listener(this);
+    this->disable_loop();
+  }
+  // Runs only during a busy period; a busy flag that never clears keeps it running. A radio that failed meanwhile
+  // is not read again, as reading it may wait on the radio every time
+  void loop() override {
+    if (!this->radio_->is_ready() || !this->radio_->is_receiving())
+      this->radio_free_();
+  }
+
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) override {
+    // The radio is set up after the layers above, which may send from their own setup
+    if (!this->radio_->is_ready()) {
+      this->radio_not_ready_(this->radio_->is_failed());
+      return false;
+    }
+    size_t size = data.size();
+    if (size == 0)
+      return this->skip_empty_();
+    if (!this->check_size_(size) || !this->may_send_(this->radio_->is_receiving()))
+      return false;
+    if (data.is_contiguous()) {
+      auto span = data.as_span();
+      return this->radio_->transmit_packet(span.data(), span.size()) == Error::NONE;
+    }
+    uint8_t buf[MaxPacketSize];
+    data.copy_to(buf, size);
+    return this->radio_->transmit_packet(buf, size) == Error::NONE;
+  }
+
+  void on_packet(const std::vector<uint8_t> &packet, float rssi, float snr) override {
+    this->receive_(packet, rssi, snr);
+  }
+
+ protected:
+  Radio *radio_;
+};
+
+}  // namespace esphome::packet_interface

esphome/components/packet_interface/radio_link.py

@@ -0,0 +1,138 @@
+"""Shared code generation for the packet interface links over packet radios, such as the SX126x and SX127x."""
+
+from collections.abc import Callable
+import math
+
+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 CORE, EsphomeError
+from esphome.cpp_generator import MockObjClass
+import esphome.final_validate as fv
+from esphome.types import ConfigType
+
+from . import new_packet_interface, packet_interface_schema
+from .layers import CONF_ADDRESSING, CONF_PEERS, STACK_ORDER
+
+# The radios' payload length register is 8 bits
+MAX_PACKET_SIZE = 255
+
+# Listen before talk holds a send back for at most the airtime of the largest packet, and never less than this
+MIN_MAX_BUSY_MS = 2000
+# Room for the preamble detection and the radio's own delays
+MARGIN = 1.25
+
+
+def packet_airtime_ms(radio: ConfigType, size: int) -> float:
+    """The time a radio configured as in radio takes to send a packet of size bytes.
+
+    LoRa follows the Semtech formula for an explicit header; the low data rate optimization is assumed when a symbol
+    takes more than 16 ms, as the drivers enable it. Other modulations count the preamble, sync word, length byte,
+    payload and CRC at the bitrate, treating the preamble size as bytes, which gives the longer time.
+    """
+    if str(radio["modulation"]) == "LORA":
+        spreading_factor = radio["spreading_factor"]
+        bandwidth_hz = (
+            float(str(radio["bandwidth"]).removesuffix("kHz").replace("_", ".")) * 1000
+        )
+        coding_rate = int(str(radio["coding_rate"])[-1]) - 4
+        symbol_s = (1 << spreading_factor) / bandwidth_hz
+        low_data_rate = 1 if symbol_s > 0.016 else 0
+        crc = 16 if radio.get("crc_enable", False) else 0
+        payload_symbols = 8 + max(
+            math.ceil(
+                (8 * size - 4 * spreading_factor + 28 + crc)
+                / (4 * (spreading_factor - 2 * low_data_rate))
+            )
+            * (coding_rate + 4),
+            0,
+        )
+        return (radio["preamble_size"] + 4.25 + payload_symbols) * symbol_s * 1000
+    bytes_sent = (
+        radio["preamble_size"] + len(radio.get("sync_value", [])) + 1 + size + 2
+    )
+    return bytes_sent * 8 / radio["bitrate"] * 1000
+
+
+def max_busy_ms(radio: ConfigType, max_packet_size: int) -> int:
+    """How long listen before talk may hold a send back."""
+    return max(
+        MIN_MAX_BUSY_MS, math.ceil(packet_airtime_ms(radio, max_packet_size) * MARGIN)
+    )
+
+
+def radio_link_schema(
+    class_: MockObjClass, radio_id_key: str, radio_class: MockObjClass
+) -> cv.Schema:
+    """Schema for a radio link, referring to its radio by radio_id_key."""
+    return packet_interface_schema(class_, MAX_PACKET_SIZE, MAX_PACKET_SIZE).extend(
+        {cv.GenerateID(radio_id_key): cv.use_id(radio_class)}
+    )
+
+
+def _radio_config(config: ConfigType, radio_id_key: str) -> ConfigType:
+    full_config = fv.full_config.get()
+    return full_config.get_config_for_path(
+        full_config.get_path_for_id(config[radio_id_key])[:-1]
+    )
+
+
+def _two_way_layer(config: ConfigType) -> str | None:
+    """The first layer on the link or in a peer that keeps state for one peer, so needs packets back from it."""
+    for stack in (config, *config.get(CONF_ADDRESSING, {}).get(CONF_PEERS, [])):
+        for kind in STACK_ORDER:
+            if (layer := stack.get(kind.key)) is not None and kind.get_type(
+                layer
+            ).single_peer(layer):
+                return kind.key
+    return None
+
+
+def radio_link_final_validate(
+    radio_id_key: str,
+    payload_length_key: str,
+    rx_start_key: str,
+    check: Callable[[ConfigType, ConfigType], None] | None = None,
+) -> Callable[[ConfigType], ConfigType]:
+    """Final validation for a radio link: packets vary in length, the radio listens after sending when a layer needs
+    replies, and check(link, radio) checks the rest.
+    """
+
+    def validator(config: ConfigType) -> ConfigType:
+        radio = _radio_config(config, radio_id_key)
+        if radio[payload_length_key] != 0:
+            raise cv.Invalid(
+                f"Packets vary in length, so the radio's '{payload_length_key}' must be 0",
+                path=[radio_id_key],
+            )
+        if not radio[rx_start_key] and (layer := _two_way_layer(config)) is not None:
+            raise cv.Invalid(
+                f"This '{layer}' needs packets from the peer, but the radio sleeps after sending unless "
+                f"'{rx_start_key}' is true",
+                path=[radio_id_key],
+            )
+        if check is not None:
+            check(config, radio)
+        return config
+
+    return validator
+
+
+async def new_radio_link(
+    config: ConfigType, radio_id_key: str, radio_domain: str
+) -> None:
+    """Create a radio link, with its listen before talk limit worked out from the radio's settings."""
+    radio = await cg.get_variable(config[radio_id_key])
+    var = await new_packet_interface(config, radio, config[CONF_MAX_PACKET_SIZE])
+    radio_config = next(
+        (
+            conf
+            for conf in CORE.config[radio_domain]
+            if conf[CONF_ID] == config[radio_id_key]
+        ),
+        None,
+    )
+    if radio_config is None:
+        raise EsphomeError(f"No {radio_domain} radio with id '{config[radio_id_key]}'")
+    cg.add(var.set_max_busy(max_busy_ms(radio_config, config[CONF_MAX_PACKET_SIZE])))

esphome/components/sx126x/packet_interface/__init__.py

@@ -0,0 +1,50 @@
+from esphome.components import packet_interface
+from esphome.components.packet_interface.radio_link import (
+    new_radio_link,
+    radio_link_final_validate,
+    radio_link_schema,
+)
+import esphome.config_validation as cv
+from esphome.types import ConfigType
+
+from .. import (
+    CONF_MODULATION,
+    CONF_PAYLOAD_LENGTH,
+    CONF_RX_START,
+    CONF_SX126X_ID,
+    CONF_SYNC_VALUE,
+    DOMAIN,
+    SX126x,
+    SX126xListener,
+    sx126x_ns,
+)
+
+CODEOWNERS = ["@clydebarrow"]
+DEPENDENCIES = ["sx126x"]
+
+SX126xPacketInterface = sx126x_ns.class_(
+    "SX126xPacketInterface", packet_interface.PacketInterface, SX126xListener
+)
+
+CONFIG_SCHEMA = radio_link_schema(SX126xPacketInterface, CONF_SX126X_ID, SX126x)
+
+
+def _check_fsk(config: ConfigType, radio: ConfigType) -> None:
+    """Without LoRa, the radio can tell that a packet is arriving only from its sync word, which listen before talk
+    needs.
+    """
+    if radio[CONF_MODULATION] != "LORA" and not radio[CONF_SYNC_VALUE]:
+        raise cv.Invalid(
+            f"Without LoRa modulation the radio needs a '{CONF_SYNC_VALUE}', so that the link can tell when a packet "
+            "is arriving and not send over it",
+            path=[CONF_SX126X_ID],
+        )
+
+
+FINAL_VALIDATE_SCHEMA = radio_link_final_validate(
+    CONF_SX126X_ID, CONF_PAYLOAD_LENGTH, CONF_RX_START, _check_fsk
+)
+
+
+async def to_code(config: ConfigType) -> None:
+    await new_radio_link(config, CONF_SX126X_ID, DOMAIN)

esphome/components/sx126x/packet_interface/sx126x_packet_interface.h

@@ -0,0 +1,10 @@
+#pragma once
+
+#include "esphome/components/packet_interface/radio_link.h"
+#include "esphome/components/sx126x/sx126x.h"
+
+namespace esphome::sx126x {
+
+using SX126xPacketInterface = packet_interface::RadioLink<SX126x, SX126xListener, SX126xError, SX126X_MAX_PACKET_SIZE>;
+
+}  // namespace esphome::sx126x

esphome/components/sx126x/sx126x.cpp

@@ -1,6 +1,7 @@
 #include "sx126x.h"
 #include "esphome/core/application.h"
 #include "esphome/core/hal.h"
+#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
 namespace esphome::sx126x {
@@ -402,6 +403,15 @@ void SX126x::loop() {
   }
 }
 
+bool SX126x::is_receiving() {
+  // DIO1 signals a received packet, which sending would overwrite in the radio's buffer
+  if (this->dio1_pin_->digital_read())
+    return true;
+  uint8_t buf[2];
+  this->read_opcode_(RADIO_GET_IRQSTATUS, buf, 2);
+  return (encode_uint16(buf[0], buf[1]) & (IRQ_HEADER_VALID | IRQ_SYNCWORD_VALID)) != 0;
+}
+
 void SX126x::run_image_cal() {
   // the following values were taken from section 9.2.1 table 9-2
   // in rev 2.1 of the datasheet
@@ -430,10 +440,11 @@ void SX126x::run_image_cal() {
 void SX126x::set_mode_rx() {
   uint8_t buf[8];
 
-  // configure irq params
+  // configure irq params; a valid header or sync word is flagged for is_receiving(), but not signalled on DIO1
   uint16_t irq = IRQ_RX_DONE | IRQ_RX_TX_TIMEOUT | IRQ_CRC_ERROR;
-  buf[0] = (irq >> 8) & 0xFF;
-  buf[1] = (irq >> 0) & 0xFF;
+  uint16_t flags = irq | IRQ_HEADER_VALID | IRQ_SYNCWORD_VALID;
+  buf[0] = (flags >> 8) & 0xFF;
+  buf[1] = (flags >> 0) & 0xFF;
   buf[2] = (irq >> 8) & 0xFF;
   buf[3] = (irq >> 0) & 0xFF;
   buf[4] = (IRQ_RADIO_NONE >> 8) & 0xFF;

esphome/components/sx126x/sx126x.h

@@ -104,6 +104,8 @@ class SX126x final : public Component,
     return this->transmit_packet(packet.data(), packet.size());
   }
   void register_listener(SX126xListener *listener) { this->listeners_.push_back(listener); }
+  /// True while a packet is being received, or has been received and not yet read; sending would lose it.
+  bool is_receiving();
   Trigger<std::vector<uint8_t>, float, float> *get_packet_trigger() { return &this->packet_trigger_; }
 
  protected:

esphome/components/sx127x/packet_interface/__init__.py

@@ -0,0 +1,68 @@
+from esphome.components import packet_interface
+from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE
+from esphome.components.packet_interface.radio_link import (
+    new_radio_link,
+    radio_link_final_validate,
+    radio_link_schema,
+)
+import esphome.config_validation as cv
+from esphome.types import ConfigType
+
+from .. import (
+    CONF_MODULATION,
+    CONF_PACKET_MODE,
+    CONF_PAYLOAD_LENGTH,
+    CONF_RX_START,
+    CONF_SX127X_ID,
+    CONF_SYNC_VALUE,
+    DOMAIN,
+    SX127x,
+    SX127xListener,
+    sx127x_ns,
+)
+
+CODEOWNERS = ["@clydebarrow"]
+DEPENDENCIES = ["sx127x"]
+
+# The FIFO limit outside LoRa mode
+FSK_MAX_PACKET_SIZE = 64
+
+SX127xPacketInterface = sx127x_ns.class_(
+    "SX127xPacketInterface", packet_interface.PacketInterface, SX127xListener
+)
+
+CONFIG_SCHEMA = radio_link_schema(SX127xPacketInterface, CONF_SX127X_ID, SX127x)
+
+
+def _check_fsk(config: ConfigType, radio: ConfigType) -> None:
+    """Without LoRa, the radio must send packets, not a continuous bit stream, and holds fewer bytes.
+
+    It can tell that a packet is arriving only from its sync word, which listen before talk needs.
+    """
+    if radio[CONF_MODULATION] == "LORA":
+        return
+    if not radio.get(CONF_PACKET_MODE, False):
+        raise cv.Invalid(
+            f"Without LoRa modulation the radio's '{CONF_PACKET_MODE}' must be true, as a link sends packets",
+            path=[CONF_SX127X_ID],
+        )
+    if not radio.get(CONF_SYNC_VALUE):
+        raise cv.Invalid(
+            f"Without LoRa modulation the radio needs a '{CONF_SYNC_VALUE}', so that the link can tell when a "
+            "packet is arriving and not send over it",
+            path=[CONF_SX127X_ID],
+        )
+    if config[CONF_MAX_PACKET_SIZE] > FSK_MAX_PACKET_SIZE:
+        raise cv.Invalid(
+            f"Without LoRa modulation the radio sends at most {FSK_MAX_PACKET_SIZE} bytes",
+            path=[CONF_MAX_PACKET_SIZE],
+        )
+
+
+FINAL_VALIDATE_SCHEMA = radio_link_final_validate(
+    CONF_SX127X_ID, CONF_PAYLOAD_LENGTH, CONF_RX_START, _check_fsk
+)
+
+
+async def to_code(config: ConfigType) -> None:
+    await new_radio_link(config, CONF_SX127X_ID, DOMAIN)

esphome/components/sx127x/packet_interface/sx127x_packet_interface.h

@@ -0,0 +1,10 @@
+#pragma once
+
+#include "esphome/components/packet_interface/radio_link.h"
+#include "esphome/components/sx127x/sx127x.h"
+
+namespace esphome::sx127x {
+
+using SX127xPacketInterface = packet_interface::RadioLink<SX127x, SX127xListener, SX127xError, SX127X_MAX_PACKET_SIZE>;
+
+}  // namespace esphome::sx127x

esphome/components/sx127x/sx127x.cpp

@@ -354,6 +354,18 @@ void SX127x::loop() {
   }
 }
 
+bool SX127x::is_receiving() {
+  // In receive mode DIO0 signals a received packet; the transmit FIFO would overwrite it
+  if (this->dio0_pin_ != nullptr && this->dio0_pin_->digital_read())
+    return true;
+  if (this->modulation_ != MOD_LORA) {
+    // Set from the sync word until the packet has been read; the preamble flag stays set when no sync word follows
+    return (this->read_register_(REG_IRQ_FLAGS_1) & IRQ_FLAGS_1_SYNC_ADDRESS_MATCH) != 0;
+  }
+  return (this->read_register_(REG_MODEM_STAT) &
+          (MODEM_STAT_SIGNAL_DETECTED | MODEM_STAT_SIGNAL_SYNCHRONIZED | MODEM_STAT_HEADER_VALID)) != 0;
+}
+
 void SX127x::run_image_cal() {
   if (this->modulation_ == MOD_LORA) {
     this->set_mode_(MOD_FSK, MODE_SLEEP);

esphome/components/sx127x/sx127x.h

@@ -88,6 +88,8 @@ class SX127x final : public Component,
     return this->transmit_packet(packet.data(), packet.size());
   }
   void register_listener(SX127xListener *listener) { this->listeners_.push_back(listener); }
+  /// True while a packet is being received, or has been received and not yet read; sending would lose it.
+  bool is_receiving();
   Trigger<std::vector<uint8_t>, float, float> *get_packet_trigger() { return &this->packet_trigger_; }
 
  protected:

esphome/components/sx127x/sx127x_reg.h

@@ -45,6 +45,7 @@ enum SX127xReg : uint8_t {
   REG_PAYLOAD_LENGTH_LSB = 0x32,
   REG_FIFO_THRESH = 0x35,
   REG_IMAGE_CAL = 0x3B,
+  REG_IRQ_FLAGS_1 = 0x3E,
   // LoRa registers
   REG_FIFO_ADDR_PTR = 0x0D,
   REG_FIFO_TX_BASE_ADDR = 0x0E,
@@ -259,6 +260,17 @@ enum SX127xIrqFlags : uint8_t {
   CAD_DETECTED = 0x01,
 };
 
+enum SX127xIrqFlags1 : uint8_t {
+  IRQ_FLAGS_1_PREAMBLE_DETECT = 0x02,
+  IRQ_FLAGS_1_SYNC_ADDRESS_MATCH = 0x01,
+};
+
+enum SX127xModemStat : uint8_t {
+  MODEM_STAT_SIGNAL_DETECTED = 0x01,
+  MODEM_STAT_SIGNAL_SYNCHRONIZED = 0x02,
+  MODEM_STAT_HEADER_VALID = 0x08,
+};
+
 enum SX127xModemCfg1 : uint8_t {
   BW_7_8 = 0x00,
   BW_10_4 = 0x10,

esphome/components/tcp/__init__.py

@@ -0,0 +1,4 @@
+"""TCP links. The packet_interface platform carries packets over a TCP connection."""
+
+CODEOWNERS = ["@clydebarrow"]
+DOMAIN = "tcp"

esphome/components/tcp/packet_interface/__init__.py

@@ -0,0 +1,134 @@
+import esphome.codegen as cg
+from esphome.components import binary_sensor, packet_interface, socket
+from esphome.components.const import (
+    CONF_ALLOWED_IPS,
+    CONF_CONNECTED,
+    CONF_HOST,
+    CONF_RECONNECT_INTERVAL,
+    CONF_ROLE,
+)
+from esphome.components.packet_interface import (
+    CONF_MAX_PACKET_SIZE,
+    new_packet_interface,
+    packet_interface_schema,
+)
+from esphome.components.packet_interface.layers import (
+    CONF_ADDRESSING,
+    CONF_RELIABILITY,
+    CONF_SESSION,
+)
+from esphome.components.packet_interface.layers.basic_session import CONF_KEEP_ALIVE
+import esphome.config_validation as cv
+from esphome.const import (
+    CONF_ID,
+    CONF_PORT,
+    CONF_TIMEOUT,
+    DEVICE_CLASS_CONNECTIVITY,
+    ENTITY_CATEGORY_DIAGNOSTIC,
+)
+from esphome.types import ConfigType
+
+CODEOWNERS = ["@clydebarrow"]
+DEPENDENCIES = ["network"]
+
+
+def AUTO_LOAD(config: ConfigType | None = None) -> list[str]:
+    """The binary_sensor component only for the connected sensor; without a config, as tooling calls it, both."""
+    if not config or CONF_CONNECTED in config:
+        return ["socket", "binary_sensor"]
+    return ["socket"]
+
+
+# Each packet and its 2 byte length must fit the socket link's 1024 byte send buffer
+MAX_PACKET_SIZE = 1022
+
+tcp_ns = cg.esphome_ns.namespace("tcp")
+TcpPacketInterface = tcp_ns.class_(
+    "TcpPacketInterface", packet_interface.PacketInterface
+)
+
+
+def _timeout(config: ConfigType) -> ConfigType:
+    """Default the timeout to three keep-alive intervals, and require it to be longer than one."""
+    keep_alive = config[CONF_KEEP_ALIVE].total_milliseconds
+    if (timeout := config.get(CONF_TIMEOUT)) is None:
+        config[CONF_TIMEOUT] = cv.positive_time_period_milliseconds(
+            f"{keep_alive * 3}ms"
+        )
+    elif timeout.total_milliseconds <= keep_alive:
+        raise cv.Invalid("Must be longer than keep_alive", path=[CONF_TIMEOUT])
+    return config
+
+
+BASE_SCHEMA = packet_interface_schema(
+    TcpPacketInterface,
+    MAX_PACKET_SIZE,
+    MAX_PACKET_SIZE,
+    # TCP resends what is lost, joins exactly two devices, and its connection state is the session
+    provides=(CONF_ADDRESSING, CONF_RELIABILITY, CONF_SESSION),
+).extend(
+    {
+        cv.Required(CONF_PORT): cv.port,
+        cv.Optional(
+            CONF_RECONNECT_INTERVAL, default="5s"
+        ): cv.positive_time_period_milliseconds,
+        cv.Optional(CONF_KEEP_ALIVE, default="10s"): cv.All(
+            cv.positive_time_period_milliseconds,
+            cv.Range(min=cv.TimePeriod(seconds=1), max=cv.TimePeriod(hours=1)),
+        ),
+        # Nothing received for this long closes the connection; it must be longer than the peer's keep_alive
+        cv.Optional(CONF_TIMEOUT): cv.All(
+            cv.positive_time_period_milliseconds,
+            cv.Range(max=cv.TimePeriod(hours=3)),
+        ),
+        cv.Optional(CONF_CONNECTED): binary_sensor.binary_sensor_schema(
+            device_class=DEVICE_CLASS_CONNECTIVITY,
+            entity_category=ENTITY_CATEGORY_DIAGNOSTIC,
+        ),
+    }
+)
+
+CONFIG_SCHEMA = cv.All(
+    cv.typed_schema(
+        {
+            "client": cv.All(
+                BASE_SCHEMA.extend({cv.Required(CONF_HOST): socket.ipv4_host}),
+                _timeout,
+            ),
+            "server": cv.All(
+                BASE_SCHEMA.extend(
+                    {cv.Optional(CONF_ALLOWED_IPS): socket.IPV4_ALLOW_SCHEMA}
+                ),
+                _timeout,
+            ),
+        },
+        key=CONF_ROLE,
+        default_type="client",
+    ),
+    # A client uses one socket; a server one listener and one accepted socket
+    socket.consume_role_sockets("tcp.packet_interface"),
+)
+
+
+async def to_code(config: ConfigType) -> None:
+    # The receive buffer's size is known here, so it is a static array rather than allocated at run time
+    rx_buffer = f"{config[CONF_ID].id}__rx_buffer"
+    cg.add_global(
+        cg.RawStatement(f"static uint8_t {rx_buffer}[{config[CONF_MAX_PACKET_SIZE]}];")
+    )
+    var = await new_packet_interface(config, cg.RawExpression(rx_buffer))
+    if config[CONF_ROLE] == "server":
+        socket.require_tcp_listener()
+        cg.add(var.set_server(True))
+        socket.add_ipv4_allow(
+            var.set_allow, config.get(CONF_ALLOWED_IPS), config[CONF_ID]
+        )
+    else:
+        socket.require_tcp_client_link()
+        cg.add(var.set_host(config[CONF_HOST]))
+    cg.add(var.set_port(config[CONF_PORT]))
+    cg.add(var.set_reconnect_interval(config[CONF_RECONNECT_INTERVAL]))
+    cg.add(var.set_keep_alive(config[CONF_KEEP_ALIVE]))
+    cg.add(var.set_receive_timeout(config[CONF_TIMEOUT]))
+    binary_sensors = binary_sensor.sub_binary_sensors(config)
+    await binary_sensors(CONF_CONNECTED, var.set_connected_sensor)

esphome/components/tcp/packet_interface/tcp_packet_interface.cpp

@@ -0,0 +1,169 @@
+#include "tcp_packet_interface.h"
+#include "esphome/core/application.h"
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
+
+namespace esphome::tcp {
+
+using packet_interface::PacketBuffer;
+using packet_interface::PacketMetaData;
+
+ESPHOME_LOG_TAG(TAG, "tcp.packet_interface");
+
+// The packet length before each packet
+static constexpr size_t LENGTH_SIZE = 2;
+
+TcpPacketInterface::TcpPacketInterface(std::span<uint8_t> rx_buffer)
+    : packet_interface::PacketInterface(rx_buffer.size()), rx_buffer_(rx_buffer) {
+  this->capabilities_ = packet_interface::CAPABILITY_RELIABILITY | packet_interface::CAPABILITY_SESSION |
+                        packet_interface::CAPABILITY_ADDRESSING;
+}
+
+void TcpPacketInterface::setup() {
+  this->link_.begin(TAG);
+#ifdef USE_SOCKET_TCP_LISTENER
+  this->listener_.begin(TAG);
+#endif
+#ifdef USE_BINARY_SENSOR
+  if (this->connected_sensor_ != nullptr)
+    this->connected_sensor_->publish_initial_state(false);
+#endif
+}
+
+void TcpPacketInterface::dump_config() {
+  bool server = false;
+#ifdef USE_SOCKET_TCP_LISTENER
+  server = this->server_;
+#endif
+  ESP_LOGCONFIG(TAG,
+                "TCP Packet Interface '%s':\n"
+                "  Role: %s\n"
+                "  Address: %s:%u\n"
+                "  Reconnect Interval: %" PRIu32 "ms\n"
+                "  Keep Alive: %" PRIu32 "ms\n"
+                "  Timeout: %" PRIu32 "ms\n"
+                "  Max Packet Size: %zu",
+                this->name_, server ? LOG_STR_LITERAL("server") : LOG_STR_LITERAL("client"),
+                server ? LOG_STR_LITERAL("*") : this->link_.host(), this->link_.port(),
+                this->link_.reconnect_interval(), this->keep_alive_ms_, this->timeout_ms_, this->max_packet_size_);
+#ifdef USE_SOCKET_TCP_LISTENER
+  this->listener_.dump_config();
+#endif
+#ifdef USE_BINARY_SENSOR
+  LOG_BINARY_SENSOR("  ", "Connected", this->connected_sensor_);
+#endif
+}
+
+void TcpPacketInterface::on_shutdown() {
+  this->link_.close();
+#ifdef USE_SOCKET_TCP_LISTENER
+  this->listener_.close();
+#endif
+}
+
+void TcpPacketInterface::loop() {
+#ifdef USE_SOCKET_TCP_LISTENER
+  if (this->server_) {
+    // A new client is accepted only once the drop of the last has been reported
+    this->listener_.poll(this->link_, !this->connected_);
+  } else {
+    this->link_.poll();
+  }
+#else
+  this->link_.poll();
+#endif
+  if (this->link_.connected() != this->connected_)
+    this->on_connection_(this->link_.connected());
+  if (!this->connected_)
+    return;
+  this->read_();
+  if (!this->connected_)
+    return;
+  uint32_t now = App.get_loop_component_start_time();
+  if (now - this->last_rx_ms_ > this->timeout_ms_) {
+    ESP_LOGW(TAG, "%s: Nothing received for %" PRIu32 "ms, closing the connection", this->name_, this->timeout_ms_);
+    this->link_.close();
+    this->link_.note_attempt();
+    this->on_connection_(false);
+    return;
+  }
+  if (now - this->last_tx_ms_ >= this->keep_alive_ms_)
+    this->queue_frame_(nullptr, 0);
+  this->link_.flush_tx();
+}
+
+void TcpPacketInterface::on_connection_(bool connected) {
+  ESP_LOGD(TAG, "%s: Connection %s", this->name_, connected ? LOG_STR_LITERAL("up") : LOG_STR_LITERAL("down"));
+  this->connected_ = connected;
+  this->rx_have_ = 0;
+  this->last_rx_ms_ = this->last_tx_ms_ = App.get_loop_component_start_time();
+#ifdef USE_BINARY_SENSOR
+  if (this->connected_sensor_ != nullptr)
+    this->connected_sensor_->publish_state(connected);
+#endif
+  this->session_callback_.call(connected);
+}
+
+void TcpPacketInterface::read_() {
+  for (;;) {
+    ssize_t count;
+    if (this->rx_have_ < LENGTH_SIZE) {
+      count = this->link_.read(this->rx_header_ + this->rx_have_, LENGTH_SIZE - this->rx_have_);
+    } else {
+      count = this->link_.read(this->rx_buffer_.data() + this->rx_have_ - LENGTH_SIZE,
+                               this->rx_length_ + LENGTH_SIZE - this->rx_have_);
+    }
+    // Nothing more for now; a dropped link is seen on the next loop
+    if (count <= 0)
+      return;
+    this->last_rx_ms_ = App.get_loop_component_start_time();
+    this->rx_have_ += count;
+    if (this->rx_have_ < LENGTH_SIZE)
+      continue;
+    if (this->rx_have_ == LENGTH_SIZE) {
+      this->rx_length_ = encode_uint16(this->rx_header_[1], this->rx_header_[0]);
+      if (this->rx_length_ > this->max_packet_size_) {
+        // The stream cannot be resynchronised, so start a new connection
+        ESP_LOGW(TAG, "%s: Packet of %u bytes is larger than %zu, closing the connection", this->name_,
+                 this->rx_length_, this->max_packet_size_);
+        this->link_.close();
+        this->link_.note_attempt();
+        this->on_connection_(false);
+        return;
+      }
+    }
+    if (this->rx_have_ == this->rx_length_ + LENGTH_SIZE) {
+      this->rx_have_ = 0;
+      // A zero length packet is a keep-alive
+      if (this->rx_length_ != 0)
+        this->on_receive_from_interface_(PacketBuffer(this->rx_buffer_.data(), this->rx_length_), {});
+    }
+  }
+}
+
+bool TcpPacketInterface::queue_frame_(const PacketBuffer *data, size_t size) {
+  if (this->link_.tx_free() < size + LENGTH_SIZE)
+    return false;
+  uint8_t *tail = this->link_.tx_tail();
+  tail[0] = static_cast<uint8_t>(size);
+  tail[1] = static_cast<uint8_t>(size >> 8);
+  if (data != nullptr)
+    data->copy_to(tail + LENGTH_SIZE, size);
+  this->link_.tx_commit(size + LENGTH_SIZE);
+  this->last_tx_ms_ = App.get_loop_component_start_time();
+  return true;
+}
+
+bool TcpPacketInterface::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  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;
+  }
+  if (!this->connected_ || !this->queue_frame_(&data, size))
+    return false;
+  this->link_.flush_tx();
+  return true;
+}
+
+}  // namespace esphome::tcp

esphome/components/tcp/packet_interface/tcp_packet_interface.h

@@ -0,0 +1,87 @@
+#pragma once
+
+#include <cstdint>
+#include <span>
+
+#include "esphome/components/packet_interface/packet_interface.h"
+#include "esphome/components/socket/tcp_client_link.h"
+#ifdef USE_SOCKET_TCP_LISTENER
+#include "esphome/components/socket/tcp_listener.h"
+#endif
+#ifdef USE_BINARY_SENSOR
+#include "esphome/components/binary_sensor/binary_sensor.h"
+#endif
+
+namespace esphome::tcp {
+
+/**
+ * A packet interface link over one TCP connection, as a client that connects to a host, or a server that accepts
+ * one client at a time.
+ *
+ * Each packet is sent as a 2 byte length, little endian, then the packet. TCP delivers the stream in order and
+ * resends what is lost, and the connection joins exactly two devices, so the link provides the reliability,
+ * addressing and session capabilities itself. Its connection state is the session state.
+ *
+ * Each packet is written to the socket as it is sent, so the link takes packets as fast as the socket does and keeps
+ * the default, unlimited, send capacity; a packet is refused only while the connection is down or the socket is full.
+ *
+ * A zero length packet is a keep-alive, sent when nothing else has been sent for keep_alive. TCP can take minutes to
+ * notice a peer that has gone without closing the connection, so the link closes it, and reconnects or accepts again,
+ * when nothing has been received for the timeout, which must be longer than the peer's keep-alive interval.
+ */
+class TcpPacketInterface : public packet_interface::PacketInterface {
+ public:
+  /// @param rx_buffer Holds the packet being received; its size is the largest packet, and it must outlive the link
+  explicit TcpPacketInterface(std::span<uint8_t> rx_buffer);
+
+  void setup() override;
+  void loop() override;
+  void dump_config() override;
+  void on_shutdown() override;
+  bool send_to_interface(const packet_interface::PacketBuffer &data,
+                         const packet_interface::PacketMetaData &meta_data) override;
+
+  void set_host(const char *host) { this->link_.set_host(host); }
+  void set_port(uint16_t port) { this->link_.set_port(port); }
+  void set_reconnect_interval(uint32_t interval_ms) { this->link_.set_reconnect_interval(interval_ms); }
+  void set_keep_alive(uint32_t keep_alive_ms) { this->keep_alive_ms_ = keep_alive_ms; }
+  /// The connection is closed when nothing has been received for this long; the peer's keep_alive sets the pace.
+  void set_receive_timeout(uint32_t timeout_ms) { this->timeout_ms_ = timeout_ms; }
+#ifdef USE_SOCKET_TCP_LISTENER
+  void set_server(bool server) { this->server_ = server; }
+#ifdef USE_SOCKET_IPV4_ALLOW
+  void set_allow(const socket::Ipv4AllowEntry *entries, size_t count) { this->listener_.set_allow(entries, count); }
+#endif
+#endif
+#ifdef USE_BINARY_SENSOR
+  void set_connected_sensor(binary_sensor::BinarySensor *sensor) { this->connected_sensor_ = sensor; }
+#endif
+
+  bool is_connected() const { return this->connected_; }
+
+ protected:
+  void on_connection_(bool connected);
+  void read_();
+  bool queue_frame_(const packet_interface::PacketBuffer *data, size_t size);
+
+  socket::TcpClientLink link_;
+#ifdef USE_SOCKET_TCP_LISTENER
+  socket::TcpListener listener_;
+  bool server_{false};
+#endif
+#ifdef USE_BINARY_SENSOR
+  binary_sensor::BinarySensor *connected_sensor_{nullptr};
+#endif
+  const std::span<uint8_t> rx_buffer_;
+  uint32_t keep_alive_ms_{10000};
+  uint32_t timeout_ms_{30000};
+  uint32_t last_rx_ms_{0};
+  uint32_t last_tx_ms_{0};
+  uint16_t rx_length_{0};
+  // Bytes of the current frame received so far, counting the 2 byte length
+  uint16_t rx_have_{0};
+  uint8_t rx_header_[2]{};
+  bool connected_{false};
+};
+
+}  // namespace esphome::tcp

esphome/components/udp/packet_interface/__init__.py

@@ -1,3 +1,5 @@
+from ipaddress import IPv4Address
+
 import esphome.codegen as cg
 from esphome.components import packet_interface
 from esphome.components.packet_interface import (
@@ -5,15 +7,24 @@ from esphome.components.packet_interface import (
     new_packet_interface,
     packet_interface_schema,
 )
+from esphome.components.packet_interface.layers import (
+    CONF_ADDRESSING,
+    check_broadcast_link,
+)
+import esphome.config_validation as cv
 from esphome.const import CONF_ID
+from esphome.core import CORE, HexInt
+import esphome.final_validate as fv
 from esphome.types import ConfigType
 
-from .. import CONF_UDP_ID, UDP_SCHEMA, udp_ns
+from .. import CONF_ADDRESSES, CONF_UDP_ID, DOMAIN as UDP_DOMAIN, UDP_SCHEMA, udp_ns
 
 CODEOWNERS = ["@clydebarrow"]
 
 DEPENDENCIES = ["udp"]
 
+CONF_POINT_TO_POINT = "point_to_point"
+
 # Size of the receive buffer of the udp component
 MAX_PACKET_SIZE = 508
 
@@ -22,9 +33,57 @@ UdpPacketInterface = udp_ns.class_(
     packet_interface.PacketInterface,
 )
 
-CONFIG_SCHEMA = packet_interface_schema(
-    UdpPacketInterface, MAX_PACKET_SIZE, MAX_PACKET_SIZE
-).extend(UDP_SCHEMA)
+CONFIG_SCHEMA = (
+    packet_interface_schema(UdpPacketInterface, MAX_PACKET_SIZE, MAX_PACKET_SIZE)
+    .extend(UDP_SCHEMA)
+    .extend(
+        {
+            # Overrides the guess from the udp component's addresses, which cannot see the subnet
+            cv.Optional(CONF_POINT_TO_POINT): cv.boolean,
+        }
+    )
+)
+
+
+def _peer(udp_config: ConfigType, config: ConfigType) -> IPv4Address | None:
+    """The one peer, when the UDP component sends to a single unicast IPv4 address.
+
+    Unless point_to_point says otherwise, an address ending in .255 is taken as a subnet broadcast, as the subnet is
+    not known here.
+    """
+    point_to_point = config.get(CONF_POINT_TO_POINT)
+    match udp_config[CONF_ADDRESSES]:
+        case [IPv4Address() as address] if not address.is_multicast:
+            if point_to_point is None:
+                point_to_point = address.packed[3] != 255
+            return address if point_to_point else None
+    if point_to_point:
+        raise cv.Invalid(
+            f"'{CONF_POINT_TO_POINT}' needs the UDP component to send to one unicast IPv4 address",
+            path=[CONF_POINT_TO_POINT],
+        )
+    return None
+
+
+def _udp_config(full_config: ConfigType, config: ConfigType) -> ConfigType:
+    return next(
+        udp for udp in full_config[UDP_DOMAIN] if udp[CONF_ID] == config[CONF_UDP_ID]
+    )
+
+
+def _final_validate(config: ConfigType) -> ConfigType:
+    if (peer := _peer(_udp_config(fv.full_config.get(), config), config)) is None:
+        check_broadcast_link(config)
+    elif CONF_ADDRESSING in config:
+        raise cv.Invalid(
+            f"The UDP component sends only to {peer}, so packets are already addressed; "
+            f"'{CONF_ADDRESSING}' is only for broadcast or multicast",
+            path=[CONF_ADDRESSING],
+        )
+    return config
+
+
+FINAL_VALIDATE_SCHEMA = _final_validate
 
 
 async def to_code(config: ConfigType) -> None:
@@ -32,9 +91,13 @@ async def to_code(config: ConfigType) -> None:
     cg.add(udp_var.set_should_listen())
     # The udp component sends to its addresses through its broadcast socket
     cg.add(udp_var.set_should_broadcast())
+    peer = _peer(_udp_config(CORE.config, config), config)
+    peer_bytes = peer.packed if peer is not None else bytes(4)
     # The copy buffer's size is known here, so it is a static array rather than allocated at run time
     tx_buffer = f"{config[CONF_ID].id}__tx_buffer"
     cg.add_global(
         cg.RawStatement(f"static uint8_t {tx_buffer}[{config[CONF_MAX_PACKET_SIZE]}];")
     )
-    await new_packet_interface(config, udp_var, cg.RawExpression(tx_buffer))
+    await new_packet_interface(
+        config, udp_var, cg.RawExpression(tx_buffer), [HexInt(x) for x in peer_bytes]
+    )

esphome/components/udp/packet_interface/udp_packet_interface.cpp

@@ -1,6 +1,8 @@
 #include "udp_packet_interface.h"
 #ifdef USE_NETWORK
+#include <cstring>
 #include "esphome/components/network/util.h"
+#include "esphome/core/application.h"
 #include "esphome/core/log.h"
 
 namespace esphome::udp {
@@ -10,11 +12,23 @@ using packet_interface::PacketMetaData;
 
 ESPHOME_LOG_TAG(TAG, "udp.packet_interface");
 
-void UdpPacketInterface::setup() {
-  this->parent_->add_listener([this](std::span<const uint8_t> data) { this->receive_(data); });
+UdpPacketInterface::UdpPacketInterface(UDPComponent *udp_component, std::span<uint8_t> tx_buffer,
+                                       std::array<uint8_t, 4> peer)
+    : packet_interface::PacketInterface(tx_buffer.size()), parent_(udp_component), tx_buffer_(tx_buffer), peer_(peer) {
+  if (this->peer_ != std::array<uint8_t, 4>{})
+    this->capabilities_ = packet_interface::CAPABILITY_ADDRESSING;
 }
 
-void UdpPacketInterface::receive_(std::span<const uint8_t> data) {
+void UdpPacketInterface::setup() {
+  this->parent_->add_listener(
+      [this](std::span<const uint8_t> data) { this->receive_(data, this->parent_->get_rx_source()); });
+}
+
+void UdpPacketInterface::receive_(std::span<const uint8_t> data, const sockaddr *source) {
+  if (!this->accepts_from_(source)) {
+    this->warn_dropped_(source);
+    return;
+  }
   if (data.size() > this->max_packet_size_) {
     ESP_LOGV(TAG, "%s: Packet of %zu bytes is larger than %zu, dropped", this->name_, data.size(),
              this->max_packet_size_);
@@ -23,6 +37,28 @@ void UdpPacketInterface::receive_(std::span<const uint8_t> data) {
   this->on_receive_from_interface_(PacketBuffer(data.data(), data.size()), {});
 }
 
+bool UdpPacketInterface::accepts_from_(const sockaddr *source) const {
+  if (!this->has_capability(packet_interface::CAPABILITY_ADDRESSING))
+    return true;
+  uint32_t address;
+  return socket::sockaddr_to_ipv4(source, &address) && memcmp(&address, this->peer_.data(), sizeof(address)) == 0;
+}
+
+void UdpPacketInterface::warn_dropped_(const sockaddr *source) {
+  static constexpr uint32_t WARNING_INTERVAL_MS = 60000;
+  uint32_t now = App.get_loop_component_start_time();
+  if (this->drop_warned_ && now - this->last_drop_warning_ms_ < WARNING_INTERVAL_MS)
+    return;
+  this->drop_warned_ = true;
+  this->last_drop_warning_ms_ = now;
+  char from[socket::SOCKADDR_STR_LEN];
+  socket::format_sockaddr_to(source, sizeof(sockaddr_storage), from);
+  ESP_LOGW(TAG,
+           "%s: Packet from %s dropped, as only %u.%u.%u.%u is accepted; set point_to_point: false if that is a "
+           "broadcast address",
+           this->name_, from, this->peer_[0], this->peer_[1], this->peer_[2], this->peer_[3]);
+}
+
 bool UdpPacketInterface::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) {
   size_t size = data.size();
   if (size > this->max_packet_size_) {

esphome/components/udp/packet_interface/udp_packet_interface.h

@@ -4,9 +4,11 @@
 
 // udp_component.h declares nothing without USE_NETWORK
 #ifdef USE_NETWORK
+#include <array>
 #include <span>
 
 #include "esphome/components/packet_interface/packet_interface.h"
+#include "esphome/components/socket/socket.h"
 #include "../udp_component.h"
 
 namespace esphome::udp {
@@ -14,24 +16,32 @@ namespace esphome::udp {
 /**
  * A packet interface link over UDP: each packet is one datagram, sent to every address of the UDP component.
  *
- * Every datagram received on the listen port is passed up, from any sender, so on a shared network an addressing
- * or encryption layer is needed to tell the peer's packets from others.
+ * When the UDP component sends to one unicast IPv4 address, the link joins exactly two devices: it accepts packets
+ * only from that address, and provides the addressing capability itself. Otherwise every packet is accepted, and an
+ * addressing layer can tell the devices apart. Packets from other senders are dropped with a warning at most once a
+ * minute, as they suggest that the address is a broadcast address the link took for a device.
  */
 class UdpPacketInterface : public packet_interface::PacketInterface {
  public:
   /// @param tx_buffer Joins packets split across buffers; its size is the largest packet, and it must outlive the link
-  UdpPacketInterface(UDPComponent *udp_component, std::span<uint8_t> tx_buffer)
-      : packet_interface::PacketInterface(tx_buffer.size()), parent_(udp_component), tx_buffer_(tx_buffer) {}
+  /// @param peer The IPv4 address of the one peer, in network byte order; all zeros if packets are broadcast
+  UdpPacketInterface(UDPComponent *udp_component, std::span<uint8_t> tx_buffer, std::array<uint8_t, 4> peer);
 
   void setup() override;
   bool send_to_interface(const packet_interface::PacketBuffer &data,
                          const packet_interface::PacketMetaData &meta_data) override;
 
  protected:
-  void receive_(std::span<const uint8_t> data);
+  void receive_(std::span<const uint8_t> data, const sockaddr *source);
+  /// True if packets from \p source are accepted: from anywhere, or with a peer, only from the peer.
+  bool accepts_from_(const sockaddr *source) const;
+  void warn_dropped_(const sockaddr *source);
 
   UDPComponent *parent_;
   const std::span<uint8_t> tx_buffer_;
+  std::array<uint8_t, 4> peer_;
+  uint32_t last_drop_warning_ms_{0};
+  bool drop_warned_{false};
 };
 }  // namespace esphome::udp
 #endif

esphome/components/udp/udp_component.cpp

@@ -211,10 +211,13 @@ void UDPComponent::loop() {
   }
   // One byte more than the largest packet, as a read cuts a longer datagram to fit and so cannot report it
   std::array<uint8_t, MAX_PACKET_SIZE + 1> buf;
+  struct sockaddr_storage source;
   for (;;) {
     if (!this->listen_socket_->ready())
       break;
-    auto len = this->listen_socket_->read(buf.data(), buf.size());
+    socklen_t source_len = sizeof(source);
+    auto len =
+        this->listen_socket_->recvfrom(buf.data(), buf.size(), reinterpret_cast<sockaddr *>(&source), &source_len);
     if (len <= 0)
       break;
     size_t packet_len = static_cast<size_t>(len);
@@ -223,7 +226,9 @@ void UDPComponent::loop() {
       continue;
     }
     ESP_LOGV(TAG, "Received packet of length %zu", packet_len);
+    this->rx_source_ = reinterpret_cast<sockaddr *>(&source);
     this->packet_listeners_.call(std::span<const uint8_t>(buf.data(), packet_len));
+    this->rx_source_ = nullptr;
   }
 }
 

esphome/components/udp/udp_component.h

@@ -32,6 +32,8 @@ class UDPComponent final : public Component {
   void set_ipv4_send(bool ipv4_send) { this->ipv4_send_ = ipv4_send; }
 #endif
   template<typename F> void add_listener(F &&listener) { this->packet_listeners_.add(std::forward<F>(listener)); }
+  /// The sender of the packet being passed to the listeners; valid only during a listener call.
+  const sockaddr *get_rx_source() const { return this->rx_source_; }
   void setup() override;
   void loop() override;
   void dump_config() override;
@@ -63,6 +65,7 @@ class UDPComponent final : public Component {
   FixedVector<SockaddrEntry> sockaddrs_{};
   const char *const *addresses_{nullptr};
   const char *listen_address_{nullptr};
+  const sockaddr *rx_source_{nullptr};
 #ifdef USE_SOCKET_IPV6
   uint32_t mcast_ifindex_{0};
 #endif

esphome/core/defines.h

@@ -268,6 +268,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/espnow/common-packet-interface.yaml

@@ -0,0 +1,36 @@
+espnow:
+  id: espnow_component
+  channel: 1
+
+packet_interface:
+  - platform: espnow
+    id: espnow_peer_link
+    peer_address: 11:22:33:44:55:66
+    encryption: Johnny Cash
+    reliability:
+    session:
+  - platform: espnow
+    id: espnow_broadcast_link
+    addressing:
+      address: 0x0001
+      peers:
+        - id: espnow_bus_link
+          destination: broadcast
+        - id: espnow_pump_link
+          destination: 0x0002
+          reliability:
+
+sensor:
+  - platform: template
+    id: espnow_sensor
+    lambda: return 1.0;
+
+packet_transport:
+  - platform: packet_interface
+    packet_interface: espnow_peer_link
+    sensors:
+      - espnow_sensor
+  - platform: packet_interface
+    packet_interface: espnow_bus_link
+    sensors:
+      - espnow_sensor

tests/components/espnow/test-packet-interface.esp32-idf.yaml

@@ -0,0 +1,2 @@
+packages:
+  espnow: !include common-packet-interface.yaml

tests/components/espnow/test-packet-interface.esp8266-ard.yaml

@@ -0,0 +1,2 @@
+packages:
+  espnow: !include common-packet-interface.yaml

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,226 @@
+#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 {
+
+// 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 = 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:
+  explicit MockLink(size_t max_packet_size = 64) : PacketInterface(max_packet_size) {}
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &) override {
+    this->sent.push_back(to_vector(data));
+    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(to_vector(data));
+      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, LinkTooSmallLeavesNoRoom) {
+  MockLink link(BasicAddressing::HEADER_SIZE);
+  BasicAddressing addressing(&link, 0x0001);
+  AddressingPort port(&addressing, 0x0002, true);
+  EXPECT_EQ(port.get_max_packet_size(), 0);
+}
+
+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, BroadcastFromPeerReachesBroadcastPort) {
+  Device hub(1, {2, BROADCAST}), b(2, {1, BROADCAST});
+  hub.port(1).send({0x10});  // a state every device needs
+  hub.port(0).send({0x11});  // point to point
+  hub.deliver({&b});
+  EXPECT_EQ(b.port(0).received, std::vector<Packet>{{0x11}});
+  EXPECT_EQ(b.port(1).received, std::vector<Packet>{{0x10}});
+  EXPECT_EQ(b.port(1).sources, std::vector<uint16_t>{1});
+}
+
+TEST(BasicAddressingTest, WithoutBroadcastPortPeerBroadcastReachesItsPort) {
+  Device hub(1, {2, 3}), b(2, BROADCAST);
+  b.port().send({0x20});
+  b.deliver({&hub});
+  EXPECT_EQ(hub.port(0).received, std::vector<Packet>{{0x20}});
+  EXPECT_TRUE(hub.port(1).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());
+}
+
+// Each test replaces App with a fresh one; the reliability layer's timers need its scheduler
+class AddressingReliabilityTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    App.~Application();
+    new (&App) Application();
+    static char name[] = "addressing-tests";
+    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
+  }
+};
+
+// 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(to_vector(data)); });
+    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,229 @@
+#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 {
+
+// 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 = std::vector<uint8_t>;
+
+// A link layer that records what is sent, with a limited send capacity
+class MockLink : public PacketInterface {
+ public:
+  explicit MockLink(size_t max_packet_size = 12) : PacketInterface(max_packet_size) {}
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &) override {
+    this->sent.push_back(to_vector(data));
+    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(to_vector(data)); });
+    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_;
+};
+
+// Each test replaces App with a fresh one; the reliability layer's timers need its scheduler
+class BasicFragmentationTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    App.~Application();
+    new (&App) Application();
+    static char name[] = "fragmentation-tests";
+    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
+  }
+
+  // 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);
+  // Nothing is mixed from the two, and the packet started last gets through
+  EXPECT_EQ(b.received, std::vector<Packet>{make_packet(20)});
+}
+
+TEST_F(BasicFragmentationTest, OneFragmentPacketFromAnotherSourceKeepsJoin) {
+  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(5));
+  auto from_c = std::move(this->link_a_.sent);
+  this->link_b_.inject(from_a[0], 1);
+  this->link_b_.inject(from_c[0], 2);
+  this->link_b_.inject(from_a[1], 1);
+  EXPECT_EQ(b.received, (std::vector<Packet>{make_packet(5), make_packet(20)}));
+}
+
+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(to_vector(data)); });
+  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})});
+}
+
+TEST_F(BasicFragmentationTest, LinkTooSmallFailsSetup) {
+  MockLink link(BasicFragmentation::HEADER_SIZE);
+  BasicFragmentation fragmentation(&link, 100);
+  fragmentation.setup();
+  EXPECT_TRUE(fragmentation.is_failed());
+  Packet packet{1};
+  EXPECT_FALSE(fragmentation.send_to_interface(PacketBuffer(packet), {}));
+}
+
+}  // namespace esphome::packet_interface::fragmentation_testing

tests/components/packet_interface/common.yaml

@@ -4,29 +4,55 @@ wifi:
 
 udp:
   - id: packet_interface_udp
+  - id: packet_interface_udp_peer
+    port: 18512
+    addresses: ["192.168.1.60"]
 
 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:
+            max_queue_length: 5
+          fragmentation:
+            max_packet_size: 1024
+          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:
+  # Sends to one unicast address, so the link provides addressing
+  - platform: udp
+    id: packet_interface_udp_peer_link
+    udp_id: packet_interface_udp_peer
+    reliability:
     session:
 
 sensor:
@@ -36,10 +62,14 @@ sensor:
 
 packet_transport:
   - platform: packet_interface
-    packet_interface: packet_interface_uart
+    packet_interface: packet_interface_peer_a
     sensors:
       - packet_interface_sensor
   - platform: packet_interface
     packet_interface: packet_interface_network
     sensors:
       - packet_interface_sensor
+  - platform: packet_interface
+    packet_interface: packet_interface_udp_peer_link
+    sensors:
+      - packet_interface_sensor

tests/components/packet_interface/listen_before_talk.cpp

@@ -0,0 +1,61 @@
+#include <gtest/gtest.h>
+#include "esphome/components/packet_interface/listen_before_talk.h"
+
+namespace esphome::packet_interface::lbt_testing {
+
+static constexpr uint32_t START = 10000;
+static constexpr uint32_t MAX_BUSY = ListenBeforeTalk::DEFAULT_MAX_BUSY_MS;
+
+TEST(ListenBeforeTalkTest, SendsWhenNotBusy) {
+  ListenBeforeTalk lbt;
+  EXPECT_TRUE(lbt.may_send(false, START));
+}
+
+TEST(ListenBeforeTalkTest, RefusesWhileBusyUpToLimit) {
+  ListenBeforeTalk lbt;
+  EXPECT_FALSE(lbt.may_send(true, START));
+  // Refused every 50 ms, as the reliability layer retries
+  uint32_t now = START;
+  while (now - START < MAX_BUSY - 50) {
+    now += 50;
+    EXPECT_FALSE(lbt.may_send(true, now));
+  }
+  now += 50;
+  EXPECT_TRUE(lbt.may_send(true, now));
+  EXPECT_EQ(lbt.busy_for(now), MAX_BUSY);
+}
+
+TEST(ListenBeforeTalkTest, BusyFlagThatNeverClearsReleasedForSlowSends) {
+  ListenBeforeTalk lbt;
+  // A send every 10 s, as session keep-alives, always finding the radio busy
+  EXPECT_FALSE(lbt.may_send(true, START));
+  EXPECT_TRUE(lbt.may_send(true, START + 10000));
+}
+
+TEST(ListenBeforeTalkTest, FreeRadioEndsBusyPeriod) {
+  ListenBeforeTalk lbt;
+  EXPECT_FALSE(lbt.may_send(true, START));
+  EXPECT_TRUE(lbt.may_send(false, START + 1500));
+  // A new busy period starts, so the earlier one does not count
+  EXPECT_FALSE(lbt.may_send(true, START + MAX_BUSY + 100));
+  EXPECT_EQ(lbt.busy_for(START + MAX_BUSY + 100), 0);
+}
+
+TEST(ListenBeforeTalkTest, EndBusyStartsNewPeriod) {
+  ListenBeforeTalk lbt;
+  EXPECT_FALSE(lbt.may_send(true, START));
+  // The link saw the radio free between sends, so a later reception is a new busy period
+  lbt.end_busy();
+  EXPECT_FALSE(lbt.may_send(true, START + MAX_BUSY + 100));
+  EXPECT_EQ(lbt.busy_for(START + MAX_BUSY + 100), 0);
+}
+
+TEST(ListenBeforeTalkTest, LimitCanBeLonger) {
+  ListenBeforeTalk lbt;
+  lbt.set_max_busy(8000);
+  EXPECT_FALSE(lbt.may_send(true, START));
+  EXPECT_FALSE(lbt.may_send(true, START + 7999));
+  EXPECT_TRUE(lbt.may_send(true, START + 8000));
+}
+
+}  // namespace esphome::packet_interface::lbt_testing

tests/components/packet_interface/radio_link.cpp

@@ -0,0 +1,132 @@
+#include <gtest/gtest.h>
+#include <cstdint>
+#include <vector>
+#include "esphome/components/packet_interface/radio_link.h"
+#include "esphome/core/application.h"
+
+namespace esphome::packet_interface::radio_link_testing {
+
+using Packet = std::vector<uint8_t>;
+
+static constexpr uint32_t START = 10000;
+static constexpr uint32_t MAX_BUSY = ListenBeforeTalk::DEFAULT_MAX_BUSY_MS;
+
+class FakeListener {
+ public:
+  virtual void on_packet(const std::vector<uint8_t> &packet, float rssi, float snr) = 0;
+};
+
+enum class FakeError { NONE, BUSY };
+
+class FakeRadio : public Component {
+ public:
+  bool is_receiving() { return this->receiving; }
+  void register_listener(FakeListener *listener) { this->listener = listener; }
+  void fail() { this->mark_failed(); }
+  FakeError transmit_packet(const uint8_t *data, size_t length) {
+    this->sent.emplace_back(data, data + length);
+    return FakeError::NONE;
+  }
+
+  FakeListener *listener{nullptr};
+  std::vector<Packet> sent;
+  bool receiving{false};
+};
+
+using TestLink = RadioLink<FakeRadio, FakeListener, FakeError, 16>;
+
+static void set_time(uint32_t now) { LoopBlockingGuard guard{nullptr, nullptr, now}; }
+
+class RadioLinkTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    this->radio_.call();
+    this->link_.call();
+  }
+  bool send_() {
+    static const Packet PACKET{1, 2, 3};
+    return this->link_.send_to_interface(PacketBuffer(PACKET), {});
+  }
+
+  FakeRadio radio_;
+  TestLink link_{&radio_, 16};
+};
+
+TEST_F(RadioLinkTest, LoopRunsOnlyWhileBusy) {
+  EXPECT_EQ(this->radio_.listener, &this->link_);
+  EXPECT_TRUE(this->link_.is_idle());
+  set_time(START);
+  this->radio_.receiving = true;
+  EXPECT_FALSE(this->send_());
+  EXPECT_TRUE(this->link_.is_in_loop_state());
+  this->link_.loop();
+  EXPECT_TRUE(this->link_.is_in_loop_state());
+  this->radio_.receiving = false;
+  this->link_.loop();
+  EXPECT_TRUE(this->link_.is_idle());
+  EXPECT_TRUE(this->radio_.sent.empty());
+}
+
+TEST_F(RadioLinkTest, ReceptionAfterStalePeriodIsNotSentOver) {
+  set_time(START);
+  this->radio_.receiving = true;
+  EXPECT_FALSE(this->send_());
+  // The reception ends with no send to see it
+  this->radio_.receiving = false;
+  this->link_.loop();
+  // A new reception, longer after the first than the limit, still holds the send back
+  set_time(START + MAX_BUSY + 100);
+  this->radio_.receiving = true;
+  EXPECT_FALSE(this->send_());
+  EXPECT_TRUE(this->radio_.sent.empty());
+}
+
+TEST_F(RadioLinkTest, BusyFlagThatNeverClearsIsSentOver) {
+  set_time(START);
+  this->radio_.receiving = true;
+  EXPECT_FALSE(this->send_());
+  this->link_.loop();
+  set_time(START + MAX_BUSY);
+  EXPECT_TRUE(this->send_());
+  ASSERT_EQ(this->radio_.sent.size(), 1u);
+  EXPECT_EQ(this->radio_.sent[0], (Packet{1, 2, 3}));
+}
+
+TEST_F(RadioLinkTest, EmptyPacketCountsAsSentWithoutTransmitting) {
+  static const Packet EMPTY;
+  EXPECT_TRUE(this->link_.send_to_interface(PacketBuffer(EMPTY), {}));
+  EXPECT_TRUE(this->radio_.sent.empty());
+}
+
+TEST_F(RadioLinkTest, FailedRadioRefusesSends) {
+  this->radio_.fail();
+  EXPECT_FALSE(this->send_());
+  EXPECT_FALSE(this->send_());
+  EXPECT_TRUE(this->radio_.sent.empty());
+}
+
+TEST_F(RadioLinkTest, RadioFailingWhileBusyEndsBusyPeriod) {
+  set_time(START);
+  this->radio_.receiving = true;
+  EXPECT_FALSE(this->send_());
+  EXPECT_TRUE(this->link_.is_in_loop_state());
+  this->radio_.fail();
+  this->link_.loop();
+  EXPECT_TRUE(this->link_.is_idle());
+}
+
+TEST_F(RadioLinkTest, ReceivedPacketCarriesSignal) {
+  Packet received;
+  PacketMetaData meta;
+  this->link_.add_packet_interface_listener([&](const PacketBuffer &data, const PacketMetaData &meta_data) {
+    received.resize(data.size());
+    data.copy_to(received.data(), received.size());
+    meta = meta_data;
+  });
+  this->link_.on_packet({4, 5}, -80.0f, 7.5f);
+  EXPECT_EQ(received, (Packet{4, 5}));
+  EXPECT_FLOAT_EQ(meta.rssi, -80.0f);
+  EXPECT_FLOAT_EQ(meta.snr, 7.5f);
+}
+
+}  // namespace esphome::packet_interface::radio_link_testing

tests/components/sx126x/common-packet-interface.yaml

@@ -0,0 +1,19 @@
+sx126x:
+  cs_pin: ${cs_pin}
+  rst_pin: ${rst_pin}
+  busy_pin: ${busy_pin}
+  dio1_pin: ${dio1_pin}
+  hw_version: sx1262
+  rf_switch: true
+  frequency: 917MHz
+  modulation: LORA
+  crc_enable: true
+
+packet_interface:
+  - platform: sx126x
+    id: sx126x_packet_interface
+    addressing:
+      address: 0x0001
+      destination: 0x0002
+    reliability:
+      timeout: 500ms

tests/components/sx126x/test-packet-interface.esp32-idf.yaml

@@ -0,0 +1,9 @@
+substitutions:
+  cs_pin: GPIO12
+  rst_pin: GPIO13
+  busy_pin: GPIO25
+  dio1_pin: GPIO26
+
+packages:
+  spi: !include ../../test_build_components/common/spi/esp32-idf.yaml
+  sx126x: !include common-packet-interface.yaml

tests/components/sx127x/common-packet-interface.yaml

@@ -0,0 +1,17 @@
+sx127x:
+  cs_pin: ${cs_pin}
+  rst_pin: ${rst_pin}
+  dio0_pin: ${dio0_pin}
+  frequency: 917MHz
+  modulation: LORA
+  preamble_size: 8
+  crc_enable: true
+
+packet_interface:
+  - platform: sx127x
+    id: sx127x_packet_interface
+    addressing:
+      address: 0x0001
+      destination: 0x0002
+    reliability:
+      timeout: 500ms

tests/components/sx127x/test-packet-interface.esp32-idf.yaml

@@ -0,0 +1,8 @@
+substitutions:
+  cs_pin: GPIO12
+  rst_pin: GPIO13
+  dio0_pin: GPIO26
+
+packages:
+  spi: !include ../../test_build_components/common/spi/esp32-idf.yaml
+  sx127x: !include common-packet-interface.yaml

tests/components/tcp/__init__.py

@@ -0,0 +1,18 @@
+from esphome.components import socket as socket_component
+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:
+        socket_component.require_tcp_listener()
+        # No entity platforms are configured, so count the connected sensor to emit USE_BINARY_SENSOR
+        CORE.platform_counts["binary_sensor"] += 1
+
+    manifest.to_code = to_code_testing
+    manifest.dependencies = manifest.dependencies + [
+        "binary_sensor",
+        "packet_interface",
+        "socket",
+    ]

tests/components/tcp/benchmark.yaml

@@ -0,0 +1,4 @@
+packet_interface:
+  - platform: tcp
+    role: server
+    port: 6000

tests/components/tcp/common.yaml

@@ -0,0 +1,36 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packet_interface:
+  - platform: tcp
+    id: tcp_client_link
+    host: 192.168.1.50
+    port: 6060
+    keep_alive: 5s
+    encryption: Johnny Cash
+    fragmentation:
+      max_packet_size: 2048
+    connected:
+      name: TCP link connected
+  - platform: tcp
+    id: tcp_server_link
+    role: server
+    port: 6061
+    allowed_ips:
+      - 192.168.1.0/24
+
+sensor:
+  - platform: template
+    id: tcp_sensor
+    lambda: return 1.0;
+
+packet_transport:
+  - platform: packet_interface
+    packet_interface: tcp_client_link
+    sensors:
+      - tcp_sensor
+  - platform: packet_interface
+    packet_interface: tcp_server_link
+    sensors:
+      - tcp_sensor

tests/components/tcp/tcp_packet_interface.cpp

@@ -0,0 +1,307 @@
+#include <gtest/gtest.h>
+
+#include <csignal>
+#include <cstdint>
+#include <memory>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <unistd.h>
+#include <vector>
+
+#include "esphome/components/tcp/packet_interface/tcp_packet_interface.h"
+#include "esphome/core/application.h"
+#include "esphome/core/wake.h"
+
+#ifdef USE_HOST
+
+namespace esphome::tcp::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>;
+
+// The receive buffer is a base, so it is set up before the link's constructor sees it
+struct TcpBufferStorage {
+  uint8_t rx_buffer_storage[64]{};
+};
+
+class TestTcp : public TcpBufferStorage, public TcpPacketInterface {
+ public:
+  TestTcp() : TcpPacketInterface(this->rx_buffer_storage) {
+    this->set_host("peer");
+    this->set_port(1);
+  }
+  socket::TcpClientLink &link() { return this->link_; }
+};
+
+// A layer stacked on the link, to check that the session state reaches it
+class UpperLayer : public packet_interface::PacketInterface {
+ public:
+  explicit UpperLayer(PacketInterface *lower) : packet_interface::PacketInterface(lower->get_max_packet_size()) {
+    this->stack_on_(lower, packet_interface::CAPABILITY_ENCRYPTION);
+  }
+  bool send_to_interface(const PacketBuffer &, const PacketMetaData &) override { return true; }
+};
+
+struct End {
+  End() {
+    this->link.add_packet_interface_listener(
+        [this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(to_vector(data)); });
+    this->link.add_on_session_state_callback([this](bool connected) { this->states.push_back(connected); });
+    this->link.setup();
+  }
+  bool send(const Packet &data) { return this->link.send_to_interface(PacketBuffer(data), {}); }
+
+  TestTcp link;
+  std::vector<Packet> received;
+  std::vector<bool> states;
+};
+
+class TcpPacketInterfaceTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    signal(SIGPIPE, SIG_IGN);
+    App.~Application();
+    new (&App) Application();
+    static char name[] = "tcp-tests";
+    static char friendly_name[] = "";
+    App.pre_setup(name, sizeof(name) - 1, friendly_name, 0);
+    set_time(1000);
+    int fds[2];
+    ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
+    this->a.link.link().adopt(std::make_unique<socket::Socket>(fds[0]));
+    this->b.link.link().adopt(std::make_unique<socket::Socket>(fds[1]));
+    this->loops(1);
+  }
+  void TearDown() override {
+    this->a.link.link().close();
+    this->b.link.link().close();
+    App.~Application();
+    new (&App) Application();  // later tests and the test listener use App
+  }
+  static void set_time(uint32_t now) { LoopBlockingGuard guard{nullptr, nullptr, now}; }
+  void loops(int count) {
+    for (int i = 0; i != count; i++) {
+      this->a.link.loop();
+      this->b.link.loop();
+    }
+  }
+
+  End a, b;
+};
+
+TEST_F(TcpPacketInterfaceTest, ProvidesReliabilitySessionAndAddressing) {
+  for (auto capability : {packet_interface::CAPABILITY_RELIABILITY, packet_interface::CAPABILITY_SESSION,
+                          packet_interface::CAPABILITY_ADDRESSING})
+    EXPECT_TRUE(this->a.link.has_capability(capability));
+  EXPECT_EQ(this->a.states, std::vector<bool>{true});
+  EXPECT_TRUE(this->a.link.is_connected());
+}
+
+TEST_F(TcpPacketInterfaceTest, PacketsBothWays) {
+  Packet big(64);
+  for (size_t i = 0; i != big.size(); i++)
+    big[i] = i;
+  ASSERT_TRUE(this->a.send({1, 2, 3}));
+  ASSERT_TRUE(this->a.send(big));
+  ASSERT_TRUE(this->b.send({9}));
+  this->loops(3);
+  EXPECT_EQ(this->b.received, (std::vector<Packet>{{1, 2, 3}, big}));
+  EXPECT_EQ(this->a.received, std::vector<Packet>{{9}});
+  EXPECT_FALSE(this->a.send(Packet(65)));
+}
+
+TEST_F(TcpPacketInterfaceTest, SessionReachesLayersAbove) {
+  UpperLayer upper(&this->a.link);
+  std::vector<bool> states;
+  upper.add_on_session_state_callback([&states](bool connected) { states.push_back(connected); });
+  EXPECT_TRUE(upper.has_capability(packet_interface::CAPABILITY_SESSION));
+  this->b.link.link().close();  // the peer goes away
+  this->loops(3);
+  EXPECT_EQ(states, std::vector<bool>{false});
+}
+
+TEST_F(TcpPacketInterfaceTest, KeepAliveKeepsIdleLinkUp) {
+  for (uint32_t now = 1000; now <= 61000; now += 5000) {
+    set_time(now);
+    this->loops(2);
+  }
+  EXPECT_TRUE(this->a.link.is_connected());
+  EXPECT_TRUE(this->b.link.is_connected());
+  EXPECT_TRUE(this->a.received.empty());  // keep-alives are not passed up
+}
+
+TEST_F(TcpPacketInterfaceTest, SilentPeerClosesLink) {
+  set_time(1000 + 30001);
+  this->a.link.loop();  // b has sent nothing for the default timeout of 30 s
+  EXPECT_FALSE(this->a.link.is_connected());
+  EXPECT_EQ(this->a.states, (std::vector<bool>{true, false}));
+}
+
+TEST_F(TcpPacketInterfaceTest, SlowerPeerMatchedByTimeout) {
+  // b sends keep-alives every 20 s; three of a's own 5 s intervals would close the link
+  this->a.link.set_keep_alive(5000);
+  this->a.link.set_receive_timeout(60000);
+  this->b.link.set_keep_alive(20000);
+  this->b.link.set_receive_timeout(60000);
+  for (uint32_t now = 1000; now <= 121000; now += 5000) {
+    set_time(now);
+    this->loops(2);
+  }
+  EXPECT_TRUE(this->a.link.is_connected());
+  EXPECT_TRUE(this->b.link.is_connected());
+  EXPECT_EQ(this->a.states, std::vector<bool>{true});
+}
+
+TEST_F(TcpPacketInterfaceTest, ConnectedSensorStartsFalse) {
+  TestTcp link;
+  binary_sensor::BinarySensor sensor;
+  link.set_connected_sensor(&sensor);
+  link.setup();
+  EXPECT_TRUE(sensor.has_state());
+  EXPECT_FALSE(sensor.state);
+}
+
+TEST_F(TcpPacketInterfaceTest, OversizedLengthClosesLink) {
+  End c;
+  int fds[2];
+  ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
+  c.link.link().adopt(std::make_unique<socket::Socket>(fds[0]));
+  c.link.loop();
+  uint8_t header[2] = {0xFF, 0x00};  // 255 bytes, more than 64
+  ASSERT_EQ(::write(fds[1], header, sizeof(header)), 2);
+  c.link.loop();
+  EXPECT_FALSE(c.link.is_connected());
+  EXPECT_EQ(c.states, (std::vector<bool>{true, false}));
+  ::close(fds[1]);
+}
+
+TEST_F(TcpPacketInterfaceTest, NotSentWhileDown) {
+  this->a.link.link().close();
+  this->a.link.loop();
+  EXPECT_FALSE(this->a.send({1}));
+}
+
+TEST_F(TcpPacketInterfaceTest, FrameAssembledAcrossReads) {
+  End c;
+  int fds[2];
+  ASSERT_EQ(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
+  c.link.link().adopt(std::make_unique<socket::Socket>(fds[0]));
+  c.link.loop();
+  // A keep-alive, then a 3 byte packet in three writes: half the length, the rest of it, then the payload
+  const uint8_t keep_alive[2] = {0x00, 0x00};
+  ASSERT_EQ(::write(fds[1], keep_alive, 2), 2);
+  c.link.loop();
+  const uint8_t length_low = 0x03;
+  ASSERT_EQ(::write(fds[1], &length_low, 1), 1);
+  c.link.loop();
+  const uint8_t length_high_and_start[2] = {0x00, 0xAA};
+  ASSERT_EQ(::write(fds[1], length_high_and_start, 2), 2);
+  c.link.loop();
+  EXPECT_TRUE(c.received.empty());
+  const uint8_t rest[2] = {0xBB, 0xCC};
+  ASSERT_EQ(::write(fds[1], rest, 2), 2);
+  c.link.loop();
+  EXPECT_EQ(c.received, (std::vector<Packet>{{0xAA, 0xBB, 0xCC}}));
+  EXPECT_TRUE(c.link.is_connected());
+  ::close(fds[1]);
+}
+
+// Waits in the main loop's select(), which marks the sockets that are ready to read
+static void wait_for_sockets() { esphome::internal::wakeable_delay(5); }
+
+// The server role, accepting connections on a loopback port
+class TcpServerTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    signal(SIGPIPE, SIG_IGN);
+    App.~Application();
+    new (&App) Application();
+    static char name[] = "tcp-tests";
+    static char friendly_name[] = "";
+    App.pre_setup(name, sizeof(name) - 1, friendly_name, 0);
+    set_time(1000);
+    this->port = free_port();
+    this->server.link.set_server(true);
+    this->server.link.set_port(this->port);
+  }
+  void TearDown() override {
+    this->server.link.on_shutdown();
+    for (int fd : this->clients)
+      ::close(fd);
+    App.~Application();
+    new (&App) Application();  // later tests and the test listener use App
+  }
+  static void set_time(uint32_t now) { LoopBlockingGuard guard{nullptr, nullptr, now}; }
+  static uint16_t free_port() {
+    int fd = ::socket(AF_INET, SOCK_STREAM, 0);
+    struct sockaddr_in addr{};
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    socklen_t len = sizeof(addr);
+    EXPECT_EQ(::bind(fd, reinterpret_cast<struct sockaddr *>(&addr), len), 0);
+    EXPECT_EQ(::getsockname(fd, reinterpret_cast<struct sockaddr *>(&addr), &len), 0);
+    ::close(fd);
+    return ntohs(addr.sin_port);
+  }
+  // Connects a client and runs the server's loop until it has accepted it
+  int connect_client() {
+    int fd = ::socket(AF_INET, SOCK_STREAM, 0);
+    struct sockaddr_in addr{};
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    addr.sin_port = htons(this->port);
+    EXPECT_EQ(::connect(fd, reinterpret_cast<struct sockaddr *>(&addr), sizeof(addr)), 0);
+    this->clients.push_back(fd);
+    for (int i = 0; i != 100 && !this->server.link.is_connected(); i++) {
+      this->server.link.loop();
+      wait_for_sockets();
+    }
+    return fd;
+  }
+
+  End server;
+  uint16_t port{0};
+  std::vector<int> clients;
+};
+
+TEST_F(TcpServerTest, AcceptsClientAndReceives) {
+  this->server.link.loop();  // starts listening
+  int client = this->connect_client();
+  ASSERT_TRUE(this->server.link.is_connected());
+  EXPECT_EQ(this->server.states, std::vector<bool>{true});
+  const uint8_t frame[3] = {0x01, 0x00, 0x42};
+  ASSERT_EQ(::write(client, frame, sizeof(frame)), 3);
+  for (int i = 0; i != 100 && this->server.received.empty(); i++) {
+    this->server.link.loop();
+    wait_for_sockets();
+  }
+  EXPECT_EQ(this->server.received, std::vector<Packet>{{0x42}});
+}
+
+TEST_F(TcpServerTest, AcceptsNextClientAfterDisconnect) {
+  this->server.link.loop();
+  int first = this->connect_client();
+  ASSERT_TRUE(this->server.link.is_connected());
+  ::close(first);
+  this->clients.clear();
+  for (int i = 0; i != 100 && this->server.link.is_connected(); i++) {
+    this->server.link.loop();
+    wait_for_sockets();
+  }
+  EXPECT_EQ(this->server.states, (std::vector<bool>{true, false}));
+  this->connect_client();
+  EXPECT_TRUE(this->server.link.is_connected());
+  EXPECT_EQ(this->server.states, (std::vector<bool>{true, false, true}));
+}
+
+}  // namespace esphome::tcp::testing
+
+#endif

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

@@ -0,0 +1,2 @@
+packages:
+  tcp: !include common.yaml

tests/components/udp/benchmark.yaml

@@ -1,5 +1,5 @@
 udp:
-  addresses: ["192.168.1.255"]
+  addresses: ["192.168.1.60"]
 
 packet_interface:
   - platform: udp

tests/components/udp/udp_packet_interface.cpp

@@ -1,6 +1,8 @@
 #include <gtest/gtest.h>
 
+#include <array>
 #include <cstdint>
+#include <cstring>
 #include <vector>
 
 #include "esphome/components/udp/packet_interface/udp_packet_interface.h"
@@ -25,29 +27,41 @@ struct UdpBufferStorage {
   uint8_t tx_buffer_storage[8]{};
 };
 
+static constexpr std::array<uint8_t, 4> PEER{192, 168, 1, 60};
+
 class TestUdp : public UdpBufferStorage, public UdpPacketInterface {
  public:
-  explicit TestUdp(UDPComponent *udp) : UdpPacketInterface(udp, this->tx_buffer_storage) {
+  explicit TestUdp(UDPComponent *udp, std::array<uint8_t, 4> peer = {})
+      : UdpPacketInterface(udp, this->tx_buffer_storage, peer) {
     this->add_packet_interface_listener(
         [this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(to_vector(data)); });
   }
+  using UdpPacketInterface::accepts_from_;
   using UdpPacketInterface::receive_;
 
   std::vector<Packet> received;
 };
 
+static sockaddr_in ipv4(std::array<uint8_t, 4> address) {
+  sockaddr_in addr{};
+  addr.sin_family = AF_INET;
+  memcpy(&addr.sin_addr.s_addr, address.data(), address.size());
+  return addr;
+}
+
 // The udp component is not set up, so it has no destinations and a send does nothing
 class UdpPacketInterfaceTest : public ::testing::Test {
  protected:
   UDPComponent udp_;
   TestUdp link_{&this->udp_};
+  sockaddr_in sender_{ipv4({10, 0, 0, 1})};
 };
 
 TEST_F(UdpPacketInterfaceTest, ReceivedPacketsUpToMaxPassedUp) {
   Packet packet{1, 2, 3, 4, 5, 6, 7, 8};
-  this->link_.receive_(packet);
+  this->link_.receive_(packet, reinterpret_cast<sockaddr *>(&this->sender_));
   Packet oversized{1, 2, 3, 4, 5, 6, 7, 8, 9};
-  this->link_.receive_(oversized);
+  this->link_.receive_(oversized, reinterpret_cast<sockaddr *>(&this->sender_));
   EXPECT_EQ(this->link_.received, std::vector<Packet>{packet});
 }
 
@@ -73,6 +87,29 @@ TEST_F(UdpPacketInterfaceTest, OversizedPacketNotSent) {
   EXPECT_FALSE(this->link_.send_to_interface(PacketBuffer(&header, 1, &body, nullptr, 0), {}));
 }
 
+TEST_F(UdpPacketInterfaceTest, WithoutPeerAcceptsAnySender) {
+  EXPECT_FALSE(this->link_.has_capability(packet_interface::CAPABILITY_ADDRESSING));
+  EXPECT_TRUE(this->link_.accepts_from_(reinterpret_cast<sockaddr *>(&this->sender_)));
+}
+
+TEST_F(UdpPacketInterfaceTest, WithPeerAcceptsOnlyPeer) {
+  TestUdp link(&this->udp_, PEER);
+  EXPECT_TRUE(link.has_capability(packet_interface::CAPABILITY_ADDRESSING));
+  auto peer = ipv4(PEER);
+  Packet packet{1, 2};
+  link.receive_(packet, reinterpret_cast<sockaddr *>(&this->sender_));
+  EXPECT_TRUE(link.received.empty());
+  link.receive_(packet, reinterpret_cast<sockaddr *>(&peer));
+  EXPECT_EQ(link.received, std::vector<Packet>{packet});
+}
+
+TEST_F(UdpPacketInterfaceTest, WithPeerDropsSenderNotIpv4) {
+  TestUdp link(&this->udp_, PEER);
+  sockaddr other{};
+  other.sa_family = AF_UNSPEC;
+  EXPECT_FALSE(link.accepts_from_(&other));
+}
+
 }  // namespace esphome::udp::testing
 
 #endif

tests/unit_tests/test_entity_transport.py

@@ -79,6 +79,54 @@ def test_timeout_not_with_session(setup_core, layers: dict, allowed: bool) -> No
         fv.full_config.reset(token)
 
 
+def test_timeout_not_with_session_on_peer(setup_core) -> None:
+    """The layers of an addressing peer are found by the peer's id."""
+    peers = [
+        {CONF_ID: ID("plain", is_manual=True)},
+        {CONF_ID: ID("with_session", is_manual=True), "session": {}},
+    ]
+    token = fv.full_config.set(
+        {
+            "packet_interface": [
+                {CONF_ID: ID("bus", is_manual=True), "addressing": {"peers": peers}}
+            ]
+        }
+    )
+    try:
+        FINAL_VALIDATE_SCHEMA(
+            CONFIG_SCHEMA({"packet_interface": "plain", CONF_TIMEOUT: "1min"})
+        )
+        with pytest.raises(cv.Invalid, match="session"):
+            FINAL_VALIDATE_SCHEMA(
+                CONFIG_SCHEMA(
+                    {"packet_interface": "with_session", CONF_TIMEOUT: "1min"}
+                )
+            )
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_timeout_not_with_link_session(setup_core) -> None:
+    """A link such as TCP does the session's job itself, so it has no session key but the same rule applies."""
+    from esphome.components.tcp.packet_interface import TcpPacketInterface
+
+    token = fv.full_config.set(
+        {
+            "packet_interface": [
+                {CONF_ID: ID("tcp_link", is_manual=True, type=TcpPacketInterface)}
+            ]
+        }
+    )
+    try:
+        with pytest.raises(cv.Invalid, match="session") as error:
+            FINAL_VALIDATE_SCHEMA(
+                CONFIG_SCHEMA({"packet_interface": "tcp_link", CONF_TIMEOUT: "1min"})
+            )
+        assert error.value.path == [CONF_TIMEOUT]
+    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"}

tests/unit_tests/test_espnow_packet_interface.py

@@ -0,0 +1,51 @@
+"""Unit tests for the espnow packet interface platform."""
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.espnow import packet_interface as espnow_packet_interface
+from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE
+import esphome.final_validate as fv
+
+ADDRESSING = {"address": 1, "destination": 2}
+
+
+def test_broadcast_takes_addressing(setup_core) -> None:
+    """Broadcast is the default, and an addressing layer tells the devices apart."""
+    config = espnow_packet_interface.CONFIG_SCHEMA({"addressing": ADDRESSING})
+    assert str(config["peer_address"]) == "FF:FF:FF:FF:FF:FF"
+    assert config[CONF_MAX_PACKET_SIZE] == 250
+
+
+def test_peer_rejects_addressing(setup_core) -> None:
+    """Packets to one peer are already addressed."""
+    assert espnow_packet_interface.CONFIG_SCHEMA({"peer_address": "11:22:33:44:55:66"})
+    with pytest.raises(cv.Invalid, match="addressing"):
+        espnow_packet_interface.CONFIG_SCHEMA(
+            {"peer_address": "11:22:33:44:55:66", "addressing": ADDRESSING}
+        )
+
+
+def test_broadcast_needs_addressing_for_two_device_layers(setup_core) -> None:
+    """Every device hears a broadcast link, so reliability needs an addressing layer there."""
+    with pytest.raises(cv.Invalid, match="every device hears this link"):
+        espnow_packet_interface.CONFIG_SCHEMA({"reliability": None})
+    assert espnow_packet_interface.CONFIG_SCHEMA(
+        {"reliability": None, "addressing": ADDRESSING}
+    )
+    assert espnow_packet_interface.CONFIG_SCHEMA(
+        {"peer_address": "11:22:33:44:55:66", "reliability": None}
+    )
+
+
+def test_packet_size_within_payload_size(setup_core) -> None:
+    """A packet cannot be larger than the espnow component's max_payload_size."""
+    token = fv.full_config.set({"espnow": {"max_payload_size": 250}})
+    try:
+        assert espnow_packet_interface._validate_packet_size(
+            {CONF_MAX_PACKET_SIZE: 250}
+        )
+        with pytest.raises(cv.Invalid, match="max_payload_size"):
+            espnow_packet_interface._validate_packet_size({CONF_MAX_PACKET_SIZE: 251})
+    finally:
+        fv.full_config.reset(token)

tests/unit_tests/test_lora_packet_interface.py

@@ -0,0 +1,210 @@
+"""Unit tests for the sx126x and sx127x packet interface platforms."""
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE, radio_link
+from esphome.components.sx126x import packet_interface as sx126x_packet_interface
+from esphome.components.sx127x import packet_interface as sx127x_packet_interface
+import esphome.final_validate as fv
+
+
+@pytest.mark.parametrize(
+    ("schema", "radio_key"),
+    [
+        (sx126x_packet_interface.CONFIG_SCHEMA, "sx126x_id"),
+        (sx127x_packet_interface.CONFIG_SCHEMA, "sx127x_id"),
+    ],
+)
+def test_max_packet_size_is_radio_limit(
+    setup_core, schema: cv.Schema, radio_key: str
+) -> None:
+    """A radio packet holds at most 255 bytes, which is also the default."""
+    assert schema({radio_key: "radio"})[CONF_MAX_PACKET_SIZE] == 255
+    with pytest.raises(cv.Invalid):
+        schema({radio_key: "radio", CONF_MAX_PACKET_SIZE: 256})
+
+
+def _lora(**options: object) -> dict:
+    return {
+        "modulation": "LORA",
+        "spreading_factor": 7,
+        "bandwidth": "125_0kHz",
+        "coding_rate": "CR_4_5",
+        "preamble_size": 8,
+        "crc_enable": False,
+        **options,
+    }
+
+
+@pytest.mark.parametrize(
+    ("radio", "airtime_ms"),
+    [
+        # Worked by hand from the Semtech formula, for a 255 byte packet with an explicit header
+        (_lora(spreading_factor=7), 394.5),
+        (_lora(spreading_factor=7, crc_enable=True), 399.6),
+        (_lora(spreading_factor=10), 2254.8),
+        # A symbol takes over 16 ms, so the low data rate optimization lengthens the packet
+        (_lora(spreading_factor=12), 9019.4),
+    ],
+)
+def test_lora_airtime(radio: dict, airtime_ms: float) -> None:
+    """The airtime of the largest LoRa packet follows the Semtech formula."""
+    assert radio_link.packet_airtime_ms(radio, 255) == pytest.approx(
+        airtime_ms, abs=0.5
+    )
+
+
+def test_max_busy_follows_airtime() -> None:
+    """Listen before talk waits at least 2 s, and longer than the largest packet takes."""
+    assert radio_link.max_busy_ms(_lora(), 255) == 2000
+    assert radio_link.max_busy_ms(_lora(spreading_factor=12), 255) > 9019
+    fsk = {
+        "modulation": "FSK",
+        "preamble_size": 4,
+        "bitrate": 1200,
+        "sync_value": [1, 2],
+    }
+    assert radio_link.packet_airtime_ms(fsk, 64) == pytest.approx(
+        (4 + 2 + 1 + 64 + 2) * 8 / 1200 * 1000
+    )
+
+
+class _FullConfig(dict):
+    """Enough of the full configuration for the radio links' final validation."""
+
+    def __init__(self, domain: str, radio: dict) -> None:
+        super().__init__({domain: [{"rx_start": True, **radio}]})
+        self.domain = domain
+
+    def get_path_for_id(self, id_: object) -> list:
+        return [self.domain, 0, "id"]
+
+    def get_config_for_path(self, path: list) -> dict:
+        return self[path[0]][path[1]]
+
+
+@pytest.mark.parametrize(
+    ("module", "domain", "radio", "link", "match", "fix"),
+    [
+        (
+            sx126x_packet_interface,
+            "sx126x",
+            _lora(payload_length=10),
+            {},
+            "payload_length",
+            {"payload_length": 0},
+        ),
+        (
+            sx127x_packet_interface,
+            "sx127x",
+            _lora(payload_length=10),
+            {},
+            "payload_length",
+            {"payload_length": 0},
+        ),
+        (
+            sx127x_packet_interface,
+            "sx127x",
+            {
+                "modulation": "FSK",
+                "packet_mode": False,
+                "payload_length": 0,
+                "sync_value": [0x12],
+            },
+            {},
+            "packet_mode",
+            {"packet_mode": True},
+        ),
+        (
+            sx127x_packet_interface,
+            "sx127x",
+            {
+                "modulation": "OOK",
+                "packet_mode": True,
+                "payload_length": 0,
+                "sync_value": [0x12],
+            },
+            {CONF_MAX_PACKET_SIZE: 65},
+            "at most 64",
+            None,
+        ),
+        (
+            sx127x_packet_interface,
+            "sx127x",
+            {
+                "modulation": "FSK",
+                "packet_mode": True,
+                "payload_length": 0,
+                "sync_value": [],
+            },
+            {},
+            "sync_value",
+            {"sync_value": [0x12]},
+        ),
+        (
+            sx126x_packet_interface,
+            "sx126x",
+            {"modulation": "FSK", "payload_length": 0, "sync_value": []},
+            {},
+            "sync_value",
+            {"sync_value": [0x12, 0x34]},
+        ),
+    ],
+)
+def test_radio_settings_checked(
+    module: object,
+    domain: str,
+    radio: dict,
+    link: dict,
+    match: str,
+    fix: dict | None,
+) -> None:
+    """Packets vary in length; FSK or OOK on an SX127x needs packet mode and a sync word, and holds at most 64 bytes;
+    FSK on an SX126x needs a sync word.
+    """
+    config = {f"{domain}_id": "radio", CONF_MAX_PACKET_SIZE: 64, **link}
+    token = fv.full_config.set(_FullConfig(domain, radio))
+    try:
+        with pytest.raises(cv.Invalid, match=match):
+            module.FINAL_VALIDATE_SCHEMA(config)
+        if fix is not None:
+            fv.full_config.set(_FullConfig(domain, {**radio, **fix}))
+            assert module.FINAL_VALIDATE_SCHEMA(config)
+    finally:
+        fv.full_config.reset(token)
+
+
+@pytest.mark.parametrize(
+    ("link", "match"),
+    [
+        ({}, None),
+        ({"encryption": "secret"}, None),
+        ({"reliability": {}}, "reliability"),
+        (
+            {
+                "addressing": {
+                    "address": 1,
+                    "peers": [
+                        {"id": "a", "destination": 2, "session": {}},
+                        {"id": "b", "destination": 3},
+                    ],
+                }
+            },
+            "session",
+        ),
+    ],
+)
+def test_rx_start_needed_for_replies(setup_core, link: dict, match: str | None) -> None:
+    """A radio that sleeps after sending suits only a link whose layers need no packets back."""
+    sleeping = _lora(payload_length=0, rx_start=False)
+    config = sx126x_packet_interface.CONFIG_SCHEMA({"sx126x_id": "radio", **link})
+    token = fv.full_config.set(_FullConfig("sx126x", sleeping))
+    try:
+        if match is None:
+            assert sx126x_packet_interface.FINAL_VALIDATE_SCHEMA(config)
+            return
+        with pytest.raises(cv.Invalid, match=f"'{match}'.*rx_start"):
+            sx126x_packet_interface.FINAL_VALIDATE_SCHEMA(config)
+    finally:
+        fv.full_config.reset(token)

tests/unit_tests/test_packet_interface.py

@@ -9,6 +9,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,
@@ -23,6 +25,8 @@ from esphome.components.packet_interface.layers.hdlc_reliability import (
 from esphome.components.uart import packet_interface as uart_packet_interface
 from esphome.components.udp import packet_interface as udp_packet_interface
 from esphome.const import CONF_ENCRYPTION
+from esphome.core import ID
+import esphome.final_validate as fv
 
 
 @pytest.mark.parametrize(
@@ -94,6 +98,7 @@ def test_layer_types_registered() -> None:
     assert {kind.key: set(kind.types) for kind in STACK_ORDER} == {
         CONF_ENCRYPTION: {"xxtea"},
         CONF_RELIABILITY: {"hdlc"},
+        CONF_FRAGMENTATION: {"basic"},
         CONF_SESSION: {"basic"},
     }
 
@@ -178,6 +183,149 @@ 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 | ID, 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_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(
+    ("layers", "valid"),
+    [
+        # 508 - 3 leaves fragments of 503 bytes: 1024 bytes take 3, 2048 bytes take 5
+        ({CONF_RELIABILITY: None, CONF_FRAGMENTATION: None}, True),
+        (
+            {CONF_RELIABILITY: None, CONF_FRAGMENTATION: {"max_packet_size": 2048}},
+            False,
+        ),
+        (
+            {
+                CONF_RELIABILITY: {CONF_MAX_QUEUE_LENGTH: 5},
+                CONF_FRAGMENTATION: {"max_packet_size": 2048},
+            },
+            True,
+        ),
+        ({CONF_FRAGMENTATION: {"max_packet_size": 16384}}, True),
+    ],
+)
+def test_fragments_fit_reliability_queue(setup_core, layers: dict, valid: bool) -> None:
+    """Every fragment of the largest packet must fit in the reliability queue at once."""
+    if valid:
+        uart_packet_interface.CONFIG_SCHEMA(layers)
+        return
+    with pytest.raises(cv.Invalid, match="5 fragments") as err:
+        uart_packet_interface.CONFIG_SCHEMA(layers)
+    assert err.value.path == [CONF_FRAGMENTATION, CONF_MAX_PACKET_SIZE]
+
+
+@pytest.mark.parametrize(
+    ("link", "match"),
+    [
+        # 256 - 4 for addressing - 16 for the rolling code - 3 for reliability leaves fragments of 231 bytes
+        ({}, "5 fragments of 231 bytes"),
+        # 24 - 4 - 16 - 3 leaves 1 byte, too few for the fragment header
+        ({CONF_MAX_PACKET_SIZE: 24}, "no room"),
+    ],
+)
+def test_fragments_checked_in_peers(setup_core, link: dict, match: str) -> None:
+    """Fragments are checked against the layers of each peer, below the addressing layer."""
+    peer = _peer(
+        "a",
+        2,
+        encryption={"key": "secret", "rolling_code_enable": True},
+        reliability=None,
+        fragmentation=None,
+        session=None,
+    )
+    with pytest.raises(cv.Invalid, match=match) as err:
+        uart_packet_interface.CONFIG_SCHEMA(
+            {
+                CONF_MAX_PACKET_SIZE: 256,
+                **link,
+                CONF_ADDRESSING: {"address": 1, "peers": [peer]},
+            }
+        )
+    assert err.value.path[:4] == [CONF_ADDRESSING, "peers", 0, CONF_FRAGMENTATION]
+
+
 @pytest.mark.parametrize(
     ("config", "loads"),
     [
@@ -185,6 +333,15 @@ def test_rolling_code_requires_session(setup_core) -> None:
         ({CONF_RELIABILITY: None, CONF_SESSION: None}, []),
         ({CONF_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(
@@ -214,3 +371,38 @@ def test_xxtea_key_stretched() -> None:
     assert key != xxtea_encryption.derive_key("Johnny Cash ")
     sha = hashlib.sha256(b"Johnny Cash").digest()
     assert key[0] != int.from_bytes(sha[:4], "little")
+
+
+@pytest.mark.parametrize(
+    "addressing",
+    [
+        {"address": 1, "destination": 1},
+        {"address": 1, "peers": [_peer("a", 2), _peer("b", 1)]},
+    ],
+)
+def test_addressing_own_address_rejected(setup_core, addressing: dict) -> None:
+    """A device cannot send to its own address, as its own packets are dropped as echoes."""
+    with pytest.raises(cv.Invalid, match="own address"):
+        uart_packet_interface.CONFIG_SCHEMA({CONF_ADDRESSING: addressing})
+
+
+def test_application_needs_peer_id_with_peers(setup_core) -> None:
+    """An application must use one peer's id, not the id of a link with peers."""
+    validator = packet_interface.final_validate_interface("packet_interface")
+    token = fv.full_config.set(
+        {
+            "packet_interface": [
+                {"id": ID("bus"), CONF_ADDRESSING: {"peers": [_peer(ID("a"), 2)]}},
+                {"id": ID("single"), CONF_ADDRESSING: {"destination": 2}},
+            ]
+        }
+    )
+    try:
+        with pytest.raises(
+            cv.Invalid, match="has peers, so use the id of one of them, such as 'a'"
+        ):
+            validator({"packet_interface": ID("bus")})
+        assert validator({"packet_interface": ID("single")})
+        assert validator({"packet_interface": ID("a")})
+    finally:
+        fv.full_config.reset(token)

tests/unit_tests/test_tcp_packet_interface.py

@@ -0,0 +1,84 @@
+"""Unit tests for the tcp packet interface platform."""
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components import socket
+from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE
+from esphome.components.tcp import packet_interface as tcp_packet_interface
+from esphome.core import CORE
+
+
+def test_client_and_server(setup_core) -> None:
+    """A client needs a host; a server listens on its port."""
+    client = tcp_packet_interface.CONFIG_SCHEMA({"host": "192.168.1.50", "port": 6060})
+    assert client["role"] == "client"
+    assert client[CONF_MAX_PACKET_SIZE] == 1022
+    assert tcp_packet_interface.CONFIG_SCHEMA({"role": "server", "port": 6060})
+    with pytest.raises(cv.Invalid):
+        tcp_packet_interface.CONFIG_SCHEMA({"port": 6060})
+
+
+@pytest.mark.parametrize("layer", ["reliability", "session", "addressing"])
+def test_provided_layers_rejected(setup_core, layer: str) -> None:
+    """TCP provides these itself, so they cannot be configured on it."""
+    with pytest.raises(cv.Invalid, match=f"provides '{layer}' itself"):
+        tcp_packet_interface.CONFIG_SCHEMA(
+            {"host": "192.168.1.50", "port": 6060, layer: {"address": 1}}
+        )
+
+
+def test_rolling_code_rejected(setup_core) -> None:
+    """The rolling code needs the session's keep-alives, which TCP does not send through the layers."""
+    with pytest.raises(cv.Invalid, match="cannot be used on this link"):
+        tcp_packet_interface.CONFIG_SCHEMA(
+            {
+                "host": "192.168.1.50",
+                "port": 6060,
+                "encryption": {"key": "secret", "rolling_code_enable": True},
+            }
+        )
+
+
+def test_auto_load_binary_sensor_for_connected(setup_core) -> None:
+    """The binary_sensor component is loaded only for the connected sensor."""
+    assert tcp_packet_interface.AUTO_LOAD() == ["socket", "binary_sensor"]
+    assert tcp_packet_interface.AUTO_LOAD({}) == ["socket", "binary_sensor"]
+    config = {"host": "192.168.1.50", "port": 6060}
+    assert tcp_packet_interface.AUTO_LOAD(config) == ["socket"]
+    config["connected"] = {"name": "Link"}
+    assert tcp_packet_interface.AUTO_LOAD(config) == ["socket", "binary_sensor"]
+
+
+def test_timeout(setup_core) -> None:
+    """The timeout follows keep_alive unless given, and must be longer than it."""
+    base = {"host": "192.168.1.50", "port": 6060}
+    config = tcp_packet_interface.CONFIG_SCHEMA({**base, "keep_alive": "2s"})
+    assert config["timeout"].total_milliseconds == 6000
+    # A peer that sends keep-alives less often is matched with a longer timeout
+    config = tcp_packet_interface.CONFIG_SCHEMA({**base, "timeout": "90s"})
+    assert config["timeout"].total_milliseconds == 90000
+    with pytest.raises(cv.Invalid, match="longer than keep_alive"):
+        tcp_packet_interface.CONFIG_SCHEMA(
+            {**base, "keep_alive": "5s", "timeout": "5s"}
+        )
+    with pytest.raises(cv.Invalid):
+        tcp_packet_interface.CONFIG_SCHEMA({**base, "keep_alive": "2h"})
+    with pytest.raises(cv.Invalid):
+        tcp_packet_interface.CONFIG_SCHEMA({**base, "timeout": "4h"})
+
+
+@pytest.mark.parametrize(
+    ("role", "tcp", "listen"), [("client", 1, 0), ("server", 1, 1)]
+)
+def test_sockets_reserved(setup_core, role: str, tcp: int, listen: int) -> None:
+    """A client reserves one socket; a server also reserves a listener."""
+    config = {"role": role, "port": 6060}
+    if role == "client":
+        config["host"] = "192.168.1.50"
+    tcp_packet_interface.CONFIG_SCHEMA(config)
+    name = "tcp.packet_interface"
+    assert CORE.data[socket.KEY_SOCKET_CONSUMERS_TCP][name] == tcp
+    assert (
+        CORE.data.get(socket.KEY_SOCKET_CONSUMERS_TCP_LISTEN, {}).get(name, 0) == listen
+    )

tests/unit_tests/test_udp_packet_interface.py

@@ -0,0 +1,79 @@
+"""Unit tests for the udp packet interface platform."""
+
+from ipaddress import IPv4Address, IPv6Address
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.udp import (
+    CONF_ADDRESSES,
+    packet_interface as udp_packet_interface,
+)
+from esphome.core import ID
+import esphome.final_validate as fv
+
+
+@pytest.mark.parametrize(
+    ("addresses", "point_to_point", "peer"),
+    [
+        ([IPv4Address("192.168.1.60")], None, IPv4Address("192.168.1.60")),
+        ([IPv4Address("255.255.255.255")], None, None),
+        ([IPv4Address("192.168.1.255")], None, None),
+        ([IPv4Address("239.0.60.53")], None, None),
+        ([IPv4Address("192.168.1.60"), IPv4Address("192.168.1.61")], None, None),
+        ([IPv6Address("2001:db8::1")], None, None),
+        # A directed broadcast on a /25, and a host ending in .255 on a /23
+        ([IPv4Address("192.168.1.127")], False, None),
+        ([IPv4Address("192.168.0.255")], True, IPv4Address("192.168.0.255")),
+    ],
+)
+def test_peer(
+    addresses: list, point_to_point: bool | None, peer: IPv4Address | None
+) -> None:
+    """Only one unicast IPv4 destination makes the link point to point, unless point_to_point says otherwise."""
+    config = {} if point_to_point is None else {"point_to_point": point_to_point}
+    assert udp_packet_interface._peer({CONF_ADDRESSES: addresses}, config) == peer
+
+
+@pytest.mark.parametrize(
+    "addresses",
+    [
+        [IPv4Address("239.0.60.53")],
+        [IPv4Address("192.168.1.60"), IPv4Address("192.168.1.61")],
+    ],
+)
+def test_point_to_point_needs_one_unicast_address(addresses: list) -> None:
+    """point_to_point cannot make a multicast or several addresses into one peer."""
+    with pytest.raises(cv.Invalid, match="one unicast IPv4 address"):
+        udp_packet_interface._peer(
+            {CONF_ADDRESSES: addresses}, {"point_to_point": True}
+        )
+
+
+def _final_validate(addresses: list, layers: dict) -> None:
+    config = udp_packet_interface.CONFIG_SCHEMA({"udp_id": "udp_link", **layers})
+    token = fv.full_config.set(
+        {"udp": [{"id": ID("udp_link"), CONF_ADDRESSES: addresses}]}
+    )
+    try:
+        udp_packet_interface._final_validate(config)
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_point_to_point_rejects_addressing(setup_core) -> None:
+    """Packets to one unicast address are already addressed."""
+    addressing = {"addressing": {"address": 1, "destination": 2}}
+    with pytest.raises(cv.Invalid, match="already addressed"):
+        _final_validate([IPv4Address("192.168.1.60")], addressing)
+    _final_validate([IPv4Address("192.168.1.255")], addressing)
+    _final_validate(
+        [IPv4Address("192.168.1.127")], {**addressing, "point_to_point": False}
+    )
+
+
+def test_broadcast_needs_addressing_for_two_device_layers(setup_core) -> None:
+    """Every device hears a broadcast link, so reliability needs an addressing layer there."""
+    with pytest.raises(cv.Invalid, match="every device hears this link"):
+        _final_validate([IPv4Address("192.168.1.255")], {"reliability": None})
+    _final_validate([IPv4Address("192.168.1.60")], {"reliability": None})