cbb
83a035c6d6
之前 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 时判据 按设计是静音的，那种"绿了"毫无意义，所以判据只放一个执行点。
106 lines
4.3 KiB
Python
106 lines
4.3 KiB
Python
"""守卫的"另一极"真机门槛：健康的总线一次都不许被重连。
|
||
|
||
为什么要单独一个脚本：把 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)
|