from __future__ import annotations import os, sys, sqlite3 from pathlib import Path import pytest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from sfm.database import SeismoDb ZC_COLS = ["tran_zc_freq", "vert_zc_freq", "long_zc_freq", "mic_zc_freq", "tran_zc_above_range", "vert_zc_above_range", "long_zc_above_range", "mic_zc_above_range"] def test_fresh_db_has_zc_freq_columns(tmp_path: Path): db = SeismoDb(tmp_path / "seismo_relay.db") with sqlite3.connect(str(tmp_path / "seismo_relay.db")) as conn: cols = {r[1] for r in conn.execute("PRAGMA table_info(events)").fetchall()} for c in ZC_COLS: assert c in cols, f"missing column {c}" def test_migration_adds_zc_freq_columns_to_legacy_db(tmp_path: Path): db_path = tmp_path / "old.db" # A "modern-minus-zc" events table (has blastware_* but not the zc columns). with sqlite3.connect(str(db_path)) as conn: conn.executescript(""" CREATE TABLE events ( id TEXT PRIMARY KEY, serial TEXT NOT NULL, waveform_key TEXT NOT NULL, session_id TEXT, timestamp TEXT, tran_ppv REAL, vert_ppv REAL, long_ppv REAL, peak_vector_sum REAL, mic_ppv REAL, project TEXT, client TEXT, operator TEXT, sensor_location TEXT, sample_rate INTEGER, record_type TEXT, false_trigger INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), UNIQUE(serial, timestamp) ); INSERT INTO events (id, serial, waveform_key, timestamp) VALUES ('legacy-id', 'BE11529', '01110000', '2026-04-01T12:00:00'); """) db = SeismoDb(db_path) # triggers _migrate rows = db.query_events(serial="BE11529") assert len(rows) == 1 for c in ZC_COLS: assert c in rows[0] # present assert rows[0][c] is None # NULL for the legacy row from minimateplus.models import PeakValues, Event from minimateplus import event_file_io from minimateplus.bw_ascii_report import BwAsciiReport, ChannelStats, MicStats def _report_with_freqs(): r = BwAsciiReport() r.channels["Tran"] = ChannelStats(ppv_ips=0.075, zc_freq_hz=100.0, zc_freq_above_range=True) r.channels["Vert"] = ChannelStats(ppv_ips=0.220, zc_freq_hz=85.0) r.channels["Long"] = ChannelStats(ppv_ips=0.045, zc_freq_hz=73.0) r.mic = MicStats(zc_freq_hz=51.0) return r def test_apply_report_copies_zc_freq_onto_peakvalues(): ev = Event(index=0) event_file_io.apply_report_to_event(ev, _report_with_freqs()) pv = ev.peak_values assert pv.vert_zc_freq == 85.0 assert pv.tran_zc_freq == 100.0 assert pv.tran_zc_above_range is True assert pv.vert_zc_above_range is False assert pv.mic_zc_freq == 51.0 def test_apply_bw_report_dict_copies_zc_freq(): ev = Event(index=0) bw = { "peaks": { "tran": {"ppv_ips": 0.075, "zc_freq_hz": 100.0, "zc_freq_above_range": True}, "vert": {"ppv_ips": 0.220, "zc_freq_hz": 85.0}, "long": {"ppv_ips": 0.045, "zc_freq_hz": 73.0}, }, "mic": {"zc_freq_hz": 51.0, "zc_freq_above_range": False}, } event_file_io.apply_bw_report_dict_to_event(ev, bw) pv = ev.peak_values assert pv.vert_zc_freq == 85.0 assert pv.tran_zc_above_range is True assert pv.mic_zc_freq == 51.0 def _event_with_zc(waveform_key="01110000"): from minimateplus.models import Event, Timestamp ev = Event(index=0) ev._waveform_key = bytes.fromhex(waveform_key) ev.timestamp = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=6, day=25, hour=8, minute=50, second=0) ev.record_type = "Waveform" ev.peak_values = PeakValues( tran=0.075, vert=0.220, long=0.045, peak_vector_sum=0.231, micl=0.01, tran_zc_freq=100.0, vert_zc_freq=85.0, long_zc_freq=73.0, mic_zc_freq=51.0, tran_zc_above_range=True, ) return ev def test_insert_events_persists_zc_freq(tmp_path: Path): db = SeismoDb(tmp_path / "seismo_relay.db") ins, _ = db.insert_events([_event_with_zc()], serial="BE11529") assert ins == 1 row = db.query_events(serial="BE11529")[0] assert row["vert_zc_freq"] == 85.0 assert row["tran_zc_freq"] == 100.0 assert row["tran_zc_above_range"] == 1 assert row["vert_zc_above_range"] == 0 assert row["mic_zc_freq"] == 51.0 def test_upsert_updates_zc_freq(tmp_path: Path): db = SeismoDb(tmp_path / "seismo_relay.db") db.insert_events([_event_with_zc()], serial="BE11529") ev2 = _event_with_zc() ev2.peak_values.vert_zc_freq = 91.0 # same (serial, timestamp) → UPSERT db.insert_events([ev2], serial="BE11529") row = db.query_events(serial="BE11529")[0] assert row["vert_zc_freq"] == 91.0 def test_backfill_zc_freq_from_sidecar(tmp_path: Path): import importlib mod = importlib.import_module("scripts.backfill_event_zc_freq") db = SeismoDb(tmp_path / "seismo_relay.db") # Insert an event WITHOUT freq (simulates a pre-feature row), with a filename. from minimateplus.models import Event, Timestamp ev = Event(index=0) ev._waveform_key = bytes.fromhex("01110000") ev.timestamp = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=6, day=25, hour=8, minute=50, second=0) ev.record_type = "Waveform" db.insert_events([ev], serial="BE11529", waveform_records={"01110000": {"filename": "M529AAAA.AB0T"}}) class FakeStore: def load_sidecar(self, serial, filename): return {"bw_report": { "peaks": {"vert": {"zc_freq_hz": 85.0}, "tran": {"zc_freq_hz": 100.0, "zc_freq_above_range": True}}, "mic": {"zc_freq_hz": 51.0}}} counts = mod.backfill_zc_freq(db, FakeStore()) assert counts["updated"] == 1 row = db.query_events(serial="BE11529")[0] assert row["vert_zc_freq"] == 85.0 assert row["tran_zc_above_range"] == 1 assert row["mic_zc_freq"] == 51.0 def test_backfill_skips_event_without_sidecar(tmp_path: Path): import importlib mod = importlib.import_module("scripts.backfill_event_zc_freq") db = SeismoDb(tmp_path / "seismo_relay.db") from minimateplus.models import Event, Timestamp ev = Event(index=0) ev._waveform_key = bytes.fromhex("01110000") ev.timestamp = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=6, day=25, hour=8, minute=50, second=0) db.insert_events([ev], serial="BE11529", waveform_records={"01110000": {"filename": "M529AAAA.AB0T"}}) class NoneStore: def load_sidecar(self, serial, filename): return None counts = mod.backfill_zc_freq(db, NoneStore()) assert counts["updated"] == 0 assert counts["skipped_no_sidecar"] == 1