seismo-relay/tests/test_micromate_events.py

"""Event-chain tests for the Micromate (series-4) client.

The load-bearing test here replays THOR's captured six-event download session
through `MicromateClient.iter_events()` + `get_event()` and asserts **every byte
we put on the wire matches what THOR put on the wire** — 99 frames — while also
decoding all six events and cross-checking each waveform's peak vector sum
against the one the device computed itself.

That capture is gitignored, so those tests skip on a fresh clone; the offline
tests below use embedded real response bytes and cover the same logic.
"""
from __future__ import annotations

import datetime
import os
import sys
from pathlib import Path

import pytest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from micromate.client import CONTENT, DecodeMismatch, MicromateClient
from micromate.framing import ACK, DLE, ETX, STX, checksum, stuff
from micromate.models import MicromateEventRef
from micromate.protocol import ProtocolError

from test_micromate_client import ScriptedTransport, SERIAL, frame

# The real 0x0C record from UM20147 (11.0BD), captured over USB 2026-09-30.
# A histogram: content[11] = 0x08.
RECORD_BD = bytes.fromhex(
    "d200000000055d4a8100001e0907eab30d1b21000000084c6f636174696f6e00"
    "0000000000000000000000000074657374320000000000000000000000000000"
    "0000000000000000000000000000000000000000000000554d32303134370000"
    "003fcb3bde00000000053f000f5472616e00003e698cdb000300005665727400"
    "003fc76e65000300004c6f6e6700003e567c21000300004d69630000003956b9"
    "7c00050000000000000000000000000000000000000000000000000000000000"
    "0000000000000000000000000000000000000000000000000000000000"
)


def chain_entry(key: bytes, size: int) -> bytes:
    """A 1E/1F response data section: 11-byte prefix then key + size."""
    return (bytes([0x08]) + bytes(7) + bytes([0xFE]) + bytes(2)
            + key + size.to_bytes(4, "big"))


def client(responses):
    t = ScriptedTransport(responses)
    return MicromateClient(t, recv_timeout=0.5), t


# ── The chain walk ────────────────────────────────────────────────────────────

def test_chain_walk_arms_before_every_entry():
    """⚠ THOR sends 0x93 before EVERY 1E/1F, and this mirrors that."""
    responses = [
        frame(0xEA, SERIAL),                                    # serial()
        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4076)),
        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes.fromhex("055d4a82"), 11032)),
        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),   # sentinel
    ]
    mm, t = client(responses)
    refs = mm.list_events(with_records=False)

    subs = [w[5] for w in t.written]
    assert subs == [0x15, 0x93, 0x1E, 0x93, 0x1F, 0x93, 0x1F]
    assert [r.key_hex for r in refs] == ["055d4a81", "055d4a82"]
    assert [r.size for r in refs] == [4076, 11032]


def test_an_all_zero_key_ends_the_chain_and_is_not_an_error():
    mm, _ = client([
        frame(0xEA, SERIAL),
        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes(4), 0)),
    ])
    assert mm.list_events(with_records=False) == []


def test_refs_carry_the_serial_because_a_key_alone_is_ambiguous():
    """⚠ UM12947 and UM20147 BOTH have an event 055d4a81.

    A store keyed on the event key alone treats one unit's event as a duplicate
    of the other's, and nothing raises.  `uid` is the safe identifier.
    """
    mm, _ = client([
        frame(0xEA, SERIAL),
        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4076)),
        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
    ])
    (ref,) = mm.list_events(with_records=False)
    assert ref.serial == "UM12947"
    assert ref.uid == "UM12947:055d4a81"


def test_a_cursor_that_never_advances_raises_rather_than_hanging():
    from micromate import client as C
    responses = [frame(0xEA, SERIAL)]
    entry = chain_entry(bytes.fromhex("055d4a81"), 4076)
    responses += [frame(0x6C, bytes(11)), frame(0xE1, entry)]      # the 1E read
    for _ in range(C._MAX_EVENTS + 2):                             # then 1F forever
        responses += [frame(0x6C, bytes(11)), frame(0xE0, entry)]
    mm, _ = client(responses)
    with pytest.raises(ProtocolError, match="not advancing"):
        mm.list_events(with_records=False)


def test_a_truncated_chain_entry_raises():
    mm, _ = client([
        frame(0xEA, SERIAL), frame(0x6C, bytes(11)), frame(0xE1, bytes(14)),
    ])
    with pytest.raises(ProtocolError, match="need 8"):
        mm.list_events(with_records=False)


