seismo-relay/bridges/mm_client_check.py

#!/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 errno
import os
import select
import sys
import termios
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 TcpTransport  # noqa: E402


class StdlibSerial:
    """Raw serial on stdlib `termios` — no pyserial.

    `minimateplus.SerialTransport` needs pyserial, and a bench host is whatever
    is to hand.  On a PEP 668 distro (Mint 22, Ubuntu 24.04, Debian 12) a plain
    `pip install pyserial` is refused outright, so a diagnostic that depends on
    it is one you cannot run at the moment you need it.  `bridges/mm_link.py`
    and `scratch/fake_unit.py` already take this approach; this is the same
    ~30 lines, and it means the tool runs on a stock Python 3 anywhere.

    Not a general replacement for SerialTransport — no flow control, no
    parity options, Linux/macOS only.  Enough for a Micromate, which is 8N1
    with no handshaking.
    """

    _BAUD = {9600: termios.B9600, 19200: termios.B19200, 38400: termios.B38400,
             57600: termios.B57600, 115200: termios.B115200}

    def __init__(self, path: str, baud: int = 115200) -> None:
        if baud not in self._BAUD:
            raise ValueError(f"unsupported baud {baud}; pick from {sorted(self._BAUD)}")
        self.path, self.baud, self.fd = path, baud, None

    def connect(self) -> None:
        if self.fd is not None:
            return
        self.fd = os.open(self.path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
        a = termios.tcgetattr(self.fd)
        a[0] = a[1] = a[3] = 0                               # raw in/out, non-canonical
        a[2] = termios.CS8 | termios.CREAD | termios.CLOCAL  # 8N1, ignore modem lines
        a[4] = a[5] = self._BAUD[self.baud]
        a[6] = list(a[6])
        a[6][termios.VMIN] = 0
        a[6][termios.VTIME] = 0
        termios.tcsetattr(self.fd, termios.TCSANOW, a)
        termios.tcflush(self.fd, termios.TCIOFLUSH)

    def disconnect(self) -> None:
        if self.fd is not None:
            os.close(self.fd)
            self.fd = None

    def is_connected(self) -> bool:
        return self.fd is not None

    def read(self, n: int) -> bytes:
        if self.fd is None:
            return b""
        r, _, _ = select.select([self.fd], [], [], 0.05)
        if not r:
            return b""
        try:
            return os.read(self.fd, n)
        except OSError as e:
            if e.errno in (errno.EAGAIN, errno.EWOULDBLOCK):
                return b""
            raise

    def write(self, data: bytes) -> None:
        if self.fd is None:
            raise OSError("port is not open")
        while data:
            data = data[os.write(self.fd, data):]


class _Timed:
    """Count bytes and time each read, so the two transports can be compared.

    With `capture`, also writes the raw byte streams to a `raw_bw_*` /
    `raw_s3_*` pair in the layout `scratch/mm_frame_parse.py` already reads --
    so a run on an unfamiliar unit can be turned into test fixtures without
    setting up a relay.
    """

    def __init__(self, inner, capture: str | None = None) -> None:
        self._inner = inner
        self.reads = 0
        self.bytes_in = 0
        self.bytes_out = 0
        self._bw = self._s3 = None
        if capture:
            stamp = time.strftime("%Y%m%d_%H%M%S")
            d = Path(capture)
            d.mkdir(parents=True, exist_ok=True)
            self.bw_path = d / f"raw_bw_{stamp}_mm_client_check.bin"
            self.s3_path = d / f"raw_s3_{stamp}_mm_client_check.bin"
            self._bw = open(self.bw_path, "wb")
            self._s3 = open(self.s3_path, "wb")

    def close_capture(self) -> None:
        for f in (self._bw, self._s3):
            if f:
                f.close()

    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)
        if self._bw:
            self._bw.write(data); self._bw.flush()
        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)
            if self._s3:
                self._s3.write(chunk); self._s3.flush()
        return chunk


