diff --git a/esphome/components/entity_transport/__init__.py b/esphome/components/entity_transport/__init__.py index b0c6a83efd1..08c6b91569f 100644 --- a/esphome/components/entity_transport/__init__.py +++ b/esphome/components/entity_transport/__init__.py @@ -7,7 +7,11 @@ import esphome.codegen as cg from esphome.components import packet_interface from esphome.components.binary_sensor import BinarySensor from esphome.components.const import CONF_PROVIDER, CONF_REMOTE_ID -from esphome.components.packet_interface import CONF_PACKET_INTERFACE, interface_layers +from esphome.components.packet_interface import ( + CONF_PACKET_INTERFACE, + final_validate_interface, + interface_layers, +) from esphome.components.packet_interface.layers import CONF_SESSION from esphome.components.sensor import Sensor import esphome.config_validation as cv @@ -108,7 +112,9 @@ def _final_validate(config: ConfigType) -> ConfigType: return config -FINAL_VALIDATE_SCHEMA = _final_validate +FINAL_VALIDATE_SCHEMA = cv.All( + final_validate_interface(CONF_PACKET_INTERFACE), _final_validate +) def _validate_remote_id(config: ConfigType) -> ConfigType: diff --git a/esphome/components/packet_interface/__init__.py b/esphome/components/packet_interface/__init__.py index d1f07018794..4cd2f031a25 100644 --- a/esphome/components/packet_interface/__init__.py +++ b/esphome/components/packet_interface/__init__.py @@ -1,3 +1,4 @@ +from collections.abc import Callable import importlib import pkgutil from typing import Any @@ -13,9 +14,16 @@ from esphome.types import ConfigType from . import layers from .layers import ( + ADDRESSING, + BROADCAST_ADDRESS, + CONF_ADDRESSING, + CONF_DESTINATION, + CONF_PEERS, + STACK_ORDER, PacketInterface, build_stack, check_session, + check_single_peer, packet_interface_ns, stack_auto_load, stack_schema, @@ -31,13 +39,17 @@ CONF_PACKET_INTERFACE = "packet_interface" for module_info in pkgutil.iter_modules(layers.__path__): importlib.import_module(f".layers.{module_info.name}", package=__package__) -LINK_LAYER_SCHEMAS = stack_schema() +LINK_LAYER_SCHEMAS = { + cv.Optional(CONF_ADDRESSING): ADDRESSING.schema(), + **stack_schema(), +} __all__ = [ "CONF_MAX_PACKET_SIZE", "CONF_PACKET_INTERFACE", "interface_layers", "PacketInterface", + "final_validate_interface", "new_packet_interface", "packet_interface_ns", "packet_interface_schema", @@ -45,7 +57,7 @@ __all__ = [ def AUTO_LOAD(config: list[ConfigType] | None = None) -> list[str]: - """Load only what the configured layers use. + """Load only what the configured layers use, on the link and in each addressing peer. Without a config, as tooling that lists dependencies calls it, everything a layer could load. """ @@ -54,22 +66,61 @@ def AUTO_LOAD(config: list[ConfigType] | None = None) -> list[str]: loads: set[str] = set() for interface in config: loads |= stack_auto_load(interface) + for peer in interface.get(CONF_ADDRESSING, {}).get(CONF_PEERS, []): + loads |= stack_auto_load(peer) return sorted(loads) def _validate_layers(config: ConfigType) -> ConfigType: + """With peers, the layers go in each peer; otherwise check them against the one destination.""" check_session(config) + if (addressing := config.get(CONF_ADDRESSING)) is None: + return config + if CONF_PEERS in addressing: + for kind in STACK_ORDER: + if kind.key in config: + raise cv.Invalid( + f"With '{CONF_PEERS}', '{kind.key}' goes in each peer", + path=[kind.key], + ) + return config + addressing.setdefault(CONF_DESTINATION, BROADCAST_ADDRESS) + check_single_peer(config, addressing[CONF_DESTINATION]) return config def interface_layers(interface_id: ID) -> ConfigType | None: - """During final validation, the configuration holding the layers of the packet interface with this id.""" + """During final validation, the configuration holding the layers of the packet interface with this id. + + The id is a link's, or an addressing peer's. + """ for interface in fv.full_config.get().get(DOMAIN, []): if interface[CONF_ID].id == interface_id.id: return interface + for peer in interface.get(CONF_ADDRESSING, {}).get(CONF_PEERS, []): + if peer[CONF_ID].id == interface_id.id: + return peer return None +def final_validate_interface(key: str) -> Callable[[ConfigType], ConfigType]: + """Final validation for an application: a link with several peers must be used through one of the peers.""" + + def validator(config: ConfigType) -> ConfigType: + interface_id: ID = config[key] + for interface in fv.full_config.get().get(DOMAIN, []): + if interface[CONF_ID].id == interface_id.id and CONF_PEERS in interface.get( + CONF_ADDRESSING, {} + ): + raise cv.Invalid( + f"'{interface_id.id}' has several peers, so use the id of one of them", + path=[key], + ) + return config + + return validator + + def packet_interface_schema( class_: MockObjClass, default_max_packet_size: int, @@ -97,10 +148,17 @@ async def new_packet_interface(config: ConfigType, *args: Any) -> MockObj: The link's constructor takes its maximum packet size, usually as a std::span over a buffer that code generation sized from the configuration, so the two cannot differ. - Returns the link layer; applications reach the top layer with PacketInterface::top(). + Returns the link layer; applications reach the top layer with PacketInterface::top(). With several addressing + peers, applications use the peers' ids instead. """ var = cg.new_Pvariable(config[CONF_ID], *args) cg.add(var.set_name(config[CONF_ID].id)) await cg.register_component(var, config) - await build_stack(config, var) + if (addressing := config.get(CONF_ADDRESSING)) is None: + await build_stack(config, var) + return var + for port, layers_config in await ADDRESSING.get_type(addressing).build( + addressing, var + ): + await build_stack(config if layers_config is None else layers_config, port) return var diff --git a/esphome/components/packet_interface/basic_addressing.cpp b/esphome/components/packet_interface/basic_addressing.cpp new file mode 100644 index 00000000000..1b5e66196a5 --- /dev/null +++ b/esphome/components/packet_interface/basic_addressing.cpp @@ -0,0 +1,87 @@ +#include "basic_addressing.h" +#ifdef USE_PACKET_INTERFACE_BASIC_ADDRESSING +#include "esphome/core/log.h" + +namespace esphome::packet_interface { + +ESPHOME_LOG_TAG(TAG, "packet_interface.addressing"); + +void BasicAddressing::setup() { + this->link_->add_packet_interface_listener( + [this](const PacketBuffer &data, const PacketMetaData &meta_data) { this->receive_(data, meta_data); }); +} + +void BasicAddressing::dump_config() { + ESP_LOGCONFIG(TAG, + "Basic Addressing on '%s':\n" + " Address: 0x%04X", + this->link_->get_name(), this->address_); +} + +bool BasicAddressing::send(const PacketBuffer &data, const PacketMetaData &meta_data, uint16_t destination) { + if (meta_data.destination_address != 0) + destination = meta_data.destination_address; + uint8_t header[HEADER_SIZE] = {static_cast(destination), static_cast(destination >> 8), + static_cast(this->address_), static_cast(this->address_ >> 8)}; + PacketBuffer packet(header, HEADER_SIZE, &data, nullptr, 0); + return this->link_->send_to_interface(packet, meta_data); +} + +void BasicAddressing::receive_(const PacketBuffer &data, const PacketMetaData &meta_data) { + // Lower layers deliver packets in one piece; the payload is passed up without copying + if (!data.is_contiguous()) { + ESP_LOGW(TAG, "%s: Packet split across buffers is not supported", this->link_->get_name()); + return; + } + auto packet = data.as_span(); + if (packet.size() < HEADER_SIZE) + return; + uint16_t destination = encode_uint16(packet[1], packet[0]); + uint16_t source = encode_uint16(packet[3], packet[2]); + if (source == this->address_ || (destination != this->address_ && destination != BROADCAST_ADDRESS)) + return; + // A device with a port of its own is heard only there; the broadcast port hears the rest + AddressingPort *target = nullptr; + for (auto *port : this->ports_) { + if (port->destination_ == source) { + target = port; + break; + } + if (port->destination_ == BROADCAST_ADDRESS) + target = port; + } + if (target == nullptr) { + ESP_LOGV(TAG, "%s: Packet from 0x%04X dropped", this->link_->get_name(), source); + return; + } + PacketMetaData received = meta_data; + received.source_address = source; + received.destination_address = destination; + target->on_receive_from_interface_(PacketBuffer(packet.data() + HEADER_SIZE, packet.size() - HEADER_SIZE), received); +} + +AddressingPort::AddressingPort(BasicAddressing *addressing, uint16_t destination, bool only_port) + : addressing_(addressing), destination_(destination) { + if (only_port) { + this->stack_on_(addressing->get_link(), CAPABILITY_ADDRESSING); + } else { + this->inherit_from_(addressing->get_link(), CAPABILITY_ADDRESSING); + } + this->max_packet_size_ = addressing->get_link()->get_max_packet_size() - BasicAddressing::HEADER_SIZE; + addressing->add_port(this); +} + +void AddressingPort::dump_config() { + ESP_LOGCONFIG(TAG, + "Addressing Port '%s':\n" + " Max Packet Size: %zu", + this->name_, this->max_packet_size_); + if (this->destination_ == BasicAddressing::BROADCAST_ADDRESS) { + ESP_LOGCONFIG(TAG, " Destination: broadcast"); + } else { + ESP_LOGCONFIG(TAG, " Destination: 0x%04X", this->destination_); + } +} + +} // namespace esphome::packet_interface +#endif diff --git a/esphome/components/packet_interface/basic_addressing.h b/esphome/components/packet_interface/basic_addressing.h new file mode 100644 index 00000000000..0dbd56d5a92 --- /dev/null +++ b/esphome/components/packet_interface/basic_addressing.h @@ -0,0 +1,68 @@ +#pragma once + +#include "esphome/core/defines.h" +#ifdef USE_PACKET_INTERFACE_BASIC_ADDRESSING +#include "esphome/core/helpers.h" +#include "packet_interface.h" + +namespace esphome::packet_interface { + +class AddressingPort; + +/** + * An addressing layer for channels shared by several devices, such as an RS485 bus or a LoRa radio. + * + * Each packet starts with a 16-bit destination address and a 16-bit source address, little endian. + * BROADCAST_ADDRESS sends to every device. A packet is dropped unless it is addressed to this device or broadcast, and + * a packet from this device's own address, such as an echo on a half-duplex bus, is dropped too. + * + * Above this layer are one or more ports, each the bottom of its own stack of layers. A port with a destination + * address is a point-to-point link to that device: it sends to it and receives only its packets, so the reliability + * and session layers and the rolling code, which work between two devices only, can be used above it. A port with + * the broadcast destination sends to every device, and receives packets from devices that have no port of their own. + */ +class BasicAddressing : public Component { + public: + static constexpr uint16_t BROADCAST_ADDRESS = 0xFFFF; + /// Destination and source addresses. + static constexpr size_t HEADER_SIZE = 4; + + BasicAddressing(PacketInterface *link, uint16_t address) : link_(link), address_(address) {} + + void setup() override; + void dump_config() override; + + PacketInterface *get_link() const { return this->link_; } + void add_port(AddressingPort *port) { this->ports_.push_back(port); } + /// Sends \p data to \p destination, or to the destination in \p meta_data if set. + bool send(const PacketBuffer &data, const PacketMetaData &meta_data, uint16_t destination); + + protected: + void receive_(const PacketBuffer &data, const PacketMetaData &meta_data); + + PacketInterface *link_; + const uint16_t address_; + StaticVector ports_; +}; + +/// The bottom of the stack of layers for one destination, or for broadcast, on an addressed channel. +class AddressingPort : public PacketInterface { + public: + /// With \p only_port, the port is the top of the link, so top() on the link reaches this port's stack. + AddressingPort(BasicAddressing *addressing, uint16_t destination, bool only_port); + + void dump_config() override; + bool send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) override { + return this->addressing_->send(data, meta_data, this->destination_); + } + uint16_t get_destination() const { return this->destination_; } + + protected: + friend class BasicAddressing; + + BasicAddressing *addressing_; + const uint16_t destination_; +}; + +} // namespace esphome::packet_interface +#endif diff --git a/esphome/components/packet_interface/basic_fragmentation.cpp b/esphome/components/packet_interface/basic_fragmentation.cpp new file mode 100644 index 00000000000..b587d1e3140 --- /dev/null +++ b/esphome/components/packet_interface/basic_fragmentation.cpp @@ -0,0 +1,138 @@ +#include "basic_fragmentation.h" +#ifdef USE_PACKET_INTERFACE_BASIC_FRAGMENTATION +#include +#include +#include "esphome/core/helpers.h" +#include "esphome/core/log.h" + +namespace esphome::packet_interface { + +ESPHOME_LOG_TAG(TAG, "packet_interface.fragmentation"); + +static constexpr uint8_t LAST_FRAGMENT = 0x80; +static constexpr uint8_t INDEX_MASK = 0x7F; + +BasicFragmentation::BasicFragmentation(PacketInterface *lower, size_t max_packet_size) : lower_(lower) { + this->stack_on_(lower, CAPABILITY_FRAGMENTATION); + this->max_packet_size_ = max_packet_size; +} + +size_t BasicFragmentation::fragments_for_(size_t size) const { + size_t fragment_size = this->fragment_size_(); + return std::max(1, (size + fragment_size - 1) / fragment_size); +} + +void BasicFragmentation::setup() { + if (this->fragments_for_(this->max_packet_size_) > MAX_FRAGMENTS) { + this->mark_failed(LOG_STR("max_packet_size needs more than 128 fragments")); + return; + } + // Checked now, while the layer below holds nothing + this->queue_too_short_ = this->fragments_for_(this->max_packet_size_) > this->lower_->get_send_capacity(); + this->tx_buffer_ = std::make_unique(this->fragment_size_()); + this->rx_buffer_ = std::make_unique(this->max_packet_size_); + // Numbering from a random start makes it unlikely that a packet joins fragments from before a restart + this->tx_number_ = random_uint32(); + this->lower_->add_packet_interface_listener( + [this](const PacketBuffer &data, const PacketMetaData &meta_data) { this->receive_(data, meta_data); }); +} + +void BasicFragmentation::dump_config() { + size_t fragments = this->fragments_for_(this->max_packet_size_); + ESP_LOGCONFIG(TAG, + "Basic Fragmentation on '%s':\n" + " Max Packet Size: %zu\n" + " Fragment Size: %zu\n" + " Fragments in the Largest Packet: %zu", + this->name_, this->max_packet_size_, this->fragment_size_(), fragments); + if (this->queue_too_short_) { + ESP_LOGW(TAG, " The reliability queue is too short for the largest packets, which cannot be sent"); + } +} + +bool BasicFragmentation::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) { + if (this->is_failed()) + return false; + size_t size = data.size(); + if (size > this->max_packet_size_) { + ESP_LOGW(TAG, "%s: Packet of %zu bytes is larger than %zu", this->name_, size, this->max_packet_size_); + return false; + } + size_t fragments = this->fragments_for_(size); + if (fragments > this->lower_->get_send_capacity()) { + ESP_LOGV(TAG, "%s: No room for %zu fragments", this->name_, fragments); + return false; + } + uint8_t header[HEADER_SIZE] = {this->tx_number_++, 0}; + if (fragments == 1) { + header[1] = LAST_FRAGMENT; + return this->lower_->send_to_interface(PacketBuffer(header, HEADER_SIZE, &data, nullptr, 0), meta_data); + } + size_t fragment_size = this->fragment_size_(); + for (size_t index = 0; index != fragments; index++) { + size_t length = data.copy_to(this->tx_buffer_.get(), fragment_size, index * fragment_size); + header[1] = index | (index + 1 == fragments ? LAST_FRAGMENT : 0); + PacketBuffer body(this->tx_buffer_.get(), length); + if (!this->lower_->send_to_interface(PacketBuffer(header, HEADER_SIZE, &body, nullptr, 0), meta_data)) { + // The peer drops the part already sent when the next packet starts + ESP_LOGV(TAG, "%s: Fragment %zu of %zu not accepted", this->name_, index, fragments); + return false; + } + } + return true; +} + +void BasicFragmentation::receive_(const PacketBuffer &data, const PacketMetaData &meta_data) { + // Lower layers deliver packets in one piece; a packet in one fragment is passed up without copying + if (!data.is_contiguous()) { + ESP_LOGW(TAG, "%s: Packet split across buffers is not supported", this->name_); + return; + } + auto fragment = data.as_span(); + if (fragment.size() < HEADER_SIZE) + return; + uint8_t number = fragment[0]; + uint8_t index = fragment[1] & INDEX_MASK; + bool last = (fragment[1] & LAST_FRAGMENT) != 0; + const uint8_t *payload = fragment.data() + HEADER_SIZE; + size_t length = fragment.size() - HEADER_SIZE; + if (index == 0) { + if (this->rx_joining_) { + ESP_LOGV(TAG, "%s: Incomplete packet %u dropped", this->name_, this->rx_number_); + } + this->rx_joining_ = false; + if (last) { + if (length > this->max_packet_size_) { + ESP_LOGV(TAG, "%s: Packet %u is larger than %zu, dropped", this->name_, number, this->max_packet_size_); + return; + } + this->on_receive_from_interface_(PacketBuffer(payload, length), meta_data); + return; + } + this->rx_joining_ = true; + this->rx_number_ = number; + this->rx_source_ = meta_data.source_address; + this->rx_next_index_ = 0; + this->rx_length_ = 0; + } else if (!this->rx_joining_ || number != this->rx_number_ || meta_data.source_address != this->rx_source_ || + index != this->rx_next_index_) { + ESP_LOGV(TAG, "%s: Fragment %u of packet %u dropped", this->name_, index, number); + this->rx_joining_ = false; + return; + } + if (this->rx_length_ + length > this->max_packet_size_) { + ESP_LOGV(TAG, "%s: Packet %u is larger than %zu, dropped", this->name_, number, this->max_packet_size_); + this->rx_joining_ = false; + return; + } + memcpy(this->rx_buffer_.get() + this->rx_length_, payload, length); + this->rx_length_ += length; + this->rx_next_index_++; + if (last) { + this->rx_joining_ = false; + this->on_receive_from_interface_(PacketBuffer(this->rx_buffer_.get(), this->rx_length_), meta_data); + } +} + +} // namespace esphome::packet_interface +#endif diff --git a/esphome/components/packet_interface/basic_fragmentation.h b/esphome/components/packet_interface/basic_fragmentation.h new file mode 100644 index 00000000000..4cc3ee3c952 --- /dev/null +++ b/esphome/components/packet_interface/basic_fragmentation.h @@ -0,0 +1,56 @@ +#pragma once + +#include "esphome/core/defines.h" +#ifdef USE_PACKET_INTERFACE_BASIC_FRAGMENTATION +#include + +#include "packet_interface.h" + +namespace esphome::packet_interface { + +/** + * A fragmentation layer: packets up to this layer's max_packet_size are split into fragments that fit the layer + * below, and joined again at the other end. + * + * Each fragment starts with a 2 byte header: an 8-bit packet number, then the fragment's index in bits 0-6, with bit + * 7 set on the last fragment. A packet that fits in one fragment still has the header, and is sent and passed up + * without being copied. + * + * A packet is started only if the layer below can take all of its fragments now (get_send_capacity()), so a full + * reliability queue does not leave the peer with part of a packet. The receiver joins one packet at a time, and drops + * it when a fragment is missing, or one arrives with another packet number or from another source address. Above a + * reliability layer, fragments arrive in order and none are missing. + */ +class BasicFragmentation : public PacketInterface { + public: + /// Packet number, then index and last flag. + static constexpr size_t HEADER_SIZE = 2; + /// The index has 7 bits. + static constexpr size_t MAX_FRAGMENTS = 128; + + BasicFragmentation(PacketInterface *lower, size_t max_packet_size); + + void setup() override; + void dump_config() override; + bool send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) override; + + protected: + void receive_(const PacketBuffer &data, const PacketMetaData &meta_data); + size_t fragment_size_() const { return this->lower_->get_max_packet_size() - HEADER_SIZE; } + size_t fragments_for_(size_t size) const; + + PacketInterface *lower_; + // One fragment being sent, and the packet being joined; both allocated in setup() + std::unique_ptr tx_buffer_; + std::unique_ptr rx_buffer_; + size_t rx_length_{0}; + uint16_t rx_source_{0}; + uint8_t tx_number_{0}; + uint8_t rx_number_{0}; + uint8_t rx_next_index_{0}; + bool rx_joining_{false}; + bool queue_too_short_{false}; +}; + +} // namespace esphome::packet_interface +#endif diff --git a/esphome/components/packet_interface/layers/__init__.py b/esphome/components/packet_interface/layers/__init__.py index b0a3667f313..51bd5d93963 100644 --- a/esphome/components/packet_interface/layers/__init__.py +++ b/esphome/components/packet_interface/layers/__init__.py @@ -2,7 +2,7 @@ Each module in this package implements one layer type and registers it with its kind, for example: - @RELIABILITY.register("hdlc", schema=HDLC_SCHEMA) + @RELIABILITY.register("hdlc", schema=HDLC_SCHEMA, single_peer=lambda config: True) async def build_hdlc(config: ConfigType, lower: MockObj) -> MockObj: ... @@ -16,17 +16,23 @@ from typing import Any import esphome.codegen as cg import esphome.config_validation as cv -from esphome.const import CONF_ENCRYPTION, CONF_KEY, CONF_TYPE +from esphome.const import CONF_ENCRYPTION, CONF_FRAGMENTATION, CONF_KEY, CONF_TYPE from esphome.cpp_generator import MockObj from esphome.types import ConfigType packet_interface_ns = cg.esphome_ns.namespace("packet_interface") PacketInterface = packet_interface_ns.class_("PacketInterface", cg.Component) +CONF_ADDRESSING = "addressing" +CONF_DESTINATION = "destination" +CONF_PEERS = "peers" CONF_RELIABILITY = "reliability" CONF_SESSION = "session" -# Builds a layer above lower and returns it +BROADCAST_ADDRESS = 0xFFFF + +# Builds a layer above lower and returns it. An addressing layer instead returns a list of (port, config) pairs, one +# for each stack it carries, where config holds the layers for that stack, or is None for the layers on the link. LayerBuilder = Callable[[ConfigType, MockObj], Awaitable[Any]] @@ -42,8 +48,11 @@ def _nothing(config: ConfigType) -> list[str]: class LayerType: """One implementation of a kind of layer.""" - schema: Any + # Returns the schema; called once every layer type is registered, as some schemas contain other layers + schema_factory: Callable[[], Any] build: LayerBuilder + # True if this configuration keeps state for one peer, so cannot be used with broadcast addressing + single_peer: Callable[[ConfigType], bool] # True if this configuration works only when the peer sends packets back, which a session layer makes sure of needs_session: Callable[[ConfigType], bool] # The components this configuration needs loaded @@ -68,11 +77,16 @@ class LayerKind: self, name: str, *, - schema: Any, + schema: Any = None, + schema_factory: Callable[[], Any] | None = None, + single_peer: Callable[[ConfigType], bool] = _never, needs_session: Callable[[ConfigType], bool] = _never, auto_load: Callable[[ConfigType], list[str]] = _nothing, ) -> Callable[[LayerBuilder], LayerBuilder]: - """Register a type with its schema.""" + """Register a type with its schema, or with a function returning a schema that contains other layers.""" + + def fixed_schema() -> Any: + return schema def decorator(build: LayerBuilder) -> LayerBuilder: if self.frozen: @@ -84,7 +98,13 @@ class LayerKind: raise ValueError( f"{self.key} layer type '{name}' is already registered" ) - self.types[name] = LayerType(schema, build, needs_session, auto_load) + self.types[name] = LayerType( + schema_factory or fixed_schema, + build, + single_peer, + needs_session, + auto_load, + ) return build return decorator @@ -103,7 +123,7 @@ class LayerKind: return cv.All( self._expand, cv.typed_schema( - {name: layer.schema for name, layer in self.types.items()}, + {name: layer.schema_factory() for name, layer in self.types.items()}, default_type=self.default_type, ), ) @@ -112,11 +132,15 @@ class LayerKind: return self.types[config[CONF_TYPE]] +# An addressing type's configuration has either a destination or a list of peers, each with an id, a destination +# and its own layers. +ADDRESSING = LayerKind(CONF_ADDRESSING, "basic") ENCRYPTION = LayerKind(CONF_ENCRYPTION, "xxtea", shorthand=CONF_KEY) RELIABILITY = LayerKind(CONF_RELIABILITY, "hdlc") +FRAGMENTATION = LayerKind(CONF_FRAGMENTATION, "basic") SESSION = LayerKind(CONF_SESSION, "basic") -# The kinds stacked above the link, from the bottom up -STACK_ORDER = (ENCRYPTION, RELIABILITY, SESSION) +# The kinds stacked above the link, or above each port of an addressing layer, from the bottom up +STACK_ORDER = (ENCRYPTION, RELIABILITY, FRAGMENTATION, SESSION) def stack_schema() -> dict: @@ -139,6 +163,21 @@ def check_session(config: ConfigType) -> None: ) +def check_single_peer(config: ConfigType, destination: int) -> None: + """Layers that keep state for one peer need a destination other than broadcast.""" + if destination != BROADCAST_ADDRESS: + return + for kind in STACK_ORDER: + if (layer := config.get(kind.key)) is not None and kind.get_type( + layer + ).single_peer(layer): + raise cv.Invalid( + f"This '{kind.key}' works between two devices only, so it needs a '{CONF_DESTINATION}' other than " + "broadcast", + path=[kind.key], + ) + + def stack_auto_load(config: ConfigType) -> set[str]: """The components that the layers configured in config need loaded.""" loads: set[str] = set() diff --git a/esphome/components/packet_interface/layers/basic_addressing.py b/esphome/components/packet_interface/layers/basic_addressing.py new file mode 100644 index 00000000000..b1003f670f5 --- /dev/null +++ b/esphome/components/packet_interface/layers/basic_addressing.py @@ -0,0 +1,107 @@ +"""Addressing layer with 16-bit addresses, carrying one stack of layers or one for each of several peers.""" + +from typing import Any + +import esphome.codegen as cg +import esphome.config_validation as cv +from esphome.const import CONF_ADDRESS, CONF_ID +from esphome.core import ID +from esphome.cpp_generator import MockObj +from esphome.types import ConfigType + +from . import ( + ADDRESSING, + BROADCAST_ADDRESS, + CONF_DESTINATION, + CONF_PEERS, + PacketInterface, + check_session, + check_single_peer, + packet_interface_ns, + stack_schema, +) + +CONF_PORT_ID = "port_id" + +BasicAddressing = packet_interface_ns.class_("BasicAddressing", cg.Component) +AddressingPort = packet_interface_ns.class_("AddressingPort", PacketInterface) + +_request_port_slot = cg.slot_counter("PACKET_INTERFACE_ADDRESSING_PORT_COUNT") + + +def _destination(value: Any) -> int: + """A device address, or broadcast.""" + if isinstance(value, str) and value.lower() == "broadcast": + return BROADCAST_ADDRESS + return cv.hex_int_range(1, BROADCAST_ADDRESS)(value) + + +def _validate_peer(config: ConfigType) -> ConfigType: + check_session(config) + check_single_peer(config, config[CONF_DESTINATION]) + return config + + +def _unique_destinations(peers: list[ConfigType]) -> list[ConfigType]: + destinations = [peer[CONF_DESTINATION] for peer in peers] + if len(set(destinations)) != len(destinations): + raise cv.Invalid("Each peer must have a different destination") + return peers + + +# A function, as each peer holds the other layers, whose schemas exist only once every layer type is registered +def _basic_addressing_schema() -> cv.Schema: + peer_schema = cv.All( + cv.Schema( + { + cv.Required(CONF_ID): cv.declare_id(AddressingPort), + cv.Required(CONF_DESTINATION): _destination, + **stack_schema(), + } + ), + _validate_peer, + ) + return cv.All( + cv.Schema( + { + cv.GenerateID(): cv.declare_id(BasicAddressing), + cv.GenerateID(CONF_PORT_ID): cv.declare_id(AddressingPort), + cv.Required(CONF_ADDRESS): cv.hex_int_range(1, BROADCAST_ADDRESS - 1), + cv.Optional(CONF_DESTINATION): _destination, + cv.Optional(CONF_PEERS): cv.All( + cv.ensure_list(peer_schema), cv.Length(min=1), _unique_destinations + ), + } + ).extend(cv.COMPONENT_SCHEMA), + cv.has_at_most_one_key(CONF_DESTINATION, CONF_PEERS), + ) + + +async def _new_port( + port_id: ID, addressing: MockObj, destination: int, only_port: bool +) -> MockObj: + _request_port_slot(str(addressing)) + port = cg.new_Pvariable(port_id, addressing, destination, only_port) + await cg.register_component(port, {}) + return port + + +@ADDRESSING.register("basic", schema_factory=_basic_addressing_schema) +async def build_basic_addressing( + config: ConfigType, link: MockObj +) -> list[tuple[MockObj, ConfigType | None]]: + cg.add_define("USE_PACKET_INTERFACE_BASIC_ADDRESSING") + hub = cg.new_Pvariable(config[CONF_ID], link, config[CONF_ADDRESS]) + await cg.register_component(hub, config) + if (peers := config.get(CONF_PEERS)) is None: + port = await _new_port( + config[CONF_PORT_ID], hub, config[CONF_DESTINATION], True + ) + return [(port, None)] + stacks = [] + # Each peer has its own stack, named after the peer for logs and saved preferences + for peer in peers: + port = await _new_port(peer[CONF_ID], hub, peer[CONF_DESTINATION], False) + cg.add(port.set_name(peer[CONF_ID].id)) + stacks.append((port, peer)) + return stacks diff --git a/esphome/components/packet_interface/layers/basic_fragmentation.py b/esphome/components/packet_interface/layers/basic_fragmentation.py new file mode 100644 index 00000000000..cb7ef02604b --- /dev/null +++ b/esphome/components/packet_interface/layers/basic_fragmentation.py @@ -0,0 +1,31 @@ +"""Fragmentation layer that splits packets larger than the layers below allow, and joins them again.""" + +import esphome.codegen as cg +import esphome.config_validation as cv +from esphome.const import CONF_ID +from esphome.cpp_generator import MockObj +from esphome.types import ConfigType + +from . import FRAGMENTATION, PacketInterface, packet_interface_ns + +CONF_MAX_PACKET_SIZE = "max_packet_size" + +BasicFragmentation = packet_interface_ns.class_("BasicFragmentation", PacketInterface) + +BASIC_FRAGMENTATION_SCHEMA = cv.Schema( + { + cv.GenerateID(): cv.declare_id(BasicFragmentation), + # The packet being joined is held in a buffer of this size + cv.Optional(CONF_MAX_PACKET_SIZE, default=1024): cv.int_range(16, 16384), + } +).extend(cv.COMPONENT_SCHEMA) + + +# Fragments from different devices are told apart by the addressing layer's source address, so several devices can +# share a channel; without addressing, two devices sending fragmented packets at once can make the receiver drop both +@FRAGMENTATION.register("basic", schema=BASIC_FRAGMENTATION_SCHEMA) +async def build_basic_fragmentation(config: ConfigType, lower: MockObj) -> MockObj: + cg.add_define("USE_PACKET_INTERFACE_BASIC_FRAGMENTATION") + var = cg.new_Pvariable(config[CONF_ID], lower, config[CONF_MAX_PACKET_SIZE]) + await cg.register_component(var, config) + return var diff --git a/esphome/components/packet_interface/layers/basic_session.py b/esphome/components/packet_interface/layers/basic_session.py index de32aaffd97..a9391385e46 100644 --- a/esphome/components/packet_interface/layers/basic_session.py +++ b/esphome/components/packet_interface/layers/basic_session.py @@ -57,6 +57,7 @@ BASIC_SESSION_SCHEMA = cv.All( @SESSION.register( "basic", schema=BASIC_SESSION_SCHEMA, + single_peer=lambda config: True, auto_load=lambda config: ["binary_sensor"] if CONF_BINARY_SENSOR in config else [], ) async def build_basic_session(config: ConfigType, lower: MockObj) -> MockObj: diff --git a/esphome/components/packet_interface/layers/hdlc_reliability.py b/esphome/components/packet_interface/layers/hdlc_reliability.py index fef0d272265..9f7c31179d1 100644 --- a/esphome/components/packet_interface/layers/hdlc_reliability.py +++ b/esphome/components/packet_interface/layers/hdlc_reliability.py @@ -47,7 +47,7 @@ HDLC_SCHEMA = cv.All( ) -@RELIABILITY.register("hdlc", schema=HDLC_SCHEMA) +@RELIABILITY.register("hdlc", schema=HDLC_SCHEMA, single_peer=lambda config: True) async def build_hdlc(config: ConfigType, lower: MockObj) -> MockObj: cg.add_define("USE_PACKET_INTERFACE_HDLC") var = cg.new_Pvariable(config[CONF_ID], lower) diff --git a/esphome/components/packet_interface/layers/xxtea_encryption.py b/esphome/components/packet_interface/layers/xxtea_encryption.py index 57ffc49baf7..73ccb89967b 100644 --- a/esphome/components/packet_interface/layers/xxtea_encryption.py +++ b/esphome/components/packet_interface/layers/xxtea_encryption.py @@ -23,10 +23,12 @@ XXTEA_SCHEMA = cv.Schema( ).extend(cv.COMPONENT_SCHEMA) +# The rolling code keeps the last counter of one peer, and accepts a packet only once the peer has sent back this +# device's boot number @ENCRYPTION.register( "xxtea", schema=XXTEA_SCHEMA, - # The rolling code accepts a packet only once the peer has sent back this device's boot number + single_peer=lambda config: config[CONF_ROLLING_CODE_ENABLE], needs_session=lambda config: config[CONF_ROLLING_CODE_ENABLE], auto_load=lambda config: ["xxtea"], ) diff --git a/esphome/components/packet_interface/packet_transport/__init__.py b/esphome/components/packet_interface/packet_transport/__init__.py index 59885ff9c3b..35eb59db2ed 100644 --- a/esphome/components/packet_interface/packet_transport/__init__.py +++ b/esphome/components/packet_interface/packet_transport/__init__.py @@ -7,7 +7,12 @@ from esphome.components.packet_transport import ( import esphome.config_validation as cv from esphome.types import ConfigType -from .. import CONF_PACKET_INTERFACE, PacketInterface, packet_interface_ns +from .. import ( + CONF_PACKET_INTERFACE, + PacketInterface, + final_validate_interface, + packet_interface_ns, +) CODEOWNERS = ["@clydebarrow"] DEPENDENCIES = ["packet_interface"] @@ -22,6 +27,8 @@ CONFIG_SCHEMA = transport_schema(PacketInterfaceTransport).extend( } ) +FINAL_VALIDATE_SCHEMA = final_validate_interface(CONF_PACKET_INTERFACE) + async def to_code(config: ConfigType) -> None: interface = await cg.get_variable(config[CONF_PACKET_INTERFACE]) diff --git a/esphome/core/defines.h b/esphome/core/defines.h index 15e46890794..5d5b984e6d1 100644 --- a/esphome/core/defines.h +++ b/esphome/core/defines.h @@ -259,6 +259,9 @@ #define USE_OTA_STATE_LISTENER #define USE_OUTPUT #define USE_OUTPUT_FLOAT_POWER_SCALING +#define PACKET_INTERFACE_ADDRESSING_PORT_COUNT 2 +#define USE_PACKET_INTERFACE_BASIC_ADDRESSING +#define USE_PACKET_INTERFACE_BASIC_FRAGMENTATION #define USE_PACKET_INTERFACE_BASIC_SESSION #define USE_PACKET_INTERFACE_HDLC #define USE_PACKET_INTERFACE_XXTEA diff --git a/tests/components/packet_interface/__init__.py b/tests/components/packet_interface/__init__.py index c3a3e781a63..0388c7b29c8 100644 --- a/tests/components/packet_interface/__init__.py +++ b/tests/components/packet_interface/__init__.py @@ -4,6 +4,9 @@ from tests.testing_helpers import ComponentManifestOverride def override_manifest(manifest: ComponentManifestOverride) -> None: async def to_code_testing(config: object) -> None: + cg.add_define("PACKET_INTERFACE_ADDRESSING_PORT_COUNT", 4) + cg.add_define("USE_PACKET_INTERFACE_BASIC_ADDRESSING") + cg.add_define("USE_PACKET_INTERFACE_BASIC_FRAGMENTATION") cg.add_define("USE_PACKET_INTERFACE_BASIC_SESSION") cg.add_define("USE_PACKET_INTERFACE_HDLC") cg.add_define("USE_PACKET_INTERFACE_XXTEA") diff --git a/tests/components/packet_interface/basic_addressing.cpp b/tests/components/packet_interface/basic_addressing.cpp new file mode 100644 index 00000000000..252fc92e594 --- /dev/null +++ b/tests/components/packet_interface/basic_addressing.cpp @@ -0,0 +1,188 @@ +#include +#include +#include +#include +#include +#include "esphome/components/packet_interface/basic_addressing.h" +#include "esphome/components/packet_interface/hdlc_reliability.h" +#include "esphome/core/application.h" + +namespace esphome::packet_interface::addressing_testing { + +using Packet = std::vector; + +static constexpr uint16_t BROADCAST = BasicAddressing::BROADCAST_ADDRESS; + +// A link layer that records what is sent, so a test can deliver it to other devices +class MockLink : public PacketInterface { + public: + MockLink() { this->set_max_packet_size(64); } + bool send_to_interface(const PacketBuffer &data, const PacketMetaData &) override { + this->sent.push_back(data.to_vector()); + return true; + } + void inject(const Packet &packet) { this->on_receive_from_interface_(PacketBuffer(packet), {}); } + + std::vector sent; +}; + +// What one port receives +struct Port { + Port(BasicAddressing *addressing, uint16_t destination, bool only_port) { + this->port = std::make_unique(addressing, destination, only_port); + this->port->add_packet_interface_listener([this](const PacketBuffer &data, const PacketMetaData &meta_data) { + this->received.push_back(data.to_vector()); + this->sources.push_back(meta_data.source_address); + }); + } + void send(const Packet &data, uint16_t destination = 0) { + PacketMetaData meta_data; + meta_data.destination_address = destination; + this->port->send_to_interface(PacketBuffer(data), meta_data); + } + + std::unique_ptr port; + std::vector received; + std::vector sources; +}; + +// A device on the bus with one port per destination given +struct Device { + Device(uint16_t address, std::initializer_list destinations) : addressing(&this->link, address) { + this->link.set_name("bus"); + for (uint16_t destination : destinations) + this->ports.push_back(std::make_unique(&this->addressing, destination, destinations.size() == 1)); + this->addressing.setup(); + } + explicit Device(uint16_t address, uint16_t destination = BROADCAST) : Device(address, {destination}) {} + Port &port(size_t index = 0) { return *this->ports[index]; } + // Every device on the bus hears every packet, including its own + void deliver(std::initializer_list bus) { + for (auto &packet : this->link.sent) { + for (auto *device : bus) + device->link.inject(packet); + } + this->link.sent.clear(); + } + + MockLink link; + BasicAddressing addressing; + std::vector> ports; +}; + +TEST(BasicAddressingTest, HeaderAndCapability) { + Device a(0x1234, 0x5678); + EXPECT_EQ(a.port().port->get_max_packet_size(), 64 - BasicAddressing::HEADER_SIZE); + EXPECT_TRUE(a.port().port->has_capability(CAPABILITY_ADDRESSING)); + EXPECT_EQ(a.link.top(), a.port().port.get()); + a.port().send({0xAA}); + a.port().send({0xBB}, 0x0042); + EXPECT_EQ(a.link.sent, (std::vector{{0x78, 0x56, 0x34, 0x12, 0xAA}, {0x42, 0x00, 0x34, 0x12, 0xBB}})); +} + +TEST(BasicAddressingTest, BroadcastReachesEveryoneButSender) { + Device a(1), b(2), c(3); + a.port().send({0x10}); + a.deliver({&a, &b, &c}); + EXPECT_TRUE(a.port().received.empty()); + EXPECT_EQ(b.port().received, std::vector{{0x10}}); + EXPECT_EQ(c.port().received, std::vector{{0x10}}); + EXPECT_EQ(b.port().sources, std::vector{1}); +} + +TEST(BasicAddressingTest, UnicastReachesOnlyDestination) { + Device a(1, 2), b(2), c(3); + a.port().send({0x20}); + a.deliver({&b, &c}); + EXPECT_EQ(b.port().received, std::vector{{0x20}}); + EXPECT_TRUE(c.port().received.empty()); +} + +TEST(BasicAddressingTest, PointToPointAcceptsOnlyPeer) { + Device a(1, 2), b(2, 1), c(3); + c.port().send({0x30}); // broadcast from a third device + c.port().send({0x31}, 2); // addressed to b, from the wrong device + a.port().send({0x32}); // from the peer + c.deliver({&b}); + a.deliver({&b}); + EXPECT_EQ(b.port().received, std::vector{{0x32}}); + EXPECT_EQ(b.port().sources, std::vector{1}); +} + +TEST(BasicAddressingTest, PeersHaveSeparateStacks) { + Device hub(1, {2, 3, BROADCAST}), b(2, 1), c(3, 1), d(4); + EXPECT_EQ(hub.link.top(), &hub.link); // with several ports, applications use the ports + EXPECT_EQ(hub.port(0).port->top(), hub.port(0).port.get()); + b.port().send({0x02}); + c.port().send({0x03}); + d.port().send({0x04}); + b.deliver({&hub}); + c.deliver({&hub}); + d.deliver({&hub}); + EXPECT_EQ(hub.port(0).received, std::vector{{0x02}}); + EXPECT_EQ(hub.port(1).received, std::vector{{0x03}}); + EXPECT_EQ(hub.port(2).received, std::vector{{0x04}}); // only devices without a port of their own + hub.port(1).send({0x13}); + hub.deliver({&b, &c, &d}); + EXPECT_EQ(c.port().received, std::vector{{0x13}}); + EXPECT_TRUE(b.port().received.empty()); + EXPECT_TRUE(d.port().received.empty()); +} + +TEST(BasicAddressingTest, WithoutBroadcastPortUnknownDevicesDropped) { + Device hub(1, {2, 3}), d(4); + d.port().send({0x04}); + d.deliver({&hub}); + EXPECT_TRUE(hub.port(0).received.empty()); + EXPECT_TRUE(hub.port(1).received.empty()); +} + +TEST(BasicAddressingTest, ShortPacketDropped) { + Device a(1); + a.link.inject({0xFF, 0xFF, 0x02}); + EXPECT_TRUE(a.port().received.empty()); +} + +// The test main does not construct App; the reliability layer's timers need its scheduler +class AddressingReliabilityTest : public ::testing::Test { + protected: + void SetUp() override { + new (&App) Application; + App.pre_setup("addressing-tests", 16, "", 0); + } + void TearDown() override { App.~Application(); } +}; + +// A reliability layer on each port, recording what it delivers +struct ReliablePort { + explicit ReliablePort(AddressingPort *port) : hdlc(port) { + this->hdlc.add_packet_interface_listener( + [this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(data.to_vector()); }); + this->hdlc.setup(); + } + void send(const Packet &data) { this->hdlc.send_to_interface(PacketBuffer(data), {}); } + HdlcReliability hdlc; + std::vector received; +}; + +TEST_F(AddressingReliabilityTest, SeveralReliablePeersOnOneBus) { + Device hub(1, {2, 3}), b(2, 1), c(3, 1); + ReliablePort hub_b(hub.port(0).port.get()), hub_c(hub.port(1).port.get()); + ReliablePort b_hub(b.port().port.get()), c_hub(c.port().port.get()); + hub_b.send({0x12}); + hub_c.send({0x13}); + b_hub.send({0x21}); + c_hub.send({0x31}); + for (int i = 0; i != 20; i++) { + for (auto *port : {&hub_b, &hub_c, &b_hub, &c_hub}) + port->hdlc.loop(); + for (auto *device : {&hub, &b, &c}) + device->deliver({&hub, &b, &c}); + } + EXPECT_EQ(b_hub.received, std::vector{{0x12}}); + EXPECT_EQ(c_hub.received, std::vector{{0x13}}); + EXPECT_EQ(hub_b.received, std::vector{{0x21}}); + EXPECT_EQ(hub_c.received, std::vector{{0x31}}); +} + +} // namespace esphome::packet_interface::addressing_testing diff --git a/tests/components/packet_interface/basic_fragmentation.cpp b/tests/components/packet_interface/basic_fragmentation.cpp new file mode 100644 index 00000000000..88d833d8e4f --- /dev/null +++ b/tests/components/packet_interface/basic_fragmentation.cpp @@ -0,0 +1,194 @@ +#include +#include +#include +#include +#include "esphome/components/packet_interface/basic_fragmentation.h" +#include "esphome/components/packet_interface/hdlc_reliability.h" +#include "esphome/core/application.h" + +namespace esphome::packet_interface::fragmentation_testing { + +using Packet = std::vector; + +// A link layer that records what is sent, with a limited send capacity +class MockLink : public PacketInterface { + public: + MockLink() { this->set_max_packet_size(12); } + bool send_to_interface(const PacketBuffer &data, const PacketMetaData &) override { + this->sent.push_back(data.to_vector()); + return true; + } + size_t get_send_capacity() const override { return this->capacity; } + void inject(const Packet &packet, uint16_t source = 0) { + PacketMetaData meta_data; + meta_data.source_address = source; + this->on_receive_from_interface_(PacketBuffer(packet), meta_data); + } + + std::vector sent; + size_t capacity{SIZE_MAX}; +}; + +struct Peer { + explicit Peer(PacketInterface *lower) : fragmentation(lower, 100) { + this->fragmentation.add_packet_interface_listener( + [this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(data.to_vector()); }); + this->fragmentation.setup(); + } + bool send(const Packet &data) { return this->fragmentation.send_to_interface(PacketBuffer(data), {}); } + + BasicFragmentation fragmentation; + std::vector received; +}; + +static Packet make_packet(size_t size) { + Packet packet(size); + for (size_t i = 0; i != size; i++) + packet[i] = i * 7 + 1; + return packet; +} + +class TestHdlc : public HdlcReliability { + public: + using HdlcReliability::HdlcReliability; + using HdlcReliability::on_timeout_; +}; + +// The test main does not construct App; the reliability layer's timers need its scheduler +class BasicFragmentationTest : public ::testing::Test { + protected: + void SetUp() override { + new (&App) Application; + App.pre_setup("fragmentation-tests", 19, "", 0); + } + void TearDown() override { App.~Application(); } + + // Deliver what a has sent to b + static void deliver(MockLink &from, MockLink &to) { + for (auto &packet : from.sent) + to.inject(packet); + from.sent.clear(); + } + + MockLink link_a, link_b; +}; + +TEST_F(BasicFragmentationTest, SizesAndCapability) { + Peer a(&this->link_a); + EXPECT_EQ(a.fragmentation.get_max_packet_size(), 100); + EXPECT_TRUE(a.fragmentation.has_capability(CAPABILITY_FRAGMENTATION)); + EXPECT_EQ(this->link_a.top(), &a.fragmentation); +} + +TEST_F(BasicFragmentationTest, SmallPacketIsOneFragment) { + Peer a(&this->link_a), b(&this->link_b); + a.send({1, 2, 3}); + ASSERT_EQ(this->link_a.sent.size(), 1); + EXPECT_EQ(this->link_a.sent[0].size(), 3 + BasicFragmentation::HEADER_SIZE); + EXPECT_EQ(this->link_a.sent[0][1], 0x80); // index 0, last + deliver(this->link_a, this->link_b); + EXPECT_EQ(b.received, (std::vector{{1, 2, 3}})); +} + +TEST_F(BasicFragmentationTest, LargePacketSplitAndJoined) { + Peer a(&this->link_a), b(&this->link_b); + for (size_t size : {10, 11, 50, 100}) { + Packet packet = make_packet(size); + ASSERT_TRUE(a.send(packet)); + EXPECT_EQ(this->link_a.sent.size(), (size + 9) / 10); // 10 bytes of payload in each 12 byte fragment + EXPECT_EQ(this->link_a.sent.back()[1] & 0x80, 0x80); + deliver(this->link_a, this->link_b); + EXPECT_EQ(b.received.back(), packet); + } + EXPECT_EQ(b.received.size(), 4); +} + +TEST_F(BasicFragmentationTest, CompositePacketSplitAcrossFragments) { + Peer a(&this->link_a), b(&this->link_b); + Packet pre = make_packet(7), mid(13, 0xAA), post(9, 0x55); + PacketBuffer body(mid.data(), mid.size()); + ASSERT_TRUE( + a.fragmentation.send_to_interface(PacketBuffer(pre.data(), pre.size(), &body, post.data(), post.size()), {})); + deliver(this->link_a, this->link_b); + Packet expected = pre; + expected.insert(expected.end(), mid.begin(), mid.end()); + expected.insert(expected.end(), post.begin(), post.end()); + EXPECT_EQ(b.received, std::vector{expected}); +} + +TEST_F(BasicFragmentationTest, MissingFragmentDropsOnlyThatPacket) { + Peer a(&this->link_a), b(&this->link_b); + a.send(make_packet(30)); + this->link_a.sent.erase(this->link_a.sent.begin() + 1); + deliver(this->link_a, this->link_b); + EXPECT_TRUE(b.received.empty()); + a.send(make_packet(25)); + deliver(this->link_a, this->link_b); + EXPECT_EQ(b.received, std::vector{make_packet(25)}); +} + +TEST_F(BasicFragmentationTest, NotStartedWithoutRoomForAllFragments) { + Peer a(&this->link_a); + this->link_a.capacity = 2; + EXPECT_FALSE(a.send(make_packet(30))); // needs 3 + EXPECT_TRUE(this->link_a.sent.empty()); + EXPECT_TRUE(a.send(make_packet(20))); + EXPECT_FALSE(a.send(make_packet(101))); +} + +TEST_F(BasicFragmentationTest, InterleavedSourcesNotMixed) { + Peer a(&this->link_a), c(&this->link_a), b(&this->link_b); + a.send(make_packet(20)); + auto from_a = std::move(this->link_a.sent); + this->link_a.sent.clear(); + c.send(make_packet(20)); + auto from_c = std::move(this->link_a.sent); + // Fragments from devices 1 and 2 arrive interleaved, perhaps with the same packet number + this->link_b.inject(from_a[0], 1); + this->link_b.inject(from_c[0], 2); + this->link_b.inject(from_a[1], 1); + this->link_b.inject(from_c[1], 2); + for (auto &packet : b.received) + EXPECT_EQ(packet, make_packet(20)); // nothing mixed from the two +} + +TEST_F(BasicFragmentationTest, OverReliabilityWithLosses) { + // Fragmentation above HDLC on each side, with every fourth frame from a dropped on the way; not every third, as + // each resend is a burst of three, so the same fragment would be lost every time + TestHdlc hdlc_a(&this->link_a), hdlc_b(&this->link_b); + hdlc_a.setup(); + hdlc_b.setup(); + // HDLC adds 3 bytes to the 12 byte link, leaving 7 bytes of payload in each fragment + Peer a(&hdlc_a), b(&hdlc_b); + Packet packet = make_packet(20); // 3 fragments, within the queue of 4 + ASSERT_TRUE(a.send(packet)); + int count = 0; + for (int i = 0; i != 30; i++) { + hdlc_a.loop(); + hdlc_b.loop(); + if (i % 5 == 4) + hdlc_a.on_timeout_(); + for (auto &frame : this->link_a.sent) { + if (++count % 4 != 0) + this->link_b.inject(frame); + } + this->link_a.sent.clear(); + deliver(this->link_b, this->link_a); + } + EXPECT_EQ(b.received, std::vector{packet}); +} + +TEST_F(BasicFragmentationTest, OversizedSingleFragmentDropped) { + // A link packet holds more than this layer's max_packet_size + BasicFragmentation fragmentation(&this->link_a, 4); + std::vector received; + fragmentation.add_packet_interface_listener( + [&received](const PacketBuffer &data, const PacketMetaData &) { received.push_back(data.to_vector()); }); + fragmentation.setup(); + this->link_a.inject({0x01, 0x80, 1, 2, 3, 4, 5}); + EXPECT_TRUE(received.empty()); + this->link_a.inject({0x02, 0x80, 1, 2, 3, 4}); + EXPECT_EQ(received, std::vector{Packet({1, 2, 3, 4})}); +} + +} // namespace esphome::packet_interface::fragmentation_testing diff --git a/tests/components/packet_interface/common.yaml b/tests/components/packet_interface/common.yaml index 0c388d95051..12d43bc6d1d 100644 --- a/tests/components/packet_interface/common.yaml +++ b/tests/components/packet_interface/common.yaml @@ -9,24 +9,40 @@ packet_interface: - platform: uart id: packet_interface_uart max_packet_size: 256 - reliability: - timeout: 300ms - ack_delay: 50ms - max_queue_length: 3 - max_retries: 4 - encryption: - key: Johnny Cash - rolling_code_enable: true - session: - keep_alive: 5s - timeout: 12s - binary_sensor: - name: UART session + addressing: + address: 0x0001 + peers: + - id: packet_interface_peer_a + destination: 0x0002 + encryption: + key: Johnny Cash + rolling_code_enable: true + reliability: + timeout: 300ms + max_queue_length: 3 + max_retries: 4 + session: + keep_alive: 5s + timeout: 12s + binary_sensor: + name: Peer A session + - id: packet_interface_peer_b + destination: 0x0003 + reliability: + fragmentation: + max_packet_size: 2048 + session: + - id: packet_interface_bus + destination: broadcast - platform: udp id: packet_interface_network udp_id: packet_interface_udp + addressing: + address: 0x0001 + destination: 0x0002 encryption: Johnny Cash reliability: + fragmentation: session: sensor: @@ -36,7 +52,7 @@ sensor: packet_transport: - platform: packet_interface - packet_interface: packet_interface_uart + packet_interface: packet_interface_peer_a sensors: - packet_interface_sensor - platform: packet_interface diff --git a/tests/unit_tests/test_packet_interface.py b/tests/unit_tests/test_packet_interface.py index 9750ea6d343..6fc1858d58f 100644 --- a/tests/unit_tests/test_packet_interface.py +++ b/tests/unit_tests/test_packet_interface.py @@ -1,4 +1,4 @@ -"""Unit tests for the packet_interface component.""" +"""Unit tests for the packet_interface component and its platforms.""" from typing import Any @@ -8,6 +8,8 @@ from esphome import config_validation as cv from esphome.components import packet_interface from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE from esphome.components.packet_interface.layers import ( + CONF_ADDRESSING, + CONF_FRAGMENTATION, CONF_RELIABILITY, CONF_SESSION, RELIABILITY, @@ -68,61 +70,6 @@ def test_reliability_queue_length_limit(setup_core) -> None: ) -def test_reliability_timeout_minimum(setup_core) -> None: - """A timeout shorter than a few loop passes cannot work.""" - uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "50ms"}}) - with pytest.raises(cv.Invalid): - uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "40ms"}}) - - -def test_ack_delay_shorter_than_timeout(setup_core) -> None: - """The acknowledgement must not wait longer than the peer's timeout.""" - reliability = uart_packet_interface.CONFIG_SCHEMA( - {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "100ms"}} - )[CONF_RELIABILITY] - assert reliability["ack_delay"].total_milliseconds == 100 - with pytest.raises(cv.Invalid, match="shorter"): - uart_packet_interface.CONFIG_SCHEMA( - {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "500ms"}} - ) - - -def test_layer_types_registered() -> None: - """Importing packet_interface registers the built-in layer types.""" - assert {kind.key: set(kind.types) for kind in STACK_ORDER} == { - CONF_ENCRYPTION: {"xxtea"}, - CONF_RELIABILITY: {"hdlc"}, - CONF_SESSION: {"basic"}, - } - - -async def _build(config: dict, lower: Any) -> Any: - return lower - - -def test_registered_type_selectable(setup_core) -> None: - """A newly registered layer type can be selected with type.""" - kind = LayerKind("test_kind", "test") - kind.register("test", schema=cv.Schema({}))(_build) - config = kind.schema()({"type": "test"}) - assert config["type"] == "test" - assert kind.get_type(config).build is _build - - -def test_duplicate_layer_type_rejected(setup_core) -> None: - """Registering a type under a name already in use is an error, not a silent replacement.""" - kind = LayerKind("test_kind", "test") - kind.register("test", schema=cv.Schema({}))(_build) - with pytest.raises(ValueError, match="already registered"): - kind.register("test", schema=cv.Schema({}))(_build) - - -def test_late_registration_rejected(setup_core) -> None: - """A type registered after the schemas were built would be missing from them, so it is an error.""" - with pytest.raises(ValueError, match="after the schemas were built"): - RELIABILITY.register("late", schema=cv.Schema({}))(_build) - - def test_session_defaults(setup_core) -> None: """An empty session key selects the basic session, timing out after three keep-alives.""" session = uart_packet_interface.CONFIG_SCHEMA({CONF_SESSION: None})[CONF_SESSION] @@ -176,13 +123,162 @@ def test_rolling_code_requires_session(setup_core) -> None: uart_packet_interface.CONFIG_SCHEMA({CONF_ENCRYPTION: "secret"}) +def test_addressing_defaults_to_broadcast(setup_core) -> None: + """Without a destination, packets are broadcast; broadcast can also be named.""" + for addressing in ({"address": 1}, {"address": 1, "destination": "broadcast"}): + config = uart_packet_interface.CONFIG_SCHEMA({CONF_ADDRESSING: addressing}) + assert config[CONF_ADDRESSING]["destination"] == 0xFFFF + with pytest.raises(cv.Invalid): + uart_packet_interface.CONFIG_SCHEMA({CONF_ADDRESSING: {"address": 0xFFFF}}) + + +@pytest.mark.parametrize( + "layers", + [ + {CONF_RELIABILITY: None}, + {CONF_SESSION: None}, + { + CONF_ENCRYPTION: {"key": "secret", "rolling_code_enable": True}, + CONF_SESSION: None, + }, + ], +) +def test_two_device_layers_need_destination(setup_core, layers: dict) -> None: + """Layers that work between two devices only cannot be used with broadcast addressing.""" + with pytest.raises(cv.Invalid, match="destination"): + uart_packet_interface.CONFIG_SCHEMA({CONF_ADDRESSING: {"address": 1}, **layers}) + uart_packet_interface.CONFIG_SCHEMA( + {CONF_ADDRESSING: {"address": 1, "destination": 2}, **layers} + ) + + +def _peer(peer_id: str, destination: Any, **layers: Any) -> dict: + return {"id": peer_id, "destination": destination, **layers} + + +def test_addressing_peers(setup_core) -> None: + """Each peer has its own destination and layers.""" + config = uart_packet_interface.CONFIG_SCHEMA( + { + CONF_ADDRESSING: { + "address": 1, + "peers": [ + _peer("peer_a", 2, reliability=None, session=None), + _peer("peer_b", 3, encryption="secret"), + _peer("everyone", "broadcast"), + ], + } + } + ) + peers = config[CONF_ADDRESSING]["peers"] + assert [peer["destination"] for peer in peers] == [2, 3, 0xFFFF] + assert CONF_RELIABILITY in peers[0] + + +@pytest.mark.parametrize( + ("addressing", "layers", "match"), + [ + ({"peers": [_peer("a", 2), _peer("b", 2)]}, {}, "different destination"), + ({"peers": [_peer("a", "broadcast", reliability=None)]}, {}, "broadcast"), + ({"peers": [_peer("a", 2)]}, {CONF_RELIABILITY: None}, "goes in each peer"), + ({"destination": 2, "peers": [_peer("a", 3)]}, {}, "destination"), + ], +) +def test_addressing_peers_invalid( + setup_core, addressing: dict, layers: dict, match: str +) -> None: + """Peers need different destinations, two-device layers need a destination, and layers go in the peers.""" + with pytest.raises(cv.Invalid, match=match): + uart_packet_interface.CONFIG_SCHEMA( + {CONF_ADDRESSING: {"address": 1, **addressing}, **layers} + ) + + +def test_layer_types_registered() -> None: + """Importing packet_interface registers the built-in layer types.""" + assert {kind.key: set(kind.types) for kind in STACK_ORDER} == { + CONF_ENCRYPTION: {"xxtea"}, + CONF_RELIABILITY: {"hdlc"}, + CONF_FRAGMENTATION: {"basic"}, + CONF_SESSION: {"basic"}, + } + + +async def _build(config: dict, lower: Any) -> Any: + return lower + + +def test_registered_type_selectable(setup_core) -> None: + """A newly registered layer type can be selected with type.""" + kind = LayerKind("test_kind", "test") + kind.register("test", schema=cv.Schema({}))(_build) + config = kind.schema()({"type": "test"}) + assert config["type"] == "test" + assert kind.get_type(config).build is _build + + +def test_duplicate_layer_type_rejected(setup_core) -> None: + """Registering a type under a name already in use is an error, not a silent replacement.""" + kind = LayerKind("test_kind", "test") + kind.register("test", schema=cv.Schema({}))(_build) + with pytest.raises(ValueError, match="already registered"): + kind.register("test", schema=cv.Schema({}))(_build) + + +def test_late_registration_rejected(setup_core) -> None: + """A type registered after the schemas were built would be missing from them, so it is an error.""" + with pytest.raises(ValueError, match="after the schemas were built"): + RELIABILITY.register("late", schema=cv.Schema({}))(_build) + + +def test_reliability_timeout_minimum(setup_core) -> None: + """A timeout shorter than a few loop passes cannot work.""" + uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "50ms"}}) + with pytest.raises(cv.Invalid): + uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "40ms"}}) + + +def test_ack_delay_shorter_than_timeout(setup_core) -> None: + """The acknowledgement must not wait longer than the peer's timeout.""" + reliability = uart_packet_interface.CONFIG_SCHEMA( + {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "100ms"}} + )[CONF_RELIABILITY] + assert reliability["ack_delay"].total_milliseconds == 100 + with pytest.raises(cv.Invalid, match="shorter"): + uart_packet_interface.CONFIG_SCHEMA( + {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "500ms"}} + ) + + +def test_fragmentation_defaults(setup_core) -> None: + """An empty fragmentation key joins packets of up to 1024 bytes.""" + fragmentation = uart_packet_interface.CONFIG_SCHEMA({CONF_FRAGMENTATION: None})[ + CONF_FRAGMENTATION + ] + assert fragmentation["type"] == "basic" + assert fragmentation["max_packet_size"] == 1024 + with pytest.raises(cv.Invalid): + uart_packet_interface.CONFIG_SCHEMA( + {CONF_FRAGMENTATION: {"max_packet_size": 8}} + ) + + @pytest.mark.parametrize( ("config", "loads"), [ ({}, []), - ({CONF_RELIABILITY: None, CONF_SESSION: None}, []), - ({CONF_SESSION: {"binary_sensor": {"name": "Link"}}}, ["binary_sensor"]), + ({CONF_RELIABILITY: None, "session": None}, []), + ({"session": {"binary_sensor": {"name": "Link"}}}, ["binary_sensor"]), ({CONF_ENCRYPTION: "secret"}, ["xxtea"]), + ( + { + CONF_ADDRESSING: { + "address": 1, + "peers": [_peer("a", 2, encryption="secret"), _peer("b", 3)], + } + }, + ["xxtea"], + ), ], ) def test_auto_load_follows_layers(