rack：单板页 Hub 行与拓扑图、单路操作拆成 RailOps；开机按键改 5 s

Co-Authored-By: Claude Opus 5.5 <[email protected]>

rack/backend/prrack/api.py

@@ -195,6 +195,16 @@ def create_app(hub: Hub, rec: RecoveryManager, static_dir: Optional[str] = None,
                                          "断电自动退出" if a == protocol.CHG_EV_DROPPED else
                                          "加钟 %d 分钟" % (a & 0x7F) if a & protocol.CHG_EV_EXTEND else
                                          "开 %d 分钟" % a)
+            elif d["type"] == 22:
+                a = d["arg"]
+                if a & protocol.HUB_EV_MODE:
+                    m = protocol.HUB_MODES[a & 0x0F] if (a & 0x0F) < len(protocol.HUB_MODES) else str(a & 0x0F)
+                    d["name"] += "（模式改为%s）" % protocol.HUB_MODE_NAMES.get(m, m)
+                elif a == protocol.HUB_EV_ON:
+                    d["name"] += "（唤醒：%s）" % protocol.HUB_WHY_NAMES.get(d["mv"], d["mv"])
+                else:
+                    d["name"] += "（休眠）"
+                d["mv"] = None  # mV 列装的是理由码，不是电压 —— 别让界面显示成「0.002 V」
             elif d["type"] == 19:
                 failed = bool(d["arg"] & protocol.BOOT_CHECK_FAILED)
                 d["name"] += "（按了 %d 次，%s）" % (d["arg"] & 0x0F, "没开起来" if failed else "开起来了")

rack/backend/prrack/protocol.py

@@ -39,8 +39,17 @@ EVENT_TYPES: Dict[int, Tuple[str, str]] = {
     19: ("BOOT_CHECK", "开机重按"),  # 0.1.2：arg bit0~3 = 按了几次，bit7 = 最终没开起来（电源键弹片接触不良）
     20: ("BLADE_MUTE", "刀片不应答关路"),  # 0.1.4：eFuse 开着、本路 INA226 + PCA9555 连续 200 ms 都不应答
     21: ("CHG_MODE", "刷机模式"),  # 0.1.5：arg 1~120 = 开（分钟），0 = 手动关，255 = 到期；0.1.6：0x80|分钟 = 加钟；0.1.13：254 = 断电自动退出
+    22: ("HUB", "Hub 启停"),  # 0.1.14：arg 0 = 休眠、1 = 唤醒、0x10|模式 = 改模式；mV 列 = 唤醒理由（HUB_WHY）
 }
 
+# LOG_EV_HUB 的 arg 与理由（firmware/main/hubctl.h）。⚠️ 冻结。
+HUB_EV_OFF, HUB_EV_ON, HUB_EV_MODE = 0, 1, 0x10
+HUB_MODES = ("auto", "on")  # hub_mode_t 的下标
+HUB_MODE_NAMES = {"auto": "按需", "on": "常开"}
+HUB_WHY_NAMES = {0: "休眠", 1: "常开", 2: "有手机开着 VBUS", 3: "刷机模式", 4: "执行器在跑序列",
+                 5: "手动保持", 6: "MQTT 断线", 7: "空闲倒计时"}
+HUB_HOLD_MAX_MIN = 240
+
 # LOG_EV_CHG_MODE 的 arg（firmware/main/chgmode.h）。⚠️ 冻结。
 CHG_EV_OFF, CHG_EV_EXPIRED, CHG_EV_EXTEND = 0, 255, 0x80
 CHG_EV_DROPPED = 254  # 0.1.13：这一路断了电（刀片拔走 / 安全判据关路），自动退出。⚠️ 带 0x80 位，要先于「加钟」判
