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
912 lines
40 KiB
Python
912 lines
40 KiB
Python
"""
|
||
waveform_codec.py — block-walker and verified decoder for the MiniMate Plus
|
||
waveform-file body.
|
||
|
||
FULLY DECODED 2026-05-11. Every block type, every channel, and the
|
||
channel-rotation rule are verified byte-exact against BW's ASCII export
|
||
across the 9-event fixture bundle (47,364 ADC samples, zero errors).
|
||
|
||
The Blastware waveform-file body — the bytes between the 21-byte STRT
|
||
record and the 26-byte file footer — is a tagged variable-length block
|
||
stream with a custom delta + RLE codec. (Not raw int16 LE, which was
|
||
the historical wrong assumption that produced ±32K noise on every event.)
|
||
|
||
Current status:
|
||
|
||
- Block framing: ✅ solved (5 block types and lengths all confirmed)
|
||
- Per-channel decode: ✅ solved (Tran / Vert / Long / MicL all byte-exact)
|
||
- Channel rotation: ✅ Tran → Vert → Long → MicL per segment
|
||
- Segment header: ✅ fully decoded (anchor pair + prev-channel extension)
|
||
- 30 NN packed-delta block: ✅ NN × 12-bit signed deltas in NN/4 groups
|
||
- MicL → dB(L) conversion: ✅ ``mic_count_to_db`` matches BW display
|
||
- Production wiring: ✅ ``client.py:_decode_a5_waveform`` uses the new
|
||
codec (via ``decode_a5_frames``). ``.h5`` sidecars now render
|
||
correctly.
|
||
|
||
Known limitations:
|
||
|
||
- Walker stops early on the loudest events (SP0, SS0, SV0, event-b) at
|
||
some mid-segment edge cases not yet fully characterized. Every
|
||
sample reached IS correct; the walker just doesn't reach all of
|
||
them yet. The cleanly-decoded subset is still ~5000–15000 samples
|
||
per loud event.
|
||
|
||
────────────────────────────────────────────────────────────────────────────
|
||
Body layout (CONFIRMED 2026-05-11 against 8 fixture events)
|
||
────────────────────────────────────────────────────────────────────────────
|
||
|
||
[7-byte preamble] [stream of tagged blocks] [trailer]
|
||
|
||
The preamble is always exactly 7 bytes:
|
||
|
||
body[0:3] = 00 02 00 magic
|
||
body[3:5] = Tran[0] int16 BE in 16-count units (LSB = 0.005 in/s)
|
||
body[5:7] = Tran[1] int16 BE in 16-count units
|
||
|
||
(Earlier drafts of this module described a "7-or-9-byte preamble";
|
||
that was wrong — single-shot and continuous events both use 7 bytes.
|
||
The "extra 2 bytes" on continuous events were the first ``00 NN`` RLE
|
||
marker, not part of the preamble.)
|
||
|
||
Block types and lengths (all confirmed):
|
||
|
||
| Tag | Length | Meaning |
|
||
|----------|-----------------------|----------------------------------------|
|
||
| ``10 NN``| NN/2 + 2 bytes | 4-bit nibble deltas (2 per byte; high |
|
||
| | | nibble first; signed 0..7 / 8..F = -8..-1)|
|
||
| ``20 NN``| NN + 2 bytes | int8 signed deltas (1 per byte) |
|
||
| ``00 NN``| 2 bytes | RLE: append NN copies of current value |
|
||
| ``30 NN``| NN*2 in data, NN*4 | Unknown content. Only in loud events. |
|
||
| | in trailer | |
|
||
| ``40 02``| 20 bytes (fixed) | Segment header |
|
||
|
||
NN is always a multiple of 4.
|
||
|
||
────────────────────────────────────────────────────────────────────────────
|
||
Tran channel, segment 0 (CONFIRMED 2026-05-11)
|
||
────────────────────────────────────────────────────────────────────────────
|
||
|
||
Segment 0 — everything before the first ``40 02`` segment header — encodes
|
||
Tran samples only. Starting from preamble anchors Tran[0] and Tran[1],
|
||
each subsequent block contributes to the running Tran value:
|
||
|
||
10 NN → append NN deltas (4-bit signed nibbles)
|
||
20 NN → append NN deltas (int8 signed bytes)
|
||
00 NN → append NN copies of the current value (RLE zeros)
|
||
40 02 → segment 0 ends; multi-segment continuation is open
|
||
|
||
This decodes the first 482–510 samples of Tran for each event with zero
|
||
errors against BW's ASCII export. The exact segment-0 sample count
|
||
varies per event (it's bounded by a fixed device-flash byte budget, not
|
||
a fixed sample count — quiet events fit more samples because zero
|
||
deltas pack into ``00 NN`` markers compactly).
|
||
|
||
Implementation: :func:`decode_tran_initial`.
|
||
|
||
────────────────────────────────────────────────────────────────────────────
|
||
Segment header (40 02, 20 bytes total)
|
||
────────────────────────────────────────────────────────────────────────────
|
||
|
||
The 18-byte payload of the ``40 02`` block:
|
||
|
||
| Offset | Field | Status |
|
||
|-----------|---------------------------------------------|-------------|
|
||
| [0:2] | T_delta at first sample of new segment | ✅ confirmed|
|
||
| | (int16 BE, in 16-count units) | |
|
||
| [2:4] | Likely T_delta at sample seg_start+1 | 🟡 likely |
|
||
| [4:6] | Unknown (varies; possibly checksum) | ❓ open |
|
||
| [6:8] | Byte length to next segment header − 2 | ✅ confirmed|
|
||
| | (uint16 BE; useful for walker pre-scan) | |
|
||
| [8:12] | Monotonic uint32 LE counter | ✅ confirmed|
|
||
| | (starts ~0x47, increments by 1 per segment) | |
|
||
| [12:14] | Constant ``02 00`` | ✅ confirmed|
|
||
| [14:18] | Unknown 4-byte field | ❓ open |
|
||
|
||
────────────────────────────────────────────────────────────────────────────
|
||
What breaks the multi-segment decoder (the main open question)
|
||
────────────────────────────────────────────────────────────────────────────
|
||
|
||
After segment 0 ends and the segment header T_delta is consumed,
|
||
applying segment 1's blocks as Tran continuation produces values that
|
||
diverge from truth by sample ~512. The block structure inside segment
|
||
1 is IDENTICAL to segment 0 (same alternating 10 NN / 00 NN pattern),
|
||
and the delta budget matches the segment size exactly (V70 segment 1
|
||
has 264 nibble-deltas + 244 RLE zeros = 508 = the segment's sample
|
||
count). But the cumulative is wrong.
|
||
|
||
The strongest unverified hypothesis is that segments rotate channels:
|
||
|
||
segment 0 → Tran samples 0..509
|
||
segment 1 → Vert samples 0..507
|
||
segment 2 → Long samples 0..507
|
||
segment 3 → Mic samples 0..507
|
||
segment 4 → Tran samples 510..N (continuation)
|
||
...
|
||
|
||
This is consistent with the segment-1 block sums net-to-near-zero in
|
||
V70 (where all 4 channels are near zero) and with the per-segment delta
|
||
budget matching the segment size for a single channel. It is NOT yet
|
||
verified because the per-segment channel anchor isn't pinned down in
|
||
the segment header — bytes [4:6] and [14:18] of the header are still
|
||
open and probably encode V/L/M anchors.
|
||
|
||
See ``docs/waveform_codec_re_status.md`` for the current working notes
|
||
and the suggested next experiment ("segment-channel scoring analyzer").
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from dataclasses import dataclass
|
||
from typing import List, Optional, Tuple
|
||
|
||
|
||
@dataclass
|
||
class WaveformBlock:
|
||
"""One tagged block parsed out of a Blastware waveform-file body."""
|
||
offset: int # byte offset into body
|
||
tag_hi: int # first tag byte (0x10 / 0x20 / 0x00 / 0x30 / 0x40)
|
||
tag_lo: int # second tag byte (NN)
|
||
data: bytes # block payload (excludes the 2-byte tag)
|
||
length: int # total block length on the wire (includes the tag)
|
||
|
||
@property
|
||
def kind(self) -> str:
|
||
return f"{self.tag_hi:02x} {self.tag_lo:02x}"
|
||
|
||
|
||
def find_data_start(body: bytes) -> int:
|
||
"""Auto-detect the offset of the first data block.
|
||
|
||
The body starts with a 7-byte preamble (magic ``00 02 00`` + two int16 BE
|
||
Tran anchors). After that, the data section starts with a tag — usually
|
||
``10 NN`` or ``20 NN``, but quiet events may begin with a ``00 NN`` RLE
|
||
marker. We return the offset of the first recognized tag.
|
||
"""
|
||
# Try fixed offset 7 first (canonical preamble length).
|
||
if len(body) >= 9:
|
||
b, nn = body[7], body[8]
|
||
# Accept the same tag vocabulary ``walk_body`` accepts, including the
|
||
# wide-NN forms (``0X``/``1X``/``2X``) and the variable-width ``40 NN``
|
||
# segment header.
|
||
if ((b & 0xF0) in (0x00, 0x10, 0x20) and nn % 4 == 0
|
||
and ((b & 0x0F) != 0 or 0 < nn <= 0xFC)) \
|
||
or (b == 0x30 and nn % 4 == 0 and 0 < nn <= 0xFC) \
|
||
or (b == 0x40 and 0 < nn <= 0x08) \
|
||
or is_tagless_segment_header(body, 7):
|
||
return 7
|
||
# Fall back to scanning the first 20 bytes.
|
||
for i in range(min(20, len(body) - 1)):
|
||
b = body[i]
|
||
nn = body[i + 1]
|
||
if b in (0x10, 0x20) and nn % 4 == 0 and 0 < nn <= 0xFC:
|
||
return i
|
||
return -1
|
||
|
||
|
||
# Channel-id byte carried in every segment header. Previously mis-read as a
|
||
# "monotonic uint32 LE counter"; it is really ``[channel][00][00][segment]``.
|
||
# Verified 2026-08-25 on 1697/1697 segment headers across the ground-truth
|
||
# corpus with zero disagreements against the decoded channel rotation.
|
||
SEGMENT_CHANNEL_IDS = {0x46: "Tran", 0x47: "Vert", 0x48: "Long", 0x49: "MicL"}
|
||
|
||
# A tagless segment header: the 14-byte tail of a ``40 NN`` header with no tag
|
||
# and no previous-channel continuation deltas (the NN=0 case).
|
||
_TAGLESS_HEADER_LEN = 14
|
||
|
||
|
||
def is_tagless_segment_header(body: bytes, i: int) -> bool:
|
||
"""True if a bare 14-byte segment header starts at *i*.
|
||
|
||
Layout ``[field2:2][len_to_next:2][channel_id:4][marker:2][anchors:4]``.
|
||
The discriminator is the 6 bytes at ``[4:10]``: a known channel id, two
|
||
zero bytes, a small segment index, and the ``01 00`` / ``02 00`` marker.
|
||
"""
|
||
if i + _TAGLESS_HEADER_LEN > len(body):
|
||
return False
|
||
return (body[i + 4] in SEGMENT_CHANNEL_IDS
|
||
and body[i + 5] == 0x00 and body[i + 6] == 0x00
|
||
and body[i + 8] in (0x01, 0x02) and body[i + 9] == 0x00)
|
||
|
||
|
||
def walk_body(body: bytes, start: Optional[int] = None) -> List[WaveformBlock]:
|
||
"""Walk the tagged-block sequence starting at *start* (auto-detected by default).
|
||
|
||
Stops when an unrecognized tag is encountered or end of body is reached.
|
||
Returned blocks are in stream order.
|
||
"""
|
||
if start is None:
|
||
start = find_data_start(body)
|
||
if start < 0:
|
||
return []
|
||
|
||
blocks: List[WaveformBlock] = []
|
||
i = start
|
||
while i + 1 < len(body):
|
||
t0 = body[i]
|
||
t1 = body[i + 1]
|
||
if t0 == 0x10 and t1 % 4 == 0 and 0 < t1 <= 0xFC:
|
||
length = t1 // 2 + 2
|
||
elif (t0 & 0xF0) == 0x10 and (t0 & 0x0F) != 0 and t1 % 4 == 0:
|
||
# Wide-NN nibble block: ``1X NN`` where X is the high nibble of a
|
||
# 12-bit NN value. NN = ((t0 & 0x0F) << 8) | t1. Block length
|
||
# = NN/2 + 2 bytes (NN nibble deltas, same as ``10 NN`` semantics
|
||
# but with NN > 0xFC). Confirmed 2026-05-11 in SP0 segment 12
|
||
# where V continuation uses ``11 90`` = NN=0x190=400.
|
||
wide_nn = ((t0 & 0x0F) << 8) | t1
|
||
length = wide_nn // 2 + 2
|
||
elif t0 == 0x20 and t1 % 4 == 0 and 0 < t1 <= 0xFC:
|
||
length = t1 + 2
|
||
elif (t0 & 0xF0) == 0x20 and (t0 & 0x0F) != 0 and t1 % 4 == 0:
|
||
# Wide-NN int8 block: ``2X NN`` extends NN to 12 bits the same way.
|
||
wide_nn = ((t0 & 0x0F) << 8) | t1
|
||
length = wide_nn + 2
|
||
elif (t0 & 0xF0) == 0x00 and t1 % 4 == 0:
|
||
# ``00 NN`` RLE zero-delta run, plus its wide form ``0X NN``
|
||
# (X != 0) which extends NN to 12 bits exactly like ``1X``/``2X``:
|
||
# NN = ((t0 & 0x0F) << 8) | t1. A narrow run maxes out at
|
||
# NN=0xFC, so quiet stretches longer than 252 samples must use
|
||
# the wide form. Confirmed 2026-08-25 against six production
|
||
# events (e.g. ``01 0c`` = 268 repeats in K558LKOF.460W).
|
||
length = 2
|
||
elif t0 == 0x30 and t1 % 4 == 0 and 0 < t1 <= 0xFC:
|
||
# Data-section ``30 NN`` blocks carry NN 12-bit signed deltas packed
|
||
# as NN/4 groups of (2-byte high-nibble field + 4 × int8 low byte).
|
||
# Length = NN/4 × 6 + 2 = NN × 1.5 + 2 (= 8 for NN=4, 14 for NN=8,
|
||
# 20 for NN=12, etc.). Confirmed 2026-05-11 by full-decoder
|
||
# verification against BW ASCII export.
|
||
#
|
||
# Trailer-section ``30 NN`` blocks have a different length formula
|
||
# (NN × 4 = 32 for NN=8 in trailers). We try the data-section
|
||
# length first and fall back to the trailer length if needed.
|
||
cand_data = t1 * 3 // 2 + 2
|
||
cand_trailer = t1 * 4
|
||
if (i + cand_data < len(body) - 1
|
||
and body[i + cand_data] in (0x10, 0x20, 0x00, 0x30, 0x40)):
|
||
length = cand_data
|
||
else:
|
||
length = cand_trailer
|
||
elif t0 == 0x40 and 0 < t1 <= 0x08:
|
||
# ``40 NN`` segment header. NN is the number of int16 BE
|
||
# continuation deltas the header carries for the PREVIOUS
|
||
# channel, so the header grows with NN:
|
||
# length = 2 (tag) + 2*NN (deltas) + 14 (fixed tail)
|
||
# ``40 02`` (20 bytes) dominates, but ``40 01`` (18) and
|
||
# ``40 03`` (22) both occur in production files. Confirmed
|
||
# 2026-08-25; the constant ``02 00`` marker moves with NN too
|
||
# (see :func:`parse_segment_header`).
|
||
length = 2 * t1 + 16
|
||
elif is_tagless_segment_header(body, i):
|
||
# Segment header with no ``40 NN`` tag (NN=0 — the previous channel
|
||
# needed no continuation deltas). Emit it as a synthetic ``40 00``
|
||
# block whose ``data`` is the whole 14-byte record, so the nd=0
|
||
# offsets in :func:`decode_waveform_v2` line up unchanged.
|
||
blocks.append(WaveformBlock(
|
||
offset=i, tag_hi=0x40, tag_lo=0x00,
|
||
data=bytes(body[i : i + _TAGLESS_HEADER_LEN]),
|
||
length=_TAGLESS_HEADER_LEN,
|
||
))
|
||
i += _TAGLESS_HEADER_LEN
|
||
continue
|
||
else:
|
||
# Unknown tag; stop. Caller can inspect ``i`` to see where.
|
||
break
|
||
|
||
if i + length > len(body):
|
||
break
|
||
|
||
data = bytes(body[i + 2 : i + length])
|
||
blocks.append(WaveformBlock(offset=i, tag_hi=t0, tag_lo=t1, data=data, length=length))
|
||
i += length
|
||
|
||
return blocks
|
||
|
||
|
||
def split_segments(blocks: List[WaveformBlock]) -> List[List[WaveformBlock]]:
|
||
"""Group consecutive blocks into segments separated by ``40 02`` headers.
|
||
|
||
The first segment is whatever runs before the first ``40 02`` header
|
||
(typically the "segment 0" preamble data after the body preamble).
|
||
Subsequent segments start with a ``40 02`` block, then have their
|
||
own data blocks until the next ``40 02``.
|
||
"""
|
||
segments: List[List[WaveformBlock]] = []
|
||
current: List[WaveformBlock] = []
|
||
for b in blocks:
|
||
if b.tag_hi == 0x40:
|
||
if current:
|
||
segments.append(current)
|
||
current = [b]
|
||
else:
|
||
current.append(b)
|
||
if current:
|
||
segments.append(current)
|
||
return segments
|
||
|
||
|
||
def parse_segment_header(block: WaveformBlock) -> Optional[dict]:
|
||
"""Decode the payload of a ``40 NN`` segment header.
|
||
|
||
NN (the tag's low byte) is the number of int16 BE continuation deltas
|
||
the header carries for the PREVIOUS channel, so every field after
|
||
those deltas shifts by ``2 * NN``. The payload is ``2 * NN + 14``
|
||
bytes. ``40 02`` is the common case; ``40 01`` and ``40 03`` also
|
||
occur in production files (confirmed 2026-08-25).
|
||
|
||
Returns a dict with the labelled fields, or None if *block* is not a
|
||
segment header or is too short.
|
||
"""
|
||
if block.tag_hi != 0x40 or block.tag_lo > 0x08:
|
||
return None
|
||
nd = block.tag_lo
|
||
if len(block.data) < 2 * nd + 14:
|
||
return None
|
||
p = block.data
|
||
counter = int.from_bytes(p[2 * nd + 4 : 2 * nd + 8], "little", signed=False)
|
||
return {
|
||
"n_prev_deltas": nd,
|
||
# ``nd`` int16 BE deltas extending the previous channel.
|
||
"prev_deltas": [
|
||
int.from_bytes(p[2 * k : 2 * k + 2], "big", signed=True)
|
||
for k in range(nd)
|
||
],
|
||
"field2": p[2 * nd : 2 * nd + 4], # 4-byte field, role unconfirmed
|
||
"counter": counter, # legacy: raw uint32 LE of the id field
|
||
"channel": SEGMENT_CHANNEL_IDS.get(p[2 * nd + 4]),
|
||
"segment_index": p[2 * nd + 7],
|
||
"marker": p[2 * nd + 8 : 2 * nd + 10], # always b"\x02\x00"
|
||
"anchors": [
|
||
int.from_bytes(p[2 * nd + 10 : 2 * nd + 12], "big", signed=True),
|
||
int.from_bytes(p[2 * nd + 12 : 2 * nd + 14], "big", signed=True),
|
||
],
|
||
}
|
||
|
||
|
||
def _s4(n: int) -> int:
|
||
"""Sign-extend a 4-bit value to signed int (0..7 → 0..7; 8..F → -8..-1)."""
|
||
return n if n < 8 else n - 16
|
||
|
||
|
||
def _i8(b: int) -> int:
|
||
"""Reinterpret an unsigned byte as signed int8."""
|
||
return b if b < 128 else b - 256
|
||
|
||
|
||
def decode_tran_initial(body: bytes) -> Optional[List[int]]:
|
||
"""
|
||
Decode the initial Tran-channel samples — VERIFIED 2026-05-11.
|
||
|
||
Returns Tran samples in **16-count units** (LSB = 0.005 in/s at Normal
|
||
range — the same quantization BW uses for its ASCII export). Returns
|
||
``None`` if the body cannot be parsed.
|
||
|
||
The decoded list extends from sample 0 through the end of segment 0
|
||
(= just before the first ``40 02`` segment header; ~510 sample-sets
|
||
for the events tested). Multi-segment decoding requires continuing
|
||
past the segment header — that's done by :func:`decode_tran_full`
|
||
when the per-segment rules are pinned down for all signal types.
|
||
|
||
Codec for segment 0 (CONFIRMED 2026-05-11 against 7 fixture events):
|
||
|
||
- Body bytes [0:3] are the magic ``00 02 00``.
|
||
- Body bytes [3:5] = ``Tran[0]`` as int16 BE in 16-count units.
|
||
- Body bytes [5:7] = ``Tran[1]`` as int16 BE in 16-count units.
|
||
- Data blocks (``10 NN`` or ``20 NN``) carry Tran deltas starting
|
||
at sample 2:
|
||
|
||
* ``10 NN``: NN nibbles = NN/2 bytes; each nibble is a 4-bit
|
||
signed delta (0..7 → 0..+7; 8..F → -8..-1). High nibble of
|
||
each byte comes first.
|
||
* ``20 NN``: NN int8 signed deltas (one delta per byte).
|
||
|
||
- ``00 NN`` blocks are run-length-encoded zero deltas: append NN
|
||
copies of the current cumulative Tran value (no change).
|
||
|
||
- ``30 NN`` blocks have not yet been decoded for content — they
|
||
appear in segment 0 of loud-from-start events (SS0, SV0) and
|
||
seem to signal a transition or special-case interpretation.
|
||
The walker steps over them but their data is ignored.
|
||
|
||
The walk stops at the first ``40 02`` segment header.
|
||
"""
|
||
if len(body) < 7 or body[0:3] != b"\x00\x02\x00":
|
||
return None
|
||
t0 = int.from_bytes(body[3:5], "big", signed=True)
|
||
t1 = int.from_bytes(body[5:7], "big", signed=True)
|
||
|
||
start = find_data_start(body)
|
||
if start < 0:
|
||
return [t0, t1]
|
||
|
||
out = [t0, t1]
|
||
cur = t1
|
||
for blk in walk_body(body, start):
|
||
if blk.tag_hi == 0x40:
|
||
# Segment boundary — stop. Multi-segment decode is decode_tran_full.
|
||
break
|
||
if blk.tag_hi == 0x10:
|
||
for byte in blk.data:
|
||
for nib in ((byte >> 4) & 0xF, byte & 0xF):
|
||
cur += _s4(nib)
|
||
out.append(cur)
|
||
elif blk.tag_hi == 0x20:
|
||
for byte in blk.data:
|
||
cur += _i8(byte)
|
||
out.append(cur)
|
||
elif blk.tag_hi == 0x00:
|
||
# RLE zero deltas: append NN copies of current Tran value.
|
||
for _ in range(blk.tag_lo):
|
||
out.append(cur)
|
||
# 30 NN: unknown content; skip.
|
||
return out
|
||
|
||
|
||
def decode_waveform_legacy(body: bytes) -> Optional[dict]:
|
||
"""
|
||
SUPERSEDED 2026-08-25 — the tag-dispatch / segment-header model.
|
||
|
||
Retained because ``micromate/idf_file.py`` trial-decodes Thor IDFW bodies
|
||
at many candidate offsets and keeps whichever yields the most samples;
|
||
the record-chain decoder returns None where this one returned garbage,
|
||
which shifts that heuristic's winner. Thor is pinned here until its own
|
||
body-offset search is reworked. Do not use for series-3.
|
||
|
||
Decode the body into per-channel sample arrays.
|
||
|
||
Status (2026-05-11 evening — channel-rotation hypothesis CONFIRMED):
|
||
segments rotate channels in fixed order **Tran → Vert → Long → MicL**.
|
||
Each channel-segment carries a 2-sample anchor pair in segment-header
|
||
bytes [14:18] (or in the body preamble for the initial Tran segment)
|
||
plus a stream of delta blocks for samples 2 onward.
|
||
|
||
Returns ``{"Tran": [...], "Vert": [...], "Long": [...], "MicL": [...]}``
|
||
with each channel's decoded samples in 16-count units (LSB = 0.005
|
||
in/s at Normal range). Returns ``None`` if the body cannot be
|
||
parsed.
|
||
"""
|
||
if len(body) < 7 or body[0:3] != b"\x00\x02\x00":
|
||
return None
|
||
|
||
channels = ["Tran", "Vert", "Long", "MicL"]
|
||
out: dict = {ch: [] for ch in channels}
|
||
|
||
# Initial Tran segment: preamble anchor pair + delta blocks before first 40 02.
|
||
t0 = int.from_bytes(body[3:5], "big", signed=True)
|
||
t1 = int.from_bytes(body[5:7], "big", signed=True)
|
||
out["Tran"].extend([t0, t1])
|
||
|
||
start = find_data_start(body)
|
||
if start < 0:
|
||
return out
|
||
|
||
blocks = walk_body(body, start)
|
||
seg_idx = [i for i, b in enumerate(blocks) if b.tag_hi == 0x40]
|
||
|
||
def apply_blocks(channel: str, anchor: int,
|
||
block_start: int, block_end: int) -> int:
|
||
"""Apply delta blocks [block_start, block_end) to *channel*'s sample
|
||
list, starting from *anchor*. Returns the final cumulative value."""
|
||
cur = anchor
|
||
for bi in range(block_start, block_end):
|
||
blk = blocks[bi]
|
||
if (blk.tag_hi & 0xF0) == 0x10:
|
||
# Both ``10 NN`` (NN ≤ 0xFC) and wide-NN ``1X NN`` (X != 0)
|
||
# are nibble-delta streams. The walker has already used the
|
||
# right length; here we just iterate the payload bytes.
|
||
for byte in blk.data:
|
||
for nib in ((byte >> 4) & 0xF, byte & 0xF):
|
||
cur += _s4(nib)
|
||
out[channel].append(cur)
|
||
elif (blk.tag_hi & 0xF0) == 0x20:
|
||
# ``20 NN`` and wide ``2X NN`` both carry int8 deltas.
|
||
for byte in blk.data:
|
||
cur += _i8(byte)
|
||
out[channel].append(cur)
|
||
elif (blk.tag_hi & 0xF0) == 0x00:
|
||
# RLE zero-delta run. Wide form ``0X NN`` carries the high
|
||
# nibble of a 12-bit NN in the tag byte, same as ``1X``/``2X``.
|
||
run = ((blk.tag_hi & 0x0F) << 8) | blk.tag_lo
|
||
for _ in range(run):
|
||
out[channel].append(cur)
|
||
elif blk.tag_hi == 0x30:
|
||
# 12-bit signed deltas, packed as NN/4 groups of 6 bytes each:
|
||
# bytes [0:2] = 16 bits = 4 × 4-bit high nibbles (MSB first)
|
||
# bytes [2:6] = 4 × int8 low bytes
|
||
# Each delta = sign_extend_12((high_nibble << 8) | low_byte).
|
||
# Confirmed 2026-05-11 against all 14 ``30 NN`` blocks in the
|
||
# bundled fixtures.
|
||
n_groups = blk.tag_lo // 4
|
||
for g in range(n_groups):
|
||
grp = blk.data[g * 6 : (g + 1) * 6]
|
||
if len(grp) < 6:
|
||
break
|
||
high_word = (grp[0] << 8) | grp[1]
|
||
for k in range(4):
|
||
nib = (high_word >> (12 - 4 * k)) & 0xF
|
||
v = (nib << 8) | grp[2 + k]
|
||
if v >= 0x800:
|
||
v -= 0x1000
|
||
cur += v
|
||
out[channel].append(cur)
|
||
# 40 02: should not occur in segment data.
|
||
return cur
|
||
|
||
# Initial Tran segment: deltas from start of body up to first 40 02 (or end).
|
||
first_seg = seg_idx[0] if seg_idx else len(blocks)
|
||
last_tran_value = apply_blocks("Tran", t1, 0, first_seg)
|
||
|
||
# Subsequent segments rotate channels. Each segment header carries:
|
||
# bytes [0:2] and [2:4] = 2 deltas extending the PREVIOUS channel
|
||
# bytes [14:16] and [16:18] = anchor pair for THIS segment's channel
|
||
#
|
||
# Rotation: V, L, M, T, V, L, M, T, ... (initial Tran segment is the
|
||
# implicit T in the cycle.)
|
||
rotation = ["Vert", "Long", "MicL", "Tran"]
|
||
# Track each channel's "running cumulative value" so we can apply the
|
||
# previous-channel extension deltas at every segment boundary.
|
||
last_value = {"Tran": last_tran_value, "Vert": None, "Long": None, "MicL": None}
|
||
|
||
prev_channel = "Tran"
|
||
for k, hi in enumerate(seg_idx):
|
||
header = blocks[hi]
|
||
# Channel comes from the header's own id byte, which is authoritative.
|
||
# The old rotation-by-position fallback is kept for headers whose id
|
||
# byte isn't one of the four known values — but a single missed or
|
||
# extra header would desync rotation and corrupt every later channel,
|
||
# which is exactly what tagless headers used to cause.
|
||
_nd = header.tag_lo
|
||
channel = None
|
||
if len(header.data) >= 2 * _nd + 8:
|
||
channel = SEGMENT_CHANNEL_IDS.get(header.data[2 * _nd + 4])
|
||
if channel is None:
|
||
channel = rotation[k % 4]
|
||
# ``40 NN``: NN int16 BE continuation deltas for the previous channel
|
||
# come first, so every later field shifts by 2*NN. NN is usually 2
|
||
# but 1 and 3 both occur (confirmed 2026-08-25).
|
||
nd = header.tag_lo
|
||
if len(header.data) < 2 * nd + 14:
|
||
continue
|
||
# Validate: real segment headers have the constant `02 00` marker
|
||
# right after the counter. Trailer/footer "40 NN" markers contain
|
||
# ASCII serial bytes or other non-header data there and would
|
||
# otherwise be mis-read as segment headers, adding spurious tail
|
||
# samples.
|
||
if header.data[2 * nd + 8 : 2 * nd + 10] != b"\x02\x00":
|
||
break
|
||
# Extend the PREVIOUS channel by NN more samples.
|
||
if last_value[prev_channel] is not None:
|
||
v = last_value[prev_channel]
|
||
for d in range(nd): # NB: not `k` — that's the segment index
|
||
v += int.from_bytes(
|
||
header.data[2 * d : 2 * d + 2], "big", signed=True
|
||
)
|
||
out[prev_channel].append(v)
|
||
last_value[prev_channel] = v
|
||
# Anchor pair for THIS segment's channel.
|
||
c0 = int.from_bytes(
|
||
header.data[2 * nd + 10 : 2 * nd + 12], "big", signed=True
|
||
)
|
||
c1 = int.from_bytes(
|
||
header.data[2 * nd + 12 : 2 * nd + 14], "big", signed=True
|
||
)
|
||
out[channel].extend([c0, c1])
|
||
# Apply delta blocks for this segment.
|
||
next_hi = seg_idx[k + 1] if k + 1 < len(seg_idx) else len(blocks)
|
||
last_value[channel] = apply_blocks(channel, c1, hi + 1, next_hi)
|
||
prev_channel = channel
|
||
|
||
return out
|
||
|
||
|
||
# ── ADC-scale conversion helpers ────────────────────────────────────────────
|
||
|
||
|
||
# Scaling factor: decode_waveform_v2 produces geo-channel samples in the BW
|
||
# display quantization (16-count units, LSB = 0.005 in/s at Normal range).
|
||
# The legacy consumer pipeline (sfm/event_hdf5.py) expects raw_samples in
|
||
# 1-count ADC units (× full_scale / 32768 → physical). To plug the new
|
||
# decoder in without rewriting consumers, multiply geo values by 16.
|
||
#
|
||
# Mic samples are already in raw ADC counts (decoded value 1 = 1 mic ADC count
|
||
# = -81.94 dB on the BW display). Mic values pass through unchanged.
|
||
_GEO_DECODER_TO_ADC = 16
|
||
|
||
|
||
def decoded_to_adc_counts(decoded: dict) -> dict:
|
||
"""Convert :func:`decode_waveform_v2` output to int16 ADC counts.
|
||
|
||
Geo channels are scaled by ×16 (decoder produces 16-count units,
|
||
consumer expects 1-count ADC). Mic is passed through as raw counts.
|
||
"""
|
||
if not decoded:
|
||
return {}
|
||
return {
|
||
"Tran": [v * _GEO_DECODER_TO_ADC for v in decoded.get("Tran", [])],
|
||
"Vert": [v * _GEO_DECODER_TO_ADC for v in decoded.get("Vert", [])],
|
||
"Long": [v * _GEO_DECODER_TO_ADC for v in decoded.get("Long", [])],
|
||
"MicL": list(decoded.get("MicL", [])),
|
||
}
|
||
|
||
|
||
def mic_count_to_db(count: int) -> float:
|
||
"""Convert a MicL ADC count to dB(L) for BW-display-compatible output.
|
||
|
||
Empirical formula (confirmed 2026-05-11 against V70 fixture: count=813
|
||
→ 140.1 dB; count=±1 → ±81.94 dB; count=±24 → ±109.5 dB):
|
||
|
||
dB = sign(count) × (81.94 + 20 × log10(|count|)) for |count| ≥ 1
|
||
dB = 0.0 for count == 0
|
||
|
||
The constant 81.94 corresponds to 10^(81.94/20) ≈ 12490 mic ADC counts
|
||
being the dB(L) reference level — almost certainly a calibration
|
||
constant from the device's mic.
|
||
"""
|
||
if count == 0:
|
||
return 0.0
|
||
sign = 1.0 if count > 0 else -1.0
|
||
return sign * (81.94 + 20.0 * math.log10(abs(count)))
|
||
|
||
|
||
# ── A5-frame entry point ────────────────────────────────────────────────────
|
||
|
||
|
||
def decode_a5_frames(a5_frames) -> Optional[dict]:
|
||
"""Decode a list of A5 (BULK_WAVEFORM_STREAM) frames into per-channel
|
||
int16 ADC samples.
|
||
|
||
Returns ``{"Tran": [...], "Vert": [...], "Long": [...], "MicL": [...]}``
|
||
with each channel's samples in **1-count ADC units** (the legacy
|
||
``event.raw_samples`` convention — multiply by ``full_scale / 32768``
|
||
to convert to physical units; for mic, use :func:`mic_count_to_db` or
|
||
a per-count psi factor).
|
||
|
||
Returns ``None`` if the frames cannot be parsed.
|
||
|
||
This is the wired-up production entry point. It:
|
||
1. Reconstructs the BW-binary body bytes from the A5 frames
|
||
(``blastware_file.extract_body_bytes``).
|
||
2. Runs the verified codec (``decode_waveform_v2``) on the body.
|
||
3. Converts to int16 ADC counts via :func:`decoded_to_adc_counts`.
|
||
"""
|
||
# Local import to avoid a cycle: blastware_file imports models and
|
||
# ultimately client.py imports waveform_codec.
|
||
from .blastware_file import extract_body_bytes
|
||
|
||
if not a5_frames:
|
||
return None
|
||
_strt, body, _footer = extract_body_bytes(a5_frames)
|
||
if not body:
|
||
return None
|
||
decoded = decode_waveform_v2(body)
|
||
if decoded is None:
|
||
return None
|
||
return decoded_to_adc_counts(decoded)
|
||
|
||
|
||
# ── Record-chain body model (CONFIRMED 2026-08-25) ──────────────────────────
|
||
#
|
||
# The body is NOT a flat tag-dispatch stream with ``40 NN`` segment headers.
|
||
# It is a chain of self-delimiting per-channel RECORDS:
|
||
#
|
||
# off+0 field2 uint16 purpose unknown (not a length, not a checksum)
|
||
# off+2 len uint16 BE next_record = off + 2 + len <- authoritative
|
||
# off+4 chan_id 0x46 Tran / 0x47 Vert / 0x48 Long / 0x49 MicL
|
||
# 0x06 = end of waveform stream
|
||
# off+5 0x00
|
||
# off+6 0x00
|
||
# off+7 segment index
|
||
# off+8 mode 2 bytes, a 3-valued enum (see below)
|
||
# off+10 anchors 2 x int16 BE, ABSOLUTE — present only when mode is 02 00
|
||
#
|
||
# Mode semantics, all ground-truth verified:
|
||
# 02 00 14-byte header; emit the 2 anchors, then blocks are CUMULATIVE deltas
|
||
# 01 00 10-byte header; no anchors; blocks carry ABSOLUTE sample values
|
||
# 00 03 10-byte header; NO TAGS AT ALL — the data section is raw 12-bit
|
||
# packed ABSOLUTE samples (6 bytes -> 4 samples)
|
||
#
|
||
# ``40 NN`` is an ordinary int16 BE DATA block (length 2*NN + 2), never a header.
|
||
# The previous model read it as a variable-width segment header of length
|
||
# 2*NN + 16, which is why walks drifted and channels came out unequal.
|
||
#
|
||
# Verified over the 1,388 series-3 waveform binaries in the production
|
||
# snapshot: the length chain terminates on a 0x06 record in 1,387 of them (the
|
||
# exception has an ambiguous footer, handled by the caller), and all four
|
||
# channels come out at identical length in 1,388/1,388 — against 156/1,388
|
||
# under the superseded model. Against the 75 events with a preserved
|
||
# Blastware ASCII export: sample-count exact 72/75 -> 75/75, fully exact
|
||
# 70/75 -> 73/75.
|
||
|
||
CHANNEL_IDS = {0x46: "Tran", 0x47: "Vert", 0x48: "Long", 0x49: "MicL"}
|
||
STREAM_END_ID = 0x06
|
||
|
||
MODE_DELTA = (0x02, 0x00)
|
||
MODE_ABSOLUTE = (0x01, 0x00)
|
||
MODE_RAW12 = (0x00, 0x03)
|
||
_MODES = (MODE_DELTA, MODE_ABSOLUTE, MODE_RAW12)
|
||
|
||
|
||
def _u16(b: bytes, p: int) -> int:
|
||
return (b[p] << 8) | b[p + 1]
|
||
|
||
|
||
def _i16(b: bytes, p: int) -> int:
|
||
v = _u16(b, p)
|
||
return v - 0x10000 if v >= 0x8000 else v
|
||
|
||
|
||
def data_block_len(body: bytes, p: int) -> Tuple[Optional[int], Optional[int]]:
|
||
"""``(byte_length, n_samples)`` of the data block at *p*, or ``(None, None)``.
|
||
|
||
Data-section blocks only — there is no segment-header tag in this model.
|
||
``30 NN`` has no trailer-length fallback here; that fallback corrupted
|
||
records whose ``30 NN`` sat near a record boundary.
|
||
"""
|
||
if p + 2 > len(body):
|
||
return None, None
|
||
t0, t1 = body[p], body[p + 1]
|
||
hi = t0 & 0xF0
|
||
nn = ((t0 & 0x0F) << 8) | t1
|
||
if hi == 0x40: # int16 BE data block
|
||
return (None, None) if (nn == 0 or nn > 0x08) else (2 * nn + 2, nn)
|
||
if nn == 0 or nn % 4:
|
||
return None, None
|
||
if hi == 0x00:
|
||
return 2, nn # RLE hold
|
||
if hi == 0x10:
|
||
return nn // 2 + 2, nn # 4-bit nibble
|
||
if hi == 0x20:
|
||
return nn + 2, nn # int8
|
||
if hi == 0x30:
|
||
return nn * 3 // 2 + 2, nn # 12-bit packed
|
||
return None, None
|
||
|
||
|
||
def unpack12(data: bytes) -> List[int]:
|
||
"""Raw 12-bit packed samples: 6 bytes -> 4 signed values."""
|
||
out: List[int] = []
|
||
for g in range(len(data) // 6):
|
||
hi = (data[6 * g] << 8) | data[6 * g + 1]
|
||
for k in range(4):
|
||
x = (((hi >> (12 - 4 * k)) & 0xF) << 8) | data[6 * g + 2 + k]
|
||
out.append(x - 0x1000 if x >= 0x800 else x)
|
||
return out
|
||
|
||
|
||
def is_record(body: bytes, p: int) -> bool:
|
||
"""True if a per-channel record header starts at *p*."""
|
||
return (p + 10 <= len(body)
|
||
and body[p + 4] in CHANNEL_IDS
|
||
and body[p + 5] == 0x00 and body[p + 6] == 0x00
|
||
and 8 <= _u16(body, p + 2) <= len(body) - p
|
||
and (body[p + 8], body[p + 9]) in _MODES)
|
||
|
||
|
||
def find_first_record(body: bytes) -> Optional[int]:
|
||
"""Offset of the first record, or None.
|
||
|
||
Under the normal ``00 02 00`` preamble the leading bytes are segment-0's
|
||
Tran blocks, so walk them. Under the ``00 00 03`` preamble that data is
|
||
raw 12-bit with no tags at all and cannot be block-walked — scan instead.
|
||
"""
|
||
if len(body) >= 3 and (body[1], body[2]) == MODE_RAW12:
|
||
scan_from = 3
|
||
else:
|
||
i = 7
|
||
while i < len(body):
|
||
if is_record(body, i):
|
||
nxt = i + 2 + _u16(body, i + 2)
|
||
if nxt + 5 <= len(body) and (is_record(body, nxt)
|
||
or body[nxt + 4] == STREAM_END_ID):
|
||
return i
|
||
length, _ = data_block_len(body, i)
|
||
if length is None:
|
||
return None
|
||
i += length
|
||
return None
|
||
for i in range(scan_from, max(scan_from, len(body) - 10)):
|
||
if is_record(body, i):
|
||
nxt = i + 2 + _u16(body, i + 2)
|
||
if nxt + 5 <= len(body) and (is_record(body, nxt)
|
||
or body[nxt + 4] == STREAM_END_ID):
|
||
return i
|
||
return None
|
||
|
||
|
||
def walk_records(body: bytes, first: Optional[int] = None) -> List[dict]:
|
||
"""Follow the length chain from *first* to the ``0x06`` terminator."""
|
||
if first is None:
|
||
first = find_first_record(body)
|
||
out: List[dict] = []
|
||
if first is None:
|
||
return out
|
||
p, seen = first, set()
|
||
while p is not None and p + 10 <= len(body):
|
||
if p in seen:
|
||
break
|
||
seen.add(p)
|
||
cid = body[p + 4]
|
||
if cid == STREAM_END_ID or cid not in CHANNEL_IDS:
|
||
break
|
||
length = _u16(body, p + 2)
|
||
if length < 8 or p + 2 + length > len(body):
|
||
break
|
||
out.append({"offset": p, "channel": CHANNEL_IDS[cid],
|
||
"segment_index": body[p + 7],
|
||
"mode": (body[p + 8], body[p + 9]),
|
||
"end": p + 2 + length})
|
||
p += 2 + length
|
||
return out
|
||
|
||
|
||
def decode_waveform_v2(body: bytes) -> Optional[dict]:
|
||
"""Decode a Blastware waveform body into per-channel sample arrays.
|
||
|
||
Returns ``{"Tran": [...], "Vert": [...], "Long": [...], "MicL": [...]}``
|
||
in 16-count units (LSB = 0.005 in/s at Normal range), or None if *body*
|
||
is not a decodable waveform body.
|
||
|
||
Implements the record-chain model documented above.
|
||
"""
|
||
if len(body) < 8 or body[0] != 0x00:
|
||
return None
|
||
preamble = (body[1], body[2])
|
||
if preamble not in (MODE_DELTA, MODE_RAW12):
|
||
return None
|
||
first = find_first_record(body)
|
||
if first is None:
|
||
return None
|
||
|
||
out: dict = {c: [] for c in ("Tran", "Vert", "Long", "MicL")}
|
||
|
||
def run(channel: str, start: int, end: int, absolute: bool) -> None:
|
||
cur = out[channel][-1] if out[channel] else 0
|
||
i = start
|
||
while i < end:
|
||
length, nn = data_block_len(body, i)
|
||
if length is None or i + length > end:
|
||
return # stop this record; the chain resyncs at end
|
||
hi = body[i] & 0xF0
|
||
if hi == 0x00:
|
||
vals = [None] * nn
|
||
elif hi == 0x10:
|
||
vals = []
|
||
for k in range(nn):
|
||
byte = body[i + 2 + k // 2]
|
||
v = (byte >> 4) if k % 2 == 0 else (byte & 0xF)
|
||
vals.append(v - 16 if v >= 8 else v)
|
||
elif hi == 0x20:
|
||
vals = [v - 256 if v >= 128 else v
|
||
for v in body[i + 2:i + 2 + nn]]
|
||
elif hi == 0x30:
|
||
vals = unpack12(body[i + 2:i + length])
|
||
else:
|
||
vals = [_i16(body, i + 2 + 2 * k) for k in range(nn)]
|
||
for v in vals:
|
||
if v is None:
|
||
pass # RLE hold, in delta AND absolute modes
|
||
elif absolute:
|
||
cur = v
|
||
else:
|
||
cur += v
|
||
out[channel].append(cur)
|
||
i += length
|
||
|
||
# Segment 0 is an implicit Tran record carried in the preamble.
|
||
if preamble == MODE_DELTA:
|
||
out["Tran"].extend([_i16(body, 3), _i16(body, 5)])
|
||
run("Tran", 7, first, absolute=False)
|
||
else:
|
||
out["Tran"].extend(unpack12(body[3:first]))
|
||
|
||
for rec in walk_records(body, first):
|
||
ch, off, mode, end = (rec["channel"], rec["offset"],
|
||
rec["mode"], rec["end"])
|
||
if mode == MODE_DELTA:
|
||
out[ch].extend([_i16(body, off + 10), _i16(body, off + 12)])
|
||
run(ch, off + 14, end, absolute=False)
|
||
elif mode == MODE_ABSOLUTE:
|
||
run(ch, off + 10, end, absolute=True)
|
||
elif mode == MODE_RAW12:
|
||
out[ch].extend(unpack12(body[off + 10:end]))
|
||
return out
|