Files

106 lines
4.3 KiB
Python
Raw Permalink Normal View History

"""守卫的"另一极"真机门槛:健康的总线一次都不许被重连。
为什么要单独一个脚本:把 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)