@@ -59,7 +68,7 @@ BOOT_CHECK_FAILED = 0x80  # LOG_EV_BOOT_CHECK 的 arg bit7
 # cmd_id_t（firmware/main/cmd.h），CMD_REJECT 事件的 arg 就是它。
 CMD_NAMES = {1: "status", 2: "set_voltage", 3: "power_cycle", 4: "key", 5: "edl", 6: "vbus",
              7: "fan", 8: "events", 9: "power", 10: "wifi", 11: "mqtt", 12: "net",
-             13: "ota", 14: "flash_mode"}
+             13: "ota", 14: "flash_mode", 15: "hub"}
 
 KEYS = ("pwr", "vup", "vdn")
 KEY_MAX_MS = 15000
@@ -128,6 +137,22 @@ def build_command(req: dict) -> dict:
                 raise CmdError("刷机模式时长必须是 1~%d 分钟" % FLASH_MODE_MAX_MIN)
             out["min"] = m
         return out
+    if cmd == "hub":
+        # 0.1.14 Hub 按需启停（整板命令）：mode = auto / on（板子存 NVS）；min = 保持 1~240 分钟，
+        # 0 = 取消保持并结束空闲倒计时（没别的理由就马上休眠）。两个都可以给，至少给一个。
+        out = {"cmd": cmd}
+        if req.get("mode") is not None:
+            if req["mode"] not in HUB_MODES:
+                raise CmdError("Hub 模式只能是 auto / on")
+            out["mode"] = req["mode"]
+        if req.get("min") is not None:
+            m = req["min"]
+            if not isinstance(m, int) or isinstance(m, bool) or not 0 <= m <= HUB_HOLD_MAX_MIN:
+                raise CmdError("Hub 保持时长必须是 0~%d 分钟" % HUB_HOLD_MAX_MIN)
+            out["min"] = m
+        if len(out) == 1:
+            raise CmdError("hub 至少要给 mode 或 min")
+        return out
     if cmd == "fan":
         duty = req.get("duty")
         if not isinstance(duty, int) or isinstance(duty, bool) or not 0 <= duty <= 1000:

rack/backend/prrack/simboard.py

@@ -51,6 +51,7 @@ class SimBoard:
         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）
+        self.hub = {"on": True, "mode": "auto", "why": "idle", "hold": 0, "idle": 600}  # 0.1.14，只回显不计时
         bus.on_message = self._on_message
         self._ev(0, 1, 0)  # BOOT
 
@@ -141,6 +142,7 @@ class SimBoard:
                 "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},
+                "hub": dict(self.hub),
                 "ota": dict(self.ota), "ev_next": len(self.events), "ev_pending": 0,
                 "cmd": {"lines": len(self.received), "dups": 0, "rej": 0}}
 
@@ -183,6 +185,29 @@ class SimBoard:
             return {"ok": True, "floor": self.fan_floor}
         if req.get("seq") is None:
             return {"ok": False, "err": "no_seq"}
+        if cmd == "hub":  # 0.1.14：固件 hubctl.h 的命令形状；不模拟倒计时
+            h = self.hub
+            if "mode" in req:
+                if req["mode"] not in ("auto", "on"):
+                    return {"ok": False, "err": "arg"}
+                if req["mode"] != h["mode"]:
+                    self._ev(0, 22, 0x10 | ("auto", "on").index(req["mode"]))
+                h["mode"] = req["mode"]
+            if "min" in req:
+                if not 0 <= req["min"] <= 240:
+                    return {"ok": False, "err": "range"}
+                h["hold"] = req["min"] * 60
+                if req["min"] == 0:
+                    h["idle"] = 0
+            if "mode" not in req and "min" not in req:
+                return {"ok": False, "err": "arg"}
+            want = h["mode"] == "on" or h["hold"] > 0 or h["idle"] > 0 or any(r.vb for r in self.rails)
+            if want != h["on"]:
+                self._ev(0, 22, 1 if want else 0, 5 if want else 0)
+            h["on"] = want
+            h["why"] = ("mode" if h["mode"] == "on" else "vbus" if any(r.vb for r in self.rails) else
+                        "hold" if h["hold"] else "idle" if h["idle"] else "off") if want else "off"
+            return {"ok": True, "hub": dict(h)}
         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)

