serversdownandClaude Opus 5
904522a9c5
data_block_len() rejected any `40 NN` block with NN > 0x08. That guard had
no evidence behind it: every corpus available when it was written used only
NN in {1,2,3,4,8}, so it was never exercised. Loud UM12947 events use NN of
12, 16, 20 ... up to 196.
Because walk_body/run stop at the first unrecognised tag rather than
raising, rejecting those blocks surfaced as silently short channels -- e.g.
Tran 1812 / Vert 2132 / Long 2324 on a file whose export carries 2324 for
all three. The real bound is the buffer; the caller additionally clamps to
the record end.
Verified against Thor's own CSV exports for UM12947 (2025-07-14 .. 09-25,
167 waveforms, supplied as CSV.zip):
length mismatches 22 -> 0
per-sample exact 1,476,242 / 1,476,249
These are NOT truncated recordings, which was the competing hypothesis --
the exports carry the full sample count.
tests/test_waveform_codec.py asserted the cap as intended behaviour. That
assertion encoded an assumption, not a verified fact, and is replaced with
one pinning the opposite plus the evidence.
Across all three ground-truth corpora: 459 waveform files,
3,807,158 / 3,807,165 samples exact. Production IDFW is now 575/575 with
zero truncations and zero decode failures (median PPV error -0.0007% across
8 units). Series-3 re-verified unchanged at 14,338/14,338.
The 7 residual samples each differ by one 4th-decimal tick and are Thor's
own rounding: intersecting the per-sample rounding constraints over that
corpus is infeasible (binding pair contradict by 2.3e-11, 7e-5 relative),
so no single linear LSB reproduces every printed value. _GEO_LSB_IPS is
already pinned to ~1e-11; do not retune it to chase these.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ru8Lg9HkkYvX9VWWo65SmL
817 lines
34 KiB
Python
817 lines
34 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)
|
||
|
||
|
||
def test_forty_nn_is_not_capped_at_eight():
|
||
"""NN > 8 is a perfectly ordinary `40 NN` block.
|
||
|
||
This test previously asserted the opposite (`40 0c` -> (None, None)),
|
||
codifying a guard that had no evidence behind it: the only corpora
|
||
available then used NN in {1,2,3,4,8}, so the cap was never exercised.
|
||
Loud UM12947 events use NN of 12, 16, 20 ... up to 196, and rejecting
|
||
them halted the block walk mid-record — surfacing as silently short
|
||
channels, since the walker stops at the first unrecognised tag rather
|
||
than raising. Lifting the cap took that corpus from 22 length-mismatched
|
||
files to 0, and 1,476,242 of 1,476,249 samples now reproduce Thor's own
|
||
CSV export exactly (the 7 stragglers differ by one 4th-decimal tick).
|
||
Verified 2026-09-11; see docs/idf_protocol_reference.md.
|
||
"""
|
||
assert data_block_len(b"\x40\x0c" + bytes(24), 0) == (26, 12)
|
||
assert data_block_len(b"\x40\xc4" + bytes(392), 0) == (394, 196)
|
||
# The real bound is the buffer: a block that cannot fit is not a block.
|
||
assert data_block_len(b"\x40\xc4" + bytes(8), 0) == (None, None)
|
||
assert data_block_len(b"\x40\x00" + bytes(8), 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
|