seismo-relay/minimateplus/sensor_check.py

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