serversdownandClaude Opus 5
686ab6e7a6
Two independent bugs, both found by diffing 75 production events against
their preserved Blastware ASCII exports (<store>/<serial>/<file>_ASCII.TXT).
1. Geo full scale was wrong — every geophone reading was 2.34% low.
The codec emits geo samples in 16-count units with a documented LSB of
exactly 0.005 in/s, and decoded_to_adc_counts multiplies by 16, so one
ADC count is 0.005/16 in/s and 10.000 in/s is 10.0/(0.005/16) = 32000
counts. sfm/event_hdf5.py and minimateplus/event_file_io.py both
divided by 32768 (2^15), scaling every sample and derived peak down by
1 - 32000/32768. The error scales with amplitude, so it was invisible
on quiet events and worst on the loud ones that matter for compliance.
Mic is unaffected (it back-solves its scale from the device peak).
216 per-channel comparisons: 32768 -> 151/216 exact, worst error 0.238
in/s on a 10 in/s event; 32000 -> 216/216 exact, worst 0.005 = 1 LSB.
2. walk_body silently truncated channels on four unhandled framing cases.
An unrecognised tag ends the walk and decode_waveform_v2 returns
whatever it got, so this surfaced as short channels, never an error:
- wide-NN RLE `0X NN` (runs longer than 252 samples)
- `30 NN` with NN > 0x10 (the old cap was arbitrary)
- variable-width `40 NN` headers: NN counts previous-channel
continuation deltas, so the header is 2*NN + 16 bytes; `40 01`
and `40 03` occur alongside `40 02`
- tagless segment headers: no `40 NN` tag at all, just the 14-byte
tail [field2:2][len:2][channel_id:4][marker:2][anchors:4]
Also: the header field documented as a "monotonic uint32 LE counter" is
really [channel_id][00][00][segment_index], with 0x46=Tran 0x47=Vert
0x48=Long 0x49=MicL — verified on 1697/1697 segment headers, zero
disagreements. decode_waveform_v2 now takes the channel from that field
instead of rotation position, which was fragile: one missed header
desynced every channel after it.
parse_segment_header now returns n_prev_deltas/prev_deltas/marker/
anchors/channel/segment_index; the old fixed_pattern (02 00 00 01)
conflated the 2-byte marker with the first anchor.
Ground-truth corpus, end to end through the production path:
exact 37 -> 72, truncated 23 -> 3, full-length value errors 15 -> 0.
Store-wide, 729 of 1388 series-3 waveform events decode differently and
728 gain samples; the scale fix changes float values on all of them, so
stored .h5 files need regenerating.
Still open: 3 events truncate at a header variant with a variable-width
prefix (2/4/6 bytes) before the channel id and an `01 00` marker.
Documented in docs/instantel_protocol_reference.md with byte offsets.
+20 tests. No regressions: the byte-exact fixture suite still passes and
the full-suite failure list is unchanged from baseline (16 pre-existing
failures from gitignored fixtures).
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HgTe8CamXAHcAmaQ6QNcog
652 lines
28 KiB
Python
652 lines
28 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_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
|
||
|
||
|
||
def test_decode_wide_rle_repeats_full_run():
|
||
"""A wide RLE run repeats the running value NN times, not NN & 0xFF."""
|
||
decoded = decode_waveform_v2(_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_v2(_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_v2(_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_v2(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 [])
|