zhou yang和Claude Opus 5.5
86d9e2a7d4
- 单板页「更多 → 刷机模式（1 小时）/ 加钟 30 分钟」，chg>0 时蓝色「刷机模式 · 充电中」、FLT 显示为「反向阻断」 - 协议白名单 flash_mode；事件 21 命名：开 / 手动关 / 到期 / 加钟 / 断电自动退出 - ⚠️ 254 带 0x80 位，必须先于「加钟」判断，否则显示成「加钟 126 分钟」 48 例全绿，已部署。 Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
266 行
11 KiB
Python
266 行
11 KiB
Python
"""模拟板子：说和真板一样的 MQTT 协议（firmware/main/mlink.h + cmd.h），给测试与本机开发用。
|
|
|
|
它**不是**固件的第二份实现，只模拟服务器看得见的那部分行为：
|
|
· 回应格式、seq 判重（按值、窗口 32）、空槽拒绝、越界拒绝；
|
|
· 遥测（status 的形状）、retained 上线消息；
|
|
· 事件日志（带序号，`events` 分页 8 条）；
|
|
· 手机的电流：正常 = 在 300~900 mA 间游走；卡死 = 一条平线；关机 = ≈0。
|
|
断电重启 / 长按电源键会让「卡死」的手机恢复正常 —— 自动恢复的测试靠这个。
|
|
时间由调用方推进（`tick(now_ms)`），测试因此是确定的。
|
|
"""
|
|
import json
|
|
import random
|
|
from typing import Dict, List, Optional
|
|
|
|
from .bus import Bus
|
|
from .protocol import topic
|
|
|
|
NORMAL, HUNG, OFF = "normal", "hung", "off"
|
|
|
|
|
|
class SimRail:
|
|
def __init__(self, rail: int, present: bool = True):
|
|
self.rail = rail
|
|
self.present = present
|
|
self.en = present
|
|
self.vb = False
|
|
self.set_mv = 3900
|
|
self.phone = NORMAL if present else OFF
|
|
self.ua = 500000
|
|
self.busy_until = 0
|
|
self.act = "idle"
|
|
self.pending: Optional[str] = None # 序列结束时要做的事
|
|
self.chg_until = 0 # 刷机模式截止（模拟时钟 ms），0 = 没开
|
|
|
|
|
|
class SimBoard:
|
|
def __init__(self, bus: Bus, board_id: str, present=(True,) * 5, seed: int = 1,
|
|
fw: str = "0.1.0-sim"):
|
|
self.bus = bus
|
|
self.id = board_id
|
|
self.fw = fw
|
|
self.rails = [SimRail(i + 1, present[i]) for i in range(5)]
|
|
self.rng = random.Random(seed)
|
|
self.seen: List[tuple] = [] # (seq, cmd, rsp)
|
|
self.events: List[list] = [] # [seq, t, rail, type, arg, mv, ua]
|
|
self.now = 0
|
|
self.up_ms = 0
|
|
self.fan_floor = 0
|
|
self.received: List[dict] = []
|
|
self.drop_next_replies = 0 # 模拟网络丢回应（测试重传）
|
|
self.ota = {"st": "idle", "got": 0, "size": 0, "err": None, "pend": False, "bad": None}
|
|
self.ota_req: Optional[dict] = None
|
|
self.ota_rate = 400000 # 每次 tick 下载多少字节（真板 ≈125 KB/s）
|
|
bus.on_message = self._on_message
|
|
self._ev(0, 1, 0) # BOOT
|
|
|
|
# ---- 对外 --------------------------------------------------------------
|
|
def start(self):
|
|
self.bus.start()
|
|
self.bus.subscribe(topic(self.id, "cmd"))
|
|
self.bus.publish(topic(self.id, "state"),
|
|
json.dumps({"online": True, "fw": self.fw, "ip": "192.168.8.200"}),
|
|
retain=True)
|
|
|
|
def go_offline(self):
|
|
self.bus.publish(topic(self.id, "state"), '{"online":false}', retain=True)
|
|
self.bus.stop()
|
|
|
|
def set_phone(self, rail: int, state: str):
|
|
self.rails[rail - 1].phone = state
|
|
|
|
def tick(self, now_ms: int, publish: bool = True):
|
|
dt = now_ms - self.now if self.now else 0
|
|
self.now = now_ms
|
|
self.up_ms += dt
|
|
self._ota_step()
|
|
for r in self.rails:
|
|
if r.act != "idle" and now_ms >= r.busy_until:
|
|
self._finish(r)
|
|
r.ua = self._current(r)
|
|
if publish:
|
|
self.bus.publish(topic(self.id, "tele"), json.dumps(self.status_body()), qos=0)
|
|
|
|
# ---- 行为 --------------------------------------------------------------
|
|
def _current(self, r: SimRail) -> int:
|
|
if not r.present or not r.en or r.act == "hr-off":
|
|
return 0
|
|
if r.phone == OFF:
|
|
return 800 # 关机漏电流
|
|
if r.phone == HUNG:
|
|
return 412000 # 一条平线
|
|
return max(250000, min(950000, r.ua + self.rng.randint(-60000, 60000)))
|
|
|
|
def _finish(self, r: SimRail):
|
|
if r.pending == "restart":
|
|
r.phone = NORMAL
|
|
r.ua = 500000
|
|
r.pending = None
|
|
r.en = True if r.act == "hr-off" else r.en
|
|
r.act = "idle"
|
|
|
|
def _ota_step(self):
|
|
"""C2：下载进度；下完就绪，带 reboot 就「重启」到新版本（五路全断，按意图恢复不模拟）。"""
|
|
o = self.ota
|
|
if o["pend"]:
|
|
o["pend"] = False # 上一圈重启起来了，这一圈连上 broker ⇒ 自证健康
|
|
self._ev(0, 18, 4)
|
|
if o["st"] != "dl":
|
|
return
|
|
o["got"] = min(o["size"], o["got"] + self.ota_rate)
|
|
if o["got"] < o["size"]:
|
|
return
|
|
o["st"] = "ready"
|
|
self._ev(0, 18, 2, 0, o["size"])
|
|
if self.ota_req.get("reboot"):
|
|
self._ota_reboot()
|
|
|
|
def _ota_reboot(self):
|
|
self._ev(0, 18, 6)
|
|
self.fw = "ota-" + self.ota_req["sha256"][:8]
|
|
self.up_ms = 0
|
|
self.ota = {"st": "idle", "got": 0, "size": 0, "err": None, "pend": True, "bad": None}
|
|
self.ota_req = None
|
|
self._ev(0, 1, 3)
|
|
|
|
def _ev(self, rail: int, typ: int, arg: int, mv: int = 0, ua: int = 0):
|
|
self.events.append([len(self.events), self.up_ms, rail, typ, arg, mv, ua])
|
|
|
|
def status_body(self) -> dict:
|
|
rails = []
|
|
for r in self.rails:
|
|
ok = r.present
|
|
rails.append({"r": r.rail, "pr": r.present, "en": r.en, "vb": r.vb, "set": r.set_mv,
|
|
"mv": (r.set_mv if r.en else 0) if ok else None,
|
|
"ua": r.ua if ok else None, "pg": r.en if ok else None,
|
|
"flt": False if ok else None,
|
|
"chg": max(0, (r.chg_until - self.now + 999) // 1000) if r.chg_until else 0,
|
|
"st": r.act, "res": "-"})
|
|
return {"ok": True, "fw": self.fw, "up": self.up_ms, "in": {"mv": 12030, "ua": 1500000},
|
|
"rails": rails, "temp": [35000, 36500],
|
|
"fan": {"duty": max(300, self.fan_floor), "floor": self.fan_floor, "en": True,
|
|
"rpm": [3300, 3280], "stall": [False, False]}, "derate": False,
|
|
"safety": {"cycles": self.up_ms // 20, "skipped": 0, "i2c_err": 0, "events": 0},
|
|
"ota": dict(self.ota), "ev_next": len(self.events), "ev_pending": 0,
|
|
"cmd": {"lines": len(self.received), "dups": 0, "rej": 0}}
|
|
|
|
def _on_message(self, t: str, payload: bytes):
|
|
try:
|
|
req = json.loads(payload)
|
|
except ValueError:
|
|
return self._reply({"ok": False, "err": "parse"})
|
|
self.received.append(req)
|
|
seq = req.get("seq")
|
|
for s, c, rsp in self.seen:
|
|
if s == seq:
|
|
out = dict(rsp) if c == req.get("cmd") else {"seq": seq, "ok": False, "err": "arg"}
|
|
out["dup"] = True
|
|
return self._reply(out)
|
|
rsp = self._exec(req)
|
|
if seq is not None:
|
|
rsp = dict({"seq": seq}, **rsp)
|
|
if req.get("cmd") not in ("status", "events", "net") and seq is not None:
|
|
self.seen.append((seq, req.get("cmd"), rsp))
|
|
self.seen = self.seen[-32:]
|
|
self._reply(rsp)
|
|
|
|
def _reply(self, rsp: dict):
|
|
if self.drop_next_replies > 0:
|
|
self.drop_next_replies -= 1
|
|
return
|
|
self.bus.publish(topic(self.id, "rsp"), json.dumps(rsp))
|
|
|
|
def _exec(self, req: dict) -> dict:
|
|
cmd = req.get("cmd")
|
|
if cmd == "status":
|
|
return self.status_body()
|
|
if cmd == "events":
|
|
start = max(0, req.get("from", 0))
|
|
ev = self.events[start:start + 8]
|
|
return {"ok": True, "oldest": 0, "next": len(self.events), "ev": ev}
|
|
if cmd == "fan":
|
|
self.fan_floor = req["duty"]
|
|
return {"ok": True, "floor": self.fan_floor}
|
|
if req.get("seq") is None:
|
|
return {"ok": False, "err": "no_seq"}
|
|
if cmd == "ota":
|
|
if not (isinstance(req.get("url"), str) and req["url"].startswith("http://")
|
|
and len(req["url"]) <= 64 and isinstance(req.get("sha256"), str)
|
|
and len(req["sha256"]) == 64 and isinstance(req.get("size"), int)
|
|
and isinstance(req.get("reboot", False), bool)):
|
|
return {"ok": False, "err": "arg"}
|
|
if self.ota["st"] == "dl":
|
|
return {"ok": False, "err": "busy"}
|
|
if self.ota["pend"]:
|
|
return {"ok": False, "err": "state"}
|
|
if self.ota["st"] == "ready" and self.ota_req and self.ota_req["sha256"] == req["sha256"]:
|
|
if req.get("reboot"):
|
|
self.ota_req = dict(self.ota_req, reboot=True)
|
|
self._ota_reboot()
|
|
return {"ok": True, "act": "reboot"}
|
|
return {"ok": True, "act": "ready"}
|
|
self.ota = {"st": "dl", "got": 0, "size": req["size"], "err": None, "pend": False, "bad": None}
|
|
self.ota_req = dict(req)
|
|
self._ev(0, 18, 1, 0, req["size"])
|
|
return {"ok": True, "act": "start"}
|
|
rail = req.get("rail")
|
|
if not isinstance(rail, int) or not 1 <= rail <= 5:
|
|
return {"ok": False, "err": "range"}
|
|
r = self.rails[rail - 1]
|
|
if not r.present:
|
|
self._ev(rail, 13, 0)
|
|
return {"ok": False, "err": "empty"}
|
|
if r.act != "idle" and cmd in ("power_cycle", "key", "edl", "vbus", "power"):
|
|
return {"ok": False, "err": "busy"}
|
|
if cmd == "power_cycle":
|
|
r.en, r.act, r.busy_until, r.pending = False, "hr-off", self.now + 4000, "restart"
|
|
self._ev(rail, 12, 0x21)
|
|
return {"ok": True}
|
|
if cmd == "key":
|
|
r.act, r.busy_until = "key-hold", self.now + req["ms"]
|
|
if req["keys"] == "pwr" and req["ms"] >= 10000:
|
|
r.pending = "restart"
|
|
self._ev(rail, 12, 0x20)
|
|
return {"ok": True}
|
|
if cmd == "edl":
|
|
r.act, r.busy_until = "edl-enum", self.now + 3000
|
|
return {"ok": True}
|
|
if cmd == "vbus":
|
|
if req["on"] and not r.en:
|
|
return {"ok": False, "err": "state"}
|
|
r.vb = req["on"]
|
|
return {"ok": True}
|
|
if cmd == "power":
|
|
r.en = req["on"]
|
|
if not r.en:
|
|
r.vb = False
|
|
if r.chg_until > self.now:
|
|
r.chg_until = 0
|
|
self._ev(rail, 21, 0)
|
|
return {"ok": True}
|
|
if cmd == "flash_mode": # 固件 chgmode.h 的规则：最多 2 路、关着电不许开、关电一并退出
|
|
if "add" in req: # 0.1.6 加钟：只给开着的加，剩余截断到 120 分钟
|
|
if r.chg_until <= self.now:
|
|
return {"ok": False, "err": "state"}
|
|
r.chg_until = min(r.chg_until + req["add"] * 60000, self.now + 120 * 60000)
|
|
self._ev(rail, 21, 0x80 | req["add"])
|
|
return {"ok": True, "left": (r.chg_until - self.now + 999) // 1000}
|
|
if not req["on"]:
|
|
if r.chg_until > self.now:
|
|
self._ev(rail, 21, 0)
|
|
r.chg_until = 0
|
|
return {"ok": True, "left": 0}
|
|
if not r.en:
|
|
return {"ok": False, "err": "state"}
|
|
active = sum(1 for x in self.rails if x is not r and x.chg_until > self.now)
|
|
if active >= 2:
|
|
return {"ok": False, "err": "state"}
|
|
m = req.get("min", 60)
|
|
r.chg_until = self.now + m * 60000
|
|
self._ev(rail, 21, m)
|
|
return {"ok": True, "left": m * 60}
|
|
if cmd == "set_voltage":
|
|
r.set_mv = req["mv"]
|
|
return {"ok": True, "set": r.set_mv}
|
|
return {"ok": False, "err": "unknown_cmd"}
|