diff --git a/micromate/client.py b/micromate/client.py index 8149863..030e167 100644 --- a/micromate/client.py +++ b/micromate/client.py @@ -32,11 +32,14 @@ from __future__ import annotations import datetime import logging +import math +import os +import struct from typing import Optional from minimateplus.transport import BaseTransport -from .models import MicromateDeviceInfo, MicromateState +from .models import MicromateDeviceInfo, MicromateEventRef, MicromateState from .protocol import MicromateProtocol, ProtocolError log = logging.getLogger(__name__) @@ -57,6 +60,27 @@ _MS_BATTERY = slice(34, 36) # uint16 BE, volts × 100 _MS_MEM_TOTAL = slice(36, 40) # uint32 BE _MS_MEM_FREE = slice(40, 44) # uint32 BE +# `SUB 0x0C` record, relative to content. Established against 7 events across +# both firmware lines. +_REC_DAY, _REC_MONTH, _REC_YEAR = 0, 1, slice(2, 4) +_REC_UNKNOWN_4 = 4 # ⚠ same shape as 0x1C's content[6]; undecoded +_REC_HOUR, _REC_MIN, _REC_SEC = 5, 6, 7 +_REC_TYPE = 11 # 0x07 waveform, 0x08 histogram +_REC_LOCATION = 12 +_REC_SETUP = 34 +_REC_SERIAL = 76 +_RECORD_TYPES = {0x07: "waveform", 0x08: "histogram"} + +# The channel labels the record carries, in the order they appear. The peak +# float sits `label + 6`; the peak vector sum sits 12 bytes BEFORE "Tran". +_REC_CHANNELS = (b"Tran", b"Vert", b"Long", b"Mic") +_REC_PVS_BACK = 12 + +# A chain walk terminates on an all-zero key. The ceilings below are guards +# against a device cursor that never advances, not fleet limits. +_NULL_KEY = bytes(4) +_MAX_EVENTS = 4096 + # A setup-list walk that does not terminate is a bug, not a big fleet. The # bench unit holds 22 setups; this is a generous ceiling, not a limit. _MAX_SETUPS = 512 @@ -71,6 +95,10 @@ def _cstring(buf: bytes, offset: int = 0) -> str: return buf[offset:].split(b"\x00")[0].decode("ascii", "replace").strip() +class DecodeMismatch(ProtocolError): + """Our decoded peak disagrees with the one the device computed itself.""" + + class MicromateClient: """High-level read-only client for one Micromate. @@ -88,6 +116,7 @@ class MicromateClient: transport, recv_timeout=recv_timeout, strict_checksums=strict_checksums ) self._firmware_line: Optional[str] = None + self._serial: Optional[str] = None # ── Lifecycle ───────────────────────────────────────────────────────────── @@ -140,6 +169,7 @@ class MicromateClient: manufacturer, model = self._parse_poll(poll.data) serial = _cstring(_content(self._proto.read_serial())) + self._serial = serial monitoring = self._parse_state(self._proto.read_state()) info = MicromateDeviceInfo( @@ -293,3 +323,246 @@ class MicromateClient: f"setup list did not terminate after {_MAX_SETUPS} entries — the " f"device cursor is not advancing" ) + + # ── Events ──────────────────────────────────────────────────────────────── + + def serial(self) -> str: + """The unit's serial, cached from `connect()` or read on demand. + + Needed by anything that handles an event, because an event key is + ambiguous without it — see `MicromateEventRef`. + """ + if self._serial is None: + self._serial = _cstring(_content(self._proto.read_serial())) + return self._serial + + def list_events(self, *, with_records: bool = True) -> list[MicromateEventRef]: + """Walk the event chain. `0x93 → 1E`, then `0x93 → 1F` until the sentinel. + + THOR sends `0x93` before **every** chain read, and this mirrors that. + The chain ends on an all-zero key. + + ⚠ `with_records=True` costs **one extra round trip per event** for the + `0x0C` read, and over cellular a round trip is ~0.65 s regardless of + size. On a unit with 40 events that is the difference between ~52 s and + ~78 s. Pass False when you only need "what is here and how big" — but + note the **record is where the type and timestamp live**, so without it + `ref.filename` is None and `get_event()` cannot pick a suffix. + + ⚠ **To download, prefer `iter_events()`.** This walks the whole chain + first; THOR interleaves, downloading each event before advancing. + `0x5A` addresses an event by key, so downloading afterwards *should* + work — but "should" is doing real work in that sentence, and the + interleaved order is the one with captures behind it. See + `iter_events()`. + """ + serial = self.serial() + refs: list[MicromateEventRef] = [] + + for i in range(_MAX_EVENTS): + self._proto.arm_event() + raw = (self._proto.read_event_first() if i == 0 + else self._proto.read_event_next()) + c = _content(raw) + if len(c) < 8: + raise ProtocolError( + f"chain entry {i}: {len(c)} B of content, need 8 (key + size)" + ) + key, size = c[0:4], int.from_bytes(c[4:8], "big") + if key == _NULL_KEY: + return refs # the sentinel, not an error + + ref = MicromateEventRef(index=i, key=key, size=size, serial=serial) + if with_records: + self._read_record_into(ref) + refs.append(ref) + + raise ProtocolError( + f"event chain did not terminate after {_MAX_EVENTS} entries — the " + f"device cursor is not advancing" + ) + + def iter_events(self, *, with_records: bool = True): + """Walk the chain, yielding each event **at the cursor position THOR uses.** + + for ref in mm.iter_events(): + if ref.is_histogram: + continue + data = mm.download_event(ref) # ← safe here + + Why this exists alongside `list_events()`: THOR's captured order is + + 0x93 → 1E → 0C → 5A×n → 0x93 → 1F → 0C → 5A×n → … + + — it downloads each event *before* advancing the chain. `list_events()` + walks to the end first, which is fine for browsing (the browse walk is + separately attested) but means a later download happens with the device + cursor parked past the event. `0x5A` is key-addressed, so it very + probably does not care; nothing observed says it does, and nothing + observed says it does not. + + Downloading inside this loop reproduces THOR's sequence exactly, so it + is the path to use when it matters. ⚠ Do not advance the generator + before finishing with the event it yielded. + """ + serial = self.serial() + for i in range(_MAX_EVENTS): + self._proto.arm_event() + raw = (self._proto.read_event_first() if i == 0 + else self._proto.read_event_next()) + c = _content(raw) + if len(c) < 8: + raise ProtocolError( + f"chain entry {i}: {len(c)} B of content, need 8 (key + size)" + ) + key, size = c[0:4], int.from_bytes(c[4:8], "big") + if key == _NULL_KEY: + return + ref = MicromateEventRef(index=i, key=key, size=size, serial=serial) + if with_records: + self._read_record_into(ref) + yield ref + + raise ProtocolError( + f"event chain did not terminate after {_MAX_EVENTS} entries — the " + f"device cursor is not advancing" + ) + + def _read_record_into(self, ref: MicromateEventRef) -> None: + """`SUB 0x0C` — 210 B of content: timestamp, type, names, peaks.""" + raw = self._proto.read_event_record(ref.key) + c = _content(raw) + if len(c) <= _REC_SERIAL: + raise ProtocolError(f"event record is {len(c)} B, too short to decode") + ref.raw_record = raw + + ref.record_type = _RECORD_TYPES.get(c[_REC_TYPE]) + if ref.record_type is None: + # Worth saying out loud rather than filing the event as a waveform: + # the suffix decides which codec runs. + log.warning("event %s: unknown record type 0x%02x at content[%d]", + ref.key_hex, c[_REC_TYPE], _REC_TYPE) + + try: + ref.timestamp = datetime.datetime( + year=int.from_bytes(c[_REC_YEAR], "big"), + month=c[_REC_MONTH], day=c[_REC_DAY], + hour=c[_REC_HOUR], minute=c[_REC_MIN], second=c[_REC_SEC], + ) + except ValueError as e: + log.warning("event %s: bad timestamp (%s): %s", + ref.key_hex, e, c[:8].hex(" ")) + + ref.sensor_location = _cstring(c, _REC_LOCATION) or None + ref.setup = _cstring(c, _REC_SETUP) or None + # Prefer the record's own serial over the cached one — they have always + # agreed, but the record is the event's own account of where it came from. + if rec_serial := _cstring(c, _REC_SERIAL): + ref.serial = rec_serial + + peaks: dict[str, float] = {} + for label in _REC_CHANNELS: + i = c.find(label) + if i < 0 or i + len(label) + 10 > len(c): + continue + peaks[label.decode()] = struct.unpack( + ">f", c[i + len(label) + 2: i + len(label) + 6])[0] + ref.peaks_ips = peaks or None + + tran = c.find(b"Tran") + if tran >= _REC_PVS_BACK: + ref.peak_vector_sum_ips = struct.unpack( + ">f", c[tran - _REC_PVS_BACK: tran - _REC_PVS_BACK + 4])[0] + + def download_event(self, ref: MicromateEventRef) -> bytes: + """The raw `.IDFW`/`.IDFH` bytes, exactly as THOR would have stored them. + + Feeds `micromate.idf_file.read_idf_file()` and `/db/import/idf_file` + unchanged — no new codec work is needed for a directly downloaded event. + """ + return self._proto.read_event_file(ref.key, ref.size) + + def get_event(self, ref: MicromateEventRef, *, verify: bool = True, + tolerance: float = 0.01): + """Download and decode one event. + + Returns the codec's `IdfReadResult`. Needs `ref.record_type`, since + `read_idf_file()` dispatches on the filename suffix and there is no + filename on the wire — so call `list_events(with_records=True)` first. + + ⚠ `verify=True` re-computes the **peak vector sum** from the decoded + samples and compares it against the float the *device* put in the `0x0C` + record. Those are two independent computations over the same samples — + the device's from its own firmware, ours from our codec — so a + disagreement means our decode is wrong. Measured agreement on the bench + events is **0.000%**. + + This is 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. + + Histograms are not verified: `samples` is empty for them. + """ + if not ref.record_type: + raise ValueError( + f"event {ref.key_hex}: record_type is unknown, so the codec " + f"cannot be dispatched. Use list_events(with_records=True)." + ) + blob = self.download_event(ref) + + import tempfile + from .idf_file import read_idf_file + + # read_idf_file dispatches on the suffix, so the bytes need a name. + with tempfile.NamedTemporaryFile(suffix=ref.suffix, delete=False) as f: + f.write(blob) + tmp = f.name + try: + result = read_idf_file(tmp) + finally: + os.unlink(tmp) + + if verify and not ref.is_histogram: + err = self.decode_error(ref, result) + if err is not None and abs(err) > tolerance: + raise DecodeMismatch( + f"event {ref.uid}: decoded peak vector sum differs from the " + f"device's own by {100 * err:+.3f}% (tolerance " + f"{100 * tolerance:.1f}%) — the decode is suspect, not the " + f"device. Stored {ref.peak_vector_sum_ips:.5f} in/s." + ) + if err is not None: + log.debug("event %s: PVS agrees to %+.4f%%", ref.uid, 100 * err) + return result + + @staticmethod + def decode_error(ref: MicromateEventRef, result) -> Optional[float]: + """Relative error between our decoded PVS and the device's stored one. + + None when either side is unavailable. Positive means the device's + figure is higher than ours. + """ + from .idf_file import geo_count_to_ips + + stored = ref.peak_vector_sum_ips + if not stored or not getattr(result, "samples", None): + return None + + ch = {k.lower(): v for k, v in result.samples.items()} + try: + t, v, l = ch["tran"], ch["vert"], ch["long"] + except KeyError: + return None + n = min(len(t), len(v), len(l)) + if not n: + return None + + pvs = max( + math.sqrt(geo_count_to_ips(t[i]) ** 2 + + geo_count_to_ips(v[i]) ** 2 + + geo_count_to_ips(l[i]) ** 2) + for i in range(n) + ) + return (stored - pvs) / pvs if pvs else None diff --git a/micromate/models.py b/micromate/models.py index 49e4c25..7b56c2d 100644 --- a/micromate/models.py +++ b/micromate/models.py @@ -478,3 +478,76 @@ class MicromateState: if frac is not None: bits.append(f"memory {frac * 100:.1f}% used") return " ".join(bits) + + +@dataclass +class MicromateEventRef: + """One entry in a unit's event chain, from `1E`/`1F` and optionally `0x0C`. + + ⚠ **`key` is NOT unique across units.** The event counter starts from the + same value on every Micromate — UM12947 and UM20147 both have an event + `055d4a81`, with different sizes and different contents. Use `uid`, or key + on `(serial, key_hex)`, for anything that stores or deduplicates. A store + keyed on the event key alone silently treats one unit's event as a duplicate + of another's, and nothing raises. + """ + + index: int + key: bytes # 4-byte event key from the chain walk + size: int # bytes the device will send for this event + serial: Optional[str] = None # the unit, because `key` alone is ambiguous + + # From `SUB 0x0C` — one extra round trip per event, so optional. + record_type: Optional[str] = None # "waveform" | "histogram" + timestamp: Optional[datetime.datetime] = None + setup: Optional[str] = None # setup file name, no extension + sensor_location: Optional[str] = None + peak_vector_sum_ips: Optional[float] = None # per-sample PVS, device-computed + peaks_ips: Optional[Dict[str, float]] = None # {"Tran": …, "Vert": …, …} + raw_record: Optional[bytes] = field(default=None, repr=False) + + @property + def key_hex(self) -> str: + return self.key.hex() + + @property + def uid(self) -> str: + """`SERIAL:key` — safe to use as a primary key. See the class note.""" + return f"{self.serial or '?'}:{self.key_hex}" + + @property + def is_histogram(self) -> Optional[bool]: + if self.record_type is None: + return None + return self.record_type == "histogram" + + @property + def suffix(self) -> Optional[str]: + return {"waveform": ".IDFW", "histogram": ".IDFH"}.get(self.record_type or "") + + @property + def filename(self) -> Optional[str]: + """The name THOR would have given this event. + + `_.IDF{W,H}` — e.g. + `UM12947_20260923163319.IDFW`. Verified against the production store + for all five bench events. + + ⚠ The type comes from the **protocol**, not the payload, so it has to be + carried here from the `0x0C` read. Returns None without it: guessing + the suffix would file a histogram as a waveform, and `read_idf_file()` + dispatches on exactly that. + """ + if not (self.serial and self.timestamp and self.suffix): + return None + return f"{self.serial}_{self.timestamp:%Y%m%d%H%M%S}{self.suffix}" + + def __str__(self) -> str: + bits = [self.uid, f"{self.size} B"] + if self.record_type: + bits.append(self.record_type) + if self.timestamp: + bits.append(self.timestamp.strftime("%Y-%m-%d %H:%M:%S")) + if self.peak_vector_sum_ips is not None: + bits.append(f"PVS {self.peak_vector_sum_ips:.4f} in/s") + return " ".join(bits) diff --git a/tests/test_micromate_events.py b/tests/test_micromate_events.py new file mode 100644 index 0000000..7423457 --- /dev/null +++ b/tests/test_micromate_events.py @@ -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")