tooling(micromate): mm_client_check -- exercise the read client on real hardware
Read-only: POLL, SERIAL, state, monitor status, the setup walk, and optionally
one event download. Never writes, erases, or changes monitoring state.
Exists because everything in micromate/{framing,protocol,client}.py is
verified against captures taken over USB on one firmware line (11.0CB). Two
things that cannot be verified that way:
- The modem path. An RX55/RV55 buffers up to ~1 s before forwarding, so one
logical response arrives as many small reads. The client reads to frame
completion rather than using read_until_idle's idle-gap detection, which
should be strictly more robust for this -- but that needed proving.
- The 11.0BD firmware line, which reports flags=0x03, a shorter model string,
and a 0x1C block 4 bytes longer. All inference from one 2026-09-23 sweep
whose captures never landed in the repo. The tool says so loudly when it
meets one, and flags an impossible battery voltage as the signature of the
from-the-end offset bug.
Run it over both paths and diff; anything differing beyond timings is a
finding. It counts reads and bytes per transport, because a higher read count
for the same bytes IS the buffering, made visible.
Smoke-tested against a scripted TCP unit replaying captured response bytes in
37-byte dribbles: 289 reads to carry 10,958 B, every frame reassembled, a
4,076 B event downloaded across 4 chunks with the assembled length exact.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ru8Lg9HkkYvX9VWWo65SmL
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,209 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
mm_client_check.py — exercise the Micromate read client against a real unit.
|
||||
|
||||
**Read-only.** It sends POLL, SERIAL, state, monitor status, the setup walk and
|
||||
(optionally) one event download. It never writes, never erases, never starts or
|
||||
stops monitoring.
|
||||
|
||||
Why it exists
|
||||
-------------
|
||||
`micromate/{framing,protocol,client}.py` are verified against captures taken
|
||||
**over USB**, on **one firmware line** (`11.0CB`). Two things that cannot be
|
||||
verified that way:
|
||||
|
||||
* **the modem path.** An RX55/RV55 bridges serial to TCP transparently, but
|
||||
it buffers up to ~1 s before forwarding, so a single logical response can
|
||||
arrive as many small reads. The client reads to frame completion rather
|
||||
than using idle-gap detection, which should be strictly more robust — but
|
||||
"should be" is the point of this script.
|
||||
* **the other firmware line.** `11.0BD` reports `flags = 0x03`, a shorter
|
||||
model string, and a `SUB 0x1C` block 4 bytes longer. Everything about that
|
||||
is currently inference from one 2026-09-23 sweep whose captures never
|
||||
landed in the repo.
|
||||
|
||||
Run it over both paths and diff the two reports. Anything that differs beyond
|
||||
timings is a finding.
|
||||
|
||||
Usage
|
||||
-----
|
||||
# over the modem
|
||||
python3 bridges/mm_client_check.py 63.45.161.30:9034
|
||||
|
||||
# over USB / direct serial
|
||||
python3 bridges/mm_client_check.py /dev/ttyACM0 --baud 115200
|
||||
|
||||
# include one event download (still read-only)
|
||||
python3 bridges/mm_client_check.py <target> --download
|
||||
|
||||
⚠ These modems bridge ONE TCP session to serial at a time. If THOR holds the
|
||||
unit, this will connect and then see nothing — that is contention, not a fault.
|
||||
`bridges/mm_probe.py` explains that case; disconnect THOR first.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from micromate.client import MicromateClient, _content # noqa: E402
|
||||
from micromate.protocol import ProtocolError # noqa: E402
|
||||
from minimateplus.transport import SerialTransport, TcpTransport # noqa: E402
|
||||
|
||||
|
||||
class _Timed:
|
||||
"""Count bytes and time each read, so the two transports can be compared."""
|
||||
|
||||
def __init__(self, inner) -> None:
|
||||
self._inner = inner
|
||||
self.reads = 0
|
||||
self.bytes_in = 0
|
||||
self.bytes_out = 0
|
||||
|
||||
def connect(self):
|
||||
return self._inner.connect()
|
||||
|
||||
def disconnect(self):
|
||||
return self._inner.disconnect()
|
||||
|
||||
def is_connected(self):
|
||||
return self._inner.is_connected()
|
||||
|
||||
def write(self, data: bytes):
|
||||
self.bytes_out += len(data)
|
||||
return self._inner.write(data)
|
||||
|
||||
def read(self, n: int) -> bytes:
|
||||
chunk = self._inner.read(n)
|
||||
if chunk:
|
||||
self.reads += 1
|
||||
self.bytes_in += len(chunk)
|
||||
return chunk
|
||||
|
||||
|
||||
def step(label: str, fn):
|
||||
"""Run one read, report how long it took and what it returned."""
|
||||
t0 = time.monotonic()
|
||||
try:
|
||||
value = fn()
|
||||
except Exception as e:
|
||||
print(f" {label:.<26} FAILED {type(e).__name__}: {e}")
|
||||
return None
|
||||
ms = 1000 * (time.monotonic() - t0)
|
||||
shown = value if isinstance(value, str) else repr(value)
|
||||
if isinstance(value, list):
|
||||
shown = f"{len(value)} entries"
|
||||
print(f" {label:.<26} {ms:7.0f} ms {shown}")
|
||||
return value
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(
|
||||
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
|
||||
)
|
||||
ap.add_argument("target", help="host:port for TCP, or a serial device path")
|
||||
ap.add_argument("--baud", type=int, default=115200,
|
||||
help="serial only; a Micromate's modem port runs at 115200")
|
||||
ap.add_argument("--timeout", type=float, default=10.0)
|
||||
ap.add_argument("--download", action="store_true",
|
||||
help="also download the first stored event (read-only)")
|
||||
ap.add_argument("--lenient", action="store_true",
|
||||
help="do not raise on a bad checksum — for diagnosis only")
|
||||
a = ap.parse_args()
|
||||
|
||||
if ":" in a.target and not Path(a.target).exists():
|
||||
host, _, port = a.target.rpartition(":")
|
||||
inner = TcpTransport(host, int(port), connect_timeout=a.timeout)
|
||||
path = f"TCP {host}:{port}"
|
||||
else:
|
||||
inner = SerialTransport(a.target, baud=a.baud)
|
||||
path = f"serial {a.target} @ {a.baud}"
|
||||
|
||||
transport = _Timed(inner)
|
||||
mm = MicromateClient(transport, recv_timeout=a.timeout,
|
||||
strict_checksums=not a.lenient)
|
||||
|
||||
print(f"\n{path} (read-only: POLL, SERIAL, state, status, setups)\n")
|
||||
t0 = time.monotonic()
|
||||
try:
|
||||
mm.open()
|
||||
except OSError as e:
|
||||
print(f" connect.................... FAILED {e}")
|
||||
return 2
|
||||
print(f" {'connect':.<26} {1000*(time.monotonic()-t0):7.0f} ms")
|
||||
|
||||
try:
|
||||
info = step("connect() identity", mm.connect)
|
||||
if info:
|
||||
print(f" serial={info.serial} model={info.model} "
|
||||
f"fw={info.firmware_line} monitoring={info.monitoring}")
|
||||
print(f" active setup={info.active_setup!r}")
|
||||
if info.firmware_line == "thor":
|
||||
print(" *** 11.0BD unit — the FIRST one this code has met. ***")
|
||||
print(" *** Check the battery and clock below carefully: ***")
|
||||
print(" *** its 0x1C block is 4 bytes longer. ***")
|
||||
|
||||
state = step("get_state()", mm.get_state)
|
||||
if state:
|
||||
print(f" {state}")
|
||||
if state.battery_volts and not 2.5 < state.battery_volts < 9.0:
|
||||
print(f" *** battery {state.battery_volts} V is impossible — "
|
||||
f"this is the from-the-end offset bug. ***")
|
||||
if state.device_time is None:
|
||||
print(" *** device clock did not decode — dump raw below. ***")
|
||||
print(f" raw 0x1C content: {_content(state.raw).hex(' ')}")
|
||||
|
||||
setups = step("list_setups()", mm.list_setups)
|
||||
if setups:
|
||||
print(f" first={setups[0]!r} last={setups[-1]!r}")
|
||||
|
||||
if a.download:
|
||||
print("\n event chain (read-only):")
|
||||
proto = mm.protocol
|
||||
proto.arm_event()
|
||||
hdr = _content(proto.read_event_first())
|
||||
key, size = hdr[0:4], int.from_bytes(hdr[4:8], "big")
|
||||
if not size:
|
||||
print(" no events stored")
|
||||
else:
|
||||
print(f" first event key={key.hex()} size={size} B")
|
||||
t1 = time.monotonic()
|
||||
blob = proto.read_event_file(key, size)
|
||||
dt = time.monotonic() - t1
|
||||
print(f" downloaded {len(blob)} B in {dt:.1f} s "
|
||||
f"({len(blob)/dt/1024:.1f} KiB/s)")
|
||||
assert len(blob) == size
|
||||
# Decode it with the existing codec to prove the bytes are real.
|
||||
try:
|
||||
from micromate.idf_file import read_idf_file
|
||||
import tempfile
|
||||
with tempfile.NamedTemporaryFile(suffix=".IDFW", delete=False) as f:
|
||||
f.write(blob)
|
||||
tmp = f.name
|
||||
ev = read_idf_file(tmp)
|
||||
print(f" decoded OK: {ev}")
|
||||
except Exception as e:
|
||||
print(f" decode failed: {type(e).__name__}: {e}")
|
||||
out = Path(f"./{key.hex()}.IDFW")
|
||||
out.write_bytes(blob)
|
||||
print(f" saved to {out} for offline analysis")
|
||||
|
||||
except ProtocolError as e:
|
||||
print(f"\n ABORTED {type(e).__name__}: {e}")
|
||||
return 3
|
||||
finally:
|
||||
mm.close()
|
||||
|
||||
print(f"\n transport: {transport.reads} reads, "
|
||||
f"{transport.bytes_in} B in, {transport.bytes_out} B out")
|
||||
print(" A modem path should show MORE reads for the same bytes than USB —")
|
||||
print(" that is the buffering, and it is exactly what needed proving.\n")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user