serversdownandClaude Opus 4.8
8d3cdba1b5
Make the sensor self-check a first-class part of the standardized decoded event
so SFM stops decoding it at report time — device-agnostic, per the store's
decoder→standardized-.h5→SFM model.
* Event gains a `sensor_check` field; both decoders attach the traces where
they set raw_samples — series-3 in event_file_io.read_blastware_file
(minimateplus.sensor_check), series-4 in waveform_store's IDF path
(micromate.sensor_check). Covers ingest and backfill (both re-decode).
* event_hdf5 bumps schema_version 1→2 and writes an optional /sensor_check
group (raw counts, int32, per channel present). read_event_hdf5 returns
it; plot_json_from_hdf5 carries it as a top-level key. Old v1 files still
read cleanly (no group → None), so nothing breaks before the backfill.
* gather_report_data reads sensor_check_waveforms from the .h5 and drops the
report-time series-3 decode — the report no longer reaches into a decoder,
and a series-4 event now lights up the same strip automatically.
Stored as raw counts (a shape diagnostic, rendered fit-to-box): the per-series
count scale differs and a physical mic unit is ill-defined, so conversion would
add complexity for no display benefit — easy to add later if a numeric use
appears.
Tests: .h5 roundtrip + backward-compat + plot_json + real series-3 decode
attaches to the Event.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YDXjZCr4RqT2U3QvMDhgzf
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
"""The event .h5 carries the sensor self-check traces (schema v2).
|
|
|
|
The sensor check is decoded by the per-series decoder and attached to the
|
|
standardized Event, so the .h5 writer persists it device-agnostically and SFM
|
|
reads it back without knowing which instrument produced it. Old v1 files (no
|
|
sensor_check group) must still read cleanly.
|
|
"""
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
from minimateplus.models import Event
|
|
from minimateplus.event_file_io import read_blastware_file
|
|
from sfm import event_hdf5
|
|
|
|
S3_FIX = Path(__file__).parent / "fixtures" / "fft-oracle-2026-09-14" / "N844LQHB.ZT0W"
|
|
|
|
|
|
def _write(ev, **kw):
|
|
d = Path(tempfile.mkdtemp())
|
|
p = d / "e.h5"
|
|
event_hdf5.write_event_hdf5(p, ev, serial="BE12844", **kw)
|
|
return p
|
|
|
|
|
|
def test_sensor_check_roundtrips_through_hdf5():
|
|
ev = Event(index=0)
|
|
ev.raw_samples = {"Tran": [1, 2, -3], "Vert": [0, 1], "Long": [2], "MicL": [5, -5]}
|
|
ev.sample_rate = 1024
|
|
sc = {"Tran": [0, -990, -500, -100], "Vert": [0, -980, -480],
|
|
"Long": [0, -986, -470], "MicL": [0, -1800, 1800, -1800]}
|
|
ev.sensor_check = sc
|
|
|
|
r = event_hdf5.read_event_hdf5(_write(ev))
|
|
assert r["schema_version"] == 2
|
|
assert set(r["sensor_check"]) == {"Tran", "Vert", "Long", "MicL"}
|
|
for ch, vals in sc.items():
|
|
assert r["sensor_check"][ch].tolist() == vals
|
|
|
|
|
|
def test_plot_json_carries_sensor_check():
|
|
ev = Event(index=0)
|
|
ev.raw_samples = {"Tran": [1, 2, 3]}
|
|
ev.sample_rate = 1024
|
|
ev.sensor_check = {"Tran": [0, -990, -500], "Vert": [0, -980],
|
|
"Long": [0, -986]} # 3-channel: no MicL
|
|
pj = event_hdf5.plot_json_from_hdf5(_write(ev))
|
|
assert pj["sensor_check"] is not None
|
|
assert "MicL" not in pj["sensor_check"]
|
|
assert pj["sensor_check"]["Tran"] == [0, -990, -500]
|
|
|
|
|
|
def test_event_without_sensor_check_still_reads_as_v2():
|
|
ev = Event(index=0)
|
|
ev.raw_samples = {"Tran": [1, 2, 3]}
|
|
ev.sample_rate = 1024
|
|
r = event_hdf5.read_event_hdf5(_write(ev))
|
|
assert r["schema_version"] == 2
|
|
assert r["sensor_check"] is None
|
|
assert event_hdf5.plot_json_from_hdf5(_write(ev))["sensor_check"] is None
|
|
|
|
|
|
def test_series3_decode_populates_event_sensor_check():
|
|
# The real series-3 decoder attaches the traces to the Event, so the
|
|
# ingest/backfill .h5 write picks them up with no extra plumbing.
|
|
ev = read_blastware_file(S3_FIX)
|
|
assert ev.sensor_check is not None
|
|
assert set(ev.sensor_check) == {"Tran", "Vert", "Long", "MicL"}
|
|
tran = np.asarray(ev.sensor_check["Tran"], dtype=float)
|
|
assert tran.min() < -800 # the geophone ring-down deflection
|