From 643f29fae6434393bea9c64cff62edf8ebe8d562 Mon Sep 17 00:00:00 2001 From: Bascht74 <66269310+Bascht74@users.noreply.github.com> Date: Mon, 12 Oct 2026 02:49:19 +0200 Subject: [PATCH] [uart] Let a UART say it has no line timing (#20227) --- esphome/components/ble_nus/__init__.py | 3 +- esphome/components/tcp_uart/__init__.py | 2 +- esphome/components/uart/__init__.py | 29 ++++++++++ esphome/components/usb_cdc_acm/__init__.py | 2 +- esphome/components/usb_uart/__init__.py | 5 +- tests/component_tests/uart/test_unclocked.py | 57 ++++++++++++++++++++ 6 files changed, 94 insertions(+), 4 deletions(-) create mode 100644 tests/component_tests/uart/test_unclocked.py diff --git a/esphome/components/ble_nus/__init__.py b/esphome/components/ble_nus/__init__.py index 54f111e4af..5d43162241 100644 --- a/esphome/components/ble_nus/__init__.py +++ b/esphome/components/ble_nus/__init__.py @@ -3,6 +3,7 @@ from esphome.components.logger import request_log_listener from esphome.components.uart import ( UARTComponent, debug_to_code, + mark_unclocked, maybe_empty_debug, uart_ns, ) @@ -42,7 +43,7 @@ def validate_rx_buffer(config: ConfigType) -> ConfigType: CONFIG_SCHEMA = cv.All( cv.Schema( { - cv.GenerateID(): cv.declare_id(BLENUS), + cv.GenerateID(): cv.All(cv.declare_id(BLENUS), mark_unclocked), cv.Optional(CONF_TYPE, default=CONF_LOGS): cv.one_of( *[CONF_LOGS, CONF_UART], lower=True ), diff --git a/esphome/components/tcp_uart/__init__.py b/esphome/components/tcp_uart/__init__.py index 35326d77d5..df9c1720ee 100644 --- a/esphome/components/tcp_uart/__init__.py +++ b/esphome/components/tcp_uart/__init__.py @@ -35,7 +35,7 @@ TcpUart = tcp_uart_ns.class_("TcpUart", uart.UARTComponent, cg.Component) BASE_SCHEMA = cv.Schema( { - cv.GenerateID(): cv.declare_id(TcpUart), + cv.GenerateID(): cv.All(cv.declare_id(TcpUart), uart.mark_unclocked), cv.Required(CONF_PORT): cv.port, cv.Optional(CONF_BAUD_RATE, default=9600): cv.int_range(min=1), cv.Optional(CONF_DATA_BITS, default=8): cv.int_range(min=5, max=8), diff --git a/esphome/components/uart/__init__.py b/esphome/components/uart/__init__.py index 2250b9d508..5eea8060ff 100644 --- a/esphome/components/uart/__init__.py +++ b/esphome/components/uart/__init__.py @@ -1,3 +1,4 @@ +from dataclasses import dataclass, field from logging import getLogger import math import re @@ -444,6 +445,34 @@ UART_DEVICE_SCHEMA = cv.Schema( KEY_UART_DEVICES = "uart_devices" +@dataclass +class UARTData: + # UARTs whose received bytes do not keep the timing of a serial line. + unclocked: list[ID] = field(default_factory=list) + + +def _get_data() -> UARTData: + if DOMAIN not in CORE.data: + CORE.data[DOMAIN] = UARTData() + return CORE.data[DOMAIN] + + +def mark_unclocked(uart_id: ID) -> ID: + """Record that the bytes of uart_id do not arrive with the timing of a serial line, e.g. over TCP or USB. + + A gap between them says nothing about where a frame ends. Use it on the UART's declared id, + cv.All(cv.declare_id(MyUart), uart.mark_unclocked), so that every to_code sees it. + """ + _get_data().unclocked.append(uart_id) + return uart_id + + +def is_unclocked(uart_id: ID) -> bool: + """Return True if uart_id was passed to mark_unclocked().""" + # By name: a generated id is named only after the schemas ran. + return any(str(uart) == str(uart_id) for uart in _get_data().unclocked) + + def final_validate_device_schema( name: str, *, diff --git a/esphome/components/usb_cdc_acm/__init__.py b/esphome/components/usb_cdc_acm/__init__.py index e49531c80b..08cd13e3d2 100644 --- a/esphome/components/usb_cdc_acm/__init__.py +++ b/esphome/components/usb_cdc_acm/__init__.py @@ -29,7 +29,7 @@ USBCDCACMInstance = usb_cdc_acm_ns.class_( # Schema for individual CDC ACM interface instances INTERFACE_SCHEMA = cv.Schema( { - cv.GenerateID(): cv.declare_id(USBCDCACMInstance), + cv.GenerateID(): cv.All(cv.declare_id(USBCDCACMInstance), uart.mark_unclocked), } ) diff --git a/esphome/components/usb_uart/__init__.py b/esphome/components/usb_uart/__init__.py index 581727b326..0875966071 100644 --- a/esphome/components/usb_uart/__init__.py +++ b/esphome/components/usb_uart/__init__.py @@ -6,6 +6,7 @@ from esphome.components.uart import ( CONF_FLUSH_TIMEOUT, UARTComponent, UARTParityOptions, + mark_unclocked, ) from esphome.components.usb_host import ( get_max_packet_size, @@ -162,7 +163,9 @@ def channel_schema(type_: "Type") -> cv.Schema: cv.ensure_list( cv.Schema( { - cv.GenerateID(): cv.declare_id(USBUartChannel), + cv.GenerateID(): cv.All( + cv.declare_id(USBUartChannel), mark_unclocked + ), cv.Optional(CONF_BUFFER_SIZE, default=256): cv.int_range( min=64, max=8192 ), diff --git a/tests/component_tests/uart/test_unclocked.py b/tests/component_tests/uart/test_unclocked.py new file mode 100644 index 0000000000..bbd131037d --- /dev/null +++ b/tests/component_tests/uart/test_unclocked.py @@ -0,0 +1,57 @@ +"""Tests for marking UARTs whose bytes do not arrive with the timing of a serial line.""" + +from esphome.components import ble_nus, tcp_uart, uart, usb_cdc_acm +from esphome.const import CONF_ID, PlatformFramework +from esphome.core import ID +from tests.component_tests.types import SetCoreConfigCallable + + +def test_only_a_marked_uart_is_unclocked() -> None: + marked = ID("link", is_declaration=True) + assert uart.mark_unclocked(marked) is marked + assert uart.is_unclocked(ID("link")) + assert not uart.is_unclocked(ID("bus")) + + +def test_a_generated_id_is_found_once_named() -> None: + generated = uart.mark_unclocked( + ID(None, is_declaration=True, type=uart.UARTComponent) + ) + name = generated.resolve([]) + assert uart.is_unclocked(ID(name)) + + +def test_tcp_uart_is_unclocked() -> None: + tcp_uart.CONFIG_SCHEMA({CONF_ID: "link", "role": "server", "port": 6638}) + assert uart.is_unclocked(ID("link")) + + +def test_every_usb_uart_channel_is_unclocked( + set_core_config: SetCoreConfigCallable, +) -> None: + set_core_config(PlatformFramework.ESP32_IDF, platform_data={"variant": "ESP32S3"}) + # Its schema is built on import and needs the target. + from esphome.components import usb_uart + + schema = usb_uart.channel_schema(usb_uart._TYPES_BY_NAME["FT2232"]) + schema( + { + "channels": [ + {CONF_ID: "ch_a", "baud_rate": 9600}, + {CONF_ID: "ch_b", "baud_rate": 115200}, + ] + } + ) + assert uart.is_unclocked(ID("ch_a")) + assert uart.is_unclocked(ID("ch_b")) + + +def test_usb_cdc_acm_interface_is_unclocked() -> None: + usb_cdc_acm.INTERFACE_SCHEMA({CONF_ID: "cdc"}) + assert uart.is_unclocked(ID("cdc")) + + +def test_ble_nus_is_unclocked(set_core_config: SetCoreConfigCallable) -> None: + set_core_config(PlatformFramework.NRF52_ZEPHYR) + ble_nus.CONFIG_SCHEMA({CONF_ID: "nus", "type": "uart"}) + assert uart.is_unclocked(ID("nus"))