seismo-relay/scratch/mm_stream_probe.py

#!/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, THOR_CHUNK_SIZE  # 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=3.0,
                    help="seconds of silence that end a read.  A 16 KB response "
                         "is ~1.4 s of serial time at 115200 BEFORE the modem's "
                         "~1 s forwarding delay, so this is deliberately roomy; "
                         "too small and a slow link looks like a clamp.")
    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 / THOR_CHUNK_SIZE)
        print(f"\n  target: {ref.key_hex}  {ref.size} B  {ref.record_type}  "
              f"({n_chunks} requests at THOR's chunk size)")

        # ── 1. the control: THOR's 1024-byte chunk loop ───────────────────────
        # ⚠ Pinned to THOR_CHUNK_SIZE on purpose.  The library default is now
        # 16,384, and using it here would make the experiment circular -- the
        # "known-good" reference would share any fault with the sizes under test.
        # 1024 is the size with THOR's own captures behind it and the one proven
        # over a modem, so it is the control.
        t0 = time.monotonic()
        chunked = mm.protocol.read_event_file(ref.key, ref.size,
                                              chunk_size=THOR_CHUNK_SIZE)
        dt_chunked = time.monotonic() - t0
        print(f"\n  [1] control ......... {len(chunked)} B in {dt_chunked:.2f} s "
              f"({n_chunks} requests at THOR's 1024 B)")

        # ── 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"
            # ⚠ dt includes the idle gap: collect() waits `idle_gap` after the
            # last byte before deciding the response is over.  Subtract it to
            # compare against the control, which reads to frame completion.
            print(f"      {n:6} B  {len(got)} frame(s)  {len(body):6} B back  "
                  f"{dt:5.2f} s ({max(dt - a.idle_gap, 0):4.2f} s net)  {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 / THOR_CHUNK_SIZE)
            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 > THOR_CHUNK_SIZE:
                # ⚠ cost is ~0.21 s per request PLUS ~2,350 B/s over a modem --
                # a download is throughput-bound, so do not promise a saving
                # proportional to the drop in request count.
                fixed, rate = 0.212, 2348.0
                t_now = now * fixed + ref.size / rate
                t_then = then * fixed + ref.size / rate
                print(f"  Modelled over cellular (0.21 s/request + "
                      f"{rate:.0f} B/s): {t_now:.1f} s -> {t_then:.1f} s, "
                      f"a {100 * (1 - t_then / t_now):.0f}% saving.")
                print("  Throughput-bound, not round-trip bound — most of that is")
                print("  bytes on the wire and batching does not touch it.")
            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())