v0.11.0 — SQLite persistence layer (SeismoDb)
sfm/database.py (new)
- SeismoDb class: three tables keyed by unit serial number
- ach_sessions: one row per ACH call-home
- events: one row per triggered event, deduped by (serial, waveform_key)
- monitor_log: one row per monitoring interval, deduped by (serial, waveform_key)
- WAL mode, per-request connections, silent dedup via UNIQUE constraint
- Query helpers: query_events(), query_monitor_log(), get_sessions(), query_units()
- false_trigger flag on events for future review UI / report filtering
bridges/ach_server.py
- Import SeismoDb; create shared instance at startup pointed at
bridges/captures/seismo_relay.db
- After each call-home: insert_events() + insert_monitor_log() + insert_ach_session()
- DB failures logged as warnings, never abort the session
sfm/server.py
- Import SeismoDb; lazy singleton via _get_db()
- New DB read endpoints: GET /db/units, /db/events, /db/monitor_log, /db/sessions
- PATCH /db/events/{id}/false_trigger for manual review flagging
CLAUDE.md / CHANGELOG.md
- Document DB schema, SFM DB endpoints, architecture decision (unit-keyed only)
- Version bump to v0.11.0
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
+406
@@ -0,0 +1,406 @@
|
||||
"""
|
||||
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+key)
|
||||
monitor_log — one row per monitoring interval (deduped by serial+key)
|
||||
|
||||
The DB file lives at:
|
||||
<output_dir>/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)
|
||||
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
|
||||
UNIQUE(serial, waveform_key)
|
||||
);
|
||||
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, waveform_key)
|
||||
);
|
||||
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)
|
||||
|
||||
@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,
|
||||
) -> tuple[int, int]:
|
||||
"""
|
||||
Insert triggered events. Silently skips duplicates (serial+waveform_key).
|
||||
Returns (inserted, skipped).
|
||||
"""
|
||||
inserted = skipped = 0
|
||||
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
|
||||
|
||||
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)
|
||||
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,
|
||||
),
|
||||
)
|
||||
inserted += 1
|
||||
except sqlite3.IntegrityError:
|
||||
skipped += 1
|
||||
|
||||
log.debug("insert_events serial=%s inserted=%d skipped=%d",
|
||||
serial, inserted, skipped)
|
||||
return inserted, skipped
|
||||
|
||||
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 set_false_trigger(self, event_id: str, value: bool) -> bool:
|
||||
"""Set or clear the false_trigger flag on an event. Returns True if found."""
|
||||
with self._connect() as conn:
|
||||
cur = conn.execute(
|
||||
"UPDATE events SET false_trigger=? WHERE id=?",
|
||||
(1 if value else 0, event_id),
|
||||
)
|
||||
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+waveform_key).
|
||||
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:
|
||||
last_seen, total_events, total_monitor_entries.
|
||||
"""
|
||||
with self._connect() as conn:
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT
|
||||
s.serial,
|
||||
MAX(s.session_time) AS last_seen,
|
||||
SUM(s.events_downloaded) AS total_events,
|
||||
SUM(s.monitor_entries) AS total_monitor_entries,
|
||||
COUNT(*) AS total_sessions
|
||||
FROM ach_sessions s
|
||||
GROUP BY s.serial
|
||||
ORDER BY last_seen DESC
|
||||
"""
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
Reference in New Issue
Block a user