#!/usr/bin/env python3 """ mm_stream_probe.py — how many bytes will `SUB 0x5A` serve in one request? Settled 2026-10-02: there is NO streaming mode ---------------------------------------------- This script started out asking whether `offset_hi = 0x10` meant "stream until done", because our 2026-09-23 notes recorded a single request appearing to return an entire 11 KB event. Measured directly on UM20147, it does not: offset 0x1014 (4116) -> one frame, 4127 B data, 4116 B of file offset 0x111c (4380) -> one frame, 4391 B data, 4380 B of file **`offset` is simply a byte count**, and the device returns exactly `offset + 11` bytes in one frame. `0x1000` is not a marker; it is part of the number. The 2026-09-23 reading was wrong, and the protocol reference now says so. The useful question it turned up ------------------------------- **1024 bytes per request is THOR's choice, not the device's limit.** The unit served 4,380 bytes in a single frame without being asked twice. Since a round trip over cellular costs ~0.65 s regardless of payload, and UM20147's 72,560-byte event is 71 chunks ≈ 46 seconds, the ceiling on one request is worth knowing precisely: every doubling halves the dominant cost. So this now walks ascending request sizes against one event and **checks each against a known-good chunk-loop download** — a pass means byte-identical output, not merely a plausible length. The offset field is a uint16, so 65,535 is the structural maximum. ⚠ **Read-only.** `0x5A` is a read we have sent thousands of times; the only new thing is a larger value in its offset field. Nothing here writes, erases or changes monitoring state. It re-POLLs at the end, because the honest risk is leaving the session in an odd state and the script should say so. Usage ----- python3 scratch/mm_stream_probe.py /dev/ttyACM1 python3 scratch/mm_stream_probe.py /dev/ttyACM1 --event largest """ from __future__ import annotations import argparse import math import sys import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "bridges")) from micromate.client import MicromateClient, _content # noqa: E402 from micromate.framing import MicromateFrameParser, build_request # noqa: E402 from micromate.protocol import SUB_BULK_DOWNLOAD # noqa: E402 from mm_client_check import StdlibSerial # noqa: E402 from minimateplus.transport import TcpTransport # noqa: E402 _CHUNK_PREFIX = 11 def collect(transport, parser, *, idle_gap: float, deadline: float) -> list: """Read until `idle_gap` seconds pass with no new bytes, or `deadline`.""" frames, last = [], time.monotonic() while time.monotonic() < deadline: chunk = transport.read(4096) if chunk: frames += parser.feed(chunk) last = time.monotonic() continue if time.monotonic() - last > idle_gap: break time.sleep(0.005) return frames def main() -> int: ap = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter ) ap.add_argument("target", help="host:port, or a serial device path") ap.add_argument("--baud", type=int, default=115200) ap.add_argument("--event", default="smallest", help="a key in hex, 'smallest' (default — the gentlest " "first test) or 'largest' (the one that matters)") ap.add_argument("--idle-gap", type=float, default=2.0, help="seconds of silence that end a streaming read; 2.0 is " "generous for a modem, which buffers ~1 s") ap.add_argument("--timeout", type=float, default=90.0) 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=10.0) label = f"TCP {host}:{port}" else: inner = StdlibSerial(a.target, baud=a.baud) label = f"serial {a.target} @ {a.baud}" mm = MicromateClient(inner, recv_timeout=20.0) print(f"\n{label} READ-ONLY: chain walk + two downloads of one event\n") mm.open() try: info = mm.connect(with_active_setup=False) print(f" {info}\n") refs = mm.list_events() if not refs: print(" no events stored — nothing to download. Record one first.") return 1 for r in refs: print(f" {r}") which = a.event.strip().lower() if which == "smallest": ref = min(refs, key=lambda r: r.size) elif which == "largest": ref = max(refs, key=lambda r: r.size) else: matches = [r for r in refs if r.key_hex.lower() == which] if not matches: print(f"\n --event {a.event!r} matched nothing") return 2 ref = matches[0] n_chunks = math.ceil(ref.size / 1024) print(f"\n target: {ref.key_hex} {ref.size} B {ref.record_type} " f"({n_chunks} chunks the known-good way)") # ── 1. the known-good chunk loop ────────────────────────────────────── t0 = time.monotonic() chunked = mm.protocol.read_event_file(ref.key, ref.size) dt_chunked = time.monotonic() - t0 print(f"\n [1] chunk loop ...... {len(chunked)} B in {dt_chunked:.2f} s " f"({n_chunks} requests)") # ── 2. how many bytes will it serve in ONE frame? ───────────────── # The streaming hypothesis is dead (see the module docstring): `offset` # is simply a BYTE COUNT, and the device returns `offset + 11` bytes in # one frame. So the real question is the ceiling -- because 1024 is # THOR's choice, not the device's limit, and every doubling halves the # round trips that dominate a cellular download. print("\n [2] chunk-size ceiling — ascending single requests") print(" each asks for N bytes from offset 0 and is checked against") print(" the known-good download, so a pass means identical bytes.\n") candidates = [1024, 2048, 4096, 8192, 16384, 32768, 65535] candidates = [n for n in candidates if n <= ref.size] or [ref.size] if ref.size not in candidates and ref.size < 65536: candidates.append(ref.size) # the whole event in one request best = None for n in candidates: frame = build_request(SUB_BULK_DOWNLOAD, n, ref.key + bytes(6)) parser = MicromateFrameParser() t0 = time.monotonic() mm.protocol._send(frame) got = collect(inner, parser, idle_gap=a.idle_gap, deadline=t0 + a.timeout) dt = time.monotonic() - t0 if not got: print(f" {n:6} B no answer ({dt:.2f} s)") continue body = b"".join(f.data[_CHUNK_PREFIX:] for f in got) ok = body == chunked[:n] flag = "OK " if ok else "MISMATCH" print(f" {n:6} B {len(got)} frame(s) {len(body):6} B back " f"{dt:5.2f} s {flag}" + ("" if ok or not body else f" (first diff at {next((i for i in range(min(len(body), n)) if body[i] != chunked[i]), None)})")) if ok and len(body) == n: best = n print() if best is None: print(" VERDICT: nothing above the current chunk size verified.") else: now = math.ceil(ref.size / 1024) then = math.ceil(ref.size / best) print(f" VERDICT: the device serves at least {best} B per request,") print(f" verified byte-identical. For this {ref.size} B event that") print(f" is {then} request(s) instead of {now}.") if best > 1024: print(f" Over cellular at ~0.65 s per round trip: " f"~{now * 0.65:.0f} s -> ~{then * 0.65:.1f} s.") if best >= ref.size: print(" The WHOLE EVENT fits in one request.") # ── 3. is the unit still healthy? ───────────────────────────────────── # The real risk of this experiment is leaving the session wedged, so # check rather than assume. print() try: p = mm.protocol.poll() print(f" [3] unit still answering POLL (SUB 0x{p.sub:02x}) — " f"session is healthy") except Exception as e: print(f" [3] ⚠ POLL FAILED after the probe: {type(e).__name__}: {e}") print(" Reconnect; if that does not help, power-cycle the unit.") return 4 finally: mm.close() return 0 if __name__ == "__main__": raise SystemExit(main())