serversdown/seismo-relay
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(micromate): the event chain -- walk, records, download, and a self-check

Browse Source 
Step 4 of docs/micromate_client_spec.md.  MicromateEventRef plus list_events(),
iter_events(), download_event(), get_event() and decode_error().  14 new tests;
112 micromate tests total.

The load-bearing test replays THOR's captured six-event download session through
iter_events() + get_event() and asserts EVERY BYTE WE EMIT MATCHES THOR'S, in
order -- 74 frames -- while decoding all six events and cross-checking each
waveform's peak vector sum against the float the device computed itself.

Two design decisions worth recording:

1. iter_events() EXISTS BECAUSE THOR INTERLEAVES.  Its captured order is
   0x93 -> 1E -> 0C -> 5A*n -> 0x93 -> 1F -> 0C -> 5A*n, downloading each event
   before advancing the chain.  list_events() walks to the end first, which is
   fine for browsing but leaves the device cursor parked past the event a later
   download addresses.  0x5A is key-addressed so it very probably does not care
   -- but nothing observed says either way, so the interleaved path is the one
   offered for downloads, and it is the one the replay test exercises.

2. get_event(verify=True) RECOMPUTES THE PEAK VECTOR SUM from the decoded
   samples and compares it against the device's own 0x0C float.  Two independent
   computations over the same samples, so a disagreement means our decode is
   wrong.  Agreement on the bench events is 0.000%.  Cheap insurance in a
   codebase whose decode failures have historically been silent -- unhandled
   block tags shorten a channel and nothing raises.  It is a decode-correctness
   check, NOT a truncation detector: a channel cut after its peak still yields
   the right PVS, and the docstring says so.  A test corrupts a stored peak to
   prove the check actually fires.

MicromateEventRef.uid is SERIAL:key, because the key alone is ambiguous across
units, and .filename generates THOR's name (<serial>_<YYYYMMDDHHMMSS>.IDFW/H)
-- returning None rather than guessing when the record type is unknown, since
read_idf_file() dispatches on exactly that suffix.

Recorded as a CANDIDATE, not used: 0x06 content[0:4] looks like the EVENT COUNT
-- 6 on a unit holding 6 events, zeros on an empty one, and THOR reads it BEFORE
the walk then downloads exactly six events without ever reading the chain
sentinel.  If it holds it lets a caller decide whether to walk at all, which
over cellular is the useful part.  Two samples on one unit, and content[4:8]
reads 9 unexplained, so list_events() still walks to the sentinel: slower by one
round trip and correct on evidence rather than inference.

Full suite: 465 passed, 16 pre-existing failures unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ru8Lg9HkkYvX9VWWo65SmL
This commit is contained in:
serversdown
2026-09-30 18:29:32 -04:00
co-authored by Claude Opus 5
parent e95d8e3b8c
commit d7d72cadf8
3 changed files with 690 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_micromate_events.py
+343
View File
@@ -0,0 +1,343 @@
"""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")