# ⚠ HYPOTHESIS, 6 events.  content[11] of the 0x0C record separated 4 waveforms
# from 2 histograms cleanly and was constant within each group.  A 4/2 split is
# thin evidence for a byte that could be anything, so _decode() below does NOT
# trust it -- it tries the other suffix on failure and says when the guess was
# wrong.  The protocol reference states no type field is known; this may be it.
_TYPE_BYTE = 11
_TYPES = {0x07: ".IDFW", 0x08: ".IDFH"}


def _event_type(record: bytes) -> str:
    if len(record) <= _TYPE_BYTE:
        return "?"
    b = record[_TYPE_BYTE]
    return {0x07: "waveform", 0x08: "histogram"}.get(b, f"unknown(0x{b:02x})")


def _decode(blob: bytes, key: bytes, record: bytes) -> None:
    """Decode the downloaded bytes, proving they are a real event file.

    read_idf_file() picks waveform vs histogram from the FILENAME SUFFIX, and a
    wire download has no filename -- so the suffix has to come from somewhere.
    This tries the 0x0C type byte first and the other suffix second; getting a
    decode either way proves the chunk assembly, and which one worked is itself
    the finding.
    """
    import tempfile
    from micromate.idf_file import read_idf_file

    guess = _TYPES.get(record[_TYPE_BYTE] if len(record) > _TYPE_BYTE else -1, ".IDFW")
    order = [guess] + [e for e in (".IDFW", ".IDFH") if e != guess]

    for n, ext in enumerate(order):
        with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as f:
            f.write(blob)
            tmp = f.name
        try:
            res = read_idf_file(tmp)
            samples = sum(len(v) for v in getattr(res, "samples", {}).values())
            note = "" if n == 0 else f"  *** the 0x0C type byte guessed {guess} — WRONG ***"
            print(f"       decoded OK as {ext}: {samples} samples{note}")
            os.unlink(tmp)
            return
        except Exception as e:
            last = f"{ext}: {type(e).__name__}: {e}"
        finally:
            if os.path.exists(tmp):
                os.unlink(tmp)

    print(f"       decode failed BOTH ways — last: {last}")
    out = Path(f"./{key.hex()}.bin")
    out.write_bytes(blob)
    print(f"       saved to {out} for offline analysis")


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; the USB-A/FTDI path runs at 115200. "
                         "Ignored by the USB-B 'PC' port, which is CDC-ACM "
                         "and negotiates its own rate.")
    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("--capture", metavar="DIR",
                    help="also write a raw_bw_*/raw_s3_*.bin pair to DIR, so "
                         "this run can become a test fixture.  Worth doing on "
                         "any unit whose firmware line is new to us.")
    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 = StdlibSerial(a.target, baud=a.baud)
        path = f"serial {a.target} @ {a.baud}"

    transport = _Timed(inner, capture=a.capture)
    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:
                rec = _content(proto.read_event_record(key))
                print(f"       first event key={key.hex()} size={size} B "
                      f"type={_event_type(rec)}")
                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(blob, key, rec)

    except ProtocolError as e:
        print(f"\n  ABORTED  {type(e).__name__}: {e}")
        return 3
    finally:
        mm.close()
        transport.close_capture()
        if a.capture:
            print(f"\n  capture written:\n    {transport.bw_path}\n    {transport.s3_path}")
            print("  parse it with: python3 scratch/mm_frame_parse.py "
                  f"{transport.bw_path} {transport.s3_path}")

    elapsed = time.monotonic() - t0
    print(f"\n  transport: {transport.reads} reads, "
          f"{transport.bytes_in} B in, {transport.bytes_out} B out, "
          f"{elapsed:.1f} s total")
    print("  Measured 2026-09-29, UM12947, same unit both ways:")
    print("     USB-B (CDC-ACM)   83 reads   list_setups  0.46 s   download 394 KiB/s")
    print("     RX55 (TCP)        36 reads   list_setups 16.05 s   download 1.6 KiB/s")
    print("  The modem needs FEWER reads, not more -- it buffers ~1 s and then")
    print("  forwards one large segment, where CDC-ACM delivers many small ones.")
    print("  Cost is ~0.65 s PER ROUND TRIP regardless of payload size, so what")
    print("  matters over cellular is the number of commands, not the bytes.\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())