def test_iter_events_does_not_read_ahead():
    """It must yield at the cursor position, one arm/advance per event."""
    mm, t = client([
        frame(0xEA, SERIAL),
        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4076)),
        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
    ])
    it = mm.iter_events(with_records=False)
    first = next(it)
    assert [w[5] for w in t.written] == [0x15, 0x93, 0x1E], "no read-ahead"
    assert first.key_hex == "055d4a81"
    assert list(it) == []


# ── The 0x0C record ───────────────────────────────────────────────────────────

def test_record_decode_on_real_bd_bytes():
    mm, _ = client([
        frame(0xEA, SERIAL),
        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4796)),
        frame(0xF3, RECORD_BD),
        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
    ])
    (ref,) = mm.list_events()

    assert ref.record_type == "histogram"      # content[11] = 0x08
    assert ref.is_histogram is True
    assert ref.suffix == ".IDFH"
    assert ref.timestamp == datetime.datetime(2026, 9, 30, 13, 27, 33)
    assert ref.sensor_location == "Location"
    assert ref.setup == "test2"
    assert ref.serial == "UM20147", "the record's own serial wins over the cache"
    assert ref.peak_vector_sum_ips == pytest.approx(1.587765, abs=1e-5)
    assert ref.peaks_ips["Tran"] == pytest.approx(0.228076, abs=1e-5)
    assert ref.peaks_ips["Vert"] == pytest.approx(1.558056, abs=1e-5)
    assert ref.peaks_ips["Long"] == pytest.approx(0.209458, abs=1e-5)


def test_the_pvs_is_not_reconstructible_from_the_reported_peaks():
    """⚠ Why the PVS field matters: it cannot be recomputed from the peaks.

    It is the PER-SAMPLE peak vector sum.  `sqrt(Σpeak²)` is only an upper
    bound, because the channel maxima do not occur at the same instant, and
    `max(T,V,L)` is a lower bound.  On THIS event the two happen to be within
    0.05%, which is exactly the coincidence that made the field look like
    `sqrt(Σpeak²)` on first inspection.  Six other events separated them.
    """
    import math
    mm, _ = client([
        frame(0xEA, SERIAL),
        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4796)),
        frame(0xF3, RECORD_BD),
        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
    ])
    (ref,) = mm.list_events()
    p = ref.peaks_ips
    upper = math.sqrt(p["Tran"] ** 2 + p["Vert"] ** 2 + p["Long"] ** 2)
    lower = max(p["Tran"], p["Vert"], p["Long"])
    assert lower < ref.peak_vector_sum_ips < upper


def test_filename_matches_thors_convention():
    ref = MicromateEventRef(index=0, key=bytes.fromhex("055d4a82"), size=11032,
                            serial="UM12947", record_type="waveform",
                            timestamp=datetime.datetime(2026, 9, 23, 16, 33, 19))
    assert ref.filename == "UM12947_20260923163319.IDFW"
    ref.record_type = "histogram"
    assert ref.filename == "UM12947_20260923163319.IDFH"


def test_filename_is_none_without_a_type_rather_than_guessing():
    """⚠ Guessing would file a histogram as a waveform, and read_idf_file()
    dispatches on exactly that suffix."""
    ref = MicromateEventRef(index=0, key=bytes(4), size=1, serial="UM12947",
                            timestamp=datetime.datetime(2026, 1, 1))
    assert ref.record_type is None
    assert ref.suffix is None
    assert ref.filename is None


def test_an_unknown_record_type_warns_and_leaves_it_none(caplog):
    bad = bytearray(RECORD_BD)
    bad[CONTENT + 11] = 0x99
    mm, _ = client([
        frame(0xEA, SERIAL),
        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 10)),
        frame(0xF3, bytes(bad)),
        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
    ])
    with caplog.at_level("WARNING"):
        (ref,) = mm.list_events()
    assert ref.record_type is None
    assert "unknown record type 0x99" in caplog.text


def test_get_event_refuses_without_a_record_type():
    mm, _ = client([])
    ref = MicromateEventRef(index=0, key=bytes.fromhex("055d4a81"), size=4076)
    with pytest.raises(ValueError, match="record_type is unknown"):
        mm.get_event(ref)


# ── Against the real captured session ─────────────────────────────────────────

_CAPTURES = (
    Path(__file__).resolve().parents[1]
    / "bridges" / "captures" / "9-24-26 - micromate2"
)
_DOWNLOAD = "raw_bw_20260925_011403_Download_events_then_delete_1_event.bin"


