DCU_DesktopController/scripts/check_recover_after_reset.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""GD32 断电/复位后一键恢复 EtherCAT，并**验收**恢复是否真生效。

用法（GD32 上电稳定后）:
  python scripts/check_recover_after_reset.py
  python scripts/check_recover_after_reset.py --skip-e2e

与旧版的差别：旧版全程只 print、没有任何失败路径（--skip-e2e 时无条件退 0），
所以"跑过了"和"恢复了"是两件事。现在每一步都有断言，退出码 0/1/2 是真门槛：
  0=全部通过  1=有断言失败（别去点页面，先修）  2=前置不满足（native/页面没起）
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import time
import urllib.request

ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, os.path.join(ROOT, "scripts"))
from guard_probe import Blocked, bridge_of, open_link, guard_of  # noqa: E402

ESP = os.environ.get("DCU_ESP_IP", "192.168.31.127")

FAILS: list[str] = []


def ck(cond, msg):
    print(("  OK   " if cond else "  FAIL ") + msg)
    if not cond:
        FAILS.append(msg)


def main() -> int:
    skip_e2e = "--skip-e2e" in sys.argv
    print("=== 1) WiFi motReset（可选，拉一下 GD32 应用层） ===")
    try:
        cfg = bytearray(24)
        cfg[0] = 1  # motReset bit0
        body = json.dumps({"ch": 2, "frames": [{"id": 0x00030301, "dlc": 24,
                                                "data": cfg.hex()}]}).encode()
        req = urllib.request.Request("http://%s/can/send_batch" % ESP, data=body,
                                     headers={"Content-Type": "application/json"},
                                     method="POST")
        t0 = time.perf_counter()
        with urllib.request.urlopen(req, timeout=6) as r:
            j = json.loads(r.read().decode("utf-8", "replace") or "{}")
        print("  wifi motReset %.0fms %s" % ((time.perf_counter() - t0) * 1000, j.get("status")))
    except Exception as e:
        print("  wifi motReset skip: %s" % e)     # 可选步骤，不作为门槛
    time.sleep(1.5)

    lk = open_link()
    print("=== 2) force_reopen + DC250 motor/start ===")
    try:
        h = lk.health()
        print("  before: dc=%s us=%s loop=%s op=%s wkc=%s"
              % (h.get("dc"), h.get("dc_cycle_us"), h.get("loop"), h.get("op"), h.get("wkc")))
    except Exception as e:
        print("  health fail: %s" % e)

    r = lk.api("/api/motor/start", {"period_ms": 0.25, "dc_cycle_us": 250,
                                    "dc_shift_us": 0, "force_reopen": True})
    print("  motor/start force_reopen ok=%s period=%s" % (r.get("ok"), r.get("period_ms")))
    print("  log: %s" % (r.get("log"),))
    ck(bool(r.get("ok")), "motor/start 回执 ok=true")

    # force_reopen 之后 loop 需要一点时间起来（冷启动见过 loop=0 的采样竞态），
    # 所以要"轮询到恢复"而不是只读一次；上限 15s。
    good = None
    t_end = time.time() + 15.0
    while time.time() < t_end:
        h = lk.health()
        if (h.get("dc") == 1 and h.get("dc_cycle_us") == 250 and h.get("loop") == 1
                and h.get("op") == 1 and (h.get("wkc") or 0) >= 3):
            good = h
            break
        time.sleep(0.5)
    ck(good is not None,
       "15s 内 health 达到 dc=1/250us/loop=1/op=1/wkc>=3（现在 %s）"
       % ({k: (good or h).get(k) for k in ("dc", "dc_cycle_us", "loop", "op", "wkc")}))

    g = guard_of(lk.health())
    t0 = g["alive_ticks"]
    time.sleep(1.0)
    g1 = guard_of(lk.health())
    ck(g1["alive_ticks"] - t0 >= 3,
       "守卫心跳 +%d/1s，armed=%s fires=%s（fires 若在本步之后才涨，说明是守卫救回来的）"
       % (g1["alive_ticks"] - t0, g1["armed"], g1["fires"]))

    print("=== 3) 发一帧看 GD32 出口 ===")
    pre = bridge_of(lk.state())
    br_pre = (pre.get("tx_count"), pre.get("ack"))
    r, ms = lk.send()
    print("  send %.1fms ok=%s queued=%s" % (ms, r.get("ok"), r.get("queued")))
    t_sent = time.perf_counter()
    outlet = None
    while time.perf_counter() - t_sent < 3.0:
        br = bridge_of(lk.state())
        if (br.get("tx_count"), br.get("ack")) != br_pre and br.get("tx_count") is not None:
            outlet = (time.perf_counter() - t_sent) * 1000.0
            print("  出口 %.0fms tx/ack %s → %s"
                  % (outlet, br_pre, (br.get("tx_count"), br.get("ack"))))
            break
        time.sleep(0.002)
    ck(outlet is not None,
       "3s 内 GD32 出口跟上（否则就是输出 PDO 没被消费，不是网页慢）")
    if outlet is None:
        print("  → 仍失败：先确认 GD32 已完整上电跑 APP；若 health 全绿而出口恒冻，"
              "等 8s 看 guard.fires 是否 +1（守卫会自动 close→open），"
              "或跑 tools/ecat/check_guard_regression.py 定位。")

    if not skip_e2e:
        print("\n=== 4) 网页全链路自测 ===")
        py = os.environ.get("MIMO_PYTHON") or sys.executable
        script = os.path.join(ROOT, "scripts", "check_web_send_latency.py")
        rc = subprocess.call([py, script, "--n", "6"], cwd=ROOT)
        ck(rc == 0, "check_web_send_latency.py 退出码 %d" % rc)

    print("\nVERDICT %s%s" % ("PASS" if not FAILS else "FAIL",
                              "" if not FAILS else " " + "; ".join(FAILS[:3])))
    return 0 if not FAILS else 1


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Blocked as e:
        print("BLOCKED 前置不满足：%s" % e)
        print("VERDICT BLOCKED")
        sys.exit(2)