diff --git a/Cargo.lock b/Cargo.lock index 6b9c187..388d26d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -567,6 +567,12 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "panic-halt" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a513e167849a384b7f9b746e517604398518590a9142f4846a32e3c2a4de7b11" + [[package]] name = "panic-probe" version = "1.0.0" @@ -1195,6 +1201,21 @@ dependencies = [ "windows-link", ] +[[package]] +name = "wiredsensor-bridge" +version = "0.1.0" +dependencies = [ + "cortex-m", + "cortex-m-rt", + "embedded-hal 1.0.0", + "fugit", + "panic-halt", + "rp2040-boot2", + "rp2040-hal", + "usb-device", + "usbd-serial", +] + [[package]] name = "wiredsensor-core" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index fe1be51..d86d1b0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [workspace] resolver = "2" -members = ["core", "firmware"] +members = ["core", "firmware", "bridge"] [workspace.package] version = "0.1.0" diff --git a/bridge/Cargo.toml b/bridge/Cargo.toml new file mode 100644 index 0000000..2721c4c --- /dev/null +++ b/bridge/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "wiredsensor-bridge" +description = "Turns a spare RP2040 into a USB-to-RS485 bridge for testing the bus" +version.workspace = true +edition.workspace = true +license.workspace = true + +[[bin]] +name = "wiredsensor-bridge" +path = "src/main.rs" +test = false +bench = false + +[dependencies] +rp2040-hal = { version = "0.12", features = ["rt", "critical-section-impl"] } +rp2040-boot2 = "0.3" + +cortex-m = "0.7" +cortex-m-rt = "0.7" +embedded-hal = "1.0" +fugit = "0.3" + +usb-device = "0.3" +usbd-serial = "0.2" + +panic-halt = "1.0" diff --git a/bridge/build.rs b/bridge/build.rs new file mode 100644 index 0000000..bab4389 --- /dev/null +++ b/bridge/build.rs @@ -0,0 +1,18 @@ +use std::{env, fs, path::PathBuf}; + +fn main() { + // Copy memory.x somewhere the linker will find it. Relying on the linker's + // working directory does not work in a workspace, where cargo runs rustc + // from the workspace root rather than this package's directory. + let out = PathBuf::from(env::var("OUT_DIR").expect("OUT_DIR")); + fs::copy("memory.x", out.join("memory.x")).expect("copy memory.x into OUT_DIR"); + println!("cargo:rustc-link-search={}", out.display()); + + // --nmagic disables page alignment of sections; without it the vector table + // is pushed away from the start of flash and the boot ROM cannot find it. + println!("cargo:rustc-link-arg-bins=--nmagic"); + println!("cargo:rustc-link-arg-bins=-Tlink.x"); + + println!("cargo:rerun-if-changed=memory.x"); + println!("cargo:rerun-if-changed=build.rs"); +} diff --git a/bridge/memory.x b/bridge/memory.x new file mode 100644 index 0000000..1bab8a0 --- /dev/null +++ b/bridge/memory.x @@ -0,0 +1,21 @@ +/* RP2040 with 2 MiB of QSPI flash. + * + * The boot ROM reads the first 256 bytes of flash into RAM and executes them to + * bring up XIP, so .boot2 must sit at the very start and the vector table must + * follow immediately after it. */ +MEMORY { + BOOT2 : ORIGIN = 0x10000000, LENGTH = 0x100 + FLASH : ORIGIN = 0x10000100, LENGTH = 2048K - 0x100 + + /* The four 64 KiB SRAM banks are striped, so treat them as one region. */ + RAM : ORIGIN = 0x20000000, LENGTH = 256K +} + +EXTERN(BOOT2_FIRMWARE) + +SECTIONS { + .boot2 ORIGIN(BOOT2) : + { + KEEP(*(.boot2)); + } > BOOT2 +} INSERT BEFORE .text; diff --git a/bridge/src/main.rs b/bridge/src/main.rs new file mode 100644 index 0000000..85d9bea --- /dev/null +++ b/bridge/src/main.rs @@ -0,0 +1,177 @@ +//! USB-to-RS485 bridge: turns a spare RP2040 into a PC interface for the bus. +//! +//! Deliberately **protocol-agnostic**. Bytes arriving from USB go out on the +//! differential pair; bytes arriving from the pair go back up USB. It knows +//! nothing about frames, addresses or CRCs, which is what makes it usable as a +//! test instrument: any bug it might have cannot be a bug that happens to agree +//! with the node's own protocol implementation. +//! +//! Unlike the sensor node this needs no register-level access. It never has to +//! delimit a frame, so the receive-timeout interrupt is irrelevant, and the one +//! thing it does need — knowing when the last stop bit has left the shifter, so +//! the driver can be released — `rp2040-hal` exposes directly as +//! `uart_is_busy()`. +//! +//! # Limitation worth knowing +//! +//! Each USB write from the PC becomes exactly one transmission, bracketed by DE. +//! A frame split across two USB packets would therefore be sent as two bursts +//! with an inter-frame gap between them, and the node would see two malformed +//! frames rather than one good one. USB full-speed bulk packets are 64 bytes, so +//! keep frames at or under 64 bytes — every real command in this protocol is at +//! most 21, and only an oversized `PING` payload could reach the limit. + +#![no_std] +#![no_main] + +use cortex_m_rt::entry; +use embedded_hal::digital::OutputPin; +use fugit::RateExtU32; +use panic_halt as _; +use rp2040_hal::{ + clocks::init_clocks_and_plls, + gpio::{FunctionUart, Pins, PullNone, PullUp}, + pac, + uart::{DataBits, StopBits, UartConfig, UartPeripheral}, + usb::UsbBus, + watchdog::Watchdog, + Clock, Sio, +}; +use usb_device::{ + bus::UsbBusAllocator, + device::{StringDescriptors, UsbDeviceBuilder, UsbVidPid}, +}; +use usbd_serial::SerialPort; + +/// Line rate. Must match the sensor node's `board::BAUD_RATE`. +const BAUD_RATE: u32 = 19_200; + +/// Crystal fitted to the board. +const XTAL_FREQ_HZ: u32 = 12_000_000; + +/// Product ID differs from the node's `0x27dd` so the two boards are tellable +/// apart when both are plugged into the same host. +const USB_VID: u16 = 0x16c0; +/// See [`USB_VID`]. +const USB_PID: u16 = 0x27de; + +/// Second-stage bootloader; see the node's firmware for the details. +#[link_section = ".boot2"] +#[no_mangle] +#[used] +pub static BOOT2_FIRMWARE: [u8; 256] = rp2040_boot2::BOOT_LOADER_W25Q080; + +#[entry] +fn main() -> ! { + let mut pac = pac::Peripherals::take().unwrap(); + let mut watchdog = Watchdog::new(pac.WATCHDOG); + let clocks = init_clocks_and_plls( + XTAL_FREQ_HZ, + pac.XOSC, + pac.CLOCKS, + pac.PLL_SYS, + pac.PLL_USB, + &mut pac.RESETS, + &mut watchdog, + ) + .expect("clock init failed — check XTAL_FREQ_HZ against the fitted crystal"); + + let sio = Sio::new(pac.SIO); + let pins = Pins::new( + pac.IO_BANK0, + pac.PADS_BANK0, + sio.gpio_bank0, + &mut pac.RESETS, + ); + + // Same pin assignment as the sensor node, so one wiring diagram covers both + // boards: GPIO0 -> DI, GPIO1 <- RO, GPIO2 -> DE and /RE. + let uart_pins = ( + pins.gpio0.reconfigure::(), + pins.gpio1.reconfigure::(), + ); + let uart = UartPeripheral::new(pac.UART0, uart_pins, &mut pac.RESETS) + .enable( + UartConfig::new(BAUD_RATE.Hz(), DataBits::Eight, None, StopBits::One), + clocks.peripheral_clock.freq(), + ) + .expect("uart baud rate not achievable from the peripheral clock"); + + let mut de = pins.gpio2.into_push_pull_output(); + de.set_low().ok(); + + let mut led = pins.gpio25.into_push_pull_output(); + led.set_low().ok(); + + let usb_bus = UsbBusAllocator::new(UsbBus::new( + pac.USBCTRL_REGS, + pac.USBCTRL_DPRAM, + clocks.usb_clock, + true, + &mut pac.RESETS, + )); + let mut serial = SerialPort::new(&usb_bus); + let mut usb_dev = UsbDeviceBuilder::new(&usb_bus, UsbVidPid(USB_VID, USB_PID)) + .strings(&[StringDescriptors::default() + .manufacturer("wiredsensor") + .product("wiredsensor RS485 bridge") + .serial_number("bridge")]) + .expect("USB string descriptors") + .device_class(usbd_serial::USB_CLASS_CDC) + .build(); + + let mut from_pc = [0u8; 128]; + let mut from_bus = [0u8; 64]; + + loop { + usb_dev.poll(&mut [&mut serial]); + + // ---- PC -> RS485 ---------------------------------------------------- + if let Ok(n) = serial.read(&mut from_pc) { + if n > 0 { + led.set_high().ok(); + de.set_high().ok(); + + let mut pending = &from_pc[..n]; + while !pending.is_empty() { + if let Ok(rest) = uart.write_raw(pending) { + pending = rest; + } + // Keep the host serviced while the line drains; a 64-byte + // burst takes 33 ms at 19200 baud, which is long enough that + // ignoring USB through it would risk the host giving up. + usb_dev.poll(&mut [&mut serial]); + } + + // Hold the driver until the final stop bit is genuinely gone. + // Releasing at FIFO-empty would truncate the last character. + while uart.uart_is_busy() {} + de.set_low().ok(); + + // Discard the turnaround glitch. While DE was asserted the + // transceiver's /RE was disabled and RO undriven, so the edge as + // it re-enables can clock in a spurious character. The node waits + // out a full inter-frame gap before replying, so this cannot eat + // any of the real response. + while uart.read_raw(&mut from_bus).is_ok() {} + + led.set_low().ok(); + } + } + + // ---- RS485 -> PC ---------------------------------------------------- + // Line errors are dropped along with their bytes: a test instrument that + // silently repaired corruption would hide exactly the faults we want to + // catch, and the node counts its own receive errors anyway. + if let Ok(n) = uart.read_raw(&mut from_bus) { + let mut pending = &from_bus[..n]; + while !pending.is_empty() { + match serial.write(pending) { + Ok(0) | Err(_) => break, + Ok(written) => pending = &pending[written..], + } + usb_dev.poll(&mut [&mut serial]); + } + } + } +} diff --git a/tools/__pycache__/wiredsensor.cpython-314.pyc b/tools/__pycache__/wiredsensor.cpython-314.pyc new file mode 100644 index 0000000..f0d6a68 Binary files /dev/null and b/tools/__pycache__/wiredsensor.cpython-314.pyc differ diff --git a/tools/wiredsensor.py b/tools/wiredsensor.py new file mode 100755 index 0000000..856b3ae --- /dev/null +++ b/tools/wiredsensor.py @@ -0,0 +1,532 @@ +#!/usr/bin/env python3 +"""RS485 master and end-to-end test suite for the wiredsensor protocol. + +This is deliberately an **independent** implementation of the wire format. The +framing and CRC below were written from the specification in README.md 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. + +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 + +MAX_PAYLOAD = 64 +HEADER_LEN = 3 +CRC_LEN = 2 +MIN_FRAME = HEADER_LEN + CRC_LEN +MAX_FRAME = HEADER_LEN + MAX_PAYLOAD + CRC_LEN + +CMD_READ_MEASUREMENT = 0x01 +CMD_READ_INFO = 0x02 +CMD_READ_STATUS = 0x03 +CMD_PING = 0x04 +CMD_SET_ADDRESS = 0x10 + +ERROR_FLAG = 0x80 + +ERR_NAMES = { + 0x01: "ILLEGAL_COMMAND", + 0x02: "ILLEGAL_LENGTH", + 0x03: "SENSOR_UNAVAILABLE", + 0x04: "SENSOR_FAULT", + 0x05: "UNSUPPORTED", +} + +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"), +] + + +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, cmd: int, payload: bytes = b"") -> bytes: + if len(payload) > MAX_PAYLOAD: + raise ValueError(f"payload of {len(payload)} exceeds {MAX_PAYLOAD}") + body = bytes([addr, cmd, len(payload)]) + payload + return body + struct.pack(" bool: + return bool(self.cmd & ERROR_FLAG) + + @property + def error_name(self) -> str | None: + if not self.is_error or len(self.payload) != 1: + return None + return ERR_NAMES.get(self.payload[0], f"unknown({self.payload[0]:#04x})") + + +class ProtocolError(Exception): + pass + + +def parse_frame(raw: bytes) -> Frame: + 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") + declared = raw[2] + if HEADER_LEN + declared + CRC_LEN != len(raw): + raise ProtocolError( + f"LEN field {declared} disagrees with {len(raw)} received bytes " + f"({raw.hex(' ')})" + ) + body, tail = raw[:-CRC_LEN], raw[-CRC_LEN:] + expected = struct.unpack(" Measurement: + if len(p) != 10: + raise ProtocolError(f"measurement payload is {len(p)} bytes, expected 10") + t, rh, age = struct.unpack(" 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, p: bytes) -> Info: + if len(p) != 16: + raise ProtocolError(f"info payload is {len(p)} bytes, expected 16") + maj, mnr, pat, proto, dev, sens, up = struct.unpack(" 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, p: bytes) -> Status: + if len(p) != 11: + raise ProtocolError(f"status payload is {len(p)} bytes, expected 11") + return cls(*struct.unpack(" list[str]: + return [name for bit, name in FLAG_NAMES if self.flags & bit] + + def __str__(self) -> str: + return ( + f"flags {self.flags:#04x} [{' '.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 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 reply, using an idle gap to decide it is complete. + + Mirrors how the node itself delimits frames, so a reply that arrives 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 request(self, cmd: int, payload: bytes = b"", addr: int | None = None) -> Frame: + """Send a command and return the parsed reply, raising if none arrives.""" + raw = self.expect_raw(cmd, payload, addr) + if not raw: + raise ProtocolError(f"no reply to command {cmd:#04x}") + return parse_frame(raw) + + def expect_raw( + self, cmd: int, payload: bytes = b"", addr: int | None = None + ) -> bytes: + self.send_raw(build_frame(self.unit if addr is None else addr, cmd, payload)) + return self.read_raw() + + def measurement(self) -> Measurement: + return Measurement.decode(self._ok(CMD_READ_MEASUREMENT).payload) + + def info(self) -> Info: + return Info.decode(self._ok(CMD_READ_INFO).payload) + + def status(self) -> Status: + return Status.decode(self._ok(CMD_READ_STATUS).payload) + + def ping(self, payload: bytes) -> bytes: + return self._ok(CMD_PING, payload).payload + + def _ok(self, cmd: int, payload: bytes = b"") -> Frame: + reply = self.request(cmd, payload) + if reply.is_error: + raise ProtocolError(f"command {cmd:#04x} failed: {reply.error_name}") + if reply.cmd != cmd: + raise ProtocolError(f"reply cmd {reply.cmd:#04x} != request {cmd:#04x}") + if reply.addr != self.unit: + raise ProtocolError(f"reply from {reply.addr:#04x} != {self.unit:#04x}") + return reply + + +# ------------------------------------------------------------------- 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 run_tests(bus: Bus) -> int: + """Exercise the parts of the protocol only a real bus can verify. + + Framing and CRC arithmetic 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 + reply intact. + """ + r = Results() + + print("\nidentity and readings") + + def t_info(): + info = bus.info() + assert info.proto_version == 1, f"protocol version {info.proto_version} != 1" + assert info.sensor_serial != 0, "sensor serial is zero — SHT31 not identified" + return str(info) + + r.check("READ_INFO", 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", 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_STATUS", t_status) + + print("\nround trip integrity") + + def t_ping_binary(): + # Bytes that would need escaping in any delimiter-based protocol. + probe = bytes([0x00, 0xFF, 0x0A, 0x0D, 0x3A, 0x7E, 0x55, 0xAA]) + echo = bus.ping(probe) + assert echo == probe, f"echo {echo.hex(' ')} != sent {probe.hex(' ')}" + return f"{len(probe)} bytes binary-transparent" + + r.check("PING echoes arbitrary bytes", t_ping_binary) + + def t_ping_long(): + # 59 payload bytes -> a 64-byte frame, the largest the bridge can send as + # a single USB packet and therefore as a single transmission. + probe = bytes(range(59)) + echo = bus.ping(probe) + assert echo == probe, f"echo of {len(echo)} bytes != sent {len(probe)}" + return "59-byte payload, 64-byte frame" + + r.check("PING echoes a maximum single-packet frame", t_ping_long) + + 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.ping(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(CMD_READ_MEASUREMENT, 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(CMD_READ_MEASUREMENT, 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, CMD_READ_MEASUREMENT)) + 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(): + frame = build_frame(bus.unit, CMD_READ_MEASUREMENT)[:-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) + + def t_length_lie(): + frame = bytearray(build_frame(bus.unit, CMD_PING, b"ab")) + frame[2] = 5 # claim more payload than is present + bus.send_raw(bytes(frame)) + raw = bus.read_raw() + assert not raw, f"answered a frame with a bad LEN: {raw.hex(' ')}" + return "ignored a frame whose LEN lies" + + r.check("inconsistent LEN is ignored", t_length_lie) + + print("\nerror replies") + + def t_unknown_cmd(): + reply = bus.request(0x7E) + assert reply.cmd == 0x7E | ERROR_FLAG, f"cmd {reply.cmd:#04x}" + assert reply.error_name == "ILLEGAL_COMMAND", reply.error_name + return "0x7E -> ILLEGAL_COMMAND" + + r.check("unknown command is refused", t_unknown_cmd) + + def t_bad_length(): + reply = bus.request(CMD_READ_MEASUREMENT, b"\xaa") + assert reply.error_name == "ILLEGAL_LENGTH", reply.error_name + return "payload on READ_MEASUREMENT -> ILLEGAL_LENGTH" + + r.check("unexpected payload is refused", t_bad_length) + + def t_set_address(): + reply = bus.request(CMD_SET_ADDRESS, b"\x22") + assert reply.error_name == "UNSUPPORTED", reply.error_name + return "SET_ADDRESS -> UNSUPPORTED, not silently ignored" + + r.check("reserved command reports UNSUPPORTED", t_set_address) + + print("\ndiagnostic counters") + + def t_crc_counter(): + before = bus.status().crc_errors + bad = bytearray(build_frame(bus.unit, CMD_READ_MEASUREMENT)) + 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 + + +# --------------------------------------------------------------------- CLI --- + + +def main() -> int: + 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") + 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("test", help="run the end-to-end test suite") + + p_ping = sub.add_parser("ping", help="echo test") + p_ping.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() + + 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()) + elif args.cmd == "info": + print(bus.info()) + elif args.cmd == "status": + print(bus.status()) + elif args.cmd == "ping": + probe = bytes(i & 0xFF for i in range(args.bytes)) + echo = bus.ping(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()}") + except ProtocolError 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 ProtocolError as exc: + return f"protocol error: {exc}" + finally: + bus.close() + + return 0 + + +if __name__ == "__main__": + sys.exit(main())