wiredsensor/tools/wiredsensor.py

#!/usr/bin/env python3
"""Modbus RTU master and end-to-end test suite for the wiredsensor node.

This is deliberately an **independent** implementation of Modbus RTU. The
framing and CRC below were written from the specification rather than sharing
code with `wiredsensor-core`. If both ends shared an implementation, a bug in it
would cancel out and every test here would pass regardless — so the duplication
is the point, not an oversight.

It is hand-written rather than built on pymodbus for one concrete reason: half
these checks inject *deliberately malformed* frames — a bad CRC, a truncated
frame, a frame for another unit — and assert the node stays silent. A conforming
client library exists precisely to make those frames unconstructable. Use
pymodbus to cross-check the happy path against a third-party stack; use this to
verify the node behaves on a bus that is misbehaving.

Talks to the bus through the USB-to-RS485 bridge firmware (`bridge/`), which is
protocol-agnostic and just moves bytes.

    ./wiredsensor.py --port /dev/ttyACM0 measure
    ./wiredsensor.py --port /dev/ttyACM0 monitor
    ./wiredsensor.py --port /dev/ttyACM0 test
"""

from __future__ import annotations

import argparse
import struct
import sys
import time
from dataclasses import dataclass

try:
    import serial
except ImportError:
    sys.exit("pyserial is required: pip install pyserial")

# ---------------------------------------------------------------- protocol ---

ADDR_BROADCAST = 0x00
ADDR_MIN, ADDR_MAX = 0x01, 0xF7

HEADER_LEN = 2  # ADDR, FC
CRC_LEN = 2
MIN_FRAME = HEADER_LEN + CRC_LEN
MAX_FRAME = 256  # the RTU limit
MAX_DATA = MAX_FRAME - HEADER_LEN - CRC_LEN

# Each host write becomes exactly one DE-bracketed transmission by the bridge,
# and a USB full-speed bulk packet is 64 bytes, so a request must fit in one.
MAX_BRIDGE_FRAME = 64

FC_READ_HOLDING = 0x03
FC_READ_INPUT = 0x04
FC_DIAGNOSTIC = 0x08

DIAG_RETURN_QUERY_DATA = 0x0000

EXCEPTION_FLAG = 0x80

EXCEPTION_NAMES = {
    0x01: "ILLEGAL_FUNCTION",
    0x02: "ILLEGAL_DATA_ADDRESS",
    0x03: "ILLEGAL_DATA_VALUE",
    0x04: "SERVER_DEVICE_FAILURE",
}

MAX_READ_REGISTERS = 125

# ------------------------------------------------------------ register map ---

REG_TEMP_MILLI_C = 0x0000  # 2 registers, signed
REG_RH_MILLI_PCT = 0x0002  # 2 registers, signed
REG_AGE_MS = 0x0004
REG_FLAGS = 0x0005
REG_SENSOR_STATUS = 0x0006
REG_I2C_ERRORS = 0x0007
REG_SENSOR_CRC_ERRORS = 0x0008
REG_FRAME_ERRORS = 0x0009
REG_CRC_ERRORS = 0x000A
REG_FW_MAJOR_MINOR = 0x000B
REG_FW_PATCH_PROTO = 0x000C
REG_DEVICE_SERIAL = 0x000D  # 2 registers
REG_SENSOR_SERIAL = 0x000F  # 2 registers
REG_UPTIME_S = 0x0011  # 2 registers
REG_COUNT = 0x0013

REG_MEASUREMENT_END = 0x0005

FLAG_NAMES = [
    (1 << 0, "SENSOR_OK"),
    (1 << 1, "DATA_STALE"),
    (1 << 2, "SENSOR_FAULT"),
    (1 << 3, "UART_ERROR"),
    (1 << 4, "EVER_MEASURED"),
    (1 << 5, "SENSOR_RESET_SEEN"),
    (1 << 6, "HEATER_ON"),
]

