"""SQLite 存储（WAL）。一个文件，备份就是拷它（同门户 / evse 的做法）。 数据量估算：一块板 5 路 × 1 条/s = 43 万行/天。原始 1 s 数据只留 `RAW_KEEP_S`（2 天）， 每分钟降采样一行（均值 / 最小 / 最大）留 `MIN_KEEP_S`（90 天）—— 电流曲线的用途是「这台手机前天夜里是不是卡死过」，一分钟的 min/max 足够看出来。 事件、审计、恢复记录**永久保留**（量小，而且是排查的唯一线索）。 ⚠️ 时间戳一律是**服务器收到的时刻**（毫秒）。板子没有墙钟（需求 §10.12： 上电从 0 计），它自己的 `t_ms` 只存在事件表的 `t_ms` 列里，作参考。 """ import sqlite3 import threading import time from typing import Iterable, List, Optional, Sequence RAW_KEEP_S = 2 * 86400 MIN_KEEP_S = 90 * 86400 SCHEMA = """ CREATE TABLE IF NOT EXISTS boards ( id TEXT PRIMARY KEY, -- STA MAC 12 位小写十六进制 name TEXT NOT NULL DEFAULT '', location TEXT NOT NULL DEFAULT '', fw TEXT NOT NULL DEFAULT '', ip TEXT NOT NULL DEFAULT '', online INTEGER NOT NULL DEFAULT 0, last_seen INTEGER NOT NULL DEFAULT 0, ev_cursor INTEGER NOT NULL DEFAULT 0, -- 下一条要拉的事件序号 created INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS phones ( board_id TEXT NOT NULL, rail INTEGER NOT NULL, label TEXT NOT NULL DEFAULT '', adb_addr TEXT NOT NULL DEFAULT '', -- 预留：服务器够得着手机时填 host:port auto_recover INTEGER NOT NULL DEFAULT 0, -- 默认关：逐台打开 notes TEXT NOT NULL DEFAULT '', PRIMARY KEY (board_id, rail) ); CREATE TABLE IF NOT EXISTS tele ( ts INTEGER NOT NULL, board_id TEXT NOT NULL, rail INTEGER NOT NULL, mv INTEGER, ua INTEGER, en INTEGER, vb INTEGER, pr INTEGER, pg INTEGER, flt INTEGER ); CREATE INDEX IF NOT EXISTS tele_idx ON tele (board_id, rail, ts); CREATE TABLE IF NOT EXISTS tele_min ( ts INTEGER NOT NULL, board_id TEXT NOT NULL, rail INTEGER NOT NULL, mv_avg INTEGER, ua_avg INTEGER, ua_min INTEGER, ua_max INTEGER, n INTEGER, PRIMARY KEY (board_id, rail, ts) ); CREATE TABLE IF NOT EXISTS board_tele ( ts INTEGER NOT NULL, board_id TEXT NOT NULL, in_mv INTEGER, in_ua INTEGER, temp0 INTEGER, temp1 INTEGER, fan_duty INTEGER ); CREATE INDEX IF NOT EXISTS board_tele_idx ON board_tele (board_id, ts); CREATE TABLE IF NOT EXISTS events ( board_id TEXT NOT NULL, seq INTEGER NOT NULL, t_ms INTEGER, rail INTEGER, type INTEGER, arg INTEGER, mv INTEGER, ua INTEGER, received INTEGER NOT NULL, PRIMARY KEY (board_id, seq) ); CREATE INDEX IF NOT EXISTS events_recv_idx ON events (received); CREATE TABLE IF NOT EXISTS audit ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, user TEXT NOT NULL, board_id TEXT NOT NULL, rail INTEGER, request TEXT NOT NULL, response TEXT NOT NULL, ok INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS recovery_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, board_id TEXT NOT NULL, rail INTEGER NOT NULL, state TEXT NOT NULL, reason TEXT NOT NULL, detail TEXT NOT NULL DEFAULT '' ); CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); """ def now_ms() -> int: return int(time.time() * 1000) class Db: """一个连接 + 一把锁。后端是单进程（uvicorn 单 worker），写入量是每秒几十行， 不值得上连接池；锁是为了 paho 线程之外偶尔还有线程池里的查询。""" def __init__(self, path: str): self.path = path self._lock = threading.Lock() self.conn = sqlite3.connect(path, check_same_thread=False, isolation_level=None) self.conn.row_factory = sqlite3.Row self.conn.execute("PRAGMA journal_mode=WAL") self.conn.execute("PRAGMA synchronous=NORMAL") self.conn.executescript(SCHEMA) def close(self): with self._lock: self.conn.close() # ---- 通用 ---------------------------------------------------------- def q(self, sql: str, args: Sequence = ()) -> List[sqlite3.Row]: with self._lock: return self.conn.execute(sql, args).fetchall() def x(self, sql: str, args: Sequence = ()) -> int: with self._lock: cur = self.conn.execute(sql, args) return cur.lastrowid def many(self, sql: str, rows: Iterable[Sequence]): with self._lock: self.conn.execute("BEGIN") try: self.conn.executemany(sql, rows) self.conn.execute("COMMIT") except Exception: self.conn.execute("ROLLBACK") raise # ---- 板子 ---------------------------------------------------------- def ensure_board(self, board: str): self.x("INSERT OR IGNORE INTO boards (id, created) VALUES (?, ?)", (board, now_ms())) self.many("INSERT OR IGNORE INTO phones (board_id, rail) VALUES (?, ?)", [(board, r) for r in range(1, 6)]) def set_board_state(self, board: str, online: bool, fw: str = None, ip: str = None): self.ensure_board(board) self.x("UPDATE boards SET online=?, fw=COALESCE(?, fw), ip=COALESCE(?, ip), last_seen=? " "WHERE id=?", (1 if online else 0, fw, ip, now_ms(), board)) def setting(self, key: str, default: str) -> str: rows = self.q("SELECT value FROM settings WHERE key=?", (key,)) return rows[0]["value"] if rows else default def set_setting(self, key: str, value: str): self.x("INSERT INTO settings (key, value) VALUES (?, ?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value", (key, value)) # ---- 遥测 ---------------------------------------------------------- def downsample(self, upto_ms: Optional[int] = None) -> int: """把还没聚合的整分钟原始数据聚成 tele_min。幂等（主键冲突就跳过）。 只聚到 `upto_ms` 之前的**整分钟**：正在进行的这一分钟数据还不全。""" upto = (upto_ms if upto_ms is not None else now_ms()) // 60000 * 60000 with self._lock: row = self.conn.execute("SELECT MAX(ts) FROM tele_min").fetchone() start = (row[0] + 60000) if row[0] is not None else 0 cur = self.conn.execute( "INSERT OR IGNORE INTO tele_min (ts, board_id, rail, mv_avg, ua_avg, ua_min, ua_max, n) " "SELECT ts/60000*60000 AS m, board_id, rail, CAST(AVG(mv) AS INTEGER), " "CAST(AVG(ua) AS INTEGER), MIN(ua), MAX(ua), COUNT(*) FROM tele " "WHERE ts >= ? AND ts < ? AND ua IS NOT NULL GROUP BY m, board_id, rail", (start, upto)) return cur.rowcount def prune(self, now: Optional[int] = None): t = now if now is not None else now_ms() with self._lock: self.conn.execute("DELETE FROM tele WHERE ts < ?", (t - RAW_KEEP_S * 1000,)) self.conn.execute("DELETE FROM board_tele WHERE ts < ?", (t - RAW_KEEP_S * 1000,)) self.conn.execute("DELETE FROM tele_min WHERE ts < ?", (t - MIN_KEEP_S * 1000,)) def history(self, board: str, rail: int, t0: int, t1: int, max_points: int = 1500) -> dict: """[t0, t1) 的电压电流曲线。跨度 ≤ 原始数据保留期且点数不多时回原始 1 s 数据， 否则回分钟聚合（带 min/max，界面画成带状）。点太多时按时间桶再抽一次。""" span = max(1, t1 - t0) raw_ok = t0 >= now_ms() - RAW_KEEP_S * 1000 and span <= 6 * 3600 * 1000 if raw_ok: bucket = max(1000, span // max_points) rows = self.q( "SELECT ts/?*? AS t, CAST(AVG(mv) AS INTEGER) mv, CAST(AVG(ua) AS INTEGER) ua, " "MIN(ua) ua_min, MAX(ua) ua_max FROM tele WHERE board_id=? AND rail=? " "AND ts>=? AND ts=? AND ts