""" scada_store.py — SQLite telemetry store for SCADA sensor readings (DVI-1095 P1). Durable store decoupling SCADA views from live mailbox re-parsing: the background collector (and, later, a LAN SCADA Agent posting to /scada/ingest) writes readings here, and /scada/status, /scada/history, and reporting read them back even when the mailbox is unreachable. Flask-independent (mirrors scada_report.py) so app.py, the report scheduler, and tests can all import it. The DB lives alongside the app's other state files (scada.db next to scada_config.json). Schema ------ readings(sensor, ts, value, raw_value, source) — one row per sensor sample. ts is a sortable text timestamp ("YYYY-MM-DD HH:MM[:SS]"); value is the float when the sample is numeric, raw_value always keeps the original string ("72.4", "On", "Off"). UNIQUE(sensor, ts, source) makes ingestion idempotent — collector cycles re-parse overlapping mailbox windows. meta(key, value) — collector health + ingest bookkeeping for UI surfacing. """ import os import re import sqlite3 import threading from datetime import datetime from pathlib import Path SCADA_DB_FILE = Path( os.environ.get("SCADA_DB_FILE", Path(__file__).resolve().parent / "scada.db")) # Serialize writes within one process; SQLite handles cross-process locking. _write_lock = threading.Lock() _US_DT_RE = re.compile( r"^(\d{1,2})/(\d{1,2})/(\d{4})\s+(\d{1,2}):(\d{2})(?::(\d{2}))?") def _connect(): conn = sqlite3.connect(str(SCADA_DB_FILE), timeout=30) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=10000") return conn def init_scada_db(): """Create tables/indexes if missing. Idempotent; call at import/startup.""" with _write_lock, _connect() as conn: conn.execute( """CREATE TABLE IF NOT EXISTS readings ( id INTEGER PRIMARY KEY AUTOINCREMENT, sensor TEXT NOT NULL, ts TEXT NOT NULL, value REAL, raw_value TEXT NOT NULL, source TEXT NOT NULL DEFAULT 'email', created_at TEXT NOT NULL DEFAULT (datetime('now')), UNIQUE(sensor, ts, source) )""") conn.execute( "CREATE INDEX IF NOT EXISTS idx_scada_readings_sensor_ts" " ON readings(sensor, ts)") conn.execute( "CREATE INDEX IF NOT EXISTS idx_scada_readings_ts ON readings(ts)") conn.execute( "CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT)") def normalize_ts(raw): """Normalize a sample timestamp to sortable "YYYY-MM-DD HH:MM[:SS]" text. SCADA emails carry US-style "MM/DD/YYYY HH:MM:SS" stamps while Graph receivedDateTime is ISO; both must land on one sortable axis. Unparseable input is returned as-is (stored verbatim, sorts best-effort). """ s = (raw or "").strip() if not s: return s m = _US_DT_RE.match(s) if m: mo, da, yr, hh, mi, ss = m.groups() out = f"{yr}-{int(mo):02d}-{int(da):02d} {int(hh):02d}:{mi}" return out + (f":{ss}" if ss else "") try: dt = datetime.fromisoformat(s.replace("Z", "+00:00")) return dt.strftime("%Y-%m-%d %H:%M:%S" if dt.second else "%Y-%m-%d %H:%M") except ValueError: return s def insert_readings(rows, source="email"): """Insert reading dicts [{sensor, ts, value}] idempotently; return the number of NEW rows stored (duplicates on (sensor, ts, source) are ignored). ``value`` may be any scalar; the float form is derived when possible.""" prepared = [] for r in rows or []: sensor = str(r.get("sensor") or "").strip() raw_value = str(r.get("value") if r.get("value") is not None else "").strip() ts = normalize_ts(str(r.get("ts") or "")) if not sensor or not raw_value or not ts: continue try: num = float(raw_value) except ValueError: num = None prepared.append((sensor, ts, num, raw_value, source)) if not prepared: return 0 with _write_lock, _connect() as conn: before = conn.total_changes conn.executemany( "INSERT OR IGNORE INTO readings (sensor, ts, value, raw_value, source)" " VALUES (?, ?, ?, ?, ?)", prepared) return conn.total_changes - before def latest_snapshot(): """Most recent stored reading per sensor: {"time": , "readings": [{label, value, ts}]} — the /scada/status fallback shape when Graph is down. Per-reading ts lets the UI age each sensor independently (a dead sensor must show stale even while its neighbors keep reporting).""" with _connect() as conn: rows = conn.execute( """SELECT r.sensor, r.raw_value, r.ts FROM readings r WHERE r.id = (SELECT r2.id FROM readings r2 WHERE r2.sensor = r.sensor ORDER BY r2.ts DESC, r2.id DESC LIMIT 1) ORDER BY r.sensor""").fetchall() readings = [{"label": s, "value": v, "ts": ts} for s, v, ts in rows] latest = max((ts for _, _, ts in rows), default=None) return {"time": latest, "readings": readings} def query_series(sensors=None, since=None, limit_per_sensor=3000): """Time series per sensor: {sensor: [{time, value}, ...]} ascending by ts. ``sensors`` optionally restricts labels; ``since`` is an inclusive normalized-ts lower bound. Each series keeps its most recent ``limit_per_sensor`` points so one runaway sensor can't bloat the payload. """ where, params = [], [] if sensors: where.append(f"sensor IN ({','.join('?' * len(sensors))})") params.extend(sensors) if since: where.append("ts >= ?") params.append(normalize_ts(since)) sql = "SELECT sensor, ts, raw_value FROM readings" if where: sql += " WHERE " + " AND ".join(where) sql += " ORDER BY sensor, ts DESC" series = {} with _connect() as conn: for sensor, ts, raw_value in conn.execute(sql, params): pts = series.setdefault(sensor, []) if len(pts) < limit_per_sensor: pts.append({"time": ts, "value": raw_value}) for pts in series.values(): pts.reverse() return series def reading_count(): with _connect() as conn: return conn.execute("SELECT COUNT(*) FROM readings").fetchone()[0] # --------------------------------------------------------------------------- # Collector health (meta) — lets the UI say WHY data stopped instead of # rendering an empty grid (the DVI-1095 P0 outage was silent for 10 days). # --------------------------------------------------------------------------- def set_meta(key, value): with _write_lock, _connect() as conn: conn.execute( "INSERT INTO meta (key, value) VALUES (?, ?)" " ON CONFLICT(key) DO UPDATE SET value = excluded.value", (key, "" if value is None else str(value))) def get_meta(key): with _connect() as conn: row = conn.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone() return row[0] if row else None def record_collector_run(ok, error=None, fetched=0, inserted=0): """Persist the outcome of one collector cycle for UI health surfacing.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") set_meta("collector_last_run", now) if ok: set_meta("collector_last_ok", now) set_meta("collector_last_error", "") set_meta("collector_last_fetched", fetched) set_meta("collector_last_inserted", inserted) else: set_meta("collector_last_error", str(error or "unknown error")) set_meta("collector_last_error_at", now) def collector_health(): """Collector health summary for API payloads: {last_run, last_ok, last_error, last_error_at, readings}. last_error is None when the most recent cycle succeeded.""" return { "last_run": get_meta("collector_last_run"), "last_ok": get_meta("collector_last_ok"), "last_error": get_meta("collector_last_error") or None, "last_error_at": get_meta("collector_last_error_at"), "readings": reading_count(), }