"""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