[uart] Let a UART say it has no line timing (#20227)

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
             ),

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),

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,
     *,

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),
     }
 )
 

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
                             ),

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"))