REGISTER_NAMES = {
    REG_TEMP_MILLI_C: "temp_milli_c (hi)",
    REG_TEMP_MILLI_C + 1: "temp_milli_c (lo)",
    REG_RH_MILLI_PCT: "rh_milli_pct (hi)",
    REG_RH_MILLI_PCT + 1: "rh_milli_pct (lo)",
    REG_AGE_MS: "age_ms",
    REG_FLAGS: "flags",
    REG_SENSOR_STATUS: "sensor_status",
    REG_I2C_ERRORS: "i2c_errors",
    REG_SENSOR_CRC_ERRORS: "sensor_crc_errors",
    REG_FRAME_ERRORS: "frame_errors",
    REG_CRC_ERRORS: "crc_errors",
    REG_FW_MAJOR_MINOR: "fw_major_minor",
    REG_FW_PATCH_PROTO: "fw_patch_proto",
    REG_DEVICE_SERIAL: "device_serial (hi)",
    REG_DEVICE_SERIAL + 1: "device_serial (lo)",
    REG_SENSOR_SERIAL: "sensor_serial (hi)",
    REG_SENSOR_SERIAL + 1: "sensor_serial (lo)",
    REG_UPTIME_S: "uptime_s (hi)",
    REG_UPTIME_S + 1: "uptime_s (lo)",
}


def crc16(data: bytes) -> int:
    """CRC-16/MODBUS: reflected poly 0xA001, init 0xFFFF, no final XOR."""
    crc = 0xFFFF
    for byte in data:
        crc ^= byte
        for _ in range(8):
            crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1
    return crc


def build_frame(addr: int, fc: int, data: bytes = b"") -> bytes:
    """An RTU ADU: address, function code, data, CRC low byte first."""
    if len(data) > MAX_DATA:
        raise ValueError(f"data field of {len(data)} exceeds {MAX_DATA}")
    body = bytes([addr, fc]) + data
    return body + struct.pack("<H", crc16(body))


def read_request(start: int, count: int) -> bytes:
    """The data field of a register read: start and count, both big-endian."""
    return struct.pack(">HH", start, count)


@dataclass
class Frame:
    addr: int
    fc: int
    data: bytes

    @property
    def is_exception(self) -> bool:
        return bool(self.fc & EXCEPTION_FLAG)

    @property
    def exception_name(self) -> str | None:
        if not self.is_exception or len(self.data) != 1:
            return None
        return EXCEPTION_NAMES.get(self.data[0], f"unknown({self.data[0]:#04x})")


class ProtocolError(Exception):
    """The bytes on the wire were not a usable response."""


class ExceptionResponse(Exception):
    """The node answered, and the answer was a Modbus exception."""

    def __init__(self, fc: int, code: int):
        self.fc = fc
        self.code = code
        name = EXCEPTION_NAMES.get(code, "unknown")
        super().__init__(f"function {fc:#04x} -> {name} ({code:#04x})")


def parse_frame(raw: bytes) -> Frame:
    """Validate a received ADU.

    There is no length field to check: in RTU the frame is delimited by the idle
    line and the CRC is the only integrity check there is.
    """
    if len(raw) < MIN_FRAME:
        raise ProtocolError(f"short frame: {len(raw)} bytes ({raw.hex(' ')})")
    if len(raw) > MAX_FRAME:
        raise ProtocolError(f"long frame: {len(raw)} bytes")
    body, tail = raw[:-CRC_LEN], raw[-CRC_LEN:]
    expected = struct.unpack("<H", tail)[0]
    actual = crc16(body)
    if actual != expected:
        raise ProtocolError(f"CRC {actual:#06x} != {expected:#06x} ({raw.hex(' ')})")
    return Frame(raw[0], raw[1], body[HEADER_LEN:])


def decode_registers(data: bytes) -> list[int]:
    """Unpack the byte-counted register block of an 0x03/0x04 response."""
    if not data:
        raise ProtocolError("register response has an empty data field")
    byte_count = data[0]
    if byte_count != len(data) - 1:
        raise ProtocolError(
            f"byte count {byte_count} disagrees with {len(data) - 1} bytes of payload"
        )
    if byte_count % 2:
        raise ProtocolError(f"odd byte count {byte_count}")
    return list(struct.unpack(f">{byte_count // 2}H", data[1:]))


