diff --git a/micromate/sensor_check.py b/micromate/sensor_check.py new file mode 100644 index 0000000..dea67e6 --- /dev/null +++ b/micromate/sensor_check.py @@ -0,0 +1,89 @@ +r"""Decode the Thor / Micromate (series-4) sensor self-check waveforms from an +IDFW event binary. + +Reverse-engineered 2026-09-15 against 4 UM (Thor) oracle events. The IDFW +binary carries the sensor self-check in its fixed-header region (before the +waveform body), as up to four records tagged ``01 0e 3c/3d/3e/3f`` — the SAME +channel ids as the series-3 MiniMate Plus (Tran / Vert / Long / MicL), which is +the physical self-test: + + * 3c / 3d / 3e = Tran / Vert / Long geophone ring-downs (a damped impulse + response — resonant frequency + damping). + * 3f = MicL pulse train (the mic's known-signal gain check). Absent + on three-channel (mic-disabled) units. + +Record framing (per record):: + + 01 0e [id:1] [flags:3] [count:2 BE] [pad:10] [int16-BE samples × count] + \___ 18-byte header ___/ + +Unlike series-3's delta-coded trailing block, series-4 stores each trace as a +raw int16 big-endian array. ``count`` (the 2-byte field at header offset +8) +is the sample count; the record is padded to a fixed stride after that. +""" +from __future__ import annotations + +import struct +from typing import Dict, List + +# Record id → channel. Same ids/order as series-3 (minimateplus.sensor_check). +_ID_TO_CHANNEL = {0x3C: "Tran", 0x3D: "Vert", 0x3E: "Long", 0x3F: "MicL"} +_CHAIN_IDS = (0x3C, 0x3D, 0x3E, 0x3F) + +_MARKER = b"\x01\x0e" # precedes the 1-byte channel id +_HEADER_LEN = 18 # bytes from the marker start to the first sample +_COUNT_OFF = 8 # 2-byte BE sample count, from the marker start +_MAX_COUNT = 4000 # sanity cap (traces are ~70-200 samples) + + +def _find_chain(raw: bytes): + """Locate the sensor-check record chain. Returns a list of + ``(offset, id, count)`` for the first run of markers whose ids run + 3c, 3d, 3e[, 3f] in order, or ``[]``. + + Records are padded to a fixed stride, so the next marker is not at + ``header + count*2``; instead collect every ``01 0e [id]`` marker with a + sane count and take the first id-ordered run. Validating the id sequence + (not a lone ``01 0e 3c``) keeps a stray marker in the waveform body from + matching — the real chain sits in the fixed header, ahead of the body. + """ + n = len(raw) + markers = [] + for p in range(n - _HEADER_LEN): + if raw[p:p + 2] == _MARKER and raw[p + 2] in _ID_TO_CHANNEL: + count = int.from_bytes(raw[p + _COUNT_OFF:p + _COUNT_OFF + 2], "big") + if 0 < count <= _MAX_COUNT: + markers.append((p, raw[p + 2], count)) + + for i, (off, rid, _c) in enumerate(markers): + if rid != 0x3C: + continue + run = [markers[i]] + for m in markers[i + 1:]: + if len(run) < len(_CHAIN_IDS) and m[1] == _CHAIN_IDS[len(run)]: + run.append(m) + else: + break + if len(run) >= 3: # 3-channel (mic-disabled) units are valid + return run + return [] + + +def decode_idf_sensor_check(raw: bytes) -> Dict[str, List[int]]: + """Decode the sensor self-check traces from a Thor/Micromate IDFW binary. + + Returns ``{"Tran": [...], "Vert": [...], "Long": [...], "MicL": [...]}`` in + raw int16 ADC counts (MicL omitted on 3-channel units), or ``{}`` if the + binary carries no sensor-check chain (a non-IDF file, or an IDFH histogram). + """ + chain = _find_chain(raw) + if not chain: + return {} + out: Dict[str, List[int]] = {} + for off, rid, count in chain: + start = off + _HEADER_LEN + blob = raw[start:start + count * 2] + if len(blob) < count * 2: + continue + out[_ID_TO_CHANNEL[rid]] = list(struct.unpack(">%dh" % count, blob)) + return out diff --git a/tests/fixtures/thor-idf-sc/UM11719_20231219162723.IDFW b/tests/fixtures/thor-idf-sc/UM11719_20231219162723.IDFW new file mode 100644 index 0000000..14620e8 Binary files /dev/null and b/tests/fixtures/thor-idf-sc/UM11719_20231219162723.IDFW differ diff --git a/tests/fixtures/thor-idf-sc/UM12947_20250806134504.IDFW b/tests/fixtures/thor-idf-sc/UM12947_20250806134504.IDFW new file mode 100644 index 0000000..a16b414 Binary files /dev/null and b/tests/fixtures/thor-idf-sc/UM12947_20250806134504.IDFW differ diff --git a/tests/fixtures/thor-idf-sc/UM13981_20220207084555.IDFW b/tests/fixtures/thor-idf-sc/UM13981_20220207084555.IDFW new file mode 100644 index 0000000..4a14035 Binary files /dev/null and b/tests/fixtures/thor-idf-sc/UM13981_20220207084555.IDFW differ diff --git a/tests/fixtures/thor-idf-sc/UM20147_20250531135901.IDFW b/tests/fixtures/thor-idf-sc/UM20147_20250531135901.IDFW new file mode 100644 index 0000000..476ddc9 Binary files /dev/null and b/tests/fixtures/thor-idf-sc/UM20147_20250531135901.IDFW differ diff --git a/tests/test_sensor_check_idf.py b/tests/test_sensor_check_idf.py new file mode 100644 index 0000000..00074b2 --- /dev/null +++ b/tests/test_sensor_check_idf.py @@ -0,0 +1,67 @@ +"""Series-4 (Thor / Micromate IDFW) sensor self-check waveform decode. + +Reverse-engineered 2026-09-15 against 4 UM (Thor) oracle events. The IDFW +binary carries the sensor self-check in its fixed-header region (before the +waveform body) as up to four records tagged ``01 0e 3c/3d/3e/3f`` — the SAME +channel ids as series-3 (Tran/Vert/Long/MicL). Unlike series-3's delta-coded +trailing block, series-4 stores each trace as a raw int16-BE array after an +18-byte record header whose sample count is a 2-byte field at offset +8. + +Three-channel (mic-disabled) Thor units carry only 3c/3d/3e — no MicL record. + +Validated by shape (geophone ring-down / mic pulse train) and cross-event +consistency, since there's no Thor Event-Report strip to exact-match against. +""" +from pathlib import Path + +import numpy as np + +from micromate.sensor_check import decode_idf_sensor_check + +FIXDIR = Path(__file__).parent / "fixtures" / "thor-idf-sc" +EVENTS = sorted(p.name for p in FIXDIR.glob("*.IDFW")) + + +def _decode(name): + return decode_idf_sensor_check((FIXDIR / name).read_bytes()) + + +def test_geo_channels_present_and_ringdown_shaped(): + # Every IDFW event has the three geophone self-checks; each is a large + # one-sided deflection (~15000 raw counts) that rings back — the geophone's + # damped impulse response. + for name in EVENTS: + sc = _decode(name) + for ch in ("Tran", "Vert", "Long"): + assert ch in sc, f"{name} missing {ch}" + tr = np.asarray(sc[ch], dtype=float) + tr = tr - tr[:4].mean() # reference to the pre-trigger baseline + assert 40 <= len(tr) <= 300, f"{name}:{ch} n={len(tr)}" + assert tr.min() < -8000, f"{name}:{ch} min {tr.min()}" + # deflects one way and rings back toward / past the baseline + assert tr.max() < abs(tr.min()), f"{name}:{ch} not one-sided" + + +def test_mic_present_only_on_four_channel_units(): + # UM11719 / UM12947 record a mic; UM13981 / UM20147 are 3-channel + # (mic-disabled) units and carry no MicL self-check. + got = {name: ("MicL" in _decode(name)) for name in EVENTS} + assert any(got.values()), "expected at least one 4-channel unit" + assert not all(got.values()), "expected at least one 3-channel unit" + for name, has_mic in got.items(): + if has_mic: + tr = np.asarray(_decode(name)["MicL"], dtype=float) + tr = tr - tr[:4].mean() + # mic self-check is a bipolar pulse train — swings both ways, wide range + assert tr.max() > 5000 and tr.min() < -5000, f"{name} mic not bipolar" + + +def test_channel_ids_and_order(): + # ids decode to the canonical channel names, geo always in Tran/Vert/Long order + sc = _decode(EVENTS[0]) + assert [c for c in ("Tran", "Vert", "Long") if c in sc] == ["Tran", "Vert", "Long"] + + +def test_returns_empty_on_non_idf_input(): + assert decode_idf_sensor_check(b"not an IDF file") == {} + assert decode_idf_sensor_check(b"") == {}