def _destuffed(blob: bytes, is_req: bool):
    i, n = 0, len(blob)
    while i < n:
        if is_req:
            if not (blob[i] == ACK and i + 1 < n and blob[i + 1] == STX):
                i += 1
                continue
            j = i + 2
        else:
            if blob[i] != STX:
                i += 1
                continue
            j = i + 1
        out = bytearray()
        while j < n:
            if blob[j] == DLE and j + 1 < n:
                out.append(blob[j + 1])
                j += 2
                continue
            if blob[j] == ETX:
                break
            out.append(blob[j])
            j += 1
        if len(out) >= 6:
            yield blob[i:j + 1], bytes(out[:-1])
        i = j + 1


@pytest.mark.skipif(
    not (_CAPTURES / _DOWNLOAD).is_file(),
    reason="capture is gitignored; present only on a dev box",
)
def test_replaying_thors_session_reproduces_every_byte_and_decodes_every_event():
    """The whole point of step 4.

    Feed THOR's own responses to `iter_events()` + `get_event()`, and assert:
      * every request byte we emit matches THOR's, in order
      * all six events decode
      * each waveform's decoded PVS matches the device's stored float
    """
    reqs = list(_destuffed((_CAPTURES / _DOWNLOAD).read_bytes(), True))
    rsps = list(_destuffed(
        (_CAPTURES / _DOWNLOAD.replace("raw_bw", "raw_s3")).read_bytes(), False))

    # THOR's session opens with commands our client does not send (POLL, 0x1C,
    # 0x06 …) and ends with a delete.  Take the contiguous run from the first
    # 0x93 to the last 0x5A -- that is the event walk, and it is what we mirror.
    subs = [p[2] for _, p in reqs]
    lo = subs.index(0x93)
    hi = len(subs) - 1 - subs[::-1].index(0x5A)
    want_wire = [w for w, _ in reqs[lo:hi + 1]]
    replay = [bytes([STX]) + stuff(p + bytes([checksum(p)])) + bytes([ETX])
              for _, p in rsps[lo:hi + 1]]

    # ⚠ THOR never reads the chain sentinel in this capture -- it downloaded
    # exactly six events and stopped, so it knew the count in advance.  It read
    # `SUB 0x06` (storage range) at frame 7, BEFORE the walk, and that response
    # begins `00 00 00 06` -- the event count.  Our generator walks until the
    # sentinel instead, so append one and compare only the overlapping frames.
    replay += [frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0))]

    # serial() would emit a 0x15 that is not in this slice, so prime the cache.
    t = ScriptedTransport(replay)
    mm = MicromateClient(t, recv_timeout=1.0)
    mm._serial = "UM12947"

    decoded, checked = 0, 0
    for ref in mm.iter_events():
        result = mm.get_event(ref)          # verify=True by default
        decoded += 1
        assert ref.serial == "UM12947"
        assert ref.filename and ref.filename.endswith(ref.suffix)
        if not ref.is_histogram:
            err = mm.decode_error(ref, result)
            assert err is not None
            assert abs(err) < 1e-4, f"{ref.uid}: PVS off by {100 * err:+.4f}%"
            checked += 1

    assert decoded == 6, "four waveforms and two histograms"
    assert checked == 4
    # Our trailing sentinel read is two frames THOR did not send; everything
    # up to it must match byte for byte, in order.
    assert t.written[:len(want_wire)] == want_wire, (
        f"we emitted {len(t.written)} frames, THOR emitted {len(want_wire)}"
    )
    assert len(t.written) == len(want_wire) + 2, "only the sentinel read is extra"


@pytest.mark.skipif(
    not (_CAPTURES / _DOWNLOAD).is_file(),
    reason="capture is gitignored; present only on a dev box",
)
def test_a_corrupted_stored_peak_is_caught_by_the_self_check():
    """Prove the verify path actually fires -- otherwise it is decoration."""
    reqs = list(_destuffed((_CAPTURES / _DOWNLOAD).read_bytes(), True))
    rsps = list(_destuffed(
        (_CAPTURES / _DOWNLOAD.replace("raw_bw", "raw_s3")).read_bytes(), False))
    subs = [p[2] for _, p in reqs]
    lo = subs.index(0x93)
    hi = len(subs) - 1 - subs[::-1].index(0x5A)
    replay = [bytes([STX]) + stuff(p + bytes([checksum(p)])) + bytes([ETX])
              for _, p in rsps[lo:hi + 1]]

    t = ScriptedTransport(replay)
    mm = MicromateClient(t, recv_timeout=1.0)
    mm._serial = "UM12947"

    for ref in mm.iter_events():
        if ref.is_histogram:
            mm.download_event(ref)          # keep the replay in step
            continue
        ref.peak_vector_sum_ips *= 1.5      # as a bad decode would look
        with pytest.raises(DecodeMismatch, match="decode is suspect"):
            mm.get_event(ref)
        return
    pytest.fail("no waveform event found in the capture")