cbb/DCU_DesktopController
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(guard): 验收链改名入库 + 补真退出码；tx 冻结守卫真机门槛两段

Browse Source 
之前 e2e_web_send_latency.py / recover_ecat_after_reset.py / measure_click_to_motion.py
全在 .gitignore 里（e2e_/measure_ 通配），换台机器 clone 就跑不了验收；且
recover 脚本零失败路径（--skip-e2e 无条件退 0）。现在统一到已跟踪的 check_ 前缀，
.gitignore 一行例外都不用加，并把"跑过了"和"恢复了"绑成一件事。

- check_recover_after_reset.py：force_reopen 后改为轮询到 dc=1/250us/loop=1/op=1/
  wkc>=3（上限 15s，替掉原来单点采样那次假 loop=0），并断言守卫心跳在涨、
  发一帧 3s 内出口必须跟上；退出码 0/1/2（前置不满足=BLOCKED=2）。
- check_web_send_latency.py：新增第 6 段读 health.guard（旧 exe 无 guard 块 /
  alive_ticks 不涨 都算 FAIL），第 5 段改调已入库的 check_click_to_motion.py。
  它每次发送只等 2s，结构上看不到 8s 判据，所以只报"守卫当时是死是活"，
  不拿它冒充触发验证。
- check_guard_idle_silence.py：健康总线的"另一极"——90s 静默 + 25s 1Hz 发送，
  要求心跳在涨、armed 恒 0、fires/wr 零增量、每次发送都被消费。这是防
  "把守卫修成重连风暴"的真机版（对应离线 T1b/T4/T5/T6）。
- check_guard_fires_hw.py：需人在场按 GD32 RESET 制造冻结，证明 armed 秒起、
  tx_stale_s 爬过 8s、fires +1、日志有 [GUARD] 全量重连、松手后 20s 内出口
  重新前进，再验证 60s 冷却内不第二次触发；--with-pm-hold 另证"静音"时心跳仍在涨。
  只认 guard.fires，不认 wr（wr 汇总所有恢复路径，会以错误的理由通过）。
- guard_probe.py：两个真机脚本共用同一份前置判据与通路选择（先 :2625 页面代理，
  再直连 native 8090..8095）。前置没满足一律 BLOCKED——loop=0 或 pm_hold=1 时判据
  按设计是静音的，那种"绿了"毫无意义，所以判据只放一个执行点。
This commit is contained in:
cbb
2026-09-21 23:30:07 +08:00
parent 8ec7b31539
commit 83a035c6d6
9 changed files with 1081 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/check_click_to_motion.py
+169
View File
@@ -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("<f", raw, off)[0]
return v if v == v else 0.0 # NaN → 0
return {
"rtAngle": f32(13),
"rtVelocity": f32(17),
"rtCurrent": f32(21),
}
def motor_sample(node: int, ch: int):
_, snap = post(ECAT + "/motor/state", {}, 5)
if not snap.get("ok"):
return None, None, None
br = snap.get("bridge") or {}
outlet = (br.get("tx_count"), br.get("ack"))
seq = snap.get("sent_seq")
for m in snap.get("motors") or []:
if m.get("node") == node and m.get("ch") == ch:
return decode_status(m.get("data") or ""), outlet, seq
return None, outlet, seq
def build_cmd(mode, voltage, current, velocity, angle, pid=100):
buf = bytearray(32)
buf[0] = mode & 0xFF
struct.pack_into("<H", buf, 1, int(round(voltage * 1000)) & 0xFFFF)
struct.pack_into("<f", buf, 3, float(current))
struct.pack_into("<f", buf, 7, float(velocity))
struct.pack_into("<f", buf, 11, float(angle))
struct.pack_into("<h", buf, 15, pid)
struct.pack_into("<h", buf, 17, pid)
struct.pack_into("<h", buf, 19, pid)
return buf.hex()
def send_cmd(ch, node, data_hex):
return post(ECAT + "/motor/send", {
"frames": [{"ch": ch, "id": (node << 16) | 0x0200, "dlc": 32, "data": data_hex}]
}, 6)
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--node", type=int, default=3)
ap.add_argument("--ch", type=int, default=2)
ap.add_argument("--vel", type=float, default=40.0)
ap.add_argument("--timeout", type=float, default=3.0)
args = ap.parse_args()
node, ch = args.node, args.ch
post(ECAT + "/motor/start",
{"period_ms": 0.25, "dc_cycle_us": 250, "dc_shift_us": 0, "force_reopen": True}, 30)
time.sleep(0.3)
base, outlet0, _ = motor_sample(node, ch)
print(f"基线 STATUS node={node} ch={ch}: {base}")
print(f"基线 GD32 outlet tx/ack: {outlet0}")
if not base:
print("FAIL: 该节点无 STATUS（请确认 EtherCAT 在线且节点有回报）")
return 1
ang0, vel0 = base["rtAngle"], base["rtVelocity"]
data = build_cmd(mode=2, voltage=24.0, current=0.0, velocity=args.vel, angle=0.0)
print(f"将发送 CMD mode=Velocity v={args.vel} → 稍后自动 Disable")
t_click = time.perf_counter()
http_ms, resp = send_cmd(ch, node, data)
t_http = time.perf_counter()
print(f"motor/send HTTP {http_ms:.1f}ms ok={resp.get('ok')} queued={resp.get('queued')}")
t_can = None
t_motion = None
motion_detail = None
can_detail = None
t0 = t_click
while time.perf_counter() - t0 < args.timeout:
st, outlet, _ = motor_sample(node, ch)
now = time.perf_counter()
if t_can is None and outlet0 and outlet != outlet0 and outlet[0] is not None:
t_can = now
can_detail = f"outlet {outlet0} → {outlet}"
if st:
d_ang = abs(st["rtAngle"] - ang0)
d_vel = abs(st["rtVelocity"] - vel0)
# 开始运动：角度变化 >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())
scripts/check_guard_fires_hw.py
+209
View File
@@ -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)
scripts/check_guard_idle_silence.py
+105
View File
@@ -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)
scripts/check_recover_after_reset.py
+134
View File
@@ -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)
scripts/check_web_send_latency.py
+288
View File
@@ -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())
scripts/guard_probe.py
+158
View File
@@ -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)