feat(series3): decode sensor self-check waveforms from the binary
The Blastware Event Report draws a "Sensor Check" strip on the right of the
waveform panel — the little traces the unit records when it pulses each sensor
before monitoring. Those live in the series-3 binary's trailing block, after
the main waveform record-chain and the per-channel calibration records, as four
length-prefixed records tagged 0x3c-0x3f (Tran/Vert/Long geophone ring-downs +
MicL pulse train). Reverse-engineered against 7 BE12844 oracle events.
New minimateplus/sensor_check.py: decode_sensor_check(raw) locates the record
chain (validated by walking the ids 0x3c->0x3f via their length prefixes) and
decodes each record's delta stream (payload[20:len-8]) with the same 10/20/30/00
delta-block tags as the main waveform codec, from an anchor of 0. Returns
{Tran,Vert,Long,MicL: [samples]} in raw 16-count units, or {} when absent.
Validated: mic pulse-train zero-crossing frequency = 20.1 Hz (exact match to
BW's mic Channel Test freq); geophone ring-downs are consistent ~-990 raw
deflections that damp to a ~-310 settle across all 7 events (a fixed
calibration pulse, so near-identical every run).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YDXjZCr4RqT2U3QvMDhgzf
This commit is contained in:
@@ -0,0 +1,146 @@
|
||||
r"""Decode the Blastware sensor self-check waveforms from a series-3 event binary.
|
||||
|
||||
Reverse-engineered 2026-09-15 against 7 BE12844 (MiniMate Plus) oracle events.
|
||||
After the main waveform record-chain and the trailing metadata / per-channel
|
||||
calibration records, the binary carries four length-prefixed records tagged
|
||||
0x3c-0x3f: the sensor self-check traces the unit records when it pulses each
|
||||
sensor before monitoring. Blastware draws these as the little waveforms in the
|
||||
"Sensor Check" strip on the right of the Event Report.
|
||||
|
||||
* 0x3c / 0x3d / 0x3e = Tran / Vert / Long geophone ring-downs (a damped
|
||||
oscillation at the geophone's resonance, ~7-8 Hz at 1024 sps).
|
||||
* 0x3f = MicL, a pulse train at the mic self-test frequency
|
||||
(~20 Hz), whose zero-crossing frequency is BW's mic "Channel Test" freq.
|
||||
|
||||
Record framing (per record, all four chained by their length prefix)::
|
||||
|
||||
[len:2 BE][id:1][00 00][Nchan:1][12-byte header][delta stream][40 02][6B]
|
||||
\_________________ payload (len bytes) _______________________________/
|
||||
|
||||
The delta stream is ``payload[20 : len-8]`` (the ``40 02`` terminator sits at
|
||||
``len-8``, followed by 6 trailing bytes). It uses the exact same 10/20/30/00
|
||||
delta-block tags as the main waveform codec
|
||||
(:mod:`minimateplus.waveform_codec`), decoded here from an implicit anchor of 0
|
||||
— so the traces come out in the same 16-count raw units as the main waveform
|
||||
(LSB = 0.005 in/s at Normal range for the geophones).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Dict, List
|
||||
|
||||
from minimateplus.waveform_codec import walk_body
|
||||
|
||||
# Record id → channel. Order mirrors the trailing per-channel calibration
|
||||
# records (Tran / Vert / Long / MicL), confirmed against BW's sensor-check
|
||||
# frequencies on all 7 oracle events.
|
||||
_ID_TO_CHANNEL = {0x3C: "Tran", 0x3D: "Vert", 0x3E: "Long", 0x3F: "MicL"}
|
||||
_CHAIN_IDS = (0x3C, 0x3D, 0x3E, 0x3F)
|
||||
|
||||
_HEADER_LEN = 20 # payload bytes before the delta stream
|
||||
_TRAILER_LEN = 8 # 40 02 terminator + 6 trailing bytes after the stream
|
||||
|
||||
|
||||
def _s4(nib: int) -> int:
|
||||
"""Sign-extend a 4-bit nibble delta."""
|
||||
return nib - 16 if nib >= 8 else nib
|
||||
|
||||
|
||||
def _i8(byte: int) -> int:
|
||||
"""Sign-extend an 8-bit int delta."""
|
||||
return byte - 256 if byte >= 128 else byte
|
||||
|
||||
|
||||
def _decode_delta_stream(buf: bytes) -> List[int]:
|
||||
"""Accumulate a 10/20/30/00 delta-block stream from an anchor of 0,
|
||||
stopping at the 0x40 terminator.
|
||||
|
||||
Mirrors the block semantics in
|
||||
:func:`minimateplus.waveform_codec.decode_waveform_v2` (fully decoded &
|
||||
byte-exact as of 2026-05-11); see that module for the format details.
|
||||
"""
|
||||
out: List[int] = []
|
||||
cur = 0
|
||||
for blk in walk_body(buf, 0):
|
||||
fam = blk.tag_hi & 0xF0
|
||||
if fam == 0x10:
|
||||
# nibble deltas, high nibble first
|
||||
for byte in blk.data:
|
||||
for nib in ((byte >> 4) & 0xF, byte & 0xF):
|
||||
cur += _s4(nib)
|
||||
out.append(cur)
|
||||
elif fam == 0x20:
|
||||
# int8 deltas
|
||||
for byte in blk.data:
|
||||
cur += _i8(byte)
|
||||
out.append(cur)
|
||||
elif fam == 0x30:
|
||||
# 12-bit signed deltas, packed as tag_lo/4 groups of 6 bytes
|
||||
for g in range(blk.tag_lo // 4):
|
||||
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.append(cur)
|
||||
elif fam == 0x00:
|
||||
# RLE zero-delta run (wide form carries the high nibble in the tag)
|
||||
run = ((blk.tag_hi & 0x0F) << 8) | blk.tag_lo
|
||||
out.extend([cur] * run)
|
||||
elif fam == 0x40:
|
||||
# segment / record terminator
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def _find_chain(body: bytes):
|
||||
"""Locate the four length-prefixed sensor-check records.
|
||||
|
||||
Returns a list of ``(offset, id, length)`` or ``None``. The chain is
|
||||
validated by walking the ids 0x3c → 0x3d → 0x3e → 0x3f via their own length
|
||||
prefixes, so a stray 0x3c byte in the waveform data cannot match.
|
||||
"""
|
||||
for p in range(len(body) - 6):
|
||||
if body[p + 2] == 0x3C and body[p + 3] == 0 and body[p + 4] == 0:
|
||||
q = p
|
||||
recs = []
|
||||
ok = True
|
||||
for expect in _CHAIN_IDS:
|
||||
if q + 3 > len(body) or body[q + 2] != expect:
|
||||
ok = False
|
||||
break
|
||||
length = int.from_bytes(body[q:q + 2], "big")
|
||||
recs.append((q, expect, length))
|
||||
q = q + 2 + length
|
||||
if ok and len(recs) == 4:
|
||||
return recs
|
||||
return None
|
||||
|
||||
|
||||
def decode_sensor_check(raw: bytes) -> Dict[str, List[int]]:
|
||||
"""Decode the four sensor self-check traces from a series-3 event binary.
|
||||
|
||||
Returns ``{"Tran": [...], "Vert": [...], "Long": [...], "MicL": [...]}`` in
|
||||
raw decode units (same 16-count LSB as the main waveform), or ``{}`` if the
|
||||
binary carries no sensor-check block (a histogram event, a non-series-3
|
||||
file, or a unit/firmware that doesn't store it).
|
||||
"""
|
||||
strt = raw.find(b"STRT")
|
||||
if strt < 0 or len(raw) < strt + 21 + 26:
|
||||
return {}
|
||||
body = raw[strt + 21: len(raw) - 26]
|
||||
chain = _find_chain(body)
|
||||
if not chain:
|
||||
return {}
|
||||
out: Dict[str, List[int]] = {}
|
||||
for off, rid, length in chain:
|
||||
payload = body[off + 2: off + 2 + length]
|
||||
if len(payload) < _HEADER_LEN + _TRAILER_LEN:
|
||||
continue
|
||||
stream = payload[_HEADER_LEN: length - _TRAILER_LEN]
|
||||
out[_ID_TO_CHANNEL[rid]] = _decode_delta_stream(stream)
|
||||
return out
|
||||
Reference in New Issue
Block a user