[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

@@ -10,7 +10,7 @@ from esphome.components.const import CONF_PROVIDER, CONF_REMOTE_ID
 from esphome.components.packet_interface import (
     CONF_PACKET_INTERFACE,
     final_validate_interface,
-    interface_layers,
+    interface_has_layer,
 )
 from esphome.components.packet_interface.layers import CONF_SESSION
 from esphome.components.sensor import Sensor
@@ -102,11 +102,13 @@ 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

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

@@ -19,12 +19,14 @@ from .layers import (
     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,
@@ -40,6 +42,9 @@ CONF_PACKET_INTERFACE = "packet_interface"
 for module_info in pkgutil.iter_modules(layers.__path__):
     importlib.import_module(f".layers.{module_info.name}", package=__package__)
 
+# 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(),
@@ -48,6 +53,7 @@ LINK_LAYER_SCHEMAS = {
 __all__ = [
     "CONF_MAX_PACKET_SIZE",
     "CONF_PACKET_INTERFACE",
+    "interface_has_layer",
     "interface_layers",
     "PacketInterface",
     "final_validate_interface",
@@ -114,6 +120,16 @@ def interface_layers(interface_id: ID) -> ConfigType | None:
     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."""
 
@@ -132,25 +148,46 @@ def final_validate_interface(key: str) -> Callable[[ConfigType], ConfigType]:
     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:

esphome/components/packet_interface/layers/__init__.py

@@ -172,19 +172,24 @@ 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):
-            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",
-                path=[kind.key],
-            )
+            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:
@@ -222,6 +227,21 @@ def check_packet_sizes(config: ConfigType, size: int) -> None:
         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],
+            )
+
+
 def stack_auto_load(config: ConfigType) -> set[str]:
     """The components that the layers configured in config need loaded."""
     loads: set[str] = set()

esphome/components/packet_interface/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/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

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/common.yaml

@@ -4,6 +4,9 @@ wifi:
 
 udp:
   - id: packet_interface_udp
+  - id: packet_interface_udp_peer
+    port: 18512
+    addresses: ["192.168.1.60"]
 
 packet_interface:
   - platform: uart
@@ -45,6 +48,12 @@ packet_interface:
     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:
   - platform: template
@@ -60,3 +69,7 @@ packet_transport:
     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

@@ -106,6 +106,27 @@ def test_timeout_not_with_session_on_peer(setup_core) -> None:
         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_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})