[modbus] Fix timeout for non-hardware UARTs (e.g., USB UART) (#14614)

Co-authored-by: Clyde Stubbs <2366188+clydebarrow@users.noreply.github.com>
2026-05-20 01:16:26 +08:00 · 2026-03-08 14:31:42 -10:00
parent 93d7ec4d72
commit 88536ff72b
8 changed files with 213 additions and 8 deletions
@@ -11,6 +11,11 @@ static const char *const TAG = "modbus";
 // Maximum bytes to log for Modbus frames (truncated if larger)
 static constexpr size_t MODBUS_MAX_LOG_BYTES = 64;

+// Approximate bits per character on the wire (depends on parity/stop bit config)
+static constexpr uint32_t MODBUS_BITS_PER_CHAR = 11;
+// Milliseconds per second
+static constexpr uint32_t MS_PER_SEC = 1000;
+
 void Modbus::setup() {
  if (this->flow_control_pin_ != nullptr) {
    this->flow_control_pin_->setup();
@@ -19,10 +24,17 @@ void Modbus::setup() {
  this->frame_delay_ms_ =
      std::max(2,  // 1750us minimum per spec - rounded up to 2ms.
                   // 3.5 characters * 11 bits per character * 1000ms/sec / (bits/sec) (Standard modbus frame delay)
-               (uint16_t) (3.5 * 11 * 1000 / this->parent_->get_baud_rate()) + 1);
+               (uint16_t) (3.5 * MODBUS_BITS_PER_CHAR * MS_PER_SEC / this->parent_->get_baud_rate()) + 1);

+  // When rx_full_threshold is configured (non-zero), the UART has a hardware FIFO with a
+  // meaningful threshold (e.g., ESP32 native UART), so we can calculate a precise delay.
+  // Otherwise (e.g., USB UART), use 50ms to handle data arriving in chunks.
+  static constexpr uint16_t DEFAULT_LONG_RX_BUFFER_DELAY_MS = 50;
+  size_t rx_threshold = this->parent_->get_rx_full_threshold();
  this->long_rx_buffer_delay_ms_ =
-      (this->parent_->get_rx_full_threshold() * 11 * 1000 / this->parent_->get_baud_rate()) + 1;
+      rx_threshold != uart::UARTComponent::RX_FULL_THRESHOLD_UNSET
+          ? (rx_threshold * MODBUS_BITS_PER_CHAR * MS_PER_SEC / this->parent_->get_baud_rate()) + 1
+          : DEFAULT_LONG_RX_BUFFER_DELAY_MS;
 }

 void Modbus::loop() {
@@ -290,7 +302,7 @@ void Modbus::send_next_frame_() {
    this->last_send_tx_offset_ = 0;
  } else {
    this->write_array(frame.data.get(), frame.size);
-    this->last_send_tx_offset_ = frame.size * 11 * 1000 / this->parent_->get_baud_rate() + 1;
+    this->last_send_tx_offset_ = frame.size * MODBUS_BITS_PER_CHAR * MS_PER_SEC / this->parent_->get_baud_rate() + 1;
  }

 #if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
@@ -39,6 +39,8 @@ enum class FlushResult {

 class UARTComponent {
 public:
+  static constexpr size_t RX_FULL_THRESHOLD_UNSET = 0;
+
  // Writes an array of bytes to the UART bus.
  // @param data A vector of bytes to be written.
  void write_array(const std::vector<uint8_t> &data) { this->write_array(&data[0], data.size()); }
@@ -201,7 +203,9 @@ class UARTComponent {
  InternalGPIOPin *rx_pin_{};
  InternalGPIOPin *flow_control_pin_{};
  size_t rx_buffer_size_{};
-  size_t rx_full_threshold_{1};
+  // ESP32 (both Arduino and ESP-IDF) always sets this at codegen time via set_rx_full_threshold().
+  // Other platforms (USB UART, Arduino, etc.) leave it unset.
+  size_t rx_full_threshold_{RX_FULL_THRESHOLD_UNSET};
  size_t rx_timeout_{0};
  uint32_t baud_rate_{0};
  uint8_t stop_bits_{0};
@@ -62,6 +62,7 @@ CONFIG_INJECT_RX_SCHEMA = cv.maybe_simple_value(
    {
        cv.GenerateID(): cv.use_id(MockUartComponent),
        cv.Required("data"): cv.templatable(validate_raw_data),
+        cv.Optional(CONF_DELAY): cv.positive_time_period_milliseconds,
    },
    key=CONF_DATA,
 )
@@ -87,7 +88,7 @@ CONFIG_SCHEMA = cv.Schema(
        cv.GenerateID(): cv.declare_id(MockUartComponent),
        cv.Required(CONF_BAUD_RATE): cv.int_range(min=1),
        cv.Optional(CONF_RX_BUFFER_SIZE, default=256): cv.validate_bytes,
-        cv.Optional(CONF_RX_FULL_THRESHOLD, default=10): cv.int_range(min=1, max=120),
+        cv.Optional(CONF_RX_FULL_THRESHOLD): cv.int_range(min=1, max=120),
        cv.Optional(CONF_RX_TIMEOUT, default=2): cv.int_range(min=0, max=92),
        cv.Optional(CONF_STOP_BITS, default=1): cv.one_of(1, 2, int=True),
        cv.Optional(CONF_DATA_BITS, default=8): cv.int_range(min=5, max=8),
@@ -126,6 +127,8 @@ async def inject_rx_to_code(config, action_id, template_arg, args):
        arr_id = ID(f"{action_id}_data", is_declaration=True, type=cg.uint8)
        arr = cg.static_const_array(arr_id, cg.ArrayInitializer(*data))
        cg.add(var.set_data_static(arr, len(data)))
+    if CONF_DELAY in config:
+        cg.add(var.set_delay(config[CONF_DELAY]))
    return var


@@ -135,7 +138,8 @@ async def to_code(config):

    cg.add(var.set_baud_rate(config[CONF_BAUD_RATE]))
    cg.add(var.set_rx_buffer_size(config[CONF_RX_BUFFER_SIZE]))
-    cg.add(var.set_rx_full_threshold(config[CONF_RX_FULL_THRESHOLD]))
+    if CONF_RX_FULL_THRESHOLD in config:
+        cg.add(var.set_rx_full_threshold(config[CONF_RX_FULL_THRESHOLD]))
    cg.add(var.set_rx_timeout(config[CONF_RX_TIMEOUT]))
    cg.add(var.set_stop_bits(config[CONF_STOP_BITS]))
    cg.add(var.set_data_bits(config[CONF_DATA_BITS]))
@@ -22,18 +22,30 @@ template<typename... Ts> class MockUartInjectRXAction : public Action<Ts...>, pu
    this->len_ = len;  // Length >= 0 indicates static mode
  }

+  void set_delay(uint32_t delay_ms) { this->delay_ms_ = delay_ms; }
+
  void play(const Ts &...x) override {
    if (this->len_ >= 0) {
      // Static mode: use pointer and length
-      this->parent_->inject_to_rx_buffer(this->code_.data, static_cast<size_t>(this->len_));
+      if (this->delay_ms_ > 0) {
+        std::vector<uint8_t> data(this->code_.data, this->code_.data + this->len_);
+        this->parent_->inject_to_rx_buffer_delayed(data, this->delay_ms_);
+      } else {
+        this->parent_->inject_to_rx_buffer(this->code_.data, static_cast<size_t>(this->len_));
+      }
    } else {
      // Template mode: call function
      auto val = this->code_.func(x...);
-      this->parent_->inject_to_rx_buffer(val);
+      if (this->delay_ms_ > 0) {
+        this->parent_->inject_to_rx_buffer_delayed(val, this->delay_ms_);
+      } else {
+        this->parent_->inject_to_rx_buffer(val);
+      }
    }
  }

 protected:
+  uint32_t delay_ms_{0};
  ssize_t len_{-1};  // -1 = template mode, >=0 = static mode with length
  union Code {
    std::vector<uint8_t> (*func)(Ts...);  // Function pointer (stateless lambdas)
@@ -53,6 +53,15 @@ void MockUartComponent::loop() {
    }
  }

+  // Process staged RX - deliver bytes whose delay has elapsed
+  uint32_t now_ms = millis();
+  while (!this->staged_rx_.empty() && (static_cast<int32_t>(now_ms - this->staged_rx_.front().available_at_ms) >= 0)) {
+    auto &staged = this->staged_rx_.front();
+    ESP_LOGD(TAG, "Delivering %zu staged RX bytes", staged.data.size());
+    this->inject_to_rx_buffer(staged.data);
+    this->staged_rx_.pop_front();
+  }
+
  // Process delayed responses
  for (auto &response : this->responses_) {
    if (response.delay_ms > 0 && response.last_match_ms > 0 && now - response.last_match_ms >= response.delay_ms) {
@@ -209,4 +218,15 @@ void MockUartComponent::inject_to_rx_buffer(const std::vector<uint8_t> &data) {
  }
 }

+void MockUartComponent::inject_to_rx_buffer_delayed(const std::vector<uint8_t> &data, uint32_t delay_ms) {
+  if (!data.empty() && data.size() <= 64) {
+    char hex_buf[format_hex_pretty_size(64)];
+    ESP_LOGD(TAG, "Staging %zu RX bytes with %ums delay: %s", data.size(), delay_ms,
+             format_hex_pretty_to(hex_buf, sizeof(hex_buf), data.data(), data.size()));
+  } else if (data.size() > 64) {
+    ESP_LOGD(TAG, "Staging %zu RX bytes with %ums delay (too large to log inline)", data.size(), delay_ms);
+  }
+  this->staged_rx_.push_back({data, millis() + delay_ms});
+}
+
 }  // namespace esphome::uart_mock
@@ -43,6 +43,8 @@ class MockUartComponent : public uart::UARTComponent, public Component {
  void set_tx_hook(std::function<void(const std::vector<uint8_t> &)> &&cb) { this->tx_hook_ = std::move(cb); }
  void inject_to_rx_buffer(const std::vector<uint8_t> &data);
  void inject_to_rx_buffer(const uint8_t *data, size_t len);
+  // Stage bytes for delayed delivery - simulates transport-level latency (e.g., USB packets)
+  void inject_to_rx_buffer_delayed(const std::vector<uint8_t> &data, uint32_t delay_ms);

 protected:
  void check_logger_conflict() override {}
@@ -82,6 +84,14 @@ class MockUartComponent : public uart::UARTComponent, public Component {
  };
  std::vector<PeriodicRx> periodic_rx_;

+  // Staged RX - bytes that are pending delivery after a delay
+  // Simulates transport-level latency (e.g., USB packet delivery)
+  struct StagedRx {
+    std::vector<uint8_t> data;
+    uint32_t available_at_ms;  // millis() time when bytes become available
+  };
+  std::deque<StagedRx> staged_rx_;
+
  // Observability
  uint32_t tx_count_{0};
  uint32_t rx_count_{0};
@@ -0,0 +1,64 @@
+esphome:
+  name: uart-mock-modbus-no-thresh
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+# Simulate a non-hardware UART (e.g., USB UART) by not setting rx_full_threshold.
+# This leaves it at the default sentinel value (0), triggering the 50ms fallback timeout.
+uart_mock:
+  - id: virtual_uart_dev
+    baud_rate: 9600
+    auto_start: false
+    debug:
+    on_tx:
+      - then:
+          - if:
+              condition: #Read 80 input registers on device 2, starting at address 0 (SDM meter request)
+                lambda: "return data == std::vector<uint8_t>({0x02,0x04,0x00,0x00,0x00,0x50,0xF0,0x05});"
+              then:
+                - uart_mock.inject_rx: # First USB packet: SDM meter response part 1
+                    !lambda return {0x02,0x04,0xA0,0x43,0x73,0x19,0x9A,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x3F,0x80,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00};
+                - uart_mock.inject_rx: # Second USB packet: rest of response (staged with 40ms latency)
+                    delay: 40ms
+                    data: !lambda return{0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                      0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x42,0x6F,0xCC,0xCD,0x43,0x7C,0xB8,0x10,0x3D,0x38,0x51,0xEC,
+                      0x43,0x81,0x1B,0xE7,0x3B,0x03,0x12,0x6F,0x50,0x1B};
+
+modbus:
+  uart_id: virtual_uart_dev
+  turnaround_time: 10ms
+
+sensor:
+  - platform: sdm_meter
+    address: 2
+    update_interval: 1s
+    phase_a:
+      voltage:
+        name: sdm_voltage
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: 'id(virtual_uart_dev).start_scenario();'
@@ -5,6 +5,10 @@ test_uart_mock_modbus :
  1. Read a single register and parse successfully (basic_register)
  2. Read multiple registers from SDM meter and parse successfully (sdm_voltage), with some intermediate delay to simulate UART buffer time.

+test_uart_mock_modbus_no_threshold :
+  Test modbus with no rx_full_threshold set (simulating USB UART / non-hardware UART).
+  Verifies the 50ms fallback timeout handles chunked data with USB packet gaps.
+
 """

 from __future__ import annotations
@@ -218,3 +222,78 @@ async def test_uart_mock_modbus_timing(
                f"Timeout waiting for SDM voltage change. Received sensor states:\n"
                f"  sdm_voltage: {sensor_states['sdm_voltage']}\n"
            )
+
+
+@pytest.mark.asyncio
+async def test_uart_mock_modbus_no_threshold(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test modbus with no rx_full_threshold (simulating USB UART).
+
+    Without the 50ms fallback timeout, the chunked response with a 40ms gap
+    between USB packets would cause a false timeout and CRC failure cascade.
+    """
+    # Replace external component path placeholder
+    external_components_path = str(
+        Path(__file__).parent / "fixtures" / "external_components"
+    )
+    yaml_config = yaml_config.replace(
+        "EXTERNAL_COMPONENT_PATH", external_components_path
+    )
+
+    loop = asyncio.get_running_loop()
+
+    # Track sensor state updates (after initial state is swallowed)
+    sensor_states: dict[str, list[float]] = {
+        "sdm_voltage": [],
+    }
+
+    voltage_changed = loop.create_future()
+
+    def on_state(state: EntityState) -> None:
+        if isinstance(state, SensorState) and not state.missing_state:
+            sensor_name = key_to_sensor.get(state.key)
+            if sensor_name and sensor_name in sensor_states:
+                sensor_states[sensor_name].append(state.state)
+                # Check if this is a good voltage reading (243V)
+                if (
+                    sensor_name == "sdm_voltage"
+                    and state.state > 200.0
+                    and not voltage_changed.done()
+                ):
+                    voltage_changed.set_result(True)
+
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected() as client,
+    ):
+        entities, _ = await client.list_entities_services()
+
+        # Build key mappings for all sensor types
+        all_names = list(sensor_states.keys())
+        key_to_sensor = build_key_to_entity_mapping(entities, all_names)
+
+        # Set up initial state helper
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+
+        try:
+            await initial_state_helper.wait_for_initial_states()
+        except TimeoutError:
+            pytest.fail("Timeout waiting for initial states")
+
+        # Start the UART mock scenario now that we're subscribed
+        start_btn = find_entity(entities, "start_scenario", ButtonInfo)
+        assert start_btn is not None, "Start Scenario button not found"
+        client.button_command(start_btn.key)
+
+        # Wait for voltage to be updated with successful parse
+        try:
+            await asyncio.wait_for(voltage_changed, timeout=2.0)
+        except TimeoutError:
+            pytest.fail(
+                f"Timeout waiting for SDM voltage change. Received sensor states:\n"
+                f"  sdm_voltage: {sensor_states['sdm_voltage']}\n"
+            )