From ddfda14085f3afe340e62961a96cb55cc345ffbd Mon Sep 17 00:00:00 2001 From: Oliver Walter Date: Tue, 28 Jul 2026 20:31:30 +0200 Subject: [PATCH] Add USB-RS485 bridge and PC-side test suite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Makes end-to-end verification of the bus possible with a spare RP2040 and no dedicated RS485 dongle. bridge/ turns a second Pico into a transparent USB-to-RS485 bridge. It is deliberately protocol-agnostic: bytes from USB go out on the pair, bytes from the pair go back up USB, and it knows nothing about frames, addresses or CRCs. A protocol-aware bridge could have a bug that happens to agree with the node's own, which would defeat the point of using it as a test instrument. It also needs no register-level code — it never has to delimit a frame, and the one thing it does need, knowing that the final stop bit has cleared the shifter before releasing DE, rp2040-hal exposes as uart_is_busy(). tools/wiredsensor.py is an independent implementation of the wire format, written from the specification in README.md rather than sharing code with wiredsensor-core. Shared code would let a framing or CRC bug cancel out and every test pass regardless. Cross-checked: it reproduces the CRC-16/MODBUS catalogue vector and builds byte-identical frames to the Rust side for all five documented examples. Its `test` subcommand covers what host tests structurally cannot, all of it timing-dependent: that malformed and mis-addressed frames leave the node silent rather than colliding with whoever was actually addressed, that the driver-enable turnaround leaves replies intact, and that no state leaks between back-to-back frames. Uses one wiring diagram for both boards, since the bridge mirrors the node's GPIO0/1/2 assignment. Frames must fit one 64-byte USB packet, as each host write becomes exactly one DE-bracketed transmission; every real command is at most 21 bytes. Co-Authored-By: Claude Opus 5 (1M context) --- Cargo.lock | 21 + Cargo.toml | 2 +- bridge/Cargo.toml | 26 + bridge/build.rs | 18 + bridge/memory.x | 21 + bridge/src/main.rs | 177 ++++++ tools/__pycache__/wiredsensor.cpython-314.pyc | Bin 0 -> 35380 bytes tools/wiredsensor.py | 532 ++++++++++++++++++ 8 files changed, 796 insertions(+), 1 deletion(-) create mode 100644 bridge/Cargo.toml create mode 100644 bridge/build.rs create mode 100644 bridge/memory.x create mode 100644 bridge/src/main.rs create mode 100644 tools/__pycache__/wiredsensor.cpython-314.pyc create mode 100755 tools/wiredsensor.py 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 0000000000000000000000000000000000000000..f0d6a68974785d57803a5718588368cc73da059f GIT binary patch literal 35380 zcmdVD3v^q@c`mvEYycY|NPsUd8W!L74QH|{N!x;1^{bIhD3jojqc?(GW-D2v`vZ{3r2-Q%^+y^A`s ztJLXTx!*tgfei|jToI&wJ1R&;S4P&p(3=S(#1_?xTMhI{Pop9QSj2(Hu+R zIMdxGGslf{UM|A*a9-0vQ>!^*E~T*fpe15qPd>u4r!`__Pg}%>r{$o%Rfq_sT(Oed z5pncz5m%3?=iZn(k{QYB*&A{9SR$Stb0qt;=X7?J);6Az9K^VdF^kf3drXl$l*~sB z%X;`oL60Z0yk|U8*kkfqd(2*2kHu^6;k`nS)$6FWMv7`UZ~0MEq}b~$;Jg_H+^{E7 z;$7`^;VqNihE0(br-gj3jPqt8&b=AmO>VRD;36wong?mwr%N|ml<0`)%|T2qtFKHg zpNE+I`^v9UQtj12k3a{<@><|2+PZ7#?L` z(LWX%*1sXlqC%sAV5CY6j#8E4C~6D`28SXv+P>kZkVh$bW{k2+qho_Ztl06+eFQ%R zu4kk`{~(|jkVek=(2t5QC^5BPsH_st4F&p!05zuu$ZFqUa5NkV^s#R^I|k^Fpxv-| z<^oUw4U2F!wT5x430)AYt3#twM6B)iKT{itTxi_iRws`5ec>_5|2Ybb1_O~%>Cp=i zZaOPXa)vL65n{j(x8vQ>?~C~QhB4C7^`Q%4zZCEd6J|>Or^g6)8^ll`1ki;ezTsg} z%OSfD9y!u{pz%oW{-bSejqOcxmKJfOx&1&_tDMu>+}?TA+k3pdvAgl$k;Z*Tnq`+7 z*V1_WNSB;(yuI^yN5@fbS94PzO;&Znay-HyPo}#f0^?ja*JIxP+nD-@E8&irhD=_Q z*X*@;`C3fuO6$i&=wzP0L$su+ z0cJ+{)J8@}hr`AgKumV^_Nt@i@9mXxPzSgT7X~Rp_wI3allu;Nlhy@##LC&*U8HjC zyC^_T{Vwa>DZGFsnL?&+&cn5!vXJ>iPE{dgQe4P%N*fKXN6-io@@_z0OSh@VRHdhL z+&Z;Xe>RN}OLb^9fnu4fT*drUT)zw!4H!!Q~&bPqP?8D><0)67|oo}eY0dc!m_YcW?Od;lj~kygJ|qTyiUd=Pg$(i$fZp3-0&(*k_Qh3|JYYDG-BRSN>JLq$ML56JLFCFEzo1i z!XDFp_L%jvH02Y}W!>2kVxRE;A~4P+ai`m?;m#>l9N@U^FJUf~5~iH8YBhM%mT>_` zljU68Mx8YdnNCrg%$+dcYO=i=i`H_t*(e49do8DE$eHU<#Wd}mo)BQnr)cDu>(=Yh zgal)43{NxEF(>vIb%_4+eSYv<@iX(z)=Il<8TJR|jJC#;y&a8@A356CBs;r(!(;wt zNg9e{s`qGp^K3k>)768H5> zYmtkA@)QE&+-hBy*hEL~Ex zAyz=FBPG`Ebw&hl28P@vTU*#S6W$vJis3wsR?7S$#AC=wgSiwruVo~M%_%l(XuL;< zA2fI59M*>#BW*s77Q!^ai7cC}8}5)7Tg@KHiH~&Vc=+g;F;cOZqulyXCkto; z-DBE*P7?;az?xIk7IPg2rl-ZFud9cvBId0+ulbR3Su!~z1jeKRWW-k%U-3b8JrB`eAOtZTIra&t7~s*7fSC zmrqTXeYe+2*%;aa3JUpjw%rbX%Z0vB;$}@id;~>z5zX!^Lu+C zJ3>hA^+h65;LI2#QAkn56_qX!&zFe*F|Qke@MG}429BTk{K0vL>-pCC%$(=jKeA?i zV#a(1OSfe$P-wx1&?ja>NEWFCm0GF7KK$uUoh_!j%GA<{Qk9qJ-Y zVGq>Hb^voHp3*Mzc9XY+y$%y6Z9GLRnz;@VKo>cPT22U9Dzi(aiXr|KaZ=`zVqM63 zqKkbUhOtKZk|1}KJWLO%GX-o=kv22$ZHH#IM8H^jqj<$pCy)9mF z0!Aog%aH#(A|ao$NZHv6eN3~r7fs7{R73%(9M#IUv7j!XGFn+_lUgW2==Wmp6@h~w;T4ibLEXd7L*sqGO$C1d$#7^L|BED{V{`^ao zbN1rfS;euDH}=n1{&uCM$NTe^6tpiwwMr2#WKN9NggGOYiH`EtVR;o7Npucw{{G0P4Q^>%H zPT=bvBuUE44gE_I+b2*`*QW4RS^-<@C&#enwboA+vo~c_RYHaA9H)zz{}C@KtjVMi z$sQ*oRSH{Bp+L(Z0}0mif;5|rY#}M_cjy8pYnOQ-oY2B(g!3HgmdZnXH`cD*P!m*T zBy)B{PnGh{8wiv2V`4SY0#*G-fD^UFEOc^rfwn-f!%@ol_cL_1F)}XMKORc6NEZu`4bfPvmsRUER0yi?5!3;WTDa zeImd5qI1GD5&n?o`{Xmf(-4!UTE6|o>HP_JU0hICu*5et-bsMPAe!zzY>jx$Vpt_FMQ<=?%iQ_f(j;}L&kXtZA%pw11Gw~xFEF&dOJq4*E<`Fj;zyPT=)4r#B#SSS*}a_cvz&TR4<==0O~ zxChM^Ih#Hvg)sj2{U0dVB)y@2M8=dm_7fU2s8R3l}pXi zUxhGDDoeZWE}Mb@bwR)zFVWh&V6tP4<3|!C{^i@&fuZY zS<8s;tZW$xO0p#c)@x;8mn{G{X@7!Y**q2sla7KBU`%mEjmV~roWTZPA-%tcJmF4w z_mf@{i=O`8)|rZULt9*IPvjhpyN=$j+5G0ncSmLqy>Cj?9J(QFn6ONi$NHvoi2m-6 z^-b=XYM$W}?#JT7V@Z1VxoNKypih0!TJt@Cctybq9@`&2vk!&34%u#I& zS^g^mdURat9 zX~O+!sL{UB4YU)YmVtAkY^s6ETG<*FpTQynTfzqFAV#g_b@g@UDItYUkfK{uNCmAf z!q8DjXiN%|7a?zqyl2QGQfp8(F$yia)XB~m2ZI>_5BjC7?8((v*IikMt(5hXC9z|% z$gAgGJ~#FBx1+J)>9XmjHxFDpFw^)aho*PVT)3WhtFiM&V`qGC7sT}ArWNxm%5JTw zy|JQp#u_i#inwiyqGkfu`S&v4&7A%6`)3xc=1i|?!NNNf#9*k^E(NL2gcl0EXX+tF zyU2S1f8ibQz&nJJ1Ya_ryDY&kOYyV7Q&djUpmUOjdoB$bm3pbR^nBzSNHOq{Ym4{@ z>C+W0I)xODLLr5tP)Ok@6nX$hIStVrI(BBsp=;iSq&-K(;mwLTy@V+0C>b{@R(Z0-W1|tl(%T=G2%f@F2QLsetk|u9OLsW1Ijp_WBi1~|7Z?r zJz;C`5P7~nYTvo9m7Ky*%t?k4rf*%VY!@g@B2yhXg-!hLNDh-y@_sh4`R&fx?C*8Y zw8nQHjf)+LoMUmr&ITCTRe&^A>$eJqhzd*e41#vz%~`h{)SL>w zCgr53Q7G`&dzaiRI&67J?OF_y2d*lK&jw$+w0 zfeB^M`okAe4`=2CsUP4^XXZ9y=J1G_<8eQjvXsUhcmEp#OEYt?39J(a933=FiyT15 zFN(sewxHi(N(CEiyLd3vrP#0y?bw%O!=!Q5HND9;QJ(`Tu4IfBLW&b9^p%Sow#KtC z^E0A0Mw&%wD>8sBKPj~$Ae*YAS*O9khazk%1VlkPNEu+ggFPM*gnJj*IVFQE?~2+L zJtW1!Sc%0|3Jld`{w$1~q?b`x`Z{@+$RmCweUrRzk(ZX;XX!MTF?DSfLnQPFs-qBUO8dc8Bg?jYFtAyd(O(aKvzRX2*N zrn};W8xgmuV7{RE>X)y4IlgS&)P&`^Z})Ffb-W-ZlJ%Nx2Du zAtUSgL#m|U$$kFld_hryFPhKEPw@FrFg|~j2@PO4s?dPYf|Knq5mll{w(Znnx@Xg- zf+9*#JAL@mor>6|yF{W5qQv{^UQ?0R4AyU>~3{FN``7&h+LZ3=>X-v%D+3 z1$cwWCM)mu7DhZ?ClGn1Y}q#!j_z*M3>{&6NY;#D{~+n8s-XDx^$kJqz1Wb>Xhm0( zLYPz%Zo0dg4at&LHq6GR)`CuG7B*xpC!IwpnID6xks<@xv6?vT9CeME?mfkT|sXw0a;Y z17U{&OJdVD6(PF0AQ5QhPKZIB{+*}`>a8SqG}6%^GLF%xVd)Y(|vN0Af!&^X~DbnI%<^u5FYjyP)4?fhyDJL zoFn&qKlGwZAOfY8O97Db#(+b6tmE7# zHZHH|>aG`dP328JGxPLZ?qd`DH-!1T<==437p(k->$c#!?EJcOa`T+9d>&!pqHs$n zzaf-=AgoF9CXMAkGGn}E5Nc%pq2>|)8~QqjKb=dOtc4`T6i!CiBiR#VVovRe<8A|9 zm$HpW#~TeWgq*30BhnkF1txRDG#M6?eo~Fg4jNyU&id(Lt*QfdIqPE92g34XC$-_? zkqhYgxwfX+W|TqU^fu~v(A1P@>~7&t7yO+xcJN}&3-p-kXKzaH^ofH4>NL!}y9)hX zY6vw^p}&T0RqB%sx&u>(s{TP;x=T(huyeD)tVGbW=mD=~o7x{X3DbxX9l8aZUPw1h zU$RV^iHltZ6(&oTq80KYXFwG)RJHsOZlte;@G;la(&G3hM&wr=wO;>^dnjA&N|>1_Hx=7^^5$ z2kcryl$%e_p4$n<2e;?gx`!_i{5lK zq)?$bVU1yuO@`-$V#9mdNvERqLwO{#L}fhErYyin$dRb8ur2b16thzo#Sua+Wf(%U z@8|f_d10Ql=ne;O%DO<0nKkTjeG7|JCZRrwP)?tOU&T~-I4{&tGIgImWJwyKYB@qY zOrxz2imZVR(J!Pmg_IjuS_of$d3xVjnBZchI_hkz1n1G?TmArcG4sS^7}< zK>LP}_3#BOKzY4K2;gh1{&GgT^ggBAZAzYa>9$rG2vN6r`ww7`1K5;sZOaX#nP&Zx zDpef*5&FWVvlr3(2LmYdFWKL1te>RyWzK6cSaaw`!yN@$dh(e(ZNrG}axJB!NDC#O zUTc!*rmG)>&KCXi{iDNVr7rp)q=tqsRDs+mM`OeQPLYV1A9NA~8k7JIw!m^7`>dd7 z*uRPRLFn27WRe*chj4y|$XJk$zXT#=yveB52*N6Ts-m+fYAeX=Lz9vOYY?Jt&afY} z6=v`_T_W}k!9I{3_rX~a9HM~m91i#1jkA65->=>IY1^5OJw*Jx zD)Ey@+^o!!{sa%%F*1r1960~cCp(CuqgCGpMGSBTg1zT_fe1cWw)xNF+)uyEqq%ubTJjS_kXebZpN##W+87&} zDmxj9o{Oax(b|-Snrv`84R!<1GvkERfHDQ@|=T7jq zvvMv6zaC6vt(;mumsL5xtmx{{m7!b9DsL>SoH+20?qy^J;=JOFZGLt8%iCYw_42Oi zf<$rsoM+>0EI^-o;kiV?x)=C=%+8DD{j{+1js7=--wh^ec3;oCS=0P}?p)20c~Aa7 zd-AS2UvS2@eBddAiM2x1U)|5h`YUUuEWf}0)tZ-UrgIZT)$zibcuwuxeeaxo>+DTe z<861|Wap)5TqtCuusvmULn4BO#H9-Mp$}f&cHmhD;YxVQ28>8MFJf%d0zIbl*<-+VhgYMSJ2K^_2^(`E&uSTc;10zm{lI zCy-Y3(VQHi<8}0>oaSNkk(n$}44c}LFTf^uU;q;pf`~c+Yod8Lhr;sET-T~%S{c}Q zIu&aQLRDshC6Rgn8z>C@88B8LLa}RaK}a>40In;DZl|+RI8G&JkYugCuu4taF@n~QUwAu#@aZ4 zUX0=yMhOqbDCwv4VGNZXhDFp7RWFHCDb`Zj>@9`z1lu)umaSEun|HJ4M~a?r zNJJ%oB96P~+0b-x9oh%tU2_K0Z_5CB#~8U09o39&rY_5g91Z1?Y!=Ll+F9f@=Bh1~ zcj3f<1MuD$OCIoJGJ+Vh$UhY9v5Bv~iBi%b6E5UwUdf^4%{@ zt^T9RKd8K(cRid~)-@*_2W=y(>`Pyc3oGx3Gc6-9Ko3{g8kOzR36y%MaAzYBCe`d@ z^=q;oGR2-$p2@zTGUn7-@3=dF*A$6Po#kpi_Q0U)b2u5FTsCRoNhf`2X&Pv($_G)i zvZZ(QDLHR{TT?F`OzmxJZtOhnZEkCB?^4KPEixH)Gvy3}@s#W)AQ>Xq5R{_?EE?TK zx7=kn++~1I#cLIF8}=kt?Y*_C^~S2!g!^DzP!0D8Jv9{l^k903+RoCuiz_IoRHEG(ALkvZRm&0tJQGr-~>{Fvz|9#fx8582IaL ziQ`k)=^*?ZOun$VY zz^%iev#YV|cxQSjU^z_tkFt~CpoG~sYmk6tEITT4^5v7Sk0w@Zy|rTZjTO5S?mcl~ z&;5|#F{X@6DaDr*aLnvXF0Zg5#zxsX^pRMOn?OQxzjAtO6G37Vse854m^`*6p}`p>Gqj4dSp?MU7&&PIM{xGjj*Obpsz|%An+lQw;8@u zhu_nO)kev|rE{O@ zmaps3I=OxfYs!HFQl7JoLX@N7S>#erqaqIpOp9E%a#RclUs7tOo}|gmTH}>gjVE1^ z+pw(GngF8FU@vma*Z@PH(w;Sa4Xiz~5=v7KrKR=(!m&Lnq=>TE9~^+CKb_H!;+h0V z#3GI$i%}OHIZ(w?F`A*Iu=^jRL+CFSdN=VPgP8BnvUR9z`Emxce(mv@NW5z2?2-3#-|u?g85cVfIbCsA z7iDg^*bvKmweaP_xr|j*V8gWk&9m3e&gR9dchB};-~8TV?>_c^SA3s0S<0O^;bQ0v ziRumQd)e9;Gn^#&mP+#=9&jJ+8T?}2pC0(9=u3N$M(U0BYYJ)Cf3y$B z+h&kgBK5l>bdsp4j5^7Z(*H%?&&m5QIA{+M3cVf$etEjb;jQ6}Df8!iNP*^}&{vZ=*-hea_>jAaGb#mbnkuASh&m67b4 z)*z2upt7HD-zA1ZY==}?+PzJ$50Za=fedSC#6TI*C)x#VhWtZ*teiGLn@zO!LQkcK z=#lPpx=+?L&s0)sbiN#ND+BnkD*x6b9 zWZektIH`vsyWU3T)b#q17F?U6q(?K@Juf12?Kt)IOXO#YJlWBV+YVqcJ{pw%9g#Te z6{6dN7&d4s4MGoCqX%W2_FzZDEVv>;Q)*~dN>=RW_dO+NYPic@8It@qier&$wmF(j zALbXl*fr5IkL{4dUq2kHzv(LZ=3yw$;&~gUo2IwMb2k3{)>+%^GqXcCwjPL|c>HG0 z6Cb#q`~?msf5Mqjo)x@!_?Bzs4cE#k{`J1;^6w1AOKay`b;(+0N@n|JkH>TN-*Pqm z)YbG$*omU(XW`}O$Zs{88*SXXwv2tbmUpwa?aQ+KMV1v|kZ1aSHZtwta69Q`7kRtM zV+VE_O{NuLS~B~ul#e_jUW}G}mHX5T`f*Q4P7j4_Qmz#8#v|hM&RSBch8YD`W-Sxq z9yDun&`2%yO`ElSW>e-Jj{Hez_vjge%uLTDT|pV<)LCo2dlr550iMY-P@0|u2Vax; zowDn=3F@b!^h2j=eq2I<{E)uV=C?yh zsMl}krG7JxS_@rRA%v_;)Z*;6CTn56B;#8FTS4htbzw_KZ8-S%XD zLoeM)I>l-s3mpLm64JMt-7O?*Vfo35M+iBV$e-KoSd^bEeT0y6iTwH9&PDmj8b}Cb zERnyUJ7ZCPvKJCUt|jspcDs`K4Wm%BXcUTBEtyNyQqrB7tc8uj3YI@>iTo?Ovliv2 zOXP%*dx`vFw|+rh3SBQ-)XP<@7S9s3ly`d!t(wwtno|yK3ZjlQ>jHj3y~q(JA{P< z9@YBAd<}F(9%|P12k90+!^R+)K%n>9c4L*YIf$-63v+?KJoOeoC(I`Xa3LRUMvl-O za>@;MVO-yboAc;eL28Wl9s6l}QL6R_eb_q`>4q~O?y*CTVZV>|9nm3m9f*x1m}kI( zVwn@$oWV#SLS3SL6JI|*CETYWB|V9IVQpi&OY4=Lrg5&*lun7F^9kA!uu!hdrs)gwKg>;IT9*$PhgIJ}rx(xY~_f93RfdHQIs%)j>69$Tz zqhK4gY?oL+MhhC{RD=jaDfN=Os9zdoVpnHt*QRM4Mp{ zG*~RtYSJi(|BqLcOEQz-Ha(hTL0X6RBB z*O|F)*+J0}%wdPujRD~|$cFBwUvf%TYqhj}~H2K@{|J%;6(`#k*X_!hnm^8Uim)%&Gfs*Sh4`80-;|eqt zb% zBi&NWLbVRHSYXC@RG7HjE6n?Nb}Y(cjf<@!$uJPJOi$L1antz|W*aZ8>0 zm{Ra5&1~r5)0J1|1}LF9S>CYiZNwpq6h-UN{N`?4{Gsl6H8FAo+uCTZ@!Cv64MakB z^+0^8cE^S7@$&5n`0lbrz{_C3Q#L80`A&sb;SMOe zPc-5m1QWlI0!~__FmV1k0tC+FA_cb}TaW6Ck`UKvPG=#TWC+TpFR0MUDlHD9Dq_+;b1ZA zr7trc_Kjz83Cxsrs{fCMUK^SYe`jQ>mvp>83Vc8CzU9w?@m+24mF)?ked5{MxU9wh zA`X|vS8R>Dx5b5RY`q#SrmYr{dJS2|C*e4iguzipDm|XOf?lQKNm4}Z*`W!y+LK{T z8u%umr(=z%VU0oD(6J9NHXGvgiOfV{^oSfTsbzNdAJbXYsKBs*uBUXXfY&i$<(9{) z$%;!mCs-wJ-K<0~i!cqy`KU&Lv`C>cf=LQgGgPS3Jx21%rD00xU!c+lP!AvqEBowa zFZpBpUJN8WVtTM1of6*2o$mVM!s(2eJwMrWeeA~#*Q??yk0ykp6VHC=u2sNSnF2nX z9628t85nfJtj2HtC7Kr2?}jPXis+A*ZkgF=*hhMfQi?obhJc z8ht8G`r0l|m|d(( zwAxRy{!_pTh7hUqzmk_t0f|iFrE>C^NI=3p^oFXCnuRkO07P$Ols)CLw1{HNa+YxA zDO951N>0I~^t*)Daf2V%3Sfiiu{}23_5LWqYjo#XS>E^SK4TyDRQG zF?DGA>FK_i92{qCeXDi0^LuT$U1|08vG*>#dm(&MTCU@Pjb2-I# z1kSTv!KZxc23^ffPkkh!tDIr%dExeOSd`OM;o2McqIPF>3V!KKh0bCTet~r$wAkT} zX{#~E&3!goFnAH#AR~D z^5R#8mxZaOx#bmC`00L-nyl>0!HdDU74@^$37CEEhN)ccGFku?ZJ*7bjm)~P@4Fs; z@7%lR-VguyOV@+(<+Ki(I5(fYetOGH-gK3bd6Y47$?o0vVF8U#P`(prp^iZcRuCIt zRA(Fkh81boyh+1>B-Fi}4s)zt)}cH(<%bE>b#FS_pi?Kn+8_}sY(lEXNJW=kd(6;o0`16`@Fd9czw>8ooeb=36ajAMj+B{PvI*Yrus;1oU3a}6g4M2 zEg%j;=ERxHj>)c@Lg6I`u6)1jy6Af8cx=myr{;uJgnYvnhc5+Hq&pNZ+A+KTM+d%t z;QIbQJ2JaFzMPhvgmlV%$S$17no&&jDq}W z=6@1f7?z~LtcPNWzB{RRg4UK8NpfkZinTpjV!#j_=L+IiKQz#%L5@W^qz2TDh0I!1 zuP}=wUg#(x?I2L5AOR?fUegj3MNglk(q1&c5TQ7>|JBx)Tj$m`Bs_rUtT*o2OQ^tr zKY5&vP%Q#{bG!&<)$g2p>)dSkdtaIXk7!N^%?7|1=}oEpatQ<=kTG(zfIun&JUY4Y z>bH(+UP1yc$Qa2CFB1m!pyE_g^0qL$CQ-CE;b{Z{<>pOTCwv!$ zU!RPqYnJ`bVAE``r0HqMYEicn;>72p%lg-g`4R*_PB~(>2>Tq&m|FO30>85 zNy<}Ypfiu6dsh!(pa%w^|GrImm}lqd5uMFgeqjTY3M z-_|Tv?-sS+hEabk2Exisq$?(8q?l(GXxGfr9SYy4$%Mi}y#5%D|2&+iJQ#uCsceJDk5Q00n=Pb-ORd#r-4CA z8~;sn?oGJ&X6E>72U2}AE)V`TDw>Iv(v2kBQ*cW%#edumfAhH_EQNKjDV(Zqbe!9aWZ2Y=t zp0huo|9n;-b}^w6NhO*30w9V`u^D4|(6(ecAq@&6UEi0~*g-i&gTKfn@7>Vmolc7B z{F?m{7Y&wl3xHF)fS*)BX|+Qfv~*-I>I`1JFuK`>FDJ+Hs9_~W@wM5V_S?-TL>S!} zvNN)do+w+Y-Rv&fZZp(bR+tc|`9RR2d|)arq--SOF0jSM;3npNyhoZw)~F|WfgFBk zj#Sz_;cqn4a98n$`e=qBn<7p+lzoeDp^`^fbwH(MJL8Aag3h4&`emzc0CkDr8LTyKJA_{7v^Ids39x^x_k9f~da7x+Yw-K2c7pzUaE;S$D&;j!mAa>~A~g z+^ZH~2HKLsCYE+#a&!%y zie$?-O_UAxN$l4FVJ?xp>CfbpO|{rPR7_Q)PAA!_VplTU@R=Fbexw|NQ2(z4fFF=& zLxf^cDAMb{!>i0=LoA!{14)!hmfcbTd8^5*Ag_kJP2>@lO1sG0OWuC+4v=@4Jespo zH$2%I!4J<3OEEMcTh0w#28VLY5jChsB{|j$cn%mnE~d#kQk7 zt7z36UruGV8_HBp@0#Nu`;@m?>lQqEP1|OR=lJGNVS>GVAxDobpX!<8H?yj0=lD8m zXuIA}+*LbMGRN=!l($%G7o2*bbAA2zV zmLk=W0M0t{5<=Lo#H-@cQRI}{Bn_gwRz{=2sh8bC#jJidlLONz4wN55#O(;^2{|%` zOsBfSZ`auiV4CF>ypC)y1+PrU0lhiiTyLJ1 zi`X=PI(lVf#0w4mZ#LqbwD4n*l9Qt;)^-s9T@3;54W4|fJx3`hghf{7d!1x zb;w%Gb@3hI;&k*chi#aZko<6}Tubk=cC69D$*0!Kq>R&SYI}T%dh?g4cU`hREtIz2 z>LuzeSfbuat+wPdZM~)?>MdsVo@V2)L94^Nu7EpLt-Yb-dM%uMYNO%NFM`ryLUxQ} zZ8E18>e7GjCnO}t{PC}`9$>cPXhcO@uD7yF=&0AqB%j{(TK!Bps8W$quI;5IzO0nh zdxGjM#=m5Xr#5P}c4$MTMc|3sV>TpP@>V?N7yrYd<5;#&I@y>6WBs^{@O62_6%^) z+CB1DY1-YZHj68t`myg!zn;C~;bZe#_-#I8k41G9s@b=ili&KVQBF4&4=bY|x~XC2 zH2D45jDD!(V0%Q{?opfaBg|2k@mH9!Nud1vvuYqpi&mWZviJK3d^px7^F#jO5EZ4p zPbQCO<3h}oDUXS=8qJoV`qkAqSW21H&am5F@V1PE4VstQrHDI+eIsZ3eY@plBfdZ| zX+WyQ-=+d&F)_})o;!7H+Vr||8%@;3E-KPC11-3t@dB(7&mg-*J&l^HpNTrEt6>#O zw`R)t@kgwwNKNmLdiuvkLd@n`HAJR!Zm=GvGeuOEsp+~9$zuc~GOqbiei>S}s}_XO zRb;jZJL;qXB-tA4#*U-dtjEW)x_*wnN1-4?e*H(bDYH7{!Lbqi9D10o5u+~C0hdo{DJ2}iRuZIP<3vg2AM z`bq6*Hh$fB7_Go+Iyg2u79PG3byioC#hUbtZ+H+N0JpFwnhz_DQTpBUGycdqKYn7J z%7vpY)v_Evgcwv-Tgp{IN;xJja+^qYCCax7Q)!&*;#wdKJ)MnxNQ}4U&U52d5I8K@ znf(YX>bF?%GxTPu7@67?LDnprYwDzGL?|cH)f;LYI|u!w7L#_%1&a_#OZfk&3=yM` z(Sdm3?i*FRrk;-VzgqEo6^W`{|HjVw{#Jg0DQZ`2m!66`+YWZNHFoW9ZBXsmgf*=V zYuXytbV^U4vZzP5y=H3g2C=eC7JT^W4f>Ta>`6<5V|00J2fa$tCukaI2u(TrJ;|`_ z0`m4EJIf?z(OY=z3@a?*76W=$UX-hnPO|bWhDxa?slJ0)Wvl8BDMV&+vb&SndnXUd z5xG{DzGQM_IETx2CP3hfHg$hORS38Xae!_;m6|B^Pg$IX5o~qJ&icyPhy54MjN%Z% zLBbZ=XvI%!ntdT$P&jr5hew&=9qjY)3!6dwbToCFwB3@GJl$}I{YHqsCRP^1Jb_Vq z9{Jb;B^%5iJExB9TMqH34)Ki>x2pHtsNR#PZoFQ0v$|={(LBCyp0|DN=vR(TwoV0M zMm~OYj_**T<^|7mfJ6P>3e^YG3^l-eflS*{WJ6*S&#W$SA zQ~5J_ac6PDxpjR1N4))O?O$o1d@NpmV7z^fZ=F{ow#LPM6tQ26*dAY%8i5>pD97H9 zGCh}jF80J)ri&+f5}8}@b8Nix7GH3KFNitbu*@7IgF(Jvj&J`6M#=3L+hdQ#H|(Bh zPq_AsH_!8qTl}&c_}vE~@0Bgnuhb@T%0F!?;;;auh_{D9Mf z*^=F8g89`eU9X=0z0=bTxVj@=vTa71E&oyN_iGakhvSbOi5Ioa<+YP(Up~9-Y%^V! z_JOBTtBZ9lwfvmDylBs@yuCN_ z_EHIC`{gH2&b5N7&V@ngf}l43t&C6XoU4%HE9SGyrpl-CHr=6u;3z z;Vn5w?H2keUYhK^y_I$lds4>B4%Gtu=wZYV`&^5uMG=IOYEFUP^J`cPP7Hj()UVx{LS9 z^(x86j{uL+AqQM?Kx!C?{qXO z|9KerZbfNXVD?J*T}&<1cv32(Qtjj&CGRkKZRGs}dHCvuY8pHP(!b@h>(7l zJi38JFV2GEvB=@1r^%Zj?=|xNl{`{4GCHdqAqMv~{0@yQ>{NamdAIaC z$cO)2_$BzjWs~U+pJn1d6**J+&pG$cxXhn%&Yy7^|G<@j2AOvJjL*hY%UL#_g$wax zo8w&0xG+)iQI=;sV?Hx$+&S;e7`M;6mcu&XRWt7RsGvB`<>6M1 z@l4QwaW_cTxWGamWoC~%iFiX1T0GASac2?pSHhpDA+LHN!$wD?#&qsEaDT_d;fui7p@@|I?LpfcaA#(J^H% zJY(W`*F<#AQb-J?^|x9tc3tZJfX{<+kT3eh!l;`Qa;b|u)Z}>PV;}Gh1dSb^g|h*j UDZa_V{lrq(RBHLjN)!D5AG(w>*8l(j literal 0 HcmV?d00001 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())