diff --git a/minimateplus/sensor_check.py b/minimateplus/sensor_check.py new file mode 100644 index 0000000..7432ea8 --- /dev/null +++ b/minimateplus/sensor_check.py @@ -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 diff --git a/tests/test_sensor_check.py b/tests/test_sensor_check.py new file mode 100644 index 0000000..b39924a --- /dev/null +++ b/tests/test_sensor_check.py @@ -0,0 +1,67 @@ +"""Blastware sensor self-check waveform decode (minimateplus.sensor_check). + +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, a series-3 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. + * 0x3f = MicL, a pulse train at the mic self-test frequency. + +The self-check injects a fixed pulse, so the response is near-identical across +events — asserted here as an invariant shape (damped one-sided ring-down for +the geophones, a multi-pulse train for the mic). +""" +from pathlib import Path + +import numpy as np + +from minimateplus.sensor_check import decode_sensor_check + +FIXDIR = Path(__file__).parent / "fixtures" / "fft-oracle-2026-09-14" +EVENTS = sorted(p.name for p in FIXDIR.iterdir()) # 7 BE12844 event binaries + + +def _decode(name): + return decode_sensor_check((FIXDIR / name).read_bytes()) + + +def test_all_four_channels_present(): + for name in EVENTS: + sc = _decode(name) + assert set(sc) == {"Tran", "Vert", "Long", "MicL"}, name + + +def test_geo_channels_are_damped_ringdowns(): + # Each geophone self-check is a large one-sided deflection (~-990 raw) that + # rings back and damps toward a settled value well above the trough. + for name in EVENTS: + sc = _decode(name) + for ch in ("Tran", "Vert", "Long"): + tr = np.asarray(sc[ch], dtype=float) + assert 240 <= len(tr) <= 260, f"{name}:{ch} n={len(tr)}" + assert abs(tr[:3].mean()) < 50, f"{name}:{ch} starts off-baseline" + assert tr.min() < -800, f"{name}:{ch} min {tr.min()}" + assert tr.max() < 60, f"{name}:{ch} unexpected positive swing {tr.max()}" + # damped: settles between the trough and zero, well above the trough + assert tr.min() < tr[-1] < 0, f"{name}:{ch} end {tr[-1]} not between trough and 0" + assert abs(tr[-1]) < 0.6 * abs(tr.min()), f"{name}:{ch} not damped, end {tr[-1]}" + + +def test_mic_channel_is_a_pulse_train(): + for name in EVENTS: + tr = np.asarray(_decode(name)["MicL"], dtype=float) + assert 235 <= len(tr) <= 255, f"{name} mic n={len(tr)}" + # larger dynamic range than the geo ring-down, and swings both ways + assert tr.min() < -1500, f"{name} mic min {tr.min()}" + assert tr.max() > 100, f"{name} mic max {tr.max()}" + # multiple pulses: several deep local minima + deep = (tr[1:-1] < tr[:-2]) & (tr[1:-1] < tr[2:]) & (tr[1:-1] < -800) + assert int(deep.sum()) >= 4, f"{name} mic pulses {int(deep.sum())}" + + +def test_returns_empty_when_no_sensor_check_block(): + assert decode_sensor_check(b"not a blastware file") == {} + assert decode_sensor_check(b"") == {}