def reg_u32(regs: list[int], at: int, base: int = 0) -> int:
    """A 32-bit register pair, high word first."""
    i = at - base
    return (regs[i] << 16) | regs[i + 1]


def reg_i32(regs: list[int], at: int, base: int = 0) -> int:
    v = reg_u32(regs, at, base)
    return v - (1 << 32) if v & (1 << 31) else v


# ------------------------------------------------------------------ decode ---


@dataclass
class Measurement:
    temp_c: float
    rh_pct: float
    age_ms: int

    @classmethod
    def decode(cls, regs: list[int], base: int = REG_TEMP_MILLI_C) -> Measurement:
        if len(regs) < 5:
            raise ProtocolError(f"expected 5 measurement registers, got {len(regs)}")
        return cls(
            reg_i32(regs, REG_TEMP_MILLI_C, base) / 1000.0,
            reg_i32(regs, REG_RH_MILLI_PCT, base) / 1000.0,
            regs[REG_AGE_MS - base],
        )

    def __str__(self) -> str:
        return f"{self.temp_c:+.3f} °C   {self.rh_pct:.3f} %RH   age {self.age_ms} ms"


@dataclass
class Info:
    fw: tuple[int, int, int]
    proto_version: int
    device_serial: int
    sensor_serial: int
    uptime_s: int

    @classmethod
    def decode(cls, regs: list[int], base: int = REG_FW_MAJOR_MINOR) -> Info:
        if len(regs) < REG_COUNT - REG_FW_MAJOR_MINOR:
            raise ProtocolError(f"expected 8 info registers, got {len(regs)}")
        major_minor = regs[REG_FW_MAJOR_MINOR - base]
        patch_proto = regs[REG_FW_PATCH_PROTO - base]
        return cls(
            (major_minor >> 8, major_minor & 0xFF, patch_proto >> 8),
            patch_proto & 0xFF,
            reg_u32(regs, REG_DEVICE_SERIAL, base),
            reg_u32(regs, REG_SENSOR_SERIAL, base),
            reg_u32(regs, REG_UPTIME_S, base),
        )

    def __str__(self) -> str:
        return (
            f"fw {self.fw[0]}.{self.fw[1]}.{self.fw[2]}  proto v{self.proto_version}  "
            f"device_serial {self.device_serial:#010x}  "
            f"sensor_serial {self.sensor_serial:#010x}  uptime {self.uptime_s} s"
        )


@dataclass
class Status:
    flags: int
    sensor_status: int
    i2c_errors: int
    sensor_crc_errors: int
    frame_errors: int
    crc_errors: int

    @classmethod
    def decode(cls, regs: list[int], base: int = REG_FLAGS) -> Status:
        if len(regs) < REG_FW_MAJOR_MINOR - REG_FLAGS:
            raise ProtocolError(f"expected 6 status registers, got {len(regs)}")
        return cls(
            regs[REG_FLAGS - base],
            regs[REG_SENSOR_STATUS - base],
            regs[REG_I2C_ERRORS - base],
            regs[REG_SENSOR_CRC_ERRORS - base],
            regs[REG_FRAME_ERRORS - base],
            regs[REG_CRC_ERRORS - base],
        )

    @property
    def flag_names(self) -> list[str]:
        return [name for bit, name in FLAG_NAMES if self.flags & bit]

    def __str__(self) -> str:
        return (
            f"flags {self.flags:#06x} [{' '.join(self.flag_names) or '-'}]  "
            f"sht_status {self.sensor_status:#06x}  "
            f"err i2c={self.i2c_errors} sht_crc={self.sensor_crc_errors} "
            f"frame={self.frame_errors} bus_crc={self.crc_errors}"
        )


# ------------------------------------------------------------------ master ---


