zhou yang和Claude Opus 5.5
26c04ce878
- 后端 prrack：MQTT（phonerail-backend @ 127.0.0.1:1884）↔ 遥测入库 / 事件增量同步 / WebSocket 推送； 命令白名单 + 重传同 seq；审计记门户 X-Auth-User（只绑 127.0.0.1，main.py 拒绝非回环）。 - 自动恢复：只看电流（平线 / 过低，服务器够不到手机 adb），先长按电源再断电重启， 每小时限次后停手；默认逐台关闭。adb 探活接口 recovery.Prober 预留。 - 前端：总览 / 板子页（五路操作、电流电压曲线）/ 事件 / 操作记录 / 自动恢复；手机宽度可用。 - 部署 deploy/deploy.sh：服务器 Python 3.8 上先跑测试再替换；systemd phonerail-rack，每日备份。 - 门户应用 id 16、key rack（phonerail 已被需求文档站占用），授权 admin + staff。 测试 26 例（本机 3.14 / 服务器 3.8 均过）。线上板 b81f3fc9c8bc 在线，WS 经门户 101 + 1 条/s 遥测。 Co-Authored-By: Claude Opus 5.5 <[email protected]>
263 行
12 KiB
Python
263 行
12 KiB
Python
"""服务器侧的核心：MQTT ↔ 状态 ↔ 数据库 ↔ 网页推送。
|
||
|
||
四条规矩（都来自板侧协议，需求 §10.8）：
|
||
1. **重传用同一个 seq。** 超时之后原样再发，板子的判重兜住；换 seq 重发 =
|
||
「按 10 秒电源键」可能执行两遍。`call()` 是唯一的下发入口。
|
||
2. **失联不动作。** 板子掉线、遥测中断，服务器只把它标成离线，没有任何自动动作会因此触发
|
||
（自动恢复在板子离线时直接停摆，recovery.py）。
|
||
3. **时间一律用服务器收到的时刻。** 板子没有墙钟。
|
||
4. **事件按序号增量拉**（遥测里的 `ev_next` 比游标大就拉），一次 8 条、拉到追平为止；
|
||
板子序号比游标还小 = 板子 flash 被清过，游标归零重拉。
|
||
"""
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
from typing import Callable, Dict, List, Optional, Set
|
||
|
||
from . import protocol
|
||
from .bus import Bus
|
||
from .db import Db, now_ms
|
||
|
||
log = logging.getLogger("prrack.hub")
|
||
|
||
STALE_MS = 10000 # 这么久没遥测 = 通信中断（遥测默认 1 s 一条）
|
||
FLUSH_S = 2.0 # 遥测攒批写库
|
||
MAINT_S = 60.0 # 降采样 + 清理
|
||
|
||
|
||
class CmdTimeout(Exception):
|
||
pass
|
||
|
||
|
||
class Hub:
|
||
def __init__(self, db: Db, bus: Bus, attempt_timeout: float = 1.5, attempts: int = 3):
|
||
self.db = db
|
||
self.bus = bus
|
||
self.attempt_timeout = attempt_timeout
|
||
self.attempts = attempts
|
||
self.seq = protocol.SeqGen()
|
||
self.loop: Optional[asyncio.AbstractEventLoop] = None
|
||
self.latest: Dict[str, dict] = {} # board → {status, last_tele, online, fw, ip}
|
||
self.mqtt_up = False
|
||
self._pending: Dict[tuple, asyncio.Future] = {}
|
||
self._rows: List[tuple] = []
|
||
self._brows: List[tuple] = []
|
||
self._syncing: Set[str] = set()
|
||
self._clients: Set[asyncio.Queue] = set()
|
||
self._tasks: List[asyncio.Task] = []
|
||
self.tele_observers: List[Callable[[str, int, dict], None]] = []
|
||
|
||
# ---- 生命周期 ------------------------------------------------------------
|
||
async def start(self):
|
||
self.loop = asyncio.get_running_loop()
|
||
for r in self.db.q("SELECT id, fw, ip, online FROM boards"):
|
||
self.latest[r["id"]] = {"status": None, "last_tele": 0, "online": False,
|
||
"fw": r["fw"], "ip": r["ip"]}
|
||
self.bus.on_message = self._from_bus
|
||
self.bus.on_connect = lambda ok: self.loop.call_soon_threadsafe(self._set_mqtt, ok)
|
||
self.bus.start()
|
||
for kind in ("rsp", "tele", "state"):
|
||
self.bus.subscribe("%s/+/%s" % (protocol.TOPIC_ROOT, kind), 0 if kind == "tele" else 1)
|
||
self._tasks = [asyncio.ensure_future(self._flusher()), asyncio.ensure_future(self._maint())]
|
||
|
||
async def stop(self):
|
||
for t in self._tasks:
|
||
t.cancel()
|
||
self.flush()
|
||
self.bus.stop()
|
||
|
||
def _set_mqtt(self, ok: bool):
|
||
self.mqtt_up = ok
|
||
self.broadcast({"t": "mqtt", "up": ok})
|
||
|
||
# ---- 收：paho 线程 → asyncio ----------------------------------------------
|
||
def _from_bus(self, t: str, payload: bytes):
|
||
if self.loop is None:
|
||
return
|
||
self.loop.call_soon_threadsafe(self._handle, t, payload)
|
||
|
||
def _handle(self, t: str, payload: bytes):
|
||
parsed = protocol.parse_topic(t)
|
||
if parsed is None:
|
||
return
|
||
board, kind = parsed
|
||
try:
|
||
msg = json.loads(payload)
|
||
except ValueError:
|
||
log.warning("%s 的 %s 不是 JSON：%r", board, kind, payload[:80])
|
||
return
|
||
if kind == "rsp":
|
||
fut = self._pending.get((board, msg.get("seq")))
|
||
if fut is not None and not fut.done():
|
||
fut.set_result(msg)
|
||
elif kind == "tele":
|
||
self._on_tele(board, msg)
|
||
elif kind == "state":
|
||
self._on_state(board, msg)
|
||
|
||
def _board(self, board: str) -> dict:
|
||
b = self.latest.get(board)
|
||
if b is None:
|
||
self.db.ensure_board(board)
|
||
b = self.latest[board] = {"status": None, "last_tele": 0, "online": False, "fw": "", "ip": ""}
|
||
return b
|
||
|
||
def _on_state(self, board: str, msg: dict):
|
||
b = self._board(board)
|
||
b["online"] = bool(msg.get("online"))
|
||
if b["online"]:
|
||
b["fw"], b["ip"] = msg.get("fw", b["fw"]), msg.get("ip", b["ip"])
|
||
self.db.set_board_state(board, b["online"], msg.get("fw"), msg.get("ip"))
|
||
self.broadcast({"t": "board", "board": self.snapshot(board)})
|
||
|
||
def _on_tele(self, board: str, st: dict):
|
||
ts = now_ms()
|
||
b = self._board(board)
|
||
b["status"], b["last_tele"] = st, ts
|
||
if not b["online"]: # 遥测来了但 retained 的 state 还没到（或丢了）：以遥测为准
|
||
b["online"] = True
|
||
self.db.set_board_state(board, True, st.get("fw"))
|
||
for r in st.get("rails", []):
|
||
self._rows.append((ts, board, r.get("r"), r.get("mv"), r.get("ua"), r.get("en"),
|
||
r.get("vb"), r.get("pr"), r.get("pg"), r.get("flt")))
|
||
temp = st.get("temp") or [None, None]
|
||
self._brows.append((ts, board, (st.get("in") or {}).get("mv"), (st.get("in") or {}).get("ua"),
|
||
temp[0], temp[1] if len(temp) > 1 else None, (st.get("fan") or {}).get("duty")))
|
||
for fn in self.tele_observers:
|
||
try:
|
||
fn(board, ts, st)
|
||
except Exception: # 观察者的 bug 不能拖垮遥测
|
||
log.exception("遥测观察者出错")
|
||
self.broadcast({"t": "tele", "board": board, "ts": ts, "status": st})
|
||
ev_next = st.get("ev_next")
|
||
if isinstance(ev_next, int) and board not in self._syncing:
|
||
cur = self.db.q("SELECT ev_cursor FROM boards WHERE id=?", (board,))[0]["ev_cursor"]
|
||
if ev_next != cur:
|
||
self._syncing.add(board)
|
||
asyncio.ensure_future(self._sync_events(board, ev_next < cur))
|
||
|
||
# ---- 发 ----------------------------------------------------------------
|
||
async def call(self, board: str, req: dict, user: str, audit: bool = True) -> dict:
|
||
"""发一条命令并等回应。`req` 已经过 protocol.build_command 校验（或是内部命令）。
|
||
超时按同一个 seq 重发 `attempts` 次；全部超时抛 CmdTimeout（审计里照样记一条）。"""
|
||
seq = self.seq.next()
|
||
msg = dict(req, seq=seq)
|
||
payload = json.dumps(msg, separators=(",", ":"))
|
||
fut = self.loop.create_future()
|
||
self._pending[(board, seq)] = fut
|
||
rsp = None
|
||
try:
|
||
for _ in range(self.attempts):
|
||
self.bus.publish(protocol.topic(board, "cmd"), payload, qos=1)
|
||
try:
|
||
rsp = await asyncio.wait_for(asyncio.shield(fut), self.attempt_timeout)
|
||
break
|
||
except asyncio.TimeoutError:
|
||
continue
|
||
finally:
|
||
self._pending.pop((board, seq), None)
|
||
if audit and protocol.is_mutating(req.get("cmd", "")):
|
||
self._audit(user, board, msg, rsp)
|
||
if rsp is None:
|
||
raise CmdTimeout("板子 %s 没有回应（seq %d，重发 %d 次）" % (board, seq, self.attempts))
|
||
return rsp
|
||
|
||
def _audit(self, user: str, board: str, req: dict, rsp: Optional[dict]):
|
||
ok = bool(rsp and rsp.get("ok"))
|
||
body = rsp if rsp is not None else {"ok": False, "err": "timeout"}
|
||
rid = self.db.x("INSERT INTO audit (ts, user, board_id, rail, request, response, ok) "
|
||
"VALUES (?,?,?,?,?,?,?)", (now_ms(), user, board, req.get("rail"),
|
||
json.dumps(req), json.dumps(body), 1 if ok else 0))
|
||
self.broadcast({"t": "audit", "row": {"id": rid, "ts": now_ms(), "user": user, "board_id": board,
|
||
"rail": req.get("rail"), "request": req,
|
||
"response": body, "ok": ok}})
|
||
|
||
async def _sync_events(self, board: str, reset: bool):
|
||
try:
|
||
if reset:
|
||
log.warning("板子 %s 的事件序号倒退了（flash 被清过？）：游标归零", board)
|
||
self.db.x("UPDATE boards SET ev_cursor=0 WHERE id=?", (board,))
|
||
for _ in range(50): # 一次最多追 400 条，剩下的下一条遥测再追
|
||
cur = self.db.q("SELECT ev_cursor FROM boards WHERE id=?", (board,))[0]["ev_cursor"]
|
||
rsp = await self.call(board, {"cmd": "events", "from": cur}, "system", audit=False)
|
||
ev = rsp.get("ev") or []
|
||
if not rsp.get("ok"):
|
||
return
|
||
now = now_ms()
|
||
self.db.many("INSERT OR IGNORE INTO events (board_id, seq, t_ms, rail, type, arg, mv, ua, "
|
||
"received) VALUES (?,?,?,?,?,?,?,?,?)", [(board, *e[:7], now) for e in ev])
|
||
nxt = max([cur] + [e[0] + 1 for e in ev])
|
||
if not ev and rsp.get("oldest", 0) > cur:
|
||
nxt = rsp["oldest"] # 中间那段被板子的环形缓冲盖掉了
|
||
self.db.x("UPDATE boards SET ev_cursor=? WHERE id=?", (nxt, board))
|
||
if ev:
|
||
self.broadcast({"t": "events", "board": board, "ev": ev, "received": now})
|
||
if nxt >= rsp.get("next", nxt) or not ev:
|
||
return
|
||
except CmdTimeout:
|
||
log.warning("拉 %s 的事件超时，等下一条遥测再试", board)
|
||
finally:
|
||
self._syncing.discard(board)
|
||
|
||
# ---- 查 ----------------------------------------------------------------
|
||
def snapshot(self, board: str) -> dict:
|
||
b = self._board(board)
|
||
meta = self.db.q("SELECT name, location FROM boards WHERE id=?", (board,))
|
||
phones = self.db.q("SELECT rail, label, adb_addr, auto_recover, notes FROM phones "
|
||
"WHERE board_id=? ORDER BY rail", (board,))
|
||
stale = b["last_tele"] == 0 or now_ms() - b["last_tele"] > STALE_MS
|
||
return {"id": board, "name": meta[0]["name"] if meta else "", "location": meta[0]["location"] if meta else "",
|
||
"online": b["online"], "stale": stale, "fw": b["fw"], "ip": b["ip"],
|
||
"last_tele": b["last_tele"], "status": b["status"],
|
||
"phones": [dict(p) for p in phones]}
|
||
|
||
def boards(self) -> List[dict]:
|
||
return [self.snapshot(b) for b in sorted(self.latest)]
|
||
|
||
# ---- 推送 ----------------------------------------------------------------
|
||
def subscribe_client(self) -> asyncio.Queue:
|
||
q: asyncio.Queue = asyncio.Queue(maxsize=200)
|
||
self._clients.add(q)
|
||
return q
|
||
|
||
def unsubscribe_client(self, q: asyncio.Queue):
|
||
self._clients.discard(q)
|
||
|
||
def broadcast(self, msg: dict):
|
||
for q in list(self._clients):
|
||
try:
|
||
q.put_nowait(msg)
|
||
except asyncio.QueueFull:
|
||
# 慢客户端：丢掉它，让它重连后重新拉全量 —— 别让一个卡住的浏览器撑爆内存。
|
||
# WS 协程发现自己不在集合里了就断开（is_subscribed）。
|
||
self._clients.discard(q)
|
||
|
||
def is_subscribed(self, q: asyncio.Queue) -> bool:
|
||
return q in self._clients
|
||
|
||
# ---- 后台 ----------------------------------------------------------------
|
||
def flush(self):
|
||
rows, brows = self._rows, self._brows
|
||
self._rows, self._brows = [], []
|
||
if rows:
|
||
self.db.many("INSERT INTO tele VALUES (?,?,?,?,?,?,?,?,?,?)", rows)
|
||
if brows:
|
||
self.db.many("INSERT INTO board_tele VALUES (?,?,?,?,?,?,?)", brows)
|
||
|
||
async def _flusher(self):
|
||
while True:
|
||
await asyncio.sleep(FLUSH_S)
|
||
try:
|
||
self.flush()
|
||
except Exception:
|
||
log.exception("遥测写库失败")
|
||
|
||
async def _maint(self):
|
||
while True:
|
||
await asyncio.sleep(MAINT_S)
|
||
try:
|
||
self.db.downsample()
|
||
self.db.prune()
|
||
except Exception:
|
||
log.exception("降采样 / 清理失败")
|