diff --git a/bridges/mm_client_check.py b/bridges/mm_client_check.py index bce8921..87fe201 100644 --- a/bridges/mm_client_check.py +++ b/bridges/mm_client_check.py @@ -314,25 +314,55 @@ def main() -> int: if setups: print(f" first={setups[0]!r} last={setups[-1]!r}") + # ── SUB 0x06: is content[0:4] really the event count? ───────────── + # Two samples on one unit said yes, and THOR reads it BEFORE the chain + # walk then stops without ever reading the sentinel. A third value + # either confirms it or kills it. + claimed = None + raw06 = step("0x06 storage range", mm.protocol.read_storage_range) + if raw06: + c = _content(raw06) + claimed = int.from_bytes(c[0:4], "big") + print(f" content[0:4] = {claimed} <- CANDIDATE: event count") + print(f" content[4:8] = {int.from_bytes(c[4:8],'big')} " + f"<- unexplained (read 9 alongside a 6 on UM12947)") + 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) + print("\n event chain (read-only), via MicromateClient:") + t1 = time.monotonic() + refs = mm.list_events() # walks to the sentinel + walk = time.monotonic() - t1 + print(f" {len(refs)} events in {walk:.1f} s " + f"({walk/max(len(refs),1):.2f} s each, 3 round trips per event)") + + if claimed is not None: + verdict = ("✓ AGREES" if claimed == len(refs) + else f"✗ DISAGREES (0x06 said {claimed})") + print(f" 0x06 count vs chain length: {verdict}") + + for ref in refs: + print(f" {ref}") + print(f" would be filed as {ref.filename}") + + if refs: + # Download via iter_events, which reproduces THOR's interleaved + # order -- the one with captures behind it. + print("\n download (first event, THOR's interleaved order):") + for ref in mm.iter_events(): + t1 = time.monotonic() + try: + result = mm.get_event(ref) # verify=True + except Exception as e: + print(f" {ref.key_hex}: {type(e).__name__}: {e}") + break + dt = max(time.monotonic() - t1, 1e-6) + n = sum(len(v) for v in getattr(result, "samples", {}).values()) + err = mm.decode_error(ref, result) + check = ("PVS agrees to %+.4f%%" % (100 * err) + if err is not None else "PVS check n/a (histogram)") + print(f" {ref.key_hex} {ref.record_type:9} {ref.size:6} B " + f"in {dt:.1f} s -> {n} samples, {check}") + break except ProtocolError as e: print(f"\n ABORTED {type(e).__name__}: {e}")