""" thingpark_store.py — SQLite store for LoRaWAN uplinks from ThingPark (DVI-1136). Durable store for telemetry pushed by the Actility ThingPark Enterprise Application Server (Generic HTTPS connector → POST /thingpark/ingest). Live LoRaWAN telemetry is push-only, so Togen receives uplinks here; the Auditor table, and later the Equipment/Oversight layers, read them back — degrading to the last stored snapshot when ThingPark is unreachable (same contract as scada_store.py). Flask-independent (mirrors scada_store.py / scada_report.py) so app.py, any future scheduler, and tests can all import it. The DB lives alongside the app's other state files (thingpark.db next to thingpark_config.json). Schema ------ uplinks(dev_eui, ts, fcnt_up, ...) — one row per received uplink. Idempotent ingest via UNIQUE(dev_eui, fcnt_up, ts): ThingPark retries failed POSTs and a device's frame counter is unique within a session, so the same uplink is stored once. Decoded payload (ThingPark X driver JSON) is kept verbatim in ``decoded`` for flexible surfacing before we pin per-model columns. meta(key, value) — ingest health bookkeeping for UI surfacing (why data stopped, mirroring the SCADA collector-health pattern). """ import json import os import sqlite3 import threading from datetime import datetime from pathlib import Path THINGPARK_DB_FILE = Path( os.environ.get("THINGPARK_DB_FILE", Path(__file__).resolve().parent / "thingpark.db")) # Serialize writes within one process; SQLite handles cross-process locking. _write_lock = threading.Lock() def _connect(): conn = sqlite3.connect(str(THINGPARK_DB_FILE), timeout=30) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=10000") return conn def init_thingpark_db(): """Create tables/indexes if missing. Idempotent; call at import/startup.""" with _write_lock, _connect() as conn: conn.execute( """CREATE TABLE IF NOT EXISTS uplinks ( id INTEGER PRIMARY KEY AUTOINCREMENT, dev_eui TEXT NOT NULL, ts TEXT NOT NULL, fcnt_up INTEGER, fport INTEGER, payload_hex TEXT, rssi REAL, snr REAL, battery REAL, lat REAL, lng REAL, decoded TEXT, source TEXT NOT NULL DEFAULT 'thingpark', created_at TEXT NOT NULL DEFAULT (datetime('now')), UNIQUE(dev_eui, fcnt_up, ts) )""") conn.execute( "CREATE INDEX IF NOT EXISTS idx_tp_uplinks_dev_ts" " ON uplinks(dev_eui, ts)") conn.execute( "CREATE INDEX IF NOT EXISTS idx_tp_uplinks_ts ON uplinks(ts)") conn.execute( "CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT)") def normalize_ts(raw): """Normalize an uplink timestamp to sortable "YYYY-MM-DD HH:MM:SS" text. ThingPark stamps uplinks with ISO-8601 (``Time``/``time``, often with a timezone offset). Unparseable input is returned trimmed (stored verbatim, sorts best-effort) so we never drop an uplink over a stamp quirk. """ s = (raw or "").strip() if not s: return s try: dt = datetime.fromisoformat(s.replace("Z", "+00:00")) return dt.strftime("%Y-%m-%d %H:%M:%S") except ValueError: return s def _as_float(v): try: return float(str(v).strip()) except (TypeError, ValueError): return None def _as_int(v): try: return int(float(str(v).strip())) except (TypeError, ValueError): return None def insert_uplinks(rows, source="thingpark"): """Insert uplink dicts idempotently; return the number of NEW rows stored. Each row needs at least ``dev_eui`` and ``ts``. Numeric fields are coerced (None when absent/unparseable); ``decoded`` is JSON-serialized if given a dict/list. Duplicates on (dev_eui, fcnt_up, ts) are ignored. """ prepared = [] for r in rows or []: if not isinstance(r, dict): continue dev_eui = str(r.get("dev_eui") or "").strip() ts = normalize_ts(str(r.get("ts") or "")) if not dev_eui or not ts: continue decoded = r.get("decoded") if isinstance(decoded, (dict, list)): decoded = json.dumps(decoded, separators=(",", ":")) elif decoded is not None: decoded = str(decoded) prepared.append(( dev_eui[:32], ts, _as_int(r.get("fcnt_up")), _as_int(r.get("fport")), (str(r.get("payload_hex")).strip()[:512] if r.get("payload_hex") is not None else None), _as_float(r.get("rssi")), _as_float(r.get("snr")), _as_float(r.get("battery")), _as_float(r.get("lat")), _as_float(r.get("lng")), decoded, source)) if not prepared: return 0 with _write_lock, _connect() as conn: before = conn.total_changes conn.executemany( "INSERT OR IGNORE INTO uplinks" " (dev_eui, ts, fcnt_up, fport, payload_hex, rssi, snr, battery," " lat, lng, decoded, source)" " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", prepared) return conn.total_changes - before def _row_to_device(row): (dev_eui, ts, fcnt_up, fport, payload_hex, rssi, snr, battery, lat, lng, decoded) = row dec = None if decoded: try: dec = json.loads(decoded) except (ValueError, TypeError): dec = decoded return { "dev_eui": dev_eui, "last_seen": ts, "fcnt_up": fcnt_up, "fport": fport, "payload_hex": payload_hex, "rssi": rssi, "snr": snr, "battery": battery, "lat": lat, "lng": lng, "decoded": dec, } def latest_snapshot(): """Most recent stored uplink per device: {"time": , "devices": [{dev_eui, last_seen, rssi, snr, battery, lat, lng, decoded, ...}]}. Per-device ts lets the Auditor table age each tag independently (a dead tag must show stale even while its neighbors keep reporting).""" with _connect() as conn: rows = conn.execute( """SELECT u.dev_eui, u.ts, u.fcnt_up, u.fport, u.payload_hex, u.rssi, u.snr, u.battery, u.lat, u.lng, u.decoded FROM uplinks u WHERE u.id = (SELECT u2.id FROM uplinks u2 WHERE u2.dev_eui = u.dev_eui ORDER BY u2.ts DESC, u2.id DESC LIMIT 1) ORDER BY u.dev_eui""").fetchall() devices = [_row_to_device(r) for r in rows] latest = max((d["last_seen"] for d in devices), default=None) return {"time": latest, "devices": devices} def device_uplinks(dev_eui, limit=500): """Recent uplinks for one device, newest first (detail/history view).""" with _connect() as conn: rows = conn.execute( """SELECT dev_eui, ts, fcnt_up, fport, payload_hex, rssi, snr, battery, lat, lng, decoded FROM uplinks WHERE dev_eui = ? ORDER BY ts DESC, id DESC LIMIT ?""", (dev_eui, limit)).fetchall() return [_row_to_device(r) for r in rows] def uplink_count(): with _connect() as conn: return conn.execute("SELECT COUNT(*) FROM uplinks").fetchone()[0] def device_count(): with _connect() as conn: return conn.execute( "SELECT COUNT(DISTINCT dev_eui) FROM uplinks").fetchone()[0] # --------------------------------------------------------------------------- # Ingest health (meta) — lets the UI say WHY data stopped instead of rendering # an empty grid (mirrors scada_store collector-health, per DVI-1095 P0 lesson). # --------------------------------------------------------------------------- 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_ingest(ok, error=None, received=0, inserted=0): """Persist the outcome of one ingest POST for UI health surfacing.""" now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") set_meta("ingest_last_run", now) if ok: set_meta("ingest_last_ok", now) set_meta("ingest_last_error", "") set_meta("ingest_last_received", received) set_meta("ingest_last_inserted", inserted) else: set_meta("ingest_last_error", str(error or "unknown error")) set_meta("ingest_last_error_at", now) def ingest_health(): """Ingest health summary for API payloads.""" return { "last_run": get_meta("ingest_last_run"), "last_ok": get_meta("ingest_last_ok"), "last_error": get_meta("ingest_last_error") or None, "last_error_at": get_meta("ingest_last_error_at"), "uplinks": uplink_count(), "devices": device_count(), }