serversdownandClaude Opus 5
9bb95003e9
Supersedes the segment-header model entirely, including the fixes made earlier today. Found via multi-agent structural analysis of the 25 files that stalled the walker, then verified independently. Records are self-delimiting: off+2 is a uint16 BE length, next_record = off + 2 + len, and the chain ends on a record whose chan_id is 0x06. off+8 carries a 3-valued mode enum: 02 00 14-byte header, 2 anchors, then CUMULATIVE delta blocks 01 00 10-byte header, no anchors, blocks are ABSOLUTE values 00 03 10-byte header, NO TAGS AT ALL - raw 12-bit packed absolute `40 NN` is an ordinary int16 BE data block (2*NN + 2), never a header. Reading it as a 2*NN + 16 header is what made walks drift — the "variable-prefix segment descriptors" reported earlier today were not a format feature, just walker drift of exactly 4 - (old_stop - true_record_start), on all 25 affected files. Measured on the production snapshot: all four channels equal length 156/1388 -> 1388/1388 ASCII sample-count exact 72/75 -> 75/75 ASCII fully exact 70/75 -> 73/75 device PPV waveform (live) 1288/1306 -> 1306/1306 (mean err 0.00000) device PPV histogram (live) 4434/4459 -> 4458/4459 Also eliminates the walker-over-read class: 24 of those 35 files were histograms that read_blastware_file fed to the waveform codec first; the old walker accepted them and returned garbage (one yielded 98,923 "intervals"), while the record-chain decoder returns None so they fall through to histogram_codec. 00 03 records are DECODED, not skipped. Skipping them silently shifts the time base of everything after them on that channel — BE9558/ K558LOF2.820W had MicL displaced by exactly 512 samples with nothing marking the gap. Footer detection now prefers the 0e 08 candidate whose body yields a chain terminating on 0x06; the signature can occur inside a sample stream. Blast radius 1 file of 1388. The superseded model survives as decode_waveform_legacy, pinned by micromate/idf_file.py: its Thor IDFW body-offset search trial-decodes candidates and keeps whichever yields the most samples, so the new decoder returning None where the old returned garbage changes that heuristic's winner. Deferred until that search uses the record chain. Tests: 253 passed (+11), failure list unchanged from baseline. The 9 tests pinning the superseded model are retargeted at decode_waveform_legacy, which still implements it. NOTE: stored .h5 files need regenerating — nearly all get longer. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HgTe8CamXAHcAmaQ6QNcog
798 lines
33 KiB
Python
798 lines
33 KiB
Python
"""
|
||
Tests for minimateplus.waveform_codec — Blastware waveform-file body block walker.
|
||
|
||
These tests lock in the STRUCTURAL framing of the body codec. The byte-to-sample
|
||
mapping is open (see waveform_codec module docstring) — until that's nailed down,
|
||
:func:`decode_waveform_v2` returns ``None`` and there is no per-sample assertion
|
||
to make.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
|
||
import pytest
|
||
|
||
from minimateplus.waveform_codec import (
|
||
WaveformBlock,
|
||
decode_waveform_legacy,
|
||
decode_tran_initial,
|
||
decode_waveform_v2,
|
||
decoded_to_adc_counts,
|
||
find_data_start,
|
||
mic_count_to_db,
|
||
parse_segment_header,
|
||
split_segments,
|
||
walk_body,
|
||
)
|
||
|
||
|
||
FIXTURES = os.path.join(
|
||
os.path.dirname(__file__), "fixtures", "decode-re-5-8-26"
|
||
)
|
||
|
||
|
||
def _bw_body(path):
|
||
"""Strip the 22-byte header and 21-byte STRT and 26-byte footer to get the body."""
|
||
with open(path, "rb") as f:
|
||
binary = f.read()
|
||
return binary[43:-26]
|
||
|
||
|
||
# Fixture metadata — bundled BW binaries from a real BE11529 unit, May 8 2026.
|
||
# Each is paired with a Blastware TXT export (the ASCII ground truth).
|
||
FIXTURES_INFO = {
|
||
"event-a": {
|
||
"filename": "M529LKVQ.6S0",
|
||
"n_samples": 3328, # 3.0 s rectime + 0.25 s pretrig at 1024 sps
|
||
"rectime": 3.0,
|
||
},
|
||
"event-b": {
|
||
"filename": "M529LK5Q.RG0",
|
||
"n_samples": 2304, # 2.0 s
|
||
"rectime": 2.0,
|
||
},
|
||
"event-c": {
|
||
"filename": "M529LK44.AB0",
|
||
"n_samples": 1280, # 1.0 s
|
||
"rectime": 1.0,
|
||
},
|
||
"event-d": {
|
||
"filename": "M529LK2V.470",
|
||
"n_samples": 1280,
|
||
"rectime": 1.0,
|
||
},
|
||
}
|
||
|
||
|
||
def _fixture_path(event_name):
|
||
info = FIXTURES_INFO[event_name]
|
||
return os.path.join(FIXTURES, event_name, info["filename"])
|
||
|
||
|
||
# ── Find data start ──────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.parametrize("event_name", list(FIXTURES_INFO.keys()))
|
||
def test_find_data_start_locates_first_block(event_name):
|
||
"""The walker auto-detects the first ``10 NN`` tag within the first 20 bytes."""
|
||
path = _fixture_path(event_name)
|
||
if not os.path.exists(path):
|
||
pytest.skip(f"fixture missing: {path}")
|
||
body = _bw_body(path)
|
||
start = find_data_start(body)
|
||
assert 0 <= start < 20, f"expected start in [0, 20), got {start}"
|
||
assert body[start] in (0x00, 0x10, 0x20, 0x30, 0x40), (
|
||
f"first tag byte 0x{body[start]:02x} not a recognized block type"
|
||
)
|
||
assert body[start + 1] % 4 == 0 or (body[start] == 0x40 and body[start + 1] == 0x02)
|
||
|
||
|
||
def test_find_data_start_canonical_offset_7():
|
||
"""All events have a 7-byte preamble (3-byte magic + 4-byte Tran anchors)."""
|
||
for name in FIXTURES_INFO:
|
||
path = _fixture_path(name)
|
||
if not os.path.exists(path):
|
||
pytest.skip(f"fixture missing: {path}")
|
||
body = _bw_body(path)
|
||
# Sanity: magic
|
||
assert body[0:3] == b"\x00\x02\x00", f"{name}: bad magic"
|
||
# First tag at offset 7
|
||
assert find_data_start(body) == 7, f"{name}: expected start=7"
|
||
|
||
|
||
# ── Block walker ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_walk_body_empty_returns_empty():
|
||
assert walk_body(b"") == []
|
||
|
||
|
||
def test_walk_body_invalid_start_returns_empty():
|
||
# Body that does not begin with a recognized tag.
|
||
assert walk_body(b"\xff\xff\xff\xff", start=0) == []
|
||
|
||
|
||
@pytest.mark.parametrize("event_name", list(FIXTURES_INFO.keys()))
|
||
def test_walk_body_produces_blocks(event_name):
|
||
"""The walker should produce a non-empty stream of blocks for every fixture."""
|
||
path = _fixture_path(event_name)
|
||
if not os.path.exists(path):
|
||
pytest.skip(f"fixture missing: {path}")
|
||
body = _bw_body(path)
|
||
blocks = walk_body(body)
|
||
assert len(blocks) > 0
|
||
# All blocks have one of the known tag families. ``1X NN`` / ``2X NN``
|
||
# with X in 0..F are valid (X > 0 means wide-NN encoding).
|
||
for b in blocks:
|
||
assert (b.tag_hi & 0xF0) in (0x10, 0x20, 0x00, 0x30, 0x40), (
|
||
f"unknown tag {b.tag_hi:#04x} at offset {b.offset}"
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize("event_name", list(FIXTURES_INFO.keys()))
|
||
def test_walk_body_block_lengths_consistent(event_name):
|
||
"""Each block's recorded length matches its on-wire footprint."""
|
||
path = _fixture_path(event_name)
|
||
if not os.path.exists(path):
|
||
pytest.skip(f"fixture missing: {path}")
|
||
body = _bw_body(path)
|
||
blocks = walk_body(body)
|
||
for b in blocks:
|
||
# Tag (2 bytes) + payload should equal length.
|
||
assert 2 + len(b.data) == b.length, (
|
||
f"block at {b.offset} length mismatch: tag(2) + data({len(b.data)}) != length({b.length})"
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize("event_name", list(FIXTURES_INFO.keys()))
|
||
def test_walk_body_blocks_contiguous(event_name):
|
||
"""Block n+1 starts exactly where block n ends (no gaps, no overlaps)."""
|
||
path = _fixture_path(event_name)
|
||
if not os.path.exists(path):
|
||
pytest.skip(f"fixture missing: {path}")
|
||
body = _bw_body(path)
|
||
blocks = walk_body(body)
|
||
for i in range(1, len(blocks)):
|
||
prev = blocks[i - 1]
|
||
cur = blocks[i]
|
||
assert cur.offset == prev.offset + prev.length, (
|
||
f"gap/overlap between block {i-1} (off={prev.offset} len={prev.length}) "
|
||
f"and block {i} (off={cur.offset})"
|
||
)
|
||
|
||
|
||
# ── Segment splitting ────────────────────────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.parametrize("event_name", list(FIXTURES_INFO.keys()))
|
||
def test_split_segments_yields_at_least_one(event_name):
|
||
path = _fixture_path(event_name)
|
||
if not os.path.exists(path):
|
||
pytest.skip(f"fixture missing: {path}")
|
||
body = _bw_body(path)
|
||
blocks = walk_body(body)
|
||
segments = split_segments(blocks)
|
||
assert len(segments) > 0
|
||
|
||
|
||
def test_split_segments_segment_count_at_least_one_per_event():
|
||
"""The walker should produce at least one ``40 02`` segment header per event.
|
||
|
||
Note: the walker currently bails out partway through event-b (still an
|
||
open issue — the body codec uses block lengths the walker doesn't
|
||
handle correctly past offset ~427). The other 3 events walk farther
|
||
and have many segment headers.
|
||
"""
|
||
for name in FIXTURES_INFO:
|
||
path = _fixture_path(name)
|
||
if not os.path.exists(path):
|
||
continue
|
||
body = _bw_body(path)
|
||
blocks = walk_body(body)
|
||
n_40 = sum(1 for b in blocks if b.tag_hi == 0x40)
|
||
assert n_40 >= 1, f"{name}: no 40 02 segment header found"
|
||
|
||
|
||
# ── Segment header parsing ───────────────────────────────────────────────────
|
||
|
||
|
||
def test_parse_segment_header_returns_none_for_non_40():
|
||
block = WaveformBlock(offset=0, tag_hi=0x10, tag_lo=0x04, data=b"\x00\x00", length=4)
|
||
assert parse_segment_header(block) is None
|
||
|
||
|
||
def test_parse_segment_header_decodes_fields():
|
||
"""Decode a known 40 02 block to verify field offsets."""
|
||
# First segment header from event-c at body offset 235:
|
||
# 40 02 00 00 00 00 0a 4b 01 1e 47 00 00 00 02 00 00 01 00 01
|
||
payload = bytes.fromhex("00000000 0a4b011e 47000000 02000001 0001".replace(" ", ""))
|
||
block = WaveformBlock(
|
||
offset=235, tag_hi=0x40, tag_lo=0x02, data=payload, length=20
|
||
)
|
||
decoded = parse_segment_header(block)
|
||
assert decoded is not None
|
||
assert decoded["n_prev_deltas"] == 2
|
||
assert decoded["prev_deltas"] == [0, 0]
|
||
assert decoded["counter"] == 0x47 # uint32 LE
|
||
assert decoded["marker"] == b"\x02\x00"
|
||
assert decoded["anchors"] == [1, 1]
|
||
|
||
|
||
def test_segment_counter_increments():
|
||
"""The 4-byte counter at bytes [8:12] of each 40 02 payload increments by 1."""
|
||
path = _fixture_path("event-c")
|
||
if not os.path.exists(path):
|
||
pytest.skip("fixture missing")
|
||
body = _bw_body(path)
|
||
blocks = walk_body(body)
|
||
headers = [b for b in blocks if b.tag_hi == 0x40 and b.tag_lo == 0x02]
|
||
counters = [parse_segment_header(b)["counter"] for b in headers]
|
||
assert len(counters) >= 5, "expect at least 5 segments to verify increments"
|
||
# First few counters should be strictly monotonic (the BW counter is global,
|
||
# incrementing across the whole flash buffer; some events may share counter
|
||
# values with the previous event's tail block, so allow non-strict).
|
||
for i in range(1, min(8, len(counters))):
|
||
assert counters[i] >= counters[i - 1], (
|
||
f"counter went backwards: {counters[i-1]} → {counters[i]}"
|
||
)
|
||
|
||
|
||
# ── decode_waveform_v2: currently a stub ─────────────────────────────────────
|
||
|
||
|
||
@pytest.mark.parametrize("event_name", list(FIXTURES_INFO.keys()))
|
||
def test_decode_waveform_v2_returns_dict(event_name):
|
||
"""decode_waveform_v2 returns a dict with all 4 channels (verified 2026-05-11)."""
|
||
path = _fixture_path(event_name)
|
||
if not os.path.exists(path):
|
||
pytest.skip(f"fixture missing: {path}")
|
||
body = _bw_body(path)
|
||
result = decode_waveform_v2(body)
|
||
assert result is not None
|
||
assert set(result.keys()) == {"Tran", "Vert", "Long", "MicL"}
|
||
|
||
|
||
# Multi-channel ground-truth fixtures. Each row: (path, channel, n_to_verify).
|
||
# These lock in the channel-rotation hypothesis: segments cycle T → V → L → M,
|
||
# with each segment header carrying a 2-sample anchor pair (bytes [14:18])
|
||
# for THIS segment's channel plus 2 continuation deltas (bytes [0:4]) for
|
||
# the PREVIOUS channel.
|
||
MULTICHANNEL_FIXTURES = [
|
||
# ALL geo channels fully decoded for every event in the bundle:
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1L.V70"), "Tran", 3328),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1L.V70"), "Vert", 3328),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1L.V70"), "Long", 3328),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1L.JQ0"), "Tran", 3328),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1L.JQ0"), "Vert", 3328),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1L.JQ0"), "Long", 3328),
|
||
# SP0 (loud all-channels): NOW fully decodes after the wide-NN walker fix.
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SP0"), "Tran", 3328),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SP0"), "Vert", 3328),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SP0"), "Long", 3328),
|
||
# SS0 / SV0 (loud-from-start): walker now reaches 3072–3078 samples per
|
||
# channel (out of 3079 total). A few tail samples still missing.
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SS0"), "Tran", 3078),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SS0"), "Vert", 3072),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SS0"), "Long", 3072),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SV0"), "Tran", 3078),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SV0"), "Vert", 3072),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SV0"), "Long", 3072),
|
||
# 5-8-26 quiet bundle: events without 30 NN blocks decode FULLY across all channels.
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-a", "M529LKVQ.6S0"), "Tran", 3328),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-a", "M529LKVQ.6S0"), "Vert", 3328),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-a", "M529LKVQ.6S0"), "Long", 3328),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-c", "M529LK44.AB0"), "Tran", 1280),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-c", "M529LK44.AB0"), "Vert", 1280),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-c", "M529LK44.AB0"), "Long", 1280),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-d", "M529LK2V.470"), "Tran", 1280),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-d", "M529LK2V.470"), "Vert", 1280),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-d", "M529LK2V.470"), "Long", 1280),
|
||
# event-b: 2304 samples × 3 — now fully decodes (was the historical
|
||
# walker-stop case; fixed by wide-NN tag support).
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-b", "M529LK5Q.RG0"), "Tran", 2304),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-b", "M529LK5Q.RG0"), "Vert", 2304),
|
||
(os.path.join(os.path.dirname(__file__), "fixtures", "decode-re-5-8-26",
|
||
"event-b", "M529LK5Q.RG0"), "Long", 2304),
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize("path,channel,n", MULTICHANNEL_FIXTURES)
|
||
def test_decode_waveform_v2_channels_match_truth(path, channel, n):
|
||
"""Decoded channels match the BW ASCII export byte-exact for the verified ranges."""
|
||
if not os.path.exists(path):
|
||
pytest.skip(f"fixture missing: {path}")
|
||
with open(path, "rb") as f:
|
||
body = f.read()[43:-26]
|
||
truth = _full_truth_channel(path, channel)
|
||
decoded = decode_waveform_v2(body)
|
||
assert decoded is not None
|
||
pred = decoded[channel]
|
||
assert len(pred) >= n, f"only {len(pred)} samples decoded, expected ≥ {n}"
|
||
for i in range(n):
|
||
assert pred[i] == truth[i], (
|
||
f"{os.path.basename(path)} {channel}[{i}]: pred={pred[i]} truth={truth[i]}"
|
||
)
|
||
|
||
|
||
# ── decode_tran_initial: confirmed correct against ground truth ──────────────
|
||
|
||
# Bundled fixtures for the high-amplitude 5-11-26 events (PPV ~6-7 in/s).
|
||
# These cracked the Tran codec — see waveform_codec module docstring.
|
||
TRAN_INITIAL_FIXTURES = [
|
||
# (path, expected first N Tran samples in 16-count units, # of samples to verify)
|
||
(
|
||
os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SP0"),
|
||
[4, 4, 3, 3, 3, 2, 2, 3, 2, 2, 2, 2, 1, 1, 1, 2, 1, 1, 1, 0, 1, 0],
|
||
22,
|
||
),
|
||
(
|
||
os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SS0"),
|
||
[-89, -89, -91, -91, -92, -93, -94, -94, -94, -94],
|
||
42,
|
||
),
|
||
(
|
||
os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1A.SV0"),
|
||
[-745, -762, -771, -774, -779, -794, -808, -811, -811, -819],
|
||
46,
|
||
),
|
||
# Vert-heavy event (T near zero) — segment 0 = 510 samples, all decode correctly.
|
||
(
|
||
os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1L.JQ0"),
|
||
[0] * 4 + [-1, 0, 0, -1, -1, 0],
|
||
38,
|
||
),
|
||
# Mic-heavy event (geos all near zero) — segment 0 = 482 samples.
|
||
(
|
||
os.path.join(os.path.dirname(__file__), "fixtures", "5-11-26", "M529LL1L.V70"),
|
||
[0] * 10,
|
||
6,
|
||
),
|
||
]
|
||
|
||
|
||
def _full_truth(path):
|
||
"""Load Tran samples (in 16-count units) from the BW ASCII export."""
|
||
return _full_truth_channel(path, "Tran")
|
||
|
||
|
||
def _full_truth_channel(path, channel):
|
||
"""Load one channel's samples (in 16-count units) from the BW ASCII export."""
|
||
import glob, re
|
||
col_idx = {"Tran": 0, "Vert": 1, "Long": 2, "MicL": 3}[channel]
|
||
# event-a's TXT has a typo ("M59" vs "M529") — pick the .TXT in the same dir
|
||
# rather than assuming exact-name correspondence.
|
||
txt_path = path + ".TXT"
|
||
if not os.path.exists(txt_path):
|
||
candidates = glob.glob(os.path.join(os.path.dirname(path), "*.TXT"))
|
||
if candidates:
|
||
txt_path = candidates[0]
|
||
with open(txt_path, "r", encoding="utf-8", errors="replace") as f:
|
||
lines = f.read().splitlines()
|
||
header_idx = None
|
||
for i, line in enumerate(lines):
|
||
if "Tran" in line and "Vert" in line and "Long" in line and "MicL" in line:
|
||
header_idx = i
|
||
break
|
||
if header_idx is None:
|
||
return None
|
||
out = []
|
||
for line in lines[header_idx + 1:]:
|
||
parts = re.split(r"\s+", line.strip())
|
||
if len(parts) < 4:
|
||
continue
|
||
try:
|
||
out.append(round(float(parts[col_idx]) * 200))
|
||
except ValueError:
|
||
continue
|
||
return out
|
||
|
||
|
||
@pytest.mark.parametrize("path,expected,n_required", TRAN_INITIAL_FIXTURES)
|
||
def test_decode_tran_initial_matches_ground_truth(path, expected, n_required):
|
||
"""The Tran initial decoder produces values matching the BW ASCII export exactly."""
|
||
if not os.path.exists(path):
|
||
pytest.skip(f"fixture missing: {path}")
|
||
with open(path, "rb") as f:
|
||
raw = f.read()
|
||
body = raw[43:-26]
|
||
decoded = decode_tran_initial(body)
|
||
assert decoded is not None
|
||
# Check first len(expected) samples match exactly.
|
||
for i in range(len(expected)):
|
||
assert decoded[i] == expected[i], (
|
||
f"sample {i}: decoded={decoded[i]} expected={expected[i]}"
|
||
)
|
||
# And we got at least n_required samples decoded.
|
||
assert len(decoded) >= n_required, (
|
||
f"decoded only {len(decoded)} samples, expected at least {n_required}"
|
||
)
|
||
|
||
|
||
def test_decode_tran_initial_handles_empty():
|
||
assert decode_tran_initial(b"") is None
|
||
assert decode_tran_initial(b"not a body") is None
|
||
|
||
|
||
def test_decode_tran_initial_synthetic_body():
|
||
"""A synthetic body with preamble + one 10 04 block decodes correctly."""
|
||
# Magic + T[0]=10 + T[1]=20 in 16-count units.
|
||
# Then 10 04 block with 4 nibbles: (+1, -1, +2, -2)
|
||
# Encoded high-nibble first: 0x1F = (1, -1), 0x2E = (2, -2)
|
||
body = b"\x00\x02\x00\x00\x0a\x00\x14" + b"\x10\x04" + b"\x1f\x2e"
|
||
decoded = decode_tran_initial(body)
|
||
# T[0]=10, T[1]=20, then deltas (+1, -1, +2, -2) from T[1]=20
|
||
assert decoded == [10, 20, 21, 20, 22, 20]
|
||
|
||
|
||
def test_decode_tran_initial_with_rle():
|
||
"""A synthetic body with 00 NN RLE block runs the current Tran value forward."""
|
||
# T[0]=5, T[1]=5, then 00 08 RLE block = 8 zero deltas → T[2..9] = 5
|
||
body = b"\x00\x02\x00\x00\x05\x00\x05" + b"\x00\x08"
|
||
decoded = decode_tran_initial(body)
|
||
assert decoded == [5, 5, 5, 5, 5, 5, 5, 5, 5, 5]
|
||
|
||
|
||
def test_decode_tran_initial_full_segment_silent_events():
|
||
"""For events with near-silent Tran, segment 0 (~482-510 samples) decodes fully."""
|
||
for path, _, _ in TRAN_INITIAL_FIXTURES[3:]: # JQ0 (Vert-heavy) and V70 (Mic-heavy)
|
||
if not os.path.exists(path):
|
||
pytest.skip(f"fixture missing: {path}")
|
||
with open(path, "rb") as f:
|
||
body = f.read()[43:-26]
|
||
truth = _full_truth(path)
|
||
decoded = decode_tran_initial(body)
|
||
assert decoded is not None
|
||
# The decoder should produce a clean run of samples; check ALL of them
|
||
# match truth (segment 0 is fully solved for events where T is near zero).
|
||
n = len(decoded)
|
||
for i in range(n):
|
||
assert decoded[i] == truth[i], (
|
||
f"{os.path.basename(path)}: sample {i}: decoded={decoded[i]} truth={truth[i]}"
|
||
)
|
||
# And we should have decoded at least 400 samples (= segment 0 worth).
|
||
assert n >= 400, f"only {n} samples decoded for {path}"
|
||
|
||
|
||
# ── ADC scaling + dB conversion ──────────────────────────────────────────────
|
||
|
||
|
||
def test_decoded_to_adc_counts_geo_scales_by_16():
|
||
"""Geo channels in decoder units (16-count) should multiply by 16 to ADC."""
|
||
decoded = {"Tran": [0, 1, -2, 100], "Vert": [5], "Long": [-10], "MicL": [813]}
|
||
adc = decoded_to_adc_counts(decoded)
|
||
assert adc["Tran"] == [0, 16, -32, 1600]
|
||
assert adc["Vert"] == [80]
|
||
assert adc["Long"] == [-160]
|
||
# Mic passes through unchanged (already ADC counts).
|
||
assert adc["MicL"] == [813]
|
||
|
||
|
||
def test_decoded_to_adc_counts_empty():
|
||
assert decoded_to_adc_counts({}) == {}
|
||
assert decoded_to_adc_counts(
|
||
{"Tran": [], "Vert": [], "Long": [], "MicL": []}
|
||
) == {"Tran": [], "Vert": [], "Long": [], "MicL": []}
|
||
|
||
|
||
def test_mic_count_to_db_zero_is_zero():
|
||
assert mic_count_to_db(0) == 0.0
|
||
|
||
|
||
def test_mic_count_to_db_unit_is_reference():
|
||
"""count = ±1 → ±81.94 dB (the calibration reference)."""
|
||
assert abs(mic_count_to_db(1) - 81.94) < 0.01
|
||
assert abs(mic_count_to_db(-1) - (-81.94)) < 0.01
|
||
|
||
|
||
def test_mic_count_to_db_doubles_every_6db():
|
||
"""Each doubling of |count| adds ~6.02 dB."""
|
||
# count=2 → 87.96 dB (+ 6.02 from 81.94)
|
||
assert abs(mic_count_to_db(2) - 87.96) < 0.05
|
||
# count=4 → 93.98 dB
|
||
assert abs(mic_count_to_db(4) - 93.98) < 0.05
|
||
# count=8 → 100.00 dB
|
||
assert abs(mic_count_to_db(8) - 100.00) < 0.05
|
||
|
||
|
||
def test_mic_count_to_db_v70_peak():
|
||
"""V70 mic peak count 813 → 140.14 dB (matches BW reported PSPL 140.1)."""
|
||
assert abs(mic_count_to_db(813) - 140.14) < 0.1
|
||
# And the negative-direction equivalent
|
||
assert abs(mic_count_to_db(-813) - (-140.14)) < 0.1
|
||
|
||
|
||
# ── End-to-end: decode_a5_frames (production entry point) ───────────────────
|
||
|
||
|
||
def test_decode_a5_frames_empty():
|
||
from minimateplus.waveform_codec import decode_a5_frames
|
||
assert decode_a5_frames([]) is None
|
||
assert decode_a5_frames(None) is None
|
||
|
||
|
||
# ── Wide-NN RLE, wide 30 NN, and variable-width segment headers ──────────────
|
||
#
|
||
# Three framing cases discovered 2026-08-25 by diffing 75 production events
|
||
# against their preserved Blastware ASCII exports. Each caused ``walk_body``
|
||
# to hit its ``else: break`` mid-stream, truncating every channel decoded
|
||
# after that point (see CHANGELOG v0.25.1).
|
||
|
||
_PREAMBLE = b"\x00\x02\x00\x00\x00\x00\x00" # magic + Tran[0]=0, Tran[1]=0
|
||
_STOP = b"\xff\xff" # unrecognised tag → walker stops
|
||
|
||
|
||
def _synth(*chunks: bytes) -> bytes:
|
||
return _PREAMBLE + b"".join(chunks) + _STOP
|
||
|
||
|
||
def test_walk_body_wide_rle_block():
|
||
"""``0X NN`` is a 12-bit-NN RLE run (NN = ((t0 & 0x0F) << 8) | t1).
|
||
|
||
Observed as ``01 0c`` (NN=268) in BE9558/K558LKOF.460W and five other
|
||
production events. A narrow ``00 NN`` maxes out at NN=0xFC, so runs
|
||
longer than 252 samples must use the wide form.
|
||
"""
|
||
blocks = walk_body(_synth(b"\x01\x0c"))
|
||
assert len(blocks) == 1
|
||
assert (blocks[0].tag_hi, blocks[0].tag_lo) == (0x01, 0x0C)
|
||
assert blocks[0].length == 2
|
||
|
||
|
||
# NOTE (2026-08-25): the four tests below assert the SUPERSEDED tag-dispatch
|
||
# model — `40 NN` as a variable-width segment header, tagless headers, channel
|
||
# from rotation. The body format is really a chain of self-delimiting records
|
||
# (see the record-chain section of waveform_codec.py), so `decode_waveform_v2`
|
||
# no longer behaves this way. They are retargeted at `decode_waveform_legacy`,
|
||
# which still implements the old model and is pinned by micromate/idf_file.py
|
||
# for Thor IDFW bodies.
|
||
|
||
|
||
def test_decode_wide_rle_repeats_full_run():
|
||
"""A wide RLE run repeats the running value NN times, not NN & 0xFF."""
|
||
decoded = decode_waveform_legacy(_synth(b"\x01\x0c"))
|
||
# 2 preamble anchors + 268 repeats
|
||
assert len(decoded["Tran"]) == 2 + 268
|
||
assert set(decoded["Tran"]) == {0}
|
||
|
||
|
||
def test_walk_body_30_block_nn_above_16():
|
||
"""``30 NN`` data blocks are not capped at NN=0x10.
|
||
|
||
``30 18`` (NN=24) appears in BE18193/T193LQ45.NN0W; the old
|
||
``0 < t1 <= 0x10`` guard rejected it and stopped the walk 1033 bytes
|
||
into a 4877-byte body. Length is still NN * 1.5 + 2.
|
||
"""
|
||
payload = bytes(36) # 24 deltas × 1.5 bytes
|
||
blocks = walk_body(_synth(b"\x30\x18" + payload, b"\x00\x04"))
|
||
assert [b.length for b in blocks] == [38, 2]
|
||
|
||
|
||
@pytest.mark.parametrize("nn,hdr_len", [(1, 18), (2, 20), (3, 22)])
|
||
def test_walk_body_segment_header_width_follows_tag_lo(nn, hdr_len):
|
||
"""``40 NN``: NN is the count of previous-channel continuation deltas.
|
||
|
||
Header length = 2 * NN + 16. ``40 02`` (the only form previously
|
||
handled) is the NN=2 case at 20 bytes; ``40 01`` (18) and ``40 03``
|
||
(22) both occur in production files.
|
||
"""
|
||
data = bytearray(hdr_len - 2)
|
||
data[2 * nn + 8 : 2 * nn + 10] = b"\x02\x00" # constant marker
|
||
blocks = walk_body(_synth(bytes([0x40, nn]) + bytes(data), b"\x00\x04"))
|
||
assert [b.length for b in blocks] == [hdr_len, 2]
|
||
|
||
|
||
@pytest.mark.parametrize("nn,hdr_len", [(1, 18), (2, 20), (3, 22)])
|
||
def test_segment_header_anchors_track_header_width(nn, hdr_len):
|
||
"""Anchor pair sits at data[2*NN+10 : 2*NN+14] regardless of width."""
|
||
data = bytearray(hdr_len - 2)
|
||
data[2 * nn + 8 : 2 * nn + 10] = b"\x02\x00"
|
||
data[2 * nn + 10 : 2 * nn + 12] = (7).to_bytes(2, "big") # anchor 0
|
||
data[2 * nn + 12 : 2 * nn + 14] = (9).to_bytes(2, "big") # anchor 1
|
||
decoded = decode_waveform_legacy(_synth(bytes([0x40, nn]) + bytes(data)))
|
||
assert decoded["Vert"][:2] == [7, 9]
|
||
|
||
|
||
# ── Tagless segment headers ─────────────────────────────────────────────────
|
||
#
|
||
# A segment header can appear WITHOUT its ``40 NN`` tag: just the 14-byte tail
|
||
# ``[field2:2][len:2][channel_id:4][marker:2][anchors:4]``. This is the NN=0
|
||
# case — no continuation deltas for the previous channel, so no tag and no
|
||
# delta bytes. Found 2026-08-25: it is where the walk stopped in 7 of the 8
|
||
# remaining truncating production events.
|
||
#
|
||
# The channel_id field (previously mis-labelled a "monotonic counter") is
|
||
# ``[channel][00][00][segment_index]`` with 0x46=Tran 0x47=Vert 0x48=Long
|
||
# 0x49=MicL — verified on 1697 of 1697 segment headers across the ground-truth
|
||
# corpus, zero disagreements.
|
||
|
||
def _tagless(chan_id=0x47, seg=2, marker=b"\x02\x00", a0=0, a1=0):
|
||
return (b"\x5d\xee" + b"\x00\xd0" + bytes([chan_id, 0, 0, seg]) + marker
|
||
+ a0.to_bytes(2, "big", signed=True) + a1.to_bytes(2, "big", signed=True))
|
||
|
||
|
||
def test_walk_body_accepts_tagless_segment_header():
|
||
"""A bare 14-byte header is walked as a segment block, not a stop."""
|
||
blocks = walk_body(_synth(b"\x10\x04\x00\x00", _tagless(), b"\x00\x04"))
|
||
kinds = [(b.tag_hi, b.tag_lo, b.length) for b in blocks]
|
||
assert kinds == [(0x10, 0x04, 4), (0x40, 0x00, 14), (0x00, 0x04, 2)]
|
||
|
||
|
||
def test_tagless_header_carries_full_14_bytes_as_data():
|
||
"""The synthetic block's data includes the leading bytes (there is no tag
|
||
to strip), so decode_waveform_v2's ``2*nd + k`` offsets line up at nd=0."""
|
||
blocks = walk_body(_synth(_tagless()))
|
||
hdr = next(b for b in blocks if b.tag_hi == 0x40)
|
||
assert len(hdr.data) == 14
|
||
assert hdr.data[8:10] == b"\x02\x00" # marker at 2*0 + 8
|
||
|
||
|
||
def test_tagless_header_anchors_and_channel_id():
|
||
"""Anchors decode from data[10:14]; the channel comes from the id byte."""
|
||
decoded = decode_waveform_legacy(_synth(_tagless(chan_id=0x48, a0=11, a1=13)))
|
||
assert decoded["Long"][:2] == [11, 13] # 0x48 → Long, not rotation
|
||
assert decoded["Vert"] == []
|
||
|
||
|
||
@pytest.mark.parametrize("chan_id,name",
|
||
[(0x46, "Tran"), (0x47, "Vert"), (0x48, "Long"), (0x49, "MicL")])
|
||
def test_segment_channel_comes_from_id_not_rotation(chan_id, name):
|
||
"""Channel is taken from the header's id byte. Two headers in a row for
|
||
the SAME channel must both land on that channel — rotation-by-position
|
||
would put the second one on the next channel and corrupt both."""
|
||
body = _synth(_tagless(chan_id=chan_id, seg=1, a0=5, a1=6),
|
||
_tagless(chan_id=chan_id, seg=2, a0=7, a1=8))
|
||
decoded = decode_waveform_legacy(body)
|
||
# Tran additionally carries the body preamble's 2 anchors (both 0 here).
|
||
expected = [0, 0, 5, 6, 7, 8] if name == "Tran" else [5, 6, 7, 8]
|
||
assert decoded[name] == expected
|
||
for other in ("Tran", "Vert", "Long", "MicL"):
|
||
if other != name:
|
||
assert decoded[other] == ([0, 0] if other == "Tran" else [])
|
||
|
||
|
||
# ── Record-chain body model (2026-08-25) ────────────────────────────────────
|
||
#
|
||
# The body is a chain of self-delimiting per-channel records, not a flat
|
||
# tag-dispatch stream. Verified over 1,388 production series-3 waveform
|
||
# binaries: the chain terminates on a 0x06 record in 1,387 of them and all
|
||
# four channels come out at identical length in 1,388/1,388 (was 156/1,388).
|
||
# Against the 75 events with a preserved Blastware ASCII export: sample-count
|
||
# exact 72/75 -> 75/75, fully exact 70/75 -> 73/75.
|
||
|
||
from minimateplus.waveform_codec import ( # noqa: E402
|
||
CHANNEL_IDS,
|
||
MODE_ABSOLUTE,
|
||
MODE_DELTA,
|
||
MODE_RAW12,
|
||
STREAM_END_ID,
|
||
data_block_len,
|
||
find_first_record,
|
||
is_record,
|
||
unpack12,
|
||
walk_records,
|
||
)
|
||
|
||
|
||
def _rec(chan_id, mode, payload, seg=0, field2=b"\x00\x00", anchors=None):
|
||
"""Build one self-delimiting record."""
|
||
head = bytearray()
|
||
head += bytes([chan_id, 0x00, 0x00, seg])
|
||
head += bytes(mode)
|
||
if anchors is not None:
|
||
for a in anchors:
|
||
head += int(a).to_bytes(2, "big", signed=True)
|
||
body = bytes(head) + payload
|
||
return field2 + (len(body) + 2).to_bytes(2, "big") + body
|
||
|
||
|
||
def _terminator():
|
||
return b"\x00\x00" + (8).to_bytes(2, "big") + bytes([STREAM_END_ID, 0, 0, 0, 0, 0])
|
||
|
||
|
||
def _body(*records, preamble=b"\x00\x02\x00", seg0=b"\x00\x00\x00\x00"):
|
||
return preamble + seg0 + b"".join(records) + _terminator()
|
||
|
||
|
||
def test_forty_nn_is_a_data_block_not_a_segment_header():
|
||
"""`40 NN` is an int16 BE data block of length 2*NN + 2.
|
||
|
||
The superseded model read it as a segment header of length 2*NN + 16,
|
||
which is what made walks drift and channels come out unequal.
|
||
"""
|
||
assert data_block_len(b"\x40\x02\x00\x01\x00\x02", 0) == (6, 2)
|
||
assert data_block_len(b"\x40\x08" + bytes(16), 0) == (18, 8)
|
||
# NN > 8 is not a data block
|
||
assert data_block_len(b"\x40\x0c" + bytes(24), 0) == (None, None)
|
||
|
||
|
||
def test_record_chain_is_followed_by_length_not_by_tag_sniffing():
|
||
payload = b"\x00\x04" # RLE hold x4
|
||
body = _body(_rec(0x47, MODE_DELTA, payload, anchors=(3, 5)))
|
||
recs = walk_records(body)
|
||
assert len(recs) == 1
|
||
assert recs[0]["channel"] == "Vert"
|
||
assert recs[0]["mode"] == MODE_DELTA
|
||
|
||
|
||
def test_chain_terminates_on_channel_id_06():
|
||
body = _body(_rec(0x47, MODE_DELTA, b"\x00\x04", anchors=(1, 1)),
|
||
_rec(0x48, MODE_DELTA, b"\x00\x04", anchors=(2, 2)))
|
||
assert [r["channel"] for r in walk_records(body)] == ["Vert", "Long"]
|
||
assert not is_record(body, len(body) - 10) # the terminator is not a record
|
||
|
||
|
||
def test_mode_delta_emits_anchors_then_accumulates():
|
||
# two anchors, then an int8 block of +1,+1,+1,+1
|
||
body = _body(_rec(0x47, MODE_DELTA, b"\x20\x04\x01\x01\x01\x01",
|
||
anchors=(10, 11)))
|
||
d = decode_waveform_v2(body)
|
||
assert d["Vert"] == [10, 11, 12, 13, 14, 15]
|
||
|
||
|
||
def test_mode_absolute_replaces_rather_than_accumulates():
|
||
"""mode `01 00`: no anchors, and block values are ABSOLUTE samples."""
|
||
body = _body(_rec(0x48, MODE_ABSOLUTE, b"\x20\x04\x0a\x0b\x0c\x0d"))
|
||
d = decode_waveform_v2(body)
|
||
assert d["Long"] == [10, 11, 12, 13], "01 00 blocks are absolute, not deltas"
|
||
|
||
|
||
def test_mode_absolute_rle_holds_the_previous_value():
|
||
body = _body(_rec(0x48, MODE_ABSOLUTE, b"\x20\x04\x07\x07\x07\x07\x00\x04"))
|
||
d = decode_waveform_v2(body)
|
||
assert d["Long"] == [7, 7, 7, 7, 7, 7, 7, 7]
|
||
|
||
|
||
def test_mode_raw12_has_no_tags_at_all():
|
||
"""mode `00 03`: the whole data section is raw 12-bit absolute samples.
|
||
|
||
Decoding these matters for the time base — skipping the record would
|
||
displace every later sample on that channel (observed on
|
||
BE9558/K558LOF2.820W, MicL shifted by exactly 512).
|
||
"""
|
||
packed = bytes([0x01, 0x23, 0x04, 0x05, 0x06, 0x07]) # 4 samples
|
||
body = _body(_rec(0x49, MODE_RAW12, packed))
|
||
d = decode_waveform_v2(body)
|
||
assert d["MicL"] == unpack12(packed)
|
||
assert len(d["MicL"]) == 4
|
||
|
||
|
||
def test_unpack12_sign_extends():
|
||
assert unpack12(bytes([0x00, 0x00, 0x01, 0x02, 0x03, 0x04])) == [1, 2, 3, 4]
|
||
# high nibble 0x8 -> negative
|
||
assert unpack12(bytes([0x80, 0x00, 0x00, 0x00, 0x00, 0x00]))[0] == -2048
|
||
|
||
|
||
def test_channel_comes_from_the_record_id():
|
||
for cid, name in CHANNEL_IDS.items():
|
||
body = _body(_rec(cid, MODE_ABSOLUTE, b"\x20\x04\x01\x02\x03\x04"))
|
||
d = decode_waveform_v2(body)
|
||
assert d[name][-4:] == [1, 2, 3, 4], f"{name} misrouted"
|
||
|
||
|
||
def test_raw12_preamble_is_scanned_not_block_walked():
|
||
"""A `00 00 03` preamble carries raw 12-bit data from body[3] with no tags,
|
||
so find_first_record must scan rather than block-walk. One production file
|
||
has this (BE13121/O121L4L1.KF0W); block-walking returns None on it."""
|
||
packed = bytes([0x00, 0x00, 0x01, 0x02, 0x03, 0x04])
|
||
body = _body(_rec(0x47, MODE_DELTA, b"\x00\x04", anchors=(1, 1)),
|
||
preamble=b"\x00\x00\x03", seg0=packed)
|
||
assert find_first_record(body) == 3 + len(packed)
|
||
d = decode_waveform_v2(body)
|
||
assert d["Tran"] == unpack12(packed)
|
||
|
||
|
||
def test_returns_none_when_no_record_chain():
|
||
assert decode_waveform_v2(b"\x00\x02\x00" + bytes(40)) is None
|
||
assert decode_waveform_v2(b"") is None
|