class Bus:
    """A Modbus master on the segment, reached through the USB-RS485 bridge."""

    def __init__(self, port: str, unit: int = 0x01, verbose: bool = False):
        # The bridge is a CDC device, so the baud rate here is ignored; the RS485
        # line rate is fixed in the bridge firmware.
        self.ser = serial.Serial(port, 115200, timeout=0)
        self.unit = unit
        self.verbose = verbose
        time.sleep(0.05)
        self.ser.reset_input_buffer()

    def close(self) -> None:
        self.ser.close()

    def send_raw(self, raw: bytes) -> None:
        """Put bytes on the bus verbatim, bypassing frame construction.

        One write becomes exactly one transmission by the bridge, which is what
        lets us inject deliberately malformed frames.
        """
        if self.verbose:
            print(f"  TX {raw.hex(' ')}", file=sys.stderr)
        self.ser.reset_input_buffer()
        self.ser.write(raw)
        self.ser.flush()

    def read_raw(self, timeout: float = 0.5, gap: float = 0.02) -> bytes:
        """Collect a response, using an idle gap to decide it is complete.

        Mirrors how the node itself delimits frames, so a response arriving in
        several USB chunks is still reassembled into one frame.
        """
        deadline = time.monotonic() + timeout
        buf = bytearray()
        last = None
        while time.monotonic() < deadline:
            waiting = self.ser.in_waiting
            if waiting:
                buf.extend(self.ser.read(waiting))
                last = time.monotonic()
            elif buf and last is not None and time.monotonic() - last >= gap:
                break
            else:
                time.sleep(0.001)
        if self.verbose and buf:
            print(f"  RX {bytes(buf).hex(' ')}", file=sys.stderr)
        return bytes(buf)

    def expect_raw(self, fc: int, data: bytes = b"", addr: int | None = None) -> bytes:
        """Send a request and return whatever bytes come back, if any."""
        self.send_raw(build_frame(self.unit if addr is None else addr, fc, data))
        return self.read_raw()

    def request(self, fc: int, data: bytes = b"", addr: int | None = None) -> Frame:
        """Send a request and return the parsed response, raising if none comes."""
        raw = self.expect_raw(fc, data, addr)
        if not raw:
            raise ProtocolError(f"no response to function {fc:#04x}")
        return parse_frame(raw)

    def transact(self, fc: int, data: bytes = b"") -> Frame:
        """As `request`, but turn an exception response into a Python exception."""
        response = self.request(fc, data)
        if response.addr != self.unit:
            raise ProtocolError(
                f"response from {response.addr:#04x} != {self.unit:#04x}"
            )
        if response.is_exception:
            if response.fc != fc | EXCEPTION_FLAG:
                raise ProtocolError(
                    f"exception on function {response.fc & ~EXCEPTION_FLAG:#04x}, "
                    f"asked {fc:#04x}"
                )
            raise ExceptionResponse(fc, response.data[0])
        if response.fc != fc:
            raise ProtocolError(f"response fc {response.fc:#04x} != request {fc:#04x}")
        return response

    def read_registers(
        self, start: int, count: int, fc: int = FC_READ_INPUT
    ) -> list[int]:
        regs = decode_registers(self.transact(fc, read_request(start, count)).data)
        if len(regs) != count:
            raise ProtocolError(f"asked for {count} registers, got {len(regs)}")
        return regs

    def diagnostic(self, payload: bytes) -> bytes:
        """Return Query Data: the node echoes the data field back unchanged."""
        data = struct.pack(">H", DIAG_RETURN_QUERY_DATA) + payload
        return self.transact(FC_DIAGNOSTIC, data).data[2:]

    def measurement(self, fc: int = FC_READ_INPUT) -> Measurement:
        return Measurement.decode(
            self.read_registers(REG_TEMP_MILLI_C, REG_MEASUREMENT_END, fc)
        )

    def status(self, fc: int = FC_READ_INPUT) -> Status:
        count = REG_FW_MAJOR_MINOR - REG_FLAGS
        return Status.decode(self.read_registers(REG_FLAGS, count, fc))

    def info(self, fc: int = FC_READ_INPUT) -> Info:
        count = REG_COUNT - REG_FW_MAJOR_MINOR
        return Info.decode(self.read_registers(REG_FW_MAJOR_MINOR, count, fc))

    def all_registers(self, fc: int = FC_READ_INPUT) -> list[int]:
        return self.read_registers(0, REG_COUNT, fc)