rack/backend/tests/test_api.py

@@ -218,3 +218,23 @@ def test_flash_mode_extend_round_trip(env):
     ev = {e["seq"]: e for e in c.get("/api/events", params={"board": B}).json()}
     assert "加钟 30 分钟" in ev[9101]["name"]
     assert "断电自动退出" in ev[9102]["name"]  # 0.1.13：254 带 0x80 位，不能被当成「加钟 126 分钟」
+
+
+def test_hub_round_trip_and_event_names(env):
+    # 固件 0.1.14：hub 命令经 /cmd 放行、回应带 hub 对象；事件 22 按 arg 命名、mV 列不当电压显示
+    c, hub, sim, db = env
+    post = lambda body: c.post("/api/boards/%s/cmd" % B, json=body).json()
+    r = post({"cmd": "hub", "mode": "on"})
+    assert r["ok"] is True and r["hub"]["mode"] == "on" and r["hub"]["on"] is True
+    r = post({"cmd": "hub", "mode": "auto", "min": 0})
+    assert r["hub"]["on"] is False and r["hub"]["why"] == "off"
+    assert post({"cmd": "status"})["hub"]["mode"] == "auto"
+    assert c.post("/api/boards/%s/cmd" % B, json={"cmd": "hub", "min": 999}).status_code == 400
+    now = 10 ** 13
+    db.many("INSERT INTO events (board_id, seq, t_ms, rail, type, arg, mv, ua, received) VALUES (?,?,?,?,?,?,?,?,?)",
+            [(B, 9201, 1, 0, 22, 1, 2, 0, now), (B, 9202, 2, 0, 22, 0, 0, 0, now),
+             (B, 9203, 3, 0, 22, 0x11, 7, 0, now)])
+    ev = {e["seq"]: e for e in c.get("/api/events", params={"board": B}).json()}
+    assert "唤醒：有手机开着 VBUS" in ev[9201]["name"] and ev[9201]["mv"] is None
+    assert "休眠" in ev[9202]["name"] and ev[9202]["severe"] is False
+    assert "模式改为常开" in ev[9203]["name"]

rack/backend/tests/test_core.py

@@ -320,6 +320,19 @@ def test_flash_mode_command_and_event_numbers():
             protocol.build_command(bad)
 
 
+def test_hub_command_and_event_numbers():
+    # 固件 0.1.14：命令 15 / 事件 22（cmd.h / evlog.h 冻结编号）
+    assert protocol.CMD_NAMES[15] == "hub" and protocol.EVENT_TYPES[22][0] == "HUB"
+    assert protocol.HUB_MODES == ("auto", "on") and protocol.HUB_WHY_NAMES[2] == "有手机开着 VBUS"
+    assert protocol.build_command({"cmd": "hub", "mode": "on"}) == {"cmd": "hub", "mode": "on"}
+    assert protocol.build_command({"cmd": "hub", "min": 0}) == {"cmd": "hub", "min": 0}
+    assert protocol.build_command({"cmd": "hub", "mode": "auto", "min": 30}) ==         {"cmd": "hub", "mode": "auto", "min": 30}
+    for bad in ({"cmd": "hub"}, {"cmd": "hub", "mode": "off"}, {"cmd": "hub", "min": 241},
+                {"cmd": "hub", "min": -1}, {"cmd": "hub", "min": True}):
+        with pytest.raises(protocol.CmdError):
+            protocol.build_command(bad)
+
+
 def test_flash_mode_extend_command():
     assert protocol.build_command({"cmd": "flash_mode", "rail": 1, "add": 30}) ==         {"cmd": "flash_mode", "rail": 1, "add": 30}
     for bad in ({"cmd": "flash_mode", "rail": 1, "add": 0},