5j TdZddlZddlZddlZddlZddlmZddlmZeejjdee jjdz Z ejZej dZdZd Zd Zdd Zd Zdd ZdZdZdZddZdZy)u 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. N)datetime)Path SCADA_DB_FILEzscada.dbz=^(\d{1,2})/(\d{1,2})/(\d{4})\s+(\d{1,2}):(\d{2})(?::(\d{2}))?ctjttd}|j d|j d|S)N)timeoutzPRAGMA journal_mode=WALzPRAGMA busy_timeout=10000)sqlite3connectstrrexecuteconns "/var/www/html/togen/scada_store.py_connectr(s6 ??3}-r :DLL*+LL,- Kct5t5}|jd|jd|jd|jdddddddy#1swYxYw#1swYyxYw)zECreate tables/indexes if missing. Idempotent; call at import/startup.aCREATE 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) )zOCREATE INDEX IF NOT EXISTS idx_scada_readings_sensor_ts ON readings(sensor, ts)z@CREATE INDEX IF NOT EXISTS idx_scada_readings_ts ON readings(ts)zBCREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT)N) _write_lockrr r s r init_scada_dbr/s RhjRD     ' ( N P P R#RRRRRRs# A4AA(A4(A1 -A44A=c |xsdj}|s|Stj|}|rS|j\}}}}}}|dt |ddt |ddt |dd| } | |rd|zSdzS t j |jdd} | j| jrdSd S#t$r|cYSwxYw) a%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). -02d :Zz+00:00%Y-%m-%d %H:%M:%Sz%Y-%m-%d %H:%M) strip _US_DT_REmatchgroupsintr fromisoformatreplacestrftimesecond ValueError) rawsmmodayrhhmissoutdts r normalize_tsr2Fs A A!"BBBAc"gc]!CGC=#b'#atD""h--"--  # #AIIc8$< ={{")).RRAQRR sAC C CCcg}|xsgD]}t|jdxsdj}t|jd|jdndj}tt|jdxsd}|r|r|s t |}|j |||||f|syt5t5}|j} |jd||j| z cdddcdddS#t $rd}YywxYw#1swYnxYwdddy#1swYyxYw)zInsert 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.sensorrvalueNtsrz\INSERT OR IGNORE INTO readings (sensor, ts, value, raw_value, source) VALUES (?, ?, ?, ?, ?)) r getrr2floatr&appendrr total_changes executemany) rowssourcepreparedrr4 raw_valuer6numrbefores rinsert_readingsrC\s@H ZR >QUU8_*+113!%%.*Dg"MSSU #aeeDk/R0 1Yb   "C S)V<= >  +hj+D##  &'/ 1!!F* +++  C  +++++s6" D D?-D* D? D'&D'*D3 /D??Ec t5}|jdj}dddDcgc] \}}}|||d}}}}td|Dd}||dS#1swY=xYwcc}}}w)u"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).aSELECT 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.sensorN)labelr5r6c3(K|] \}}}| yw)N).0_r6s r z"latest_snapshot..s*Ar"*s)default)timereadings)rr fetchallmax)rr<r(vr6rMlatests rlatest_snapshotrRvs .t|| "# $,8: .DHHHxq!R!ar2HHH *T*D 9F 11..Is A'A3'A0c \gg}}|rA|jddjdt|zd|j||r+|jd|jt |d}|r|ddj|zz }|d z }i}t 5}|j ||D];\}} } |j|g} t| |ks(| j| | d = d d d |jD]} | j|S#1swY0xYw) a$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. z sensor IN (,?)zts >= ?z*SELECT sensor, ts, raw_value FROM readingsz WHERE z AND z ORDER BY sensor, ts DESC)rLr5N) r9joinlenextendr2rr setdefaultvaluesreverse) sensorssincelimit_per_sensorwhereparamssqlseriesrr4r6r@ptss r query_seriesres&6E {388C#g,,>#?"@BC g  Y l5)* 6C  y7<<... &&C F =t%)\\#v%> = !FB ##FB/C3x** B;< == }}  M==s!:D"D""D+ct5}|jdjdcdddS#1swYyxYw)NzSELECT COUNT(*) FROM readingsrrr fetchoner s r reading_countris= Kt||;<EEGJKKKs "7Ac t5t5}|jd||dn t|fddddddy#1swYxYw#1swYyxYw)NzaINSERT INTO meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.valuer)rrr r )keyr5rs rset_metarlsY 8hj8D  E  "3u: 6 8888888s! A"AAA AAct5}|jd|fj}dddr|dSdS#1swYxYw)Nz$SELECT value FROM meta WHERE key = ?rrg)rkrrows rget_metarosO VtllAC6JSSUV3q6"d"VVs ">Actjjd}td||r1td|tddtd|td|y tdt |xsdtd |y ) zCPersist the outcome of one collector cycle for UI health surfacing.rcollector_last_runcollector_last_okcollector_last_errorrcollector_last_fetchedcollector_last_insertedz unknown errorcollector_last_error_atN)rnowr$rlr )okerrorfetchedinsertedrws rrecord_collector_runr|ss ,,. ! !"5 6C !3' $c*',)73*H5'U-Eo)FG*C0rcrtdtdtdxsdtdtdS)zCollector 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.rqrrrsNrv)last_runlast_ok last_error last_error_atrM)rorirGrrcollector_healthrs< 12/056>$!";<!O  r)email)NNi )Nrr)__doc__osrer threadingrpathlibrenvironr7__file__resolveparentrLockrcompilerrrr2rCrRrerirlror|rrGrrrs, JJNN?DN$:$:$<$C$Cj$PQS inn BJJD F R.,+42$:K8# 1 r