# ------------------------------------------------------------------- tests ---


class Results:
    def __init__(self) -> None:
        self.passed = 0
        self.failed: list[tuple[str, str]] = []

    def check(self, name: str, fn) -> None:
        try:
            detail = fn()
            self.passed += 1
            print(f"  \033[32mPASS\033[0m  {name}" + (f"  — {detail}" if detail else ""))
        except Exception as exc:  # noqa: BLE001 - a failed probe is a test result
            self.failed.append((name, str(exc)))
            print(f"  \033[31mFAIL\033[0m  {name}\n          {exc}")


def expect_exception(bus: Bus, fc: int, data: bytes, want: int) -> str:
    """Assert a request draws a specific Modbus exception code."""
    try:
        bus.transact(fc, data)
    except ExceptionResponse as exc:
        want_name = EXCEPTION_NAMES[want]
        assert exc.code == want, f"got {exc}, expected {want_name}"
        return want_name
    raise AssertionError(f"function {fc:#04x} succeeded, expected {EXCEPTION_NAMES[want]}")


def run_tests(bus: Bus) -> int:
    """Exercise the parts of the protocol only a real bus can verify.

    Framing, CRC arithmetic and the register map are already covered by host
    unit tests in `core/`. What cannot be tested off-hardware is everything
    timing-dependent: that the node delimits frames by an idle gap, that it
    stays off the wire for traffic it must not answer, and that its
    driver-enable turnaround leaves the response intact.
    """
    r = Results()

    print("\nidentity and readings")

    def t_info():
        info = bus.info()
        assert info.proto_version == 2, f"protocol version {info.proto_version} != 2"
        assert info.sensor_serial != 0, "sensor serial is zero — SHT31 not identified"
        return str(info)

    r.check("read identity registers", t_info)

    def t_measure():
        m = bus.measurement()
        assert -40 <= m.temp_c <= 125, f"temperature {m.temp_c} outside SHT31 range"
        assert 0 <= m.rh_pct <= 100, f"humidity {m.rh_pct} outside 0..100"
        assert m.age_ms <= 3000, f"reading is {m.age_ms} ms old"
        return str(m)

    r.check("read measurement registers", t_measure)

    def t_status():
        s = bus.status()
        assert s.flags & 0x10, "EVER_MEASURED not set"
        assert not s.flags & 0x04, "SENSOR_FAULT is set"
        return str(s)

    r.check("read diagnostic registers", t_status)

    print("\nregister map consistency")

    def t_both_function_codes():
        # A master that only implements 0x03 must see exactly what one using
        # 0x04 sees. The two are served from one table precisely so they agree.
        holding = bus.all_registers(FC_READ_HOLDING)
        # Skip the volatile registers: age and uptime move between the two reads.
        volatile = {REG_AGE_MS, REG_UPTIME_S, REG_UPTIME_S + 1}
        inputs = bus.all_registers(FC_READ_INPUT)
        for i, (h, n) in enumerate(zip(holding, inputs)):
            if i in volatile:
                continue
            assert h == n, f"register {i} ({REGISTER_NAMES.get(i, '?')}): {h} != {n}"
        return f"0x03 and 0x04 agree across {len(holding)} registers"

    r.check("holding and input registers agree", t_both_function_codes)

    def t_subrange():
        # ESPHome coalesces adjacent registers into one read, so a sub-range must
        # return the same values as the same addresses read individually.
        whole = bus.all_registers()
        for start, count in [
            (REG_FLAGS, 1),
            (REG_SENSOR_STATUS, 4),
            (REG_DEVICE_SERIAL, 2),
            (REG_FLAGS, REG_COUNT - REG_FLAGS),
        ]:
            part = bus.read_registers(start, count)
            expect = whole[start : start + count]
            assert part == expect, f"start {start} count {count}: {part} != {expect}"
        return "4 sub-ranges match the whole-map read"

    r.check("sub-range reads are consistent", t_subrange)

    print("\nround trip integrity")

    def t_diag_binary():
        # Bytes that would need escaping in any delimiter-based protocol.
        probe = bytes([0x00, 0xFF, 0x0A, 0x0D, 0x3A, 0x7E, 0x55, 0xAA])
        echo = bus.diagnostic(probe)
        assert echo == probe, f"echo {echo.hex(' ')} != sent {probe.hex(' ')}"
        return f"{len(probe)} bytes binary-transparent"

    r.check("diagnostic echoes arbitrary bytes", t_diag_binary)

    def t_diag_long():
        # A 64-byte frame: the largest the bridge can send as a single USB packet
        # and therefore as a single transmission. Well past the 16-byte RX FIFO
        # watermark, which is the case the frame-gap logic gets wrong if the
        # deadline is measured from the ISR's timestamp.
        payload = bytes(range(MAX_BRIDGE_FRAME - MIN_FRAME - 2))
        echo = bus.diagnostic(payload)
        assert echo == payload, f"echo of {len(echo)} bytes != sent {len(payload)}"
        return f"{len(payload)}-byte payload, {MAX_BRIDGE_FRAME}-byte frame"

    r.check("diagnostic echoes a maximum single-packet frame", t_diag_long)

    def t_long_response():
        # The response side of the same concern: 19 registers is a 43-byte
        # response, comfortably past the watermark.
        regs = bus.all_registers()
        assert len(regs) == REG_COUNT, f"got {len(regs)} registers"
        return f"{REG_COUNT} registers in one {5 + 2 * REG_COUNT}-byte response"

    r.check("whole-map read returns a long response intact", t_long_response)

    def t_back_to_back():
        # Catches state leaking between frames — a stale receive buffer or a
        # gap-detection flag left set would show up here and nowhere else.
        for i in range(20):
            probe = bytes([i, 0xA5 ^ i])
            echo = bus.diagnostic(probe)
            assert echo == probe, f"iteration {i}: {echo.hex(' ')} != {probe.hex(' ')}"
        return "20 consecutive requests, no state leak"

    r.check("back-to-back requests", t_back_to_back)

    print("\nsilence where silence is required")

    def t_other_unit():
        other = 0x02 if bus.unit != 0x02 else 0x03
        raw = bus.expect_raw(FC_READ_INPUT, read_request(0, 5), addr=other)
        assert not raw, f"answered a frame addressed to {other:#04x}: {raw.hex(' ')}"
        return f"ignored a frame for unit {other:#04x}"

    r.check("frame for another unit is ignored", t_other_unit)

    def t_broadcast():
        raw = bus.expect_raw(FC_READ_INPUT, read_request(0, 5), addr=ADDR_BROADCAST)
        assert not raw, f"answered a broadcast: {raw.hex(' ')}"
        return "ignored the broadcast address"

    r.check("broadcast is not answered", t_broadcast)

    def t_bad_crc():
        frame = bytearray(build_frame(bus.unit, FC_READ_INPUT, read_request(0, 5)))
        frame[-1] ^= 0xFF
        bus.send_raw(bytes(frame))
        raw = bus.read_raw()
        assert not raw, f"answered a frame with a bad CRC: {raw.hex(' ')}"
        return "ignored a corrupted frame"

    r.check("bad CRC is ignored", t_bad_crc)

    def t_truncated():
        # With no length field, a frame one byte short can only be caught by the
        # CRC — which is exactly what must happen here.
        frame = build_frame(bus.unit, FC_READ_INPUT, read_request(0, 5))[:-1]
        bus.send_raw(frame)
        raw = bus.read_raw()
        assert not raw, f"answered a truncated frame: {raw.hex(' ')}"
        return "ignored a truncated frame"

    r.check("truncated frame is ignored", t_truncated)

    print("\nexception responses")

    def t_unknown_fc():
        # 0x06 Write Single Register is a perfectly ordinary Modbus function this
        # node does not implement, which is the interesting case: a master may
        # well try it.
        return expect_exception(bus, 0x06, b"\x00\x00\x00\x01", 0x01)

    r.check("unimplemented function is refused", t_unknown_fc)

    def t_past_end():
        return expect_exception(bus, FC_READ_INPUT, read_request(REG_COUNT, 1), 0x02)

    r.check("read past the map is ILLEGAL_DATA_ADDRESS", t_past_end)

    def t_straddles_end():
        data = read_request(REG_COUNT - 1, 2)
        return expect_exception(bus, FC_READ_INPUT, data, 0x02)

    r.check("read straddling the end is refused", t_straddles_end)

    def t_zero_count():
        return expect_exception(bus, FC_READ_INPUT, read_request(0, 0), 0x03)

    r.check("zero register count is ILLEGAL_DATA_VALUE", t_zero_count)

    def t_huge_count():
        data = read_request(0, MAX_READ_REGISTERS + 1)
        return expect_exception(bus, FC_READ_INPUT, data, 0x03)

    r.check("oversized register count is ILLEGAL_DATA_VALUE", t_huge_count)

    def t_short_request():
        return expect_exception(bus, FC_READ_INPUT, b"\x00\x00", 0x03)

    r.check("misshaped request is ILLEGAL_DATA_VALUE", t_short_request)

    def t_unknown_subfunction():
        return expect_exception(bus, FC_DIAGNOSTIC, b"\x00\x0a", 0x01)

    r.check("unknown diagnostic sub-function is refused", t_unknown_subfunction)

    print("\ndiagnostic counters")

    def t_crc_counter():
        before = bus.status().crc_errors
        bad = bytearray(build_frame(bus.unit, FC_READ_INPUT, read_request(0, 5)))
        bad[-1] ^= 0xFF
        bus.send_raw(bytes(bad))
        bus.read_raw(timeout=0.15)
        after = bus.status().crc_errors
        assert after == before + 1, f"crc_errors went {before} -> {after}, expected +1"
        return f"crc_errors {before} -> {after}"

    r.check("a corrupt frame increments crc_errors", t_crc_counter)

    total = r.passed + len(r.failed)
    print(f"\n{r.passed}/{total} passed")
    if r.failed:
        print("\nfailures:")
        for name, why in r.failed:
            print(f"  {name}: {why}")
        return 1
    return 0


