diff --git a/.gitignore b/.gitignore index d12dae2..e7afc78 100644 --- a/.gitignore +++ b/.gitignore @@ -38,6 +38,9 @@ scripts/selftest_*.py # _ 前缀：现场排障探针、预览截图、一次性驱动 scripts/_* # 一次性分析/采样/测量/追踪脚本 +# 注：验收链上的脚本一律用 check_ 前缀入库（check_web_send_latency / +# check_recover_after_reset / check_click_to_motion / check_guard_*）， +# 下面这些 e2e_/measure_/capture_ 通配只挡新的一次性草稿 scripts/anal_*.py scripts/collect_*.py scripts/capture_*.py @@ -45,7 +48,6 @@ scripts/measure_*.py scripts/e2e_*.py scripts/trace_*.py scripts/flash_gd32_ecat.py -scripts/recover_ecat_after_reset.py scripts/test_ecat_version_*.py scripts/test_gd32_restart_version.py scripts/test_motor_ui_state.* diff --git a/README.md b/README.md index ce259bb..8b3ff6c 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,8 @@ esp32_app.vbs esp32_web.vbs serve.bat 三个启动入口 | `test_app_lease.py` / `test_app_lease_e2e.mjs` | 关窗停总线 | | `test_esf_pack.py` | PDO 快发 ESF1 打包 | | `check_esf_latency.py` / `check_latency_breakdown.py` / `check_e2e_click_motion.py` | 快发与点击延迟测量 | +| `check_web_send_latency.py` / `check_recover_after_reset.py` / `check_click_to_motion.py` | 网页全链路自测（原 `e2e_web_send_latency.py`）、GD32 复位后恢复并验收、点击→STATUS 动作 | +| `check_guard_idle_silence.py` / `check_guard_fires_hw.py` / `guard_probe.py` | tx 冻结守卫真机门槛：健康总线不许被重连 / 制造冻结证明守卫真会触发；共用前置与通路（改 native 后跑，见 `tools/ecat/MIGRATION.md` 加固段） | | `browser_check_{refresh,tick,floor,period}.mjs` | 浏览器侧 DOM/节拍实测 | | `probe_ecat_slot_atomicity.py` / `probe_ecat_vs_wifi.py` | 可复用采样器（gitignore 白名单） | @@ -389,6 +391,13 @@ esp32_app.vbs esp32_web.vbs serve.bat 三个启动入口 `_fill_can_out` 逐字节对拍通过后固化的；改 `fill_can_out`/`pack_tx_batch` 后必跑。 原 16 项假从站后端自测（含窗口自适应 SparseSlave 专项）已随 Python 后端删除， 其覆盖结论记录在 `tools/ecat/MIGRATION.md`。 + - `G:\Robot_v2\DCU\tools\ecat\check_guard_regression.bat` —— **tx 冻结守卫**回归门槛。 + 现场症状：点一次「发送 CMD」后 GD32 不消费输出 PDO，health 全绿而 + `bridge.tx_count/ack` 冻死，看着像网页慢，实为 EtherCAT 楔死。默认三段 + （纯逻辑 38 项 / 源码契约 / 构建指纹）不碰网卡、十几秒出结论；`--live` 加稳定性 + 套件与 90s 空闲静默；`--live --wedge` 加真机触发（要按一次 GD32 RESET，电机停一下）。 + SKIP 不折算 PASS：有段没跑就退 2。改 `wedge_worker`/`guard_tx.h`/发送路由后必跑； + 30 秒自检与判据契约见 `tools/ecat/MIGRATION.md` 的「加固 2026-09-21」段。 - `node scripts/test_ecat_autostart.mjs` —— 自动接入链条（stub 后端， 不依赖板子，19 项覆盖开网卡/幂等/换网卡/各级失败语义）。 - `node scripts/test_motor_refresh.mjs` —— 刷新速度每链路一份（55 项， diff --git a/docs/compose/spec/host-code-quality.md b/docs/compose/spec/host-code-quality.md index 8d60c37..15bbfc3 100644 --- a/docs/compose/spec/host-code-quality.md +++ b/docs/compose/spec/host-code-quality.md @@ -48,7 +48,12 @@ commits: 2c6235d..8043d8e（含评估、P0/P2/P3、界面回滚与安全重做 s - `test_esf_pack.py` PASS；`test_motor_refresh.mjs` 62/62 PASS（前端重做后） - `node --check` main/scope/seq/ota PASS - UI 门禁（Edge→:2625/motor.html）：控制台无报错；绿灯 AL OP；N1–N3 有数据；离线遮罩 `display:none` -- 真机 `e2e_web_send_latency.py`（重做后两轮）：均为 **PASS 26 / FAIL 1**，exit=1。唯一失败项固定为 **`motor/start`（force_reopen）后立刻读 health 显示 `loop=0`**；同轮 `dc=1 dc_us=250 wkc=3`，且 `motor/send`×6 与 GD32 出口 tx/ack 均前进（HTTP p50≈1.9–5.4ms，出口 p50≈2.1–2.3ms，点击→STATUS≈14ms），WiFi 对照可用。判定为 **e2e 对 force_reopen 后 health 的采样竞态/口径问题**，不是发送链路回归；界面侧数据正常。 +- 真机 `check_web_send_latency.py`（原 `e2e_web_send_latency.py`；现随守卫加固改名入库， + 并新增第 6 段读 `health.guard` 可观测面。下表为改名前两轮的结果）：均为 **PASS 26 / FAIL 1**，exit=1。唯一失败项固定为 **`motor/start`（force_reopen）后立刻读 health 显示 `loop=0`**；同轮 `dc=1 dc_us=250 wkc=3`，且 `motor/send`×6 与 GD32 出口 tx/ack 均前进（HTTP p50≈1.9–5.4ms，出口 p50≈2.1–2.3ms，点击→STATUS≈14ms），WiFi 对照可用。判定为 **e2e 对 force_reopen 后 health 的采样竞态/口径问题**，不是发送链路回归；界面侧数据正常。 +- 2026-09-21 续：上述 `loop=0` 竞态现已由 `check_recover_after_reset.py` 改成"轮询到 + 恢复为止（上限 15s）"，不再单点采样；同轮补上守卫可见面断言（`alive_ticks` 心跳、 + `fires`、`tx_base`）。tx 冻结守卫为何会悄悄失效、以及"必须触发/不许误触发"两极 + 的测试契约，见 `tools/ecat/MIGRATION.md` 的「加固 2026-09-21」段。 **Journey log** — - 大拆 P1 曾导致界面全无数据 → 用户要求回滚 + 保留现场 + 重做 diff --git a/scripts/check_click_to_motion.py b/scripts/check_click_to_motion.py new file mode 100644 index 0000000..5e67d16 --- /dev/null +++ b/scripts/check_click_to_motion.py @@ -0,0 +1,169 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""测量「点击发送」到「电机 STATUS 显示开始运动」的端到端延迟。 + +口径说明（务必读）： + t_click = 脚本在发起 motor/send 之前的时间戳 + ≈ 网页点击处理完、即将 fetch 的时刻（未含浏览器主线程/桥调用开销） + t_http = motor/send HTTP 返回（控制面入队完成） + t_can = GD32 bridge.tx_count/ack 变化（EtherCAT 出口，命令已进 CAN） + t_motion = 电机 STATUS 中 rtAngle/rtVelocity 相对基线发生可测变化 + （软件可见的「开始动」代理，不是编码器轴端绝对时刻） + +用法: + python scripts/measure_click_to_motion.py + python scripts/measure_click_to_motion.py --node 3 --ch 2 --vel 40 --timeout 3 +""" +from __future__ import annotations + +import argparse +import json +import os +import struct +import sys +import time +import urllib.request + +PAGE = os.environ.get("DCU_PAGE_BASE", "http://127.0.0.1:2625") +ECAT = PAGE + "/ecat/api" + + +def post(url, obj, timeout=8.0): + data = json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST") + t0 = time.perf_counter() + with urllib.request.urlopen(req, timeout=timeout) as r: + body = r.read() + return (time.perf_counter() - t0) * 1000.0, json.loads(body.decode("utf-8", "replace") or "{}") + + +def decode_status(data_hex: str): + if not data_hex or len(data_hex) < 36: + return None + raw = bytes.fromhex(data_hex[:72] if len(data_hex) >= 72 else data_hex) + if len(raw) < 21: + return None + + def f32(off): + if off + 4 > len(raw): + return 0.0 + v = struct.unpack_from("0.3° 或速度偏离基线 >5 + if t_motion is None and (d_ang > 0.3 or d_vel > 5.0): + t_motion = now + motion_detail = ( + f"Δangle={d_ang:.3f}° Δvel={d_vel:.2f} " + f"angle {ang0:.2f}→{st['rtAngle']:.2f} vel {vel0:.2f}→{st['rtVelocity']:.2f}" + ) + if t_motion is not None and t_can is not None: + break + time.sleep(0.002) + + stop = build_cmd(mode=0, voltage=24.0, current=0, velocity=0, angle=0) + send_cmd(ch, node, stop) + + def ms(dt): + return None if dt is None else dt * 1000.0 + + print("\n=== 端到端（脚本 t_click ≈ 点击后即将发请求）===") + print(f" t_click → HTTP 返回 : {ms(t_http - t_click):.1f} ms") + print(f" t_click → GD32 CAN 出口 : {ms(t_can - t_click) if t_can else '未观测到'} ms {can_detail or ''}") + print(f" t_click → STATUS 开始动 : {ms(t_motion - t_click) if t_motion else '超时未观测到'} ms {motion_detail or ''}") + if t_can: + print(f" HTTP 返回 → CAN 出口 : {ms(t_can - t_http):.1f} ms") + if t_motion and t_can: + print(f" CAN 出口 → STATUS 动 : {ms(t_motion - t_can):.1f} ms ← 固件/回报段") + if t_motion: + print(f"\n软件口径「点击→STATUS 显示开始动」≈ {ms(t_motion - t_click):.1f} ms") + print("（不含真实鼠标事件前的浏览器调度；机械轴端时刻会更晚）") + return 0 + print("\nFAIL: 超时未在 STATUS 中看到运动（电机未响应或回报未更新）") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/check_guard_fires_hw.py b/scripts/check_guard_fires_hw.py new file mode 100644 index 0000000..3567c51 --- /dev/null +++ b/scripts/check_guard_fires_hw.py @@ -0,0 +1,209 @@ +"""真机正向验证：主动制造输出冻结，证明守卫**真的**会触发，且 60s 内不第二次触发。 + +这套判据历史上"修过一次又复发"，而复发时页面全绿、wr 恒 0，没有任何测试断言它 +必须触发。纯逻辑门槛（test_guard_logic）已经能离线判死，但它证明的是"函数"， +这条证明的是"函数确实挂在这台机器的总线上工作"。 + +流程（两处物理动作都要人在场，脚本会停下来等）： + 1. 前置：bus/op/loop=1、dc=250us、wkc>=3、pm_hold=0、recovering=0、心跳在涨； + 并要求 since_full_s>61（60s 冷却是进程全局的，早先任何一次恢复都会吃掉预算） + 2. 证明通路活着：发一帧，出口 tx_count/ack 必须动 + 3. → 请按 GD32 的 RESET：主站继续打包（cycles 涨）而出口冻住 = 现场症状复现 + 4. 只发一次（就是"点一下"），然后盯 guard：armed 必须在 1s 内起来， + tx_stale_s 必须爬过 8s，fires 必须 +1（预算 25s：按 GD32 复位会连输入一起冻， + 输入侧判据可能先做一次 OP-only 恢复，其 10s 冷却会把本判据的补触发往后推） + 5. → 请让 GD32 重新跑起 APP：出口必须重新前进、dc/wkc/loop 恢复、再发一帧拿回显 + 6. 冷却：从触发那一刻起每 5s 一帧共 70s，fires 增量必须仍为 1 + 7.（可选 --with-pm-hold）再来一轮，但先 /api/pm/hold=1：要求 fires 零增量**且心跳 + 在涨** —— 证明"静音"是设计而不是线程死亡 + +用法：python scripts/check_guard_fires_hw.py [--with-pm-hold] +退出码：0=PASS 1=FAIL 2=BLOCKED（前置/不在场）。末行恒为 VERDICT。 +""" +import os +import sys +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) +from guard_probe import Blocked, bridge_of, guard_of, open_link, preflight, verdict # noqa: E402 + +NATIVE_LOG = os.path.abspath(os.path.join(HERE, os.pardir, os.pardir, "ecat", + "build_native", "native_stdout.log")) + + +def ask(prompt): + if not sys.stdin or not hasattr(sys.stdin, "isatty") or not sys.stdin.isatty(): + raise Blocked("这一步要人在场，但 stdin 不是终端：%s" % prompt) + raw = input(prompt + "（回车继续 / 输入 s 放弃本测试）: ").strip().lower() + if raw == "s": + raise Blocked("用户放弃") + + +def ck(fails, cond, msg): + print((" OK " if cond else " FAIL ") + msg) + if not cond: + fails.append(msg) + + +def log_tail(n=400): + try: + with open(NATIVE_LOG, "rb") as f: + f.seek(max(0, os.path.getsize(NATIVE_LOG) - 65536)) + return f.read().decode("utf-8", "replace").splitlines()[-n:] + except OSError: + return [] + + +def wait_freeze(lk, fails, budget=12.0): + """等"主站还在打包、出口冻住"——现场症状的定义。""" + t0 = time.time() + seen = None + while time.time() - t0 < budget: + h = lk.health() + st = lk.state() + br = bridge_of(st) + now = (br.get("tx_count"), br.get("ack"), h.get("cycles")) + if seen and seen[0] == now[0] and seen[1] == now[1] and now[2] > seen[2]: + print(" 出口已冻 %s（cycles 仍在涨 %s→%s），耗时 %.1fs" + % (now[0], seen[2], now[2], time.time() - t0)) + return True + seen = now + time.sleep(0.4) + ck(fails, False, "%.0fs 内没出现「cycles 涨而出口不动」——冻结没制造成功" % budget) + return False + + +def run_round(lk, fails, tag, pm_hold=False): + g0 = guard_of(lk.health()) + st0 = lk.state() + fires0 = g0["fires"] + wr0 = st0.get("wr", 0) + if pm_hold: + lk.api("/api/pm/hold", {"hold": 1}) + print(" pm_hold=1（守卫按设计静音；心跳必须仍在涨）") + ask("请按 GD32 板上 RESET（让 APP 停止消费输出 PDO）") + if not wait_freeze(lk, fails): + ask("请让 GD32 重新跑起 APP，然后继续") + return + _r, ms = lk.send() + print(" 单次发送完成（HTTP %.1fms）；开始盯 guard" % ms) + t_fire = None + armed_at = None + stale_max = 0.0 + t0 = time.time() + while time.time() - t0 < 25.0: + g = guard_of(lk.health()) + if g["armed"] and armed_at is None: + armed_at = time.time() - t0 + stale_max = max(stale_max, g.get("tx_stale_s") or 0) + if g["fires"] > fires0: + t_fire = time.time() - t0 + break + time.sleep(0.5) + hit = [l for l in log_tail() if "GUARD" in l and "全量重连" in l] + if pm_hold: + ck(fails, t_fire is None, "pm_hold 期间不得触发（%s）" + % ("没触发 ✔" if t_fire is None else "%.1fs 就触发了 ✗" % t_fire)) + g1 = guard_of(lk.health()) + ticks0 = g0["alive_ticks"] + time.sleep(1.2) + g2 = guard_of(lk.health()) + ck(fails, g2["alive_ticks"] - g1["alive_ticks"] >= 3, + "静音中心跳仍在涨（+%d/1.2s）—— 证明是设计静音而不是线程死了" + % (g2["alive_ticks"] - g1["alive_ticks"])) + lk.api("/api/pm/hold", {"hold": 0}) + ask("请让 GD32 重新跑起 APP，然后继续") + return + ck(fails, armed_at is not None and armed_at <= 1.5, + "armed 在 %.2fs 内起来（旧判据在这里就已经断了）" % (armed_at if armed_at is not None else -1)) + ck(fails, stale_max >= 8.0, "tx_stale_s 爬过 8s（峰值 %.1fs）" % stale_max) + ck(fails, t_fire is not None, + "fires %d→%d：守卫在 %.1fs 时触发（预算 25s）" + % (fires0, guard_of(lk.health())["fires"], t_fire if t_fire is not None else -1)) + ck(fails, bool(hit), "native_stdout.log 里有 [GUARD] … 全量重连 这一行（最近命中 %d 条）" + % len(hit)) + if t_fire is None: + print(" ⇒ 判据没触发：先 curl /api/native/health 看 guard 块，" + "再确认跑的是新 exe（tools/ecat/check_build_provenance.py）") + ask("请让 GD32 重新跑起 APP（守卫已做过 close→open，恢复应该很快）") + t1 = time.time() + resumed = False + while time.time() - t1 < 30.0: + h = lk.health() + if h.get("loop") == 1 and (h.get("wkc") or 0) >= 3 and h.get("dc_cycle_us") == 250: + pre = bridge_of(lk.state()) + _r, ms2 = lk.send() + lag, ok = None, False + t2 = time.perf_counter() + while time.perf_counter() - t2 < 2.0: + bb = bridge_of(lk.state()) + if (bb.get("tx_count") != pre.get("tx_count") + or bb.get("ack") != pre.get("ack")): + ok, lag = True, (time.perf_counter() - t2) * 1000.0 + break + time.sleep(0.002) + if ok: + resumed = True + print(" 恢复：出口重新前进（发送→跟上 %.1fms），probe HTTP %.1fms" % (lag, ms2)) + break + time.sleep(0.5) + ck(fails, resumed, "松手后出口重新前进且总线回到 250us/OP") + st1 = lk.state() + ck(fails, st1.get("wr", 0) > wr0, "wr %d→%d（全量重连计一次；wr 汇总所有恢复路径，" + "不能单独当本判据的凭据）" % (wr0, st1.get("wr", 0))) + + print(" 冷却观察：每 5s 一帧共 70s，期间 fires 不得再涨") + f_now = guard_of(lk.health())["fires"] + t3 = time.time() + extra = 0 + while time.time() - t3 < 70.0: + lk.send() + time.sleep(5.0) + g = guard_of(lk.health()) + if g["fires"] > f_now and time.time() - t3 < 60.0: + extra += 1 + f_now = g["fires"] + g_end = guard_of(lk.health()) + ck(fails, extra == 0 and g_end["fires"] <= fires0 + 2, + "冷却守住：本轮 fires %d→%d，60s 内额外触发 %d 次" % (fires0, g_end["fires"], extra)) + + +def main(): + fails = [] + rounds_pm = "--with-pm-hold" in sys.argv + lk = open_link() + preflight(lk) + + g = guard_of(lk.health()) + since_full = g.get("since_full_s", -1) + if since_full is not None and since_full >= 0 and since_full < 61: + wait = 62 - since_full + print("上次全量重连在 %.0fs 前，等 %.0fs 让 60s 冷却走完" % (since_full, wait)) + time.sleep(wait) + st = lk.state() + act = [m for m in (st.get("motors") or []) if m.get("online")] + if act: + print(" 注意：有 %d 个电机槽位 online。本测试会让总线断流并强制重连，" % len(act)) + ask("请确认电机已停止/未使能（否则输入 s 放弃）") + + print("\n=== 主轮：制造冻结，要求守卫触发 ===") + run_round(lk, fails, "main") + if rounds_pm: + print("\n=== 对照轮：pm_hold=1 时同样的冻结不得触发，但心跳必须仍在涨 ===") + run_round(lk, fails, "pm_hold", pm_hold=True) + + print("\n日志：%s" % NATIVE_LOG) + return verdict(not fails, "; ".join(fails[:4])) + + +if __name__ == "__main__": + try: + sys.exit(main()) + except Blocked as e: + print("BLOCKED %s" % e) + print("VERDICT BLOCKED") + sys.exit(2) + except KeyboardInterrupt: + print("VERDICT BLOCKED 中断（别忘了 GD32 可能还在 RESET 状态）") + sys.exit(2) diff --git a/scripts/check_guard_idle_silence.py b/scripts/check_guard_idle_silence.py new file mode 100644 index 0000000..23f7e7f --- /dev/null +++ b/scripts/check_guard_idle_silence.py @@ -0,0 +1,105 @@ +"""守卫的"另一极"真机门槛：健康的总线一次都不许被重连。 + +为什么要单独一个脚本：把 8s 判据"修好"最省事的写法是让它更容易触发，而反复 +force_reopen 会踢掉从站 1000ms 过程数据看门狗，把 SSC 邮箱楔得比原故障更难恢复。 +所以"必须触发"必须配一条"不许误触发"： + 阶段 A（默认 90s） 什么都不发。要求心跳在涨、armed 恒 0、fires/wr 零增量。 + 阶段 B（默认 25s） 1Hz 持续发送。要求每次都被 GD32 消费掉，且 fires 仍为 0。 + （这一条正是 T1b 的真机版：点一下就等出口动过 → 撤销。） + +前置不满足一律 BLOCKED（退出码 2），绝不折算成 PASS —— 判据在 loop=0 或 pm_hold=1 +时按设计是静音的，那种情况下"没触发"毫无意义。 + +用法：python scripts/check_guard_idle_silence.py [--seconds 90] [--send-seconds 25] +退出码：0=PASS 1=FAIL 2=BLOCKED（前置不满足）；末行恒为 VERDICT。 +""" +import os +import sys +import time + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from guard_probe import Blocked, bridge_of, guard_of, open_link, preflight, verdict # noqa: E402 + + +def arg(name, default): + return float(sys.argv[sys.argv.index(name) + 1]) if name in sys.argv else default + + +def main(): + idle_s = arg("--seconds", 90.0) + send_s = arg("--send-seconds", 25.0) + fails = [] + + def ck(cond, msg): + print((" OK " if cond else " FAIL ") + msg) + if not cond: + fails.append(msg) + + lk = open_link() + h0, g0 = preflight(lk) + st0 = lk.state() + fires0, wr0 = g0["fires"], st0.get("wr", 0) + + print("\n阶段 A：静默 %.0fs（有人碰页面都会让这一阶段失去意义）" % idle_s) + t_end = time.time() + idle_s + ticks_prev = g0["alive_ticks"] + armed_seen = 0 + slow = 0 + while time.time() < t_end: + time.sleep(1.0) + g = guard_of(lk.health()) + d = g["alive_ticks"] - ticks_prev + ticks_prev = g["alive_ticks"] + if d < 3: + slow += 1 + if g["armed"]: + armed_seen += 1 + ck(slow == 0, "心跳每次都前进（慢拍 %d 次）" % slow) + ck(armed_seen == 0, "静默期 armed 恒 0（看到 %d 次为 1）" % armed_seen) + gA = guard_of(lk.health()) + ck(gA["fires"] == fires0, "静默期本判据触发次数 %d→%d（必须不变）" % (fires0, gA["fires"])) + ck(lk.state().get("wr", 0) == wr0, "静默期 wr 不变（其它恢复路径也没动）") + + print("\n阶段 B：1Hz 发送 %.0fs，健康总线不得被当成楔死" % send_s) + t_end = time.time() + send_s + sent = consumed = 0 + worst_ms = 0.0 + tx_prev = bridge_of(lk.state()).get("tx_count") + while time.time() < t_end: + pre = bridge_of(lk.state()) + _r, ms = lk.send() + sent += 1 + tx_now, ok = None, False + t0 = time.perf_counter() + while time.perf_counter() - t0 < 2.0: + br = bridge_of(lk.state()) + if br.get("tx_count") != pre.get("tx_count") or br.get("ack") != pre.get("ack"): + tx_now, ok = br.get("tx_count"), True + break + time.sleep(0.002) + if ok: + consumed += 1 + worst_ms = max(worst_ms, (time.perf_counter() - t0) * 1000.0) + else: + print(" 注意：第 %d 次发送 2s 内出口无变化 pre=%s" % (sent, pre.get("tx_count"))) + tx_prev = tx_now if tx_now is not None else tx_prev + time.sleep(max(0.0, 1.0 - (time.perf_counter() - t0))) + gB = guard_of(lk.health()) + ck(consumed == sent, "出口跟上 %d/%d 次（最慢 %.1fms）" % (consumed, sent, worst_ms)) + ck(gB["fires"] == fires0, + "持续发送期间本判据触发次数仍为 %d（现在 %d）" % (fires0, gB["fires"])) + if gB["fires"] != fires0: + print(" ⇒ 误触发：判据把「点一次、那拍已消费」当成了楔死。检查 guard_tx.h 的" + "撤销是否仍写在武装之后、基线是否仍是发送时快照（tx_base）。") + + print("\n小结：fires %d→%d wr=%s 消费 %d/%d" % (fires0, gB["fires"], wr0, consumed, sent)) + return verdict(not fails, "; ".join(fails[:3])) + + +if __name__ == "__main__": + try: + sys.exit(main()) + except Blocked as e: + print("BLOCKED 前置不满足：%s" % e) + print("VERDICT BLOCKED") + sys.exit(2) diff --git a/scripts/check_recover_after_reset.py b/scripts/check_recover_after_reset.py new file mode 100644 index 0000000..21748e5 --- /dev/null +++ b/scripts/check_recover_after_reset.py @@ -0,0 +1,134 @@ +#!/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) diff --git a/scripts/check_web_send_latency.py b/scripts/check_web_send_latency.py new file mode 100644 index 0000000..6c1919b --- /dev/null +++ b/scripts/check_web_send_latency.py @@ -0,0 +1,288 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""模拟网页全链路自测（改 native/前端后必跑）。 + +覆盖浏览器实际路径（同源 :2625 /ecat/api/* + 静态资源）， +以及 GD32 出口（bridge.tx_count/ack）与 WiFi 对照。 + +用法: + python scripts/e2e_web_send_latency.py + python scripts/e2e_web_send_latency.py --n 8 +退出码: 0=PASS, 1=FAIL +""" +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +import time +import urllib.error +import urllib.request + +ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +PAGE = os.environ.get("DCU_PAGE_BASE", "http://127.0.0.1:2625") +ECAT = PAGE + "/ecat/api" +ESP = os.environ.get("DCU_ESP_IP", "192.168.31.127") + +PASS: list[str] = [] +FAIL: list[str] = [] + + +def ok(msg: str) -> None: + PASS.append(msg) + print(f" ✓ {msg}") + + +def bad(msg: str) -> None: + FAIL.append(msg) + print(f" ✗ {msg}") + + +def http_json(url: str, obj=None, timeout=8.0): + data = b"{}" if obj is None else json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + req = urllib.request.Request( + url, data=data, + headers={"Content-Type": "application/json", "Accept": "application/json"}, + method="POST" if obj is not None else "GET", + ) + t0 = time.perf_counter() + try: + with urllib.request.urlopen(req, timeout=timeout) as r: + body = r.read() + ms = (time.perf_counter() - t0) * 1000.0 + try: + return ms, r.status, json.loads(body.decode("utf-8", "replace") or "{}") + except Exception: + return ms, r.status, {"_raw_len": len(body)} + except Exception as e: + ms = (time.perf_counter() - t0) * 1000.0 + return ms, None, {"ok": False, "error": str(e)} + + +def http_get(url: str, timeout=5.0): + t0 = time.perf_counter() + try: + with urllib.request.urlopen(url, timeout=timeout) as r: + body = r.read() + return (time.perf_counter() - t0) * 1000.0, r.status, body + except Exception as e: + return (time.perf_counter() - t0) * 1000.0, None, str(e).encode() + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--n", type=int, default=6) + args = ap.parse_args() + + print("=== 1) 页面与静态资源（模拟浏览器打开电机页） ===") + ms, st, body = http_get(PAGE + "/pages/motor.html") + html = body.decode("utf-8", "replace") if isinstance(body, (bytes, bytearray)) else str(body) + if st == 200 and "motor.html" or st == 200: + ok(f"motor.html HTTP {st} ({ms:.1f}ms)") + else: + bad(f"motor.html 失败 st={st} {body[:80]!r}") + css_ref = re.search(r'motor\.css\?v=([0-9a-z]+)', html) + js_ref = re.search(r'main\.js\?v=([0-9a-z]+)', html) + if css_ref: + ok(f"CSS 引用 motor.css?v={css_ref.group(1)}") + else: + bad("motor.html 无 motor.css 引用") + if js_ref: + ok(f"JS 引用 main.js?v={js_ref.group(1)}") + else: + bad("motor.html 无 main.js 引用") + if "sendLatency" in html or "send-trace" in html: + bad("页面仍含延迟埋点（应已删除）") + else: + ok("页面无延迟埋点残留") + + for rel in ( + "/css/motor.css", + "/js/motor/main.js", + "/js/motor/transport.js", + "/js/motor/transport-ecat.js", + "/js/motor/transport-wifi.js", + "/js/common/desktop-bridge.js", + "/js/common/device-link.js", + "/js/common/app-lease.js", + ): + ms, st, b = http_get(PAGE + rel) + if st == 200 and isinstance(b, (bytes, bytearray)) and len(b) > 100: + ok(f"{rel} {st} {len(b)}B ({ms:.1f}ms)") + else: + bad(f"{rel} 资源异常 st={st} {b[:60]!r}") + + print("\n=== 2) EtherCAT 控制面（网页同源 /ecat/api/*） ===") + ms, st, h = http_json(ECAT + "/native/health", {}) + if h.get("ok") and h.get("bus_open") == 1: + ok(f"native/health {ms:.1f}ms dc={h.get('dc')} dc_us={h.get('dc_cycle_us')} loop={h.get('loop')} wkc={h.get('wkc')}") + else: + bad(f"native/health 异常 {h}") + if h.get("dc") == 1 and h.get("dc_cycle_us") == 250: + ok("DC/周期为 250µs（改动生效）") + else: + bad(f"DC/周期未钉死 250µs: dc={h.get('dc')} us={h.get('dc_cycle_us')}") + + ms, st, r = http_json(ECAT + "/motor/start", + {"period_ms": 0.25, "dc_cycle_us": 250, "dc_shift_us": 0, + "force_reopen": True}, 30) + log = r.get("log") if isinstance(r, dict) else None + if r.get("ok"): + ok(f"motor/start ok period={r.get('period_ms')} ({ms:.0f}ms) log={log}") + else: + bad(f"motor/start 失败 {r}") + ms, st, h = http_json(ECAT + "/native/health", {}) + if h.get("dc") == 1 and h.get("dc_cycle_us") == 250 and h.get("loop") == 1: + ok(f"start 后 health: dc=250 loop=1 wkc={h.get('wkc')} ({ms:.1f}ms)") + else: + bad(f"start 后 health 不健康: {h}") + + ms, st, snap = http_json(ECAT + "/motor/state", {}) + if snap.get("ok"): + br = snap.get("bridge") or {} + ok(f"motor/state {ms:.1f}ms period={snap.get('period_ms')} wkc={snap.get('wkc')} " + f"tx={br.get('tx_count')} ack={br.get('ack')} online=" + f"{[m.get('node') for m in (snap.get('motors') or []) if m.get('online')]}") + else: + bad(f"motor/state 失败 {snap}") + + print("\n=== 3) 模拟网页点击「发送 CMD」×N（POST motor/send + 看 GD32 出口） ===") + frame = {"frames": [{"ch": 2, "id": 0x00030200, "dlc": 32, "data": "03" * 32}]} + + def bridge_snap(): + _, _, s = http_json(ECAT + "/motor/state", {}) + if not isinstance(s, dict) or not s.get("ok"): + return None + br = s.get("bridge") or {} + return { + "ms_tx": br.get("tx_count"), + "ack": br.get("ack"), + "sent_seq": s.get("sent_seq"), + "tx_pack": (s.get("tx_pack") or {}).get("seq"), + "txres": br.get("tx_result"), + } + + send_lats = [] + outlet_lats = [] + for i in range(args.n): + pre = bridge_snap() + ms, st, r = http_json(ECAT + "/motor/send", frame, 6.0) + send_lats.append(ms) + send_ok = isinstance(r, dict) and r.get("ok") is True + outlet = None + t0 = time.perf_counter() + while (time.perf_counter() - t0) < 2.0: + post = bridge_snap() + if pre and post and ( + post["ms_tx"] != pre["ms_tx"] + or post["ack"] != pre["ack"] + ): + outlet = (time.perf_counter() - t0) * 1000.0 + outlet_lats.append(outlet) + break + time.sleep(0.001) + if send_ok and outlet is not None: + ok(f"send[{i}] HTTP {ms:.1f}ms → GD32 出口 {outlet:.1f}ms " + f"tx {pre['ms_tx']}→{bridge_snap() and bridge_snap()['ms_tx']} " + f"ack {pre['ack']}→{'?'}") + elif send_ok and outlet is None: + post = bridge_snap() + bad(f"send[{i}] HTTP ok {ms:.1f}ms，但 2s 内 GD32 出口无变化 " + f"pre={pre} post={post} resp={r}") + else: + bad(f"send[{i}] HTTP 失败 {ms:.1f}ms {r}") + time.sleep(0.05) + + def pct(xs, q): + if not xs: + return None + xs = sorted(xs) + return xs[min(len(xs) - 1, int(q * (len(xs) - 1)))] + + if send_lats: + s_p50 = pct(send_lats, 0.5) + if s_p50 is not None and s_p50 < 50: + ok(f"控制面 HTTP send p50={s_p50:.1f}ms（网页路径）") + elif s_p50 is not None: + bad(f"控制面 HTTP send 过慢 p50={s_p50:.1f}ms") + if outlet_lats: + o_p50 = pct(outlet_lats, 0.5) + o_max = max(outlet_lats) + if o_p50 is not None and o_p50 < 50 and o_max < 200: + ok(f"GD32 CAN 出口 p50={o_p50:.1f}ms max={o_max:.1f}ms") + else: + bad(f"GD32 CAN 出口过慢 p50={o_p50} max={o_max:.1f}ms（或仅有部分命中）") + else: + bad("没有任何 send 观测到 GD32 bridge.tx/ack 变化 — 电机侧未消费输出 PDO") + + print("\n=== 4) WiFi 对照（ESP32 /can/send_batch） ===") + try: + body = {"ch": 2, "frames": [{"id": 0x00030200, "dlc": 32, "data": "03" * 32}]} + data = json.dumps(body).encode() + req = urllib.request.Request( + f"http://{ESP}/can/send_batch", data=data, + headers={"Content-Type": "application/json"}, method="POST") + t0 = time.perf_counter() + with urllib.request.urlopen(req, timeout=5) as resp: + raw = resp.read() + wms = (time.perf_counter() - t0) * 1000.0 + j = json.loads(raw.decode("utf-8", "replace") or "{}") + if j.get("status") == "ok": + ok(f"WiFi send_batch {wms:.0f}ms ok（链路可用）") + else: + bad(f"WiFi send_batch 回执异常 {wms:.0f}ms {j}") + except Exception as e: + bad(f"WiFi send_batch 失败: {e}") + + print("\n=== 5) 点击→电机 STATUS 开始动（软件端到端） ===") + print(" 调用 check_click_to_motion.py（会发一小段速度指令后自动 Disable）") + import subprocess + py = sys.executable + motion = os.path.join(ROOT, "scripts", "check_click_to_motion.py") + rc = subprocess.call([py, motion, "--node", "3", "--ch", "2", "--vel", "40", "--timeout", "3"], cwd=ROOT) + if rc == 0: + ok("点击→STATUS 开始动 已测通（详见上表）") + else: + bad("点击→STATUS 开始动 测量失败/超时") + + print("\n=== 6) tx 冻结守卫可观测面（改 native 后必查） ===") + # 这一段只回答"守卫还活着吗"，不测触发（触发要 8s+60s 冷却，见 + # scripts/check_guard_idle_silence.py 与 tools/ecat/check_guard_regression.py）。 + # 本脚本每次发送只等 2s，结构上看不到 8s 判据，所以以前"守卫被改死"与 + # "守卫健康"在这里完全同像——必须显式读 guard 块。 + try: + _, h = post(PAGE + "/ecat/api/native/health", {}, 5) + g = h.get("guard") + if not isinstance(g, dict): + bad("health 无 guard 块 ⇒ 跑的是旧 exe（不含守卫可观测面）；" + "跑 tools/ecat/build_native.bat 换二进制") + else: + t0 = g.get("alive_ticks", 0) + time.sleep(1.0) + _, h2 = post(PAGE + "/ecat/api/native/health", {}, 5) + t1 = (h2.get("guard") or {}).get("alive_ticks", 0) + if t1 - t0 >= 3: + ok(f"守卫心跳 +{t1 - t0}/1s，armed={g.get('armed')} fires={g.get('fires')} " + f"wr={h2.get('wr')} build={h2.get('guard_build')}") + else: + bad(f"守卫心跳仅 +{t1 - t0}/1s ⇒ wedge_worker 没在跑（守卫死了而页面仍全绿）") + if h2.get("recovering"): + ok("health.recovering=1：此刻正在重拉，绿灯不作数（可见面工作正常）") + except Exception as e: + bad(f"guard 可见面读取失败: {e}") + + print("\n=== 汇总 ===") + print(f"PASS {len(PASS)} FAIL {len(FAIL)}") + if FAIL: + print("失败项:") + for x in FAIL: + print(" -", x) + return 1 + print("全链路自测通过：网页资源 + ECAT 控制面 + 发送出口 + 点击→STATUS 动作延迟") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/guard_probe.py b/scripts/guard_probe.py new file mode 100644 index 0000000..63ed692 --- /dev/null +++ b/scripts/guard_probe.py @@ -0,0 +1,158 @@ +"""tx 冻结守卫真机验证的共用底座：一条通路 + 一份前置判据。 + +为什么要有这个文件：两个真机脚本（check_guard_idle_silence.py / check_guard_fires_hw.py） +必须用**同一份**前置条件。前置没满足就跑，等于什么都没测——比如 loop=0 或 pm_hold=1 +时判据按设计是静音的，这种"跑绿了其实没测"正是本次加固要根除的缺陷类别，所以 +判据只放一个执行点，两个脚本都从这里取。 + +通路优先级：先试页面代理 :2625/ecat/api/*（用户点发送走的就是这条，代理自己那套 +"不给 connect 设 timeout"的约束在 esp32_server/ecat_client 里），失败再直连 native +8090..8095（APP 没起时也能跑门槛）。 +""" +import json +import os +import sys +import time +import urllib.error +import urllib.request + +PAGE_BASE = os.environ.get("DCU_PAGE_BASE", "http://127.0.0.1:2625") +NATIVE_PORTS = list(range(8090, 8096)) +GUARD_KEYS = ("alive_ticks", "armed", "tx_stale_s", "since_send_s", "since_full_s", + "cooldown_left_s", "tx_count", "tx_base", "motor_send_hits", "fires", + "last_fire_age_s") +# 注：这里的 timeout 只作用于门槛工具自己的请求。esp32_server 的 ECAT 代理严禁设 +# socket timeout（Windows 上会让 connect 走 select 轮询，每请求白加约 10ms）， +# 那是另一条路径，见 ecat_client.py 的同名注释。 +TIMEOUT = float(os.environ.get("DCU_GUARD_TIMEOUT", "30")) + + +class Blocked(Exception): + """前置不满足：调用方必须以退出码 2 结束，绝不折算成 PASS。""" + + +def _post(url, obj): + data = json.dumps(obj).encode("utf-8") if obj is not None else b"{}" + req = urllib.request.Request(url, data=data, method="POST", + headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=TIMEOUT) as r: + return json.loads(r.read().decode("utf-8", "replace")) + + +class Link: + def __init__(self, base, how): + self.base = base + self.how = how + + def api(self, path, obj=None): + return _post(self.base + path, obj if obj is not None else {}) + + def health(self): + return self.api("/api/native/health") + + def state(self): + return self.api("/api/motor/state") + + def send(self, ch=2, ident=0x00030200, dlc=32, payload=None): + body = {"frames": [{"ch": ch, "id": ident, "dlc": dlc, + "data": payload or "03" * dlc}]} + t0 = time.perf_counter() + r = self.api("/api/motor/send", body) + return r, (time.perf_counter() - t0) * 1000.0 + + +def open_link(verbose=True): + last = None + for url in [PAGE_BASE + "/ecat/api"]: + try: + lk = Link(url, "页面代理") + h = lk.health() + if not h.get("ok"): + raise ValueError("health.ok 非真") + if verbose: + print("通路：%s（%s）" % (url, lk.how)) + return lk + except Exception as e: + last = e + for p in NATIVE_PORTS: + try: + lk = Link("http://127.0.0.1:%d" % p, "直连 native") + h = lk.health() + if h.get("ok"): + if verbose: + print("通路：http://127.0.0.1:%d（%s；页面代理不可用：%s）" % (p, lk.how, last)) + return lk + except Exception as e: + last = e + raise Blocked("native 连不上（试过 %s 与 %s）：%s" + % (PAGE_BASE + "/ecat/api", NATIVE_PORTS, last)) + + +def guard_of(health): + g = health.get("guard") + if not isinstance(g, dict): + raise Blocked("health 里没有 guard 块 ⇒ 跑的是旧 exe（不含可观测面），" + "先跑 tools/ecat/build_native.bat 换二进制") + missing = [k for k in GUARD_KEYS if k not in g] + if missing: + raise Blocked("guard 缺字段 %s ⇒ 二进制与脚本不同代" % missing) + return g + + +def preflight(lk, want_tx_alive=True): + """总线必须真的在交换数据、且守卫没被静音，否则跑了也测不到东西。""" + h = lk.health() + g = guard_of(h) + need = {"bus_open": 1, "op": 1, "loop": 1} + bad = ["%s=%s（需 1）" % (k, h.get(k)) for k, v in need.items() if h.get(k) != v] + if h.get("dc") != 1 or h.get("dc_cycle_us") != 250: + bad.append("dc=%s/%sus（需 1/250）" % (h.get("dc"), h.get("dc_cycle_us"))) + if (h.get("wkc") or 0) < 3: + bad.append("wkc=%s（需 >=3）" % h.get("wkc")) + if h.get("pm_hold"): + bad.append("pm_hold=1（休眠豁免期判据按设计静音，测不到）") + if h.get("recovering"): + bad.append("recovering=1（正在重拉，此刻的绿灯不算数）") + t0 = g["alive_ticks"] + time.sleep(1.0) + g1 = guard_of(lk.health()) + if g1["alive_ticks"] - t0 < 3: + bad.append("guard.alive_ticks +%d/1s（需 >=3）⇒ 巡检线程没在跑" + % (g1["alive_ticks"] - t0)) + if bad: + raise Blocked("；".join(bad)) + if want_tx_alive: + s = lk.state() + tx = ((s.get("bridge") or {}).get("tx_count"), (s.get("head") or {}).get("global_seq")) + print("前置 OK：bus/op/loop=1 dc=250us wkc=%s ticks=+%d/1s 当前 tx_count=%s gseq=%s" + % (h.get("wkc"), g1["alive_ticks"] - t0, tx[0], tx[1])) + else: + print("前置 OK（未要求出口活动）") + return h, g1 + + +def bridge_of(state): + return state.get("bridge") or {} + + +def verdict(ok, why=""): + """末行恒为 VERDICT，包装脚本 grep 它；退出码 0/1/2 是真门槛。""" + if ok: + print("VERDICT PASS") + return 0 + print("VERDICT FAIL " + why) + return 1 + + +__all__ = ["Link", "Blocked", "open_link", "guard_of", "preflight", "bridge_of", + "verdict", "GUARD_KEYS", "PAGE_BASE"] + + +if __name__ == "__main__": + try: + lk = open_link() + preflight(lk) + except Blocked as e: + print("BLOCKED %s" % e) + sys.exit(2) + sys.exit(0)