""" sfm/database.py — SQLite persistence layer for seismo-relay. Three tables, all keyed by unit serial number: ach_sessions — one row per inbound ACH call-home events — one row per triggered waveform event (deduped by serial+timestamp) monitor_log — one row per monitoring interval (deduped by serial+start_time) The DB file lives at: /seismo_relay.db (default: bridges/captures/seismo_relay.db) Usage ----- from sfm.database import SeismoDb db = SeismoDb("bridges/captures/seismo_relay.db") # Write a call-home session session_id = db.insert_ach_session(serial="BE11529", peer="1.2.3.4:51920", events_downloaded=3, monitor_entries=2, duration_seconds=47.3) # Write events (silently skips duplicates) db.insert_events(events, serial="BE11529", session_id=session_id) # Write monitor log entries db.insert_monitor_log(entries, session_id=session_id) # Query rows = db.query_events(serial="BE11529", from_dt=datetime(...), to_dt=datetime(...)) """ from __future__ import annotations import datetime import logging import sqlite3 import uuid from pathlib import Path from typing import Optional from minimateplus.models import Event, MonitorLogEntry log = logging.getLogger("sfm.database") # ── Schema ───────────────────────────────────────────────────────────────────── _SCHEMA = """ PRAGMA journal_mode = WAL; PRAGMA foreign_keys = ON; CREATE TABLE IF NOT EXISTS ach_sessions ( id TEXT PRIMARY KEY, -- UUID serial TEXT NOT NULL, session_time TEXT NOT NULL, -- ISO-8601 UTC peer TEXT, -- "ip:port" events_downloaded INTEGER NOT NULL DEFAULT 0, monitor_entries INTEGER NOT NULL DEFAULT 0, duration_seconds REAL, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) ); CREATE INDEX IF NOT EXISTS idx_ach_sessions_serial ON ach_sessions(serial); CREATE INDEX IF NOT EXISTS idx_ach_sessions_time ON ach_sessions(session_time); CREATE TABLE IF NOT EXISTS events ( id TEXT PRIMARY KEY, -- UUID serial TEXT NOT NULL, waveform_key TEXT NOT NULL, -- 8-hex device key (dedup field) session_id TEXT, -- FK → ach_sessions.id timestamp TEXT, -- ISO-8601 local time from device tran_ppv REAL, -- in/s vert_ppv REAL, -- in/s long_ppv REAL, -- in/s peak_vector_sum REAL, -- in/s mic_ppv REAL, -- psi or dB depending on setup project TEXT, client TEXT, operator TEXT, sensor_location TEXT, sample_rate INTEGER, record_type TEXT, -- "single_shot" | "continuous" false_trigger INTEGER NOT NULL DEFAULT 0, -- 0=no, 1=yes (manual flag) reviewed_real INTEGER NOT NULL DEFAULT 0, -- 0=no, 1=operator-confirmed real (mutually exclusive with false_trigger) blastware_filename TEXT, -- event file within waveform store; extension is per-event (AB0T encodes timestamp) blastware_filesize INTEGER, -- bytes; NULL if no event file saved a5_pickle_filename TEXT, -- ".a5.pkl" sidecar sidecar_filename TEXT, -- ".sfm.json" review/metadata sidecar device_family TEXT, -- "series3" (MiniMate Plus / BW) | "series4" (Micromate / Thor) — drives per-family UI rendering (units, labels) tran_zc_freq REAL, -- Transverse ZC frequency (Hz) vert_zc_freq REAL, -- Vertical ZC frequency (Hz) long_zc_freq REAL, -- Longitudinal ZC frequency (Hz) mic_zc_freq REAL, -- Microphone ZC frequency (Hz) tran_zc_above_range INTEGER, -- 1 = value is device ceiling, render ">N Hz" vert_zc_above_range INTEGER, long_zc_above_range INTEGER, mic_zc_above_range INTEGER, shape_crest_factor REAL, -- peak / rms of the triggering channel shape_near_peak_count INTEGER, -- samples >= 0.5 * peak (FT: few; real: many) shape_sample_count INTEGER, -- total samples (to normalize near_peak_count) shape_axis TEXT, -- geophone channel measured ("Tran"/"Vert"/"Long") shape_offset INTEGER, -- 1 = DC-offset false trigger (pre-trigger baseline off zero + flat). Meaningful for waveforms only. shape_offset_axis TEXT, -- geo channel the offset was measured on shape_offset_pre REAL, -- pre-trigger baseline median (in/s) shape_offset_spread REAL, -- max(pre,mid,end) - min(...) in in/s; small = constant/DC created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), UNIQUE(serial, timestamp) ); CREATE INDEX IF NOT EXISTS idx_events_serial ON events(serial); CREATE INDEX IF NOT EXISTS idx_events_timestamp ON events(timestamp); CREATE INDEX IF NOT EXISTS idx_events_session ON events(session_id); CREATE TABLE IF NOT EXISTS monitor_log ( id TEXT PRIMARY KEY, -- UUID serial TEXT NOT NULL, waveform_key TEXT NOT NULL, -- 8-hex device key (dedup field) session_id TEXT, -- FK → ach_sessions.id start_time TEXT, -- ISO-8601 stop_time TEXT, -- ISO-8601 duration_seconds REAL, geo_threshold_ips REAL, -- in/s created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), UNIQUE(serial, start_time) ); CREATE INDEX IF NOT EXISTS idx_monitor_log_serial ON monitor_log(serial); CREATE INDEX IF NOT EXISTS idx_monitor_log_start ON monitor_log(start_time); CREATE INDEX IF NOT EXISTS idx_monitor_log_session ON monitor_log(session_id); """ # ── SeismoDb class ───────────────────────────────────────────────────────────── class SeismoDb: """ Thin SQLite wrapper for seismo-relay persistence. Thread-safe: each call opens, uses, and closes a connection with check_same_thread=False and WAL mode enabled. For the ACH server's single-writer / occasional-reader pattern this is more than sufficient. """ def __init__(self, db_path: str | Path) -> None: self.db_path = Path(db_path) self.db_path.parent.mkdir(parents=True, exist_ok=True) self._init_schema() log.info("SeismoDb initialised at %s", self.db_path) # ── Internal helpers ─────────────────────────────────────────────────────── def _connect(self) -> sqlite3.Connection: conn = sqlite3.connect(str(self.db_path), check_same_thread=False) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode = WAL") conn.execute("PRAGMA foreign_keys = ON") return conn def _init_schema(self) -> None: with self._connect() as conn: conn.executescript(_SCHEMA) self._migrate(conn) def _migrate(self, conn: sqlite3.Connection) -> None: """Apply in-place schema migrations for existing databases.""" # Migration 1: change events UNIQUE from (serial, waveform_key) [or any # waveform_key-based variant] to (serial, timestamp). # Rationale: device key counter resets to 01110000 after every erase, so # waveform_key is not a stable dedup field across erase cycles. The event # timestamp (from the device clock) is the correct natural key. row = conn.execute( "SELECT sql FROM sqlite_master WHERE type='table' AND name='events'" ).fetchone() if row and "UNIQUE(serial, timestamp)" not in row[0]: log.info("_migrate: rebuilding events table — UNIQUE(serial, timestamp)") conn.executescript(""" ALTER TABLE events RENAME TO events_old; 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 OR IGNORE INTO events SELECT * FROM events_old; DROP TABLE events_old; CREATE INDEX IF NOT EXISTS idx_events_serial ON events(serial); CREATE INDEX IF NOT EXISTS idx_events_timestamp ON events(timestamp); CREATE INDEX IF NOT EXISTS idx_events_session ON events(session_id); """) log.info("_migrate: events table rebuilt OK") # Migration 1b: add Blastware-file columns to existing events tables. # New columns are NULLable so old rows just read NULL. existing_cols = { r[1] for r in conn.execute("PRAGMA table_info(events)").fetchall() } for col, ddl in ( ("blastware_filename", "TEXT"), ("blastware_filesize", "INTEGER"), ("a5_pickle_filename", "TEXT"), ("sidecar_filename", "TEXT"), ("device_family", "TEXT"), ("tran_zc_freq", "REAL"), ("vert_zc_freq", "REAL"), ("long_zc_freq", "REAL"), ("mic_zc_freq", "REAL"), ("tran_zc_above_range", "INTEGER"), ("vert_zc_above_range", "INTEGER"), ("long_zc_above_range", "INTEGER"), ("mic_zc_above_range", "INTEGER"), ("shape_crest_factor", "REAL"), ("shape_near_peak_count", "INTEGER"), ("shape_sample_count", "INTEGER"), ("shape_axis", "TEXT"), ("shape_offset", "INTEGER"), ("shape_offset_axis", "TEXT"), ("shape_offset_pre", "REAL"), ("shape_offset_spread", "REAL"), ("reviewed_real", "INTEGER NOT NULL DEFAULT 0"), ): if col not in existing_cols: log.info("_migrate: events ADD COLUMN %s %s", col, ddl) conn.execute(f"ALTER TABLE events ADD COLUMN {col} {ddl}") # Migration 1c: backfill device_family for existing rows by sniffing # the device-native binary filename's extension. Thor (Micromate # Series IV) writes `.IDFH` / `.IDFW`; MiniMate Plus (Series III) # writes `.AB0*` / `.N00` / `.` Blastware extensions. We do # this here rather than from sidecars so the migration is fully # self-contained (doesn't need the waveform-store root) and runs at # DB-init time. Only fills NULL device_family so re-runs are no-ops. rebackfill = conn.execute( "SELECT COUNT(*) FROM events WHERE device_family IS NULL" ).fetchone() if rebackfill and rebackfill[0] > 0: log.info("_migrate: backfilling device_family for %d events", rebackfill[0]) # Series IV (Thor IDF) — extension is exactly .IDFH or .IDFW conn.execute( """ UPDATE events SET device_family = 'series4' WHERE device_family IS NULL AND ( UPPER(blastware_filename) LIKE '%.IDFH' OR UPPER(blastware_filename) LIKE '%.IDFW' ) """ ) # Everything else with a filename → Series III (Blastware family) conn.execute( """ UPDATE events SET device_family = 'series3' WHERE device_family IS NULL AND blastware_filename IS NOT NULL """ ) # Rows with no filename (e.g. older monitor_log-derived events) # stay NULL — UI handles NULL as "unknown family". remaining = conn.execute( "SELECT COUNT(*) FROM events WHERE device_family IS NULL" ).fetchone()[0] log.info("_migrate: device_family backfill complete (remaining NULL=%d)", remaining) # Migration 2: change monitor_log UNIQUE from (serial, waveform_key) to # (serial, start_time) — same reasoning as events. row = conn.execute( "SELECT sql FROM sqlite_master WHERE type='table' AND name='monitor_log'" ).fetchone() if row and "UNIQUE(serial, start_time)" not in row[0]: log.info("_migrate: rebuilding monitor_log table — UNIQUE(serial, start_time)") conn.executescript(""" ALTER TABLE monitor_log RENAME TO monitor_log_old; CREATE TABLE monitor_log ( id TEXT PRIMARY KEY, serial TEXT NOT NULL, waveform_key TEXT NOT NULL, session_id TEXT, start_time TEXT, stop_time TEXT, duration_seconds REAL, geo_threshold_ips REAL, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), UNIQUE(serial, start_time) ); INSERT OR IGNORE INTO monitor_log SELECT * FROM monitor_log_old; DROP TABLE monitor_log_old; CREATE INDEX IF NOT EXISTS idx_monitor_log_serial ON monitor_log(serial); CREATE INDEX IF NOT EXISTS idx_monitor_log_start ON monitor_log(start_time); CREATE INDEX IF NOT EXISTS idx_monitor_log_session ON monitor_log(session_id); """) log.info("_migrate: monitor_log table rebuilt OK") @staticmethod def _iso(dt: Optional[datetime.datetime]) -> Optional[str]: return dt.isoformat() if dt is not None else None @staticmethod def _new_id() -> str: return str(uuid.uuid4()) # ── ACH sessions ────────────────────────────────────────────────────────── def insert_ach_session( self, *, serial: str, peer: Optional[str] = None, events_downloaded: int = 0, monitor_entries: int = 0, duration_seconds: Optional[float] = None, session_time: Optional[datetime.datetime] = None, ) -> str: """Insert a new ACH session row. Returns the new session UUID.""" sid = self._new_id() ts = self._iso(session_time or datetime.datetime.utcnow()) with self._connect() as conn: conn.execute( """ INSERT INTO ach_sessions (id, serial, session_time, peer, events_downloaded, monitor_entries, duration_seconds) VALUES (?, ?, ?, ?, ?, ?, ?) """, (sid, serial, ts, peer, events_downloaded, monitor_entries, duration_seconds), ) log.debug("ach_session inserted: %s serial=%s events=%d monitor=%d", sid, serial, events_downloaded, monitor_entries) return sid def get_sessions( self, serial: Optional[str] = None, limit: int = 50, ) -> list[dict]: """Return recent ACH sessions, newest first.""" with self._connect() as conn: if serial: rows = conn.execute( "SELECT * FROM ach_sessions WHERE serial=? " "ORDER BY session_time DESC LIMIT ?", (serial, limit), ).fetchall() else: rows = conn.execute( "SELECT * FROM ach_sessions ORDER BY session_time DESC LIMIT ?", (limit,), ).fetchall() return [dict(r) for r in rows] # ── Events ──────────────────────────────────────────────────────────────── def insert_events( self, events: list[Event], *, serial: str, session_id: Optional[str] = None, waveform_records: Optional[dict[str, dict]] = None, device_family: Optional[str] = None, ) -> tuple[int, int]: """ Insert triggered events. Silently skips duplicates (serial+timestamp). Returns (inserted, skipped). ``waveform_records`` (optional): dict keyed by event waveform_key (hex) whose value is a record from ``WaveformStore.save()``: {"filename": str, "filesize": int, "a5_pickle_filename": str} For events whose key is in this dict, the matching columns are populated. If a row with the same (serial, timestamp) already exists (dedup hit), the matching waveform record is upserted onto the existing row so a re-download via the live endpoint refreshes the file metadata. ``device_family`` (optional): "series3" (MiniMate Plus / Blastware) or "series4" (Micromate / Thor). Drives per-family UI rendering — most importantly the mic-unit convention (psi vs dB(L)). Set on every insert and overwritten on every UPSERT so the latest writer wins. """ inserted = skipped = 0 wave_recs = waveform_records or {} with self._connect() as conn: for ev in events: key = ev._waveform_key.hex() if ev._waveform_key else None if key is None: skipped += 1 continue ts = None if ev.timestamp: try: ts = datetime.datetime( ev.timestamp.year, ev.timestamp.month, ev.timestamp.day, ev.timestamp.hour, ev.timestamp.minute, ev.timestamp.second, ).isoformat() except Exception: ts = str(ev.timestamp) pv = ev.peak_values pi = ev.project_info rec = wave_recs.get(key) or {} try: conn.execute( """ INSERT INTO events (id, serial, waveform_key, session_id, timestamp, tran_ppv, vert_ppv, long_ppv, peak_vector_sum, mic_ppv, project, client, operator, sensor_location, sample_rate, record_type, blastware_filename, blastware_filesize, a5_pickle_filename, sidecar_filename, device_family, 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, shape_crest_factor, shape_near_peak_count, shape_sample_count, shape_axis, shape_offset, shape_offset_axis, shape_offset_pre, shape_offset_spread) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( self._new_id(), serial, key, session_id, ts, pv.tran if pv else None, pv.vert if pv else None, pv.long if pv else None, pv.peak_vector_sum if pv else None, pv.micl if pv else None, pi.project if pi else None, pi.client if pi else None, pi.operator if pi else None, pi.sensor_location if pi else None, ev.sample_rate, ev.record_type, rec.get("filename"), rec.get("filesize"), rec.get("a5_pickle_filename"), rec.get("sidecar_filename"), device_family, pv.tran_zc_freq if pv else None, pv.vert_zc_freq if pv else None, pv.long_zc_freq if pv else None, pv.mic_zc_freq if pv else None, (1 if (pv and pv.tran_zc_above_range) else 0), (1 if (pv and pv.vert_zc_above_range) else 0), (1 if (pv and pv.long_zc_above_range) else 0), (1 if (pv and pv.mic_zc_above_range) else 0), rec.get("shape_crest_factor"), rec.get("shape_near_peak_count"), rec.get("shape_sample_count"), rec.get("shape_axis"), rec.get("shape_offset"), rec.get("shape_offset_axis"), rec.get("shape_offset_pre"), rec.get("shape_offset_spread"), ), ) inserted += 1 except sqlite3.IntegrityError: skipped += 1 # UPSERT path: a row for this (serial, timestamp) already # exists. Refresh every device-authoritative field from # the new data so that a re-import with better data (e.g. # a watcher re-forward where the previous attempt missed # the paired BW ASCII report) replaces stale peaks / # project info / sample_rate. # # Preserved (not in this UPDATE): # id, waveform_key, session_id, created_at — immutable / FK # false_trigger — operator review state # # Behaviour change vs prior versions: this UPDATE used # to only refresh filename / filesize / a5_pickle / # sidecar fields. As a result, the first insert's # broken-codec peak values were locked in forever even # if subsequent re-forwards arrived with correct # report-derived values. Now every re-import lifts the # DB row up to whatever the latest Event carries. conn.execute( """ UPDATE events SET tran_ppv = ?, vert_ppv = ?, long_ppv = ?, peak_vector_sum = ?, mic_ppv = ?, project = ?, client = ?, operator = ?, sensor_location = ?, sample_rate = ?, record_type = ?, blastware_filename = ?, blastware_filesize = ?, a5_pickle_filename = ?, sidecar_filename = ?, device_family = COALESCE(?, device_family), 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 = ?, shape_crest_factor = COALESCE(?, shape_crest_factor), shape_near_peak_count = COALESCE(?, shape_near_peak_count), shape_sample_count = COALESCE(?, shape_sample_count), shape_axis = COALESCE(?, shape_axis), shape_offset = COALESCE(?, shape_offset), shape_offset_axis = COALESCE(?, shape_offset_axis), shape_offset_pre = COALESCE(?, shape_offset_pre), shape_offset_spread = COALESCE(?, shape_offset_spread) WHERE serial = ? AND timestamp = ? """, ( pv.tran if pv else None, pv.vert if pv else None, pv.long if pv else None, pv.peak_vector_sum if pv else None, pv.micl if pv else None, pi.project if pi else None, pi.client if pi else None, pi.operator if pi else None, pi.sensor_location if pi else None, ev.sample_rate, ev.record_type, rec.get("filename") if rec else None, rec.get("filesize") if rec else None, rec.get("a5_pickle_filename") if rec else None, rec.get("sidecar_filename") if rec else None, device_family, pv.tran_zc_freq if pv else None, pv.vert_zc_freq if pv else None, pv.long_zc_freq if pv else None, pv.mic_zc_freq if pv else None, (1 if (pv and pv.tran_zc_above_range) else 0), (1 if (pv and pv.vert_zc_above_range) else 0), (1 if (pv and pv.long_zc_above_range) else 0), (1 if (pv and pv.mic_zc_above_range) else 0), rec.get("shape_crest_factor") if rec else None, rec.get("shape_near_peak_count") if rec else None, rec.get("shape_sample_count") if rec else None, rec.get("shape_axis") if rec else None, rec.get("shape_offset") if rec else None, rec.get("shape_offset_axis") if rec else None, rec.get("shape_offset_pre") if rec else None, rec.get("shape_offset_spread") if rec else None, serial, ts, ), ) log.debug("insert_events serial=%s inserted=%d skipped=%d", serial, inserted, skipped) return inserted, skipped def get_event(self, event_id: str) -> Optional[dict]: """Return one event row by id, or None.""" with self._connect() as conn: row = conn.execute( "SELECT * FROM events WHERE id = ?", (event_id,), ).fetchone() return dict(row) if row else None def query_events( self, serial: Optional[str] = None, from_dt: Optional[datetime.datetime] = None, to_dt: Optional[datetime.datetime] = None, false_trigger: Optional[bool] = None, limit: int = 500, offset: int = 0, ) -> list[dict]: """Query events with optional filters. Returns newest first.""" clauses: list[str] = [] params: list = [] if serial: clauses.append("serial = ?") params.append(serial) if from_dt: clauses.append("timestamp >= ?") params.append(from_dt.isoformat()) if to_dt: clauses.append("timestamp <= ?") params.append(to_dt.isoformat()) if false_trigger is not None: clauses.append("false_trigger = ?") params.append(1 if false_trigger else 0) where = ("WHERE " + " AND ".join(clauses)) if clauses else "" params += [limit, offset] with self._connect() as conn: rows = conn.execute( f"SELECT * FROM events {where} " f"ORDER BY timestamp DESC LIMIT ? OFFSET ?", params, ).fetchall() return [dict(r) for r in rows] def find_twins(self, event_id: str, *, window_seconds: int | None = None) -> list[dict]: """ Find this event's histogram/waveform twin(s): the SAME physical event recorded both as a scheduled histogram and as a triggered waveform. A real trigger is captured twice — once as a triggered waveform (stamped at the trigger instant) and once inside the scheduled histogram whose interval contains it (stamped at the histogram's interval start, e.g. the 7am/7pm call-in). The two can be HOURS apart in time yet report the same serial and identical peak_vector_sum. Twins are therefore matched by: * same serial, * identical peak_vector_sum, * OPPOSITE record type (one histogram, one waveform), and * the waveform's timestamp falls within the histogram's interval — from a histogram's timestamp up to the next histogram (same serial). This replaces the old ±``window_seconds`` heuristic, which silently missed twins more than a few minutes apart (a histogram's interval-start stamp and the trigger instant routinely differ by hours). ``window_seconds`` is still accepted for backward compatibility but is ignored. Returns [] if the event or a required field (serial / peak_vector_sum / timestamp) is missing. Caveat: identical-PVS matching remains a proxy for "same physical event" — if the device clamps/saturates PVS (to sqrt(3) * geo_range), two distinct saturated events could share a PVS. The added opposite-type and interval constraints make a false pairing far less likely than the old time-window match, and false_trigger/reviewed_real stay re-derivable from the sidecar source of truth. """ def _parse(ts): if not ts: return None try: return datetime.datetime.fromisoformat(str(ts).replace(" ", "T")) except ValueError: return None def _is_hist(rt): return str(rt or "").lower().startswith("hist") row = self.get_event(event_id) if not row: return [] serial = row.get("serial"); pvs = row.get("peak_vector_sum") t_target = _parse(row.get("timestamp")) if serial is None or pvs is None or t_target is None: return [] target_hist = _is_hist(row.get("record_type")) with self._connect() as conn: cand_rows = [dict(r) for r in conn.execute( "SELECT * FROM events WHERE serial=? AND id!=? AND peak_vector_sum=?", (serial, event_id, pvs)).fetchall()] hist_ts = [r["timestamp"] for r in conn.execute( "SELECT timestamp FROM events WHERE serial=? AND lower(record_type) LIKE 'hist%'", (serial,)).fetchall()] # Histogram interval-start times for this serial, sorted, to bound intervals. starts = sorted(x for x in (_parse(t) for t in hist_ts) if x is not None) def _interval_end(h_start): # The next histogram strictly after h_start bounds the interval; else open-ended. for x in starts: if x > h_start: return x return None def _covers(h_start, w_time): end = _interval_end(h_start) return h_start <= w_time and (end is None or w_time < end) twins = [] for c in cand_rows: if _is_hist(c.get("record_type")) == target_hist: continue # twins are strictly cross-type (one histogram, one waveform) c_time = _parse(c.get("timestamp")) if c_time is None: continue h_start, w_time = (t_target, c_time) if target_hist else (c_time, t_target) if _covers(h_start, w_time): twins.append(c) return twins def propagate_review_to_twins(self, event_id: str, *, window_seconds: int | None = None) -> list[str]: """ Copy this event's `false_trigger`/`reviewed_real` columns onto each of its histogram/waveform twins (see `find_twins`), so flagging one twin flags both. Returns the list of twin ids updated. ``window_seconds`` is accepted for backward compatibility but ignored; twin matching is now interval-based (see `find_twins`). """ row = self.get_event(event_id) if not row: return [] ft = 1 if row.get("false_trigger") else 0 real = 1 if row.get("reviewed_real") else 0 twins = self.find_twins(event_id) moved = [] with self._connect() as conn: for tw in twins: conn.execute("UPDATE events SET false_trigger=?, reviewed_real=? WHERE id=?", (ft, real, tw["id"])) moved.append(tw["id"]) return moved def set_false_trigger(self, event_id: str, value: bool) -> bool: """ Set or clear the false_trigger flag on an event. Returns True if found. Mirrors the 3-state exclusivity enforced by `update_event_review`: setting false_trigger true also clears `reviewed_real` (a false trigger can't also be a confirmed-real event). Clearing false_trigger leaves `reviewed_real` untouched. """ with self._connect() as conn: if value: cur = conn.execute( "UPDATE events SET false_trigger=1, reviewed_real=0 WHERE id=?", (event_id,), ) else: cur = conn.execute( "UPDATE events SET false_trigger=0 WHERE id=?", (event_id,), ) return cur.rowcount > 0 def delete_event(self, event_id: str) -> Optional[dict]: """ Hard-delete one event row by id. Returns the deleted row (so the caller can clean up any on-disk files referenced by it) or None if no row matched. """ with self._connect() as conn: row = conn.execute( "SELECT * FROM events WHERE id = ?", (event_id,), ).fetchone() if row is None: return None conn.execute("DELETE FROM events WHERE id = ?", (event_id,)) return dict(row) def delete_events_bulk( self, serial: Optional[str] = None, from_dt: Optional[datetime.datetime] = None, to_dt: Optional[datetime.datetime] = None, false_trigger: Optional[bool] = None, ids: Optional[list[str]] = None, ) -> list[dict]: """ Hard-delete events matching the given filters. Returns the list of deleted row dicts. Refuses to delete with no filters at all (would wipe the whole table) — raises ValueError. Filter semantics match query_events: serial / from_dt / to_dt / false_trigger combine with AND. `ids` is an additional inclusion list (event_id IN (...)); if supplied alongside other filters, only rows matching all conditions are deleted. """ clauses: list[str] = [] params: list = [] if serial: clauses.append("serial = ?") params.append(serial) if from_dt: clauses.append("timestamp >= ?") params.append(from_dt.isoformat()) if to_dt: clauses.append("timestamp <= ?") params.append(to_dt.isoformat()) if false_trigger is not None: clauses.append("false_trigger = ?") params.append(1 if false_trigger else 0) if ids: placeholders = ",".join("?" * len(ids)) clauses.append(f"id IN ({placeholders})") params.extend(ids) if not clauses: raise ValueError( "delete_events_bulk refuses to delete with no filters " "(would wipe the entire events table)" ) where = "WHERE " + " AND ".join(clauses) with self._connect() as conn: rows = conn.execute( f"SELECT * FROM events {where}", params, ).fetchall() if rows: conn.execute(f"DELETE FROM events {where}", params) return [dict(r) for r in rows] def update_event_review(self, event_id: str, review: dict) -> bool: """ Sync derived index columns from a sidecar's `review` block. The derived indexes are `events.false_trigger` and `events.reviewed_real` — kept in sync so `/db/events` queries don't have to scan every sidecar JSON on disk. The sidecar JSON itself remains the source of truth for the full review state. The two columns are mutually exclusive: setting either one true forces the other's column to 0. Returns True when the row exists, False otherwise. No-op fields (review without `false_trigger` or `reviewed_real`) leave both columns untouched. """ if not isinstance(review, dict): return False has_ft = "false_trigger" in review has_real = "reviewed_real" in review if not has_ft and not has_real: # Nothing derived to update; just confirm the row exists. with self._connect() as conn: row = conn.execute( "SELECT 1 FROM events WHERE id=?", (event_id,), ).fetchone() return row is not None sets = {} if has_ft: sets["false_trigger"] = 1 if review.get("false_trigger") else 0 if has_real: sets["reviewed_real"] = 1 if review.get("reviewed_real") else 0 # mutual exclusivity: a true in one forces the other column to 0 if sets.get("false_trigger") == 1: sets["reviewed_real"] = 0 if sets.get("reviewed_real") == 1: sets["false_trigger"] = 0 assign = ", ".join(f"{k}=?" for k in sets) params = list(sets.values()) + [event_id] with self._connect() as conn: cur = conn.execute(f"UPDATE events SET {assign} WHERE id=?", params) return cur.rowcount > 0 # ── Monitor log ─────────────────────────────────────────────────────────── def insert_monitor_log( self, entries: list[MonitorLogEntry], *, session_id: Optional[str] = None, ) -> tuple[int, int]: """ Insert monitor log entries. Silently skips duplicates (serial+start_time). Returns (inserted, skipped). """ inserted = skipped = 0 with self._connect() as conn: for e in entries: try: conn.execute( """ INSERT INTO monitor_log (id, serial, waveform_key, session_id, start_time, stop_time, duration_seconds, geo_threshold_ips) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( self._new_id(), e.serial or "", e.key, session_id, self._iso(e.start_time), self._iso(e.stop_time), e.duration_seconds, e.geo_threshold_ips, ), ) inserted += 1 except sqlite3.IntegrityError: skipped += 1 log.debug("insert_monitor_log inserted=%d skipped=%d", inserted, skipped) return inserted, skipped def query_monitor_log( self, serial: Optional[str] = None, from_dt: Optional[datetime.datetime] = None, to_dt: Optional[datetime.datetime] = None, limit: int = 500, offset: int = 0, ) -> list[dict]: """Query monitor log entries with optional filters. Returns newest first.""" clauses: list[str] = [] params: list = [] if serial: clauses.append("serial = ?") params.append(serial) if from_dt: clauses.append("start_time >= ?") params.append(from_dt.isoformat()) if to_dt: clauses.append("start_time <= ?") params.append(to_dt.isoformat()) where = ("WHERE " + " AND ".join(clauses)) if clauses else "" params += [limit, offset] with self._connect() as conn: rows = conn.execute( f"SELECT * FROM monitor_log {where} " f"ORDER BY start_time DESC LIMIT ? OFFSET ?", params, ).fetchall() return [dict(r) for r in rows] # ── Fleet overview ──────────────────────────────────────────────────────── def query_units(self) -> list[dict]: """ Return one row per known serial with summary stats. Aggregates from BOTH source tables: - `events` — populated by every ingest path (live ACH, /db/import/blastware_file from the series3-watcher forwarder, etc.) - `ach_sessions` — only populated by the live ACH server; empty for events that came in via the BW-importer route. Earlier this method only joined on `ach_sessions`, which made watcher-forwarded units invisible to the SFM webapp's fleet overview even though their events were correctly populated in `events`. Now we union the two and surface every serial that has activity in either table. Fields: serial — unit serial number (e.g. "BE11529") last_seen — most recent of MAX(events.timestamp) and MAX(ach_sessions.session_time) total_events — COUNT(*) from `events` (the authoritative count regardless of ingest path) total_monitor_entries — from `ach_sessions`, 0 when absent total_sessions — COUNT(*) from `ach_sessions`, 0 when absent """ with self._connect() as conn: event_stats = { row["serial"]: row for row in conn.execute( """ SELECT serial, MAX(timestamp) AS last_event_at, COUNT(*) AS total_events FROM events GROUP BY serial """, ).fetchall() } session_stats = { row["serial"]: row for row in conn.execute( """ SELECT serial, MAX(session_time) AS last_session_at, SUM(monitor_entries) AS total_monitor_entries, COUNT(*) AS total_sessions FROM ach_sessions GROUP BY serial """, ).fetchall() } all_serials = set(event_stats) | set(session_stats) units = [] for serial in all_serials: e = event_stats.get(serial) s = session_stats.get(serial) last_event_at = e["last_event_at"] if e else None last_session_at = s["last_session_at"] if s else None # Prefer whichever timestamp is more recent last_seen = max( (t for t in (last_event_at, last_session_at) if t), default=None, ) units.append({ "serial": serial, "last_seen": last_seen, "total_events": e["total_events"] if e else 0, "total_monitor_entries": s["total_monitor_entries"] if s else 0, "total_sessions": s["total_sessions"] if s else 0, }) # Sort by last_seen desc; serials with no timestamp at all sink to the bottom. units.sort(key=lambda u: u.get("last_seen") or "", reverse=True) return units