# ----------------------------------------------------------------- esphome ---

ESPHOME_YAML = """\
# wiredsensor node at unit address {unit} — paste into your ESPHome config and
# set the tx_pin/rx_pin to whatever your RS485 transceiver is wired to.
uart:
  id: rs485
  tx_pin: GPIO17
  rx_pin: GPIO16
  baud_rate: {baud}
  data_bits: 8
  parity: NONE
  stop_bits: 1

modbus:
  id: rs485_bus
  uart_id: rs485

modbus_controller:
  - id: wiredsensor
    address: {unit}
    modbus_id: rs485_bus
    update_interval: 30s

sensor:
  - platform: modbus_controller
    modbus_controller_id: wiredsensor
    name: "Temperature"
    register_type: read          # function code 0x04
    address: 0x0000
    value_type: S_DWORD
    unit_of_measurement: "°C"
    device_class: temperature
    accuracy_decimals: 2
    filters:
      - multiply: 0.001

  - platform: modbus_controller
    modbus_controller_id: wiredsensor
    name: "Humidity"
    register_type: read
    address: 0x0002
    value_type: S_DWORD
    unit_of_measurement: "%"
    device_class: humidity
    accuracy_decimals: 2
    filters:
      - multiply: 0.001

  - platform: modbus_controller
    modbus_controller_id: wiredsensor
    name: "Reading age"
    register_type: read
    address: 0x0004
    value_type: U_WORD
    unit_of_measurement: "ms"
    entity_category: diagnostic

# The health flags deliberately use `holding` (function code 0x03) even though
# they are the same registers. ESPHome coalesces adjacent registers of the same
# register_type into one command, and a read overlapping 0x0000..0x0004 is
# refused outright when the sensor has never produced a reading. Asking for the
# flags under the other function code puts them in their own command, so they
# still report when the measurement cannot — which is exactly when you want them.
binary_sensor:
  - platform: modbus_controller
    modbus_controller_id: wiredsensor
    name: "Sensor OK"
    register_type: holding
    address: 0x0005
    bitmask: 0x01
    entity_category: diagnostic

  - platform: modbus_controller
    modbus_controller_id: wiredsensor
    name: "Data stale"
    register_type: holding
    address: 0x0005
    bitmask: 0x02
    entity_category: diagnostic

  - platform: modbus_controller
    modbus_controller_id: wiredsensor
    name: "Sensor fault"
    register_type: holding
    address: 0x0005
    bitmask: 0x04
    entity_category: diagnostic
"""


# --------------------------------------------------------------------- CLI ---


def main() -> int | str:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--port", default="/dev/ttyACM0", help="bridge serial port")
    ap.add_argument(
        "--unit", type=lambda s: int(s, 0), default=0x01, help="node address"
    )
    ap.add_argument("-v", "--verbose", action="store_true", help="dump bus traffic")
    ap.add_argument(
        "--holding",
        action="store_const",
        const=FC_READ_HOLDING,
        default=FC_READ_INPUT,
        dest="fc",
        help="use 0x03 Read Holding Registers instead of 0x04 Read Input Registers",
    )
    sub = ap.add_subparsers(dest="cmd", required=True)

    sub.add_parser("measure", help="read temperature and humidity once")
    sub.add_parser("info", help="read firmware and serial numbers")
    sub.add_parser("status", help="read health flags and counters")
    sub.add_parser("registers", help="dump the whole register map")
    sub.add_parser("test", help="run the end-to-end test suite")
    sub.add_parser("esphome", help="print a ready-to-paste ESPHome config")

    p_diag = sub.add_parser("diag", help="Return Query Data echo test")
    p_diag.add_argument("--bytes", type=int, default=8, help="payload length")

    p_mon = sub.add_parser("monitor", help="poll continuously")
    p_mon.add_argument(
        "--interval", type=float, default=1.0, help="seconds between polls"
    )

    args = ap.parse_args()

    # Needs no bus at all, so it works before any hardware is wired up.
    if args.cmd == "esphome":
        print(ESPHOME_YAML.format(unit=args.unit, baud=19200))
        return 0

    try:
        bus = Bus(args.port, args.unit, args.verbose)
    except serial.SerialException as exc:
        return f"cannot open {args.port}: {exc}"

    try:
        if args.cmd == "measure":
            print(bus.measurement(args.fc))
        elif args.cmd == "info":
            print(bus.info(args.fc))
        elif args.cmd == "status":
            print(bus.status(args.fc))
        elif args.cmd == "registers":
            for i, v in enumerate(bus.all_registers(args.fc)):
                name = REGISTER_NAMES.get(i, "")
                print(f"  {i:#06x}  {v:#06x}  {v:>6}  {name}")
        elif args.cmd == "diag":
            probe = bytes(i & 0xFF for i in range(args.bytes))
            echo = bus.diagnostic(probe)
            print("echo ok" if echo == probe else f"MISMATCH: {echo.hex(' ')}")
        elif args.cmd == "monitor":
            while True:
                try:
                    print(f"{time.strftime('%H:%M:%S')}  {bus.measurement(args.fc)}")
                except (ProtocolError, ExceptionResponse) as exc:
                    print(f"{time.strftime('%H:%M:%S')}  {exc}")
                time.sleep(args.interval)
        elif args.cmd == "test":
            return run_tests(bus)
    except KeyboardInterrupt:
        return 130
    except ExceptionResponse as exc:
        return f"the node refused the request: {exc}"
    except ProtocolError as exc:
        return f"protocol error: {exc}"
    finally:
        bus.close()

    return 0


if __name__ == "__main__":
    sys.exit(main())