Files
STM32F4-Base/scripts/v5_log_gap_report.py
T
wandering1 4a4ad6bc5f chore(全仓): @build/@test 旧文档别名清退为技能调用名
现状口径:/keil-build 与 /serial-boot-test 是 DSH 会话的技能调用名,
不是 shell 命令;「名 + 命令行参数」的命令形态改为可执行体
run_boot_test.py 的真实调用。

- AGENTS.md:别名说明重写为「技能调用名不是 shell 命令」,两条目改
  「编译(/keil-build)」「串口回归(/serial-boot-test)」
- keil-build / serial-boot-test 两 skill 调用名与示例同步;task-worktree
  方案文档去掉 @build/@test 惯例从句;@task 保留(不在点名范围)
- 123 文件现状陈述全改;台架捕获、变更记录/时间线、rename 映射输入
  等 41 行历史证据保留旧名
- gen_memory_svg.py 6 处措辞随改,3 张 SVG 产物同批重跑
- 顺手补并行方案同行漏网计数 51→52 钩子

验证:提交期快钩子对 123 文件实跑全 Passed;check_doc_links --all
rc=0(3038 处、无法判定 0);check_mermaid --all rc=0;YAML 解析 52
钩子不变;skill-size rc=0;py_compile 全过;字节卫生零漂移。
2026-10-30 13:03:12 +08:00

472 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
长跑复跑(V5 类)串口日志静默分析与看门狗计数工具(主机侧,纯标准库,不碰固件)。
判据口径出处(两处;改口径先改文档、再同步本脚本,自检断言会拦漂移):
- 「无源 vs 真停滞」判别:docs/子系统设计/控制台单写者日志设计说明.md 文末附录
(日志静默观察项与单写者化的合并评估,2026-10-25)——静默段贴边(前一行 / 结束后一窗)命中
「丢弃账 / 故障账」留痕行 = 真停滞嫌疑;零命中 = 无源(系统安静,正常形态)。
- 开机良性豁免:docs/测试说明/板级测试/实机测试实施指南.md §9 第 8 条——开机屏障期
(reset cause / armed / netTask: started 等锚点后一窗)偶见的单抽样 MISS start
属固有交错;且自 v1.1 起 MISS start 整体降级为**诊断行**(复合缺席边沿,不代表
停喂,v2-on-ui 看门狗 v1.5 按位分账语义)——违规判定只看 task stall 行与
真停滞嫌疑段,见下。
输入契约(serial-assistant 落盘日志 = 纯文本完整行;两种时钟源自动择一):
- 首选采集相对钟:行首 [+NNNms](串口回归/采集管线相对时戳;相对钟回退 >5s、或出现
reset cause 锚点行 = 新会话边界——烧录/复位/clear 都会造成不连续);
- 无相对钟时退化为设备绝对钟:行内 [YYYY-MM-DD HH:MM:SS](墙钟未就绪的
[----/--/-- --:--:--] 不算时戳;RTC 同步类向前跳变按会话边界处理、不计静默;
回拨钳制不产生假静默);
- 无钟行并入前一个活动时点(裸 UART 取证行本身即活动,其时间取相邻活动时点)。
用法(十行):
'.venv/Scripts/python.exe' scripts/v5_log_gap_report.py <日志路径>
[--gap-threshold-s 10] [--boot-exempt-s 10] [--post-window-s 8]
[--jump-cap-s 3600] [--top 20] [--json] [--report-only] [--selftest]
输出 = 汇总(stall 计数 / MISS 归因诊断[探针窗内/窗外 + 每分钟速率] / >=阈值
静默段清单:每段带判定与贴边证据行);--json 出机读整包;--selftest 跑内嵌
合成样例(无需台架日志)。
退出码:0=达标(零 stall、零真停滞嫌疑段;MISS 为诊断行不参与判定,v1.1 起);
1=有违规;2=文件不可读/无有效时戳行;3=用法错误;--report-only 分析成功即退 0。
创建日期:2026-10-25(道F 批 2 支援 task-19:探针窗心跳缺口修复后的长跑复跑
判读工具;台架执行归 lead,本工具只做主机侧判读)。
v1.1(2026-10-25,道F task-22 / 大窗判读 F1):违规口径 = stall>0 或真停滞嫌疑段
(MISS start 降级为诊断行并出归因:探针窗内/窗外 + 每分钟速率;v1.0 曾把运行期
MISS 计违规,经 2026-10-25 台架 70min 窗逐窗归因属收紧过头,予以纠正)。stall 行
语义随看门狗 v1.5 按位分账变化(行文本不变):每行 = 同一任务 800ms 单串,见
docs/子系统设计/看门狗设计说明.md §5.3/§6.1;使用文档
docs/测试说明/板级测试/实机测试实施指南.md §7.2。
"""
import argparse
import io
import json
import re
import sys
from datetime import datetime
# —— 行形态与固件/采集管线实际留痕逐字核对(改文案先改固件侧再同步这里)——
RE_REL = re.compile(r"^\[\+(\d+)ms\]")
RE_ABS = re.compile(r"(\d{4})-(\d{2})-(\d{2})[ T](\d{2}):(\d{2}):(\d{2})")
RE_MISS = re.compile(r"MISS start snap=0x([0-9A-Fa-f]+) req=0x([0-9A-Fa-f]+)"
r"(?: t=\d+)?")
S_STALL = "task stall:" # DBG_ERROR("[WD] task stall: mask=...")
S_RESET = "reset cause:" # 会话边界锚(复位/烧录都会带)
S_BOOT_ANCHORS = (S_RESET, "armed: timeout", "netTask: started")
# 真停滞证据 = 两本「没落地」账的留痕行(附录口径的账目清单)
S_EVIDENCE = (
"CONSOLE dropped", # 控制台环丢弃(写者恢复后补记,贴边出现)
"CONSOLE tx fault", # 僵尸通道自愈(伴随丢块)
"queue full, drop", # syslog 队列满
"preboot full, drop", # syslog 内核前缓冲满
"hold-over drop", # syslog 积压让位丢
)
S_PROBE_OPEN = "no TIME packet" # time_sync 30s 预算耗尽(探针窗起点)
S_PROBE_CLOSE = "auto-relisten ok" # 监听重建完成(探针窗终点)
ENV_ERR = 2
class Line:
__slots__ = ("no", "rel_ms", "dt", "text")
def __init__(self, no, rel_ms, dt, text):
self.no = no
self.rel_ms = rel_ms
self.dt = dt
self.text = text
def _clip(s, n=110):
s = s.strip()
return s if len(s) <= n else s[: n - 1] + "~"
def parse_stream(txt):
lines = []
for i, s in enumerate(txt.splitlines(), 1):
if not s.strip():
continue
m = RE_REL.match(s)
rel = int(m.group(1)) if m else None
ma = RE_ABS.search(s)
dt = None
if ma:
try:
dt = datetime(*[int(x) for x in ma.groups()])
except ValueError:
dt = None
lines.append(Line(i, rel, dt, s))
mode = "rel" if any(x.rel_ms is not None for x in lines) else "abs"
return lines, mode
def load_lines(path):
with open(path, "rb") as f:
raw = f.read()
return parse_stream(raw.decode("utf-8", errors="replace"))
def build_points(lines, mode, jump_cap_s):
"""归并成活动时点 [(ts, [Line...])] + 会话切分 + 边界计数(见模块头契约)。"""
points = []
splits = [0]
meta = {"jumps": 0, "clock_backs": 0, "no_ts_head": 0}
last_ts = None
for ln in lines:
boundary = False
if mode == "rel":
if ln.rel_ms is None:
if not points:
meta["no_ts_head"] += 1
continue
ts = points[-1][0]
else:
ts = ln.rel_ms / 1000.0
if last_ts is not None and ts < last_ts - 5.0:
boundary = True
else:
if ln.dt is None:
if not points:
meta["no_ts_head"] += 1
continue
ts = points[-1][0]
else:
ts = ln.dt.timestamp()
if last_ts is not None:
d = ts - last_ts
if d < 0:
meta["clock_backs"] += 1
ts = last_ts
elif d > jump_cap_s:
meta["jumps"] += 1
boundary = True
if S_RESET in ln.text and points and ts != points[-1][0]:
boundary = True
if boundary:
splits.append(len(points))
if points and points[-1][0] == ts and not boundary:
points[-1][1].append(ln)
else:
points.append((ts, [ln]))
if ln.rel_ms is not None or ln.dt is not None:
last_ts = ts
ends = splits[1:] + [len(points)]
sessions = [(a, b) for a, b in zip(splits, ends) if b > a]
return points, sessions, meta
def _markers(txt):
return any(e in txt for e in S_EVIDENCE)
def _probe_windows(sp):
wins = []
open_ts = None
for ts, ls in sp:
for l in ls:
if open_ts is None and S_PROBE_OPEN in l.text:
open_ts = ts
elif open_ts is not None and S_PROBE_CLOSE in l.text:
wins.append((open_ts, ts))
open_ts = None
if open_ts is not None and sp:
wins.append((open_ts, sp[-1][0]))
return wins
def analyze(lines, mode, args):
points, sessions, meta = build_points(lines, mode, args.jump_cap_s)
if not points:
return None
cs = {"stall": 0, "miss": 0, "miss_benign": 0, "miss_runtime": 0,
"miss_probe_win": 0, "miss_outside": 0, "miss_minutes": {},
"evidence": 0}
gaps = []
be = args.boot_exempt_s
pw = args.post_window_s
for si, (a, b) in enumerate(sessions):
sp = points[a:b]
anchors = [ts for ts, ls in sp
for l in ls if any(k in l.text for k in S_BOOT_ANCHORS)]
pwins = _probe_windows(sp)
ev = [(ts, l) for ts, ls in sp for l in ls if _markers(l.text)]
cs["evidence"] += len(ev)
for ts, ls in sp:
for l in ls:
if S_STALL in l.text:
cs["stall"] += 1
if RE_MISS.search(l.text):
cs["miss"] += 1
if any(0 <= ts - an <= be for an in anchors):
cs["miss_benign"] += 1
else:
cs["miss_runtime"] += 1
# v1.1 诊断归因:探针窗内 = 重建风暴的复合边沿(正常形态),
# 窗外 = 值得看一眼的缺席;两者都不参与违规判定
if any(op - 2.0 <= ts <= cl + 2.0 for op, cl in pwins):
cs["miss_probe_win"] += 1
else:
cs["miss_outside"] += 1
mi = int(ts // 60)
cs["miss_minutes"][mi] = cs["miss_minutes"].get(mi, 0) + 1
for i in range(1, len(sp)):
t0 = sp[i - 1][0]
t1 = sp[i][0]
if t1 - t0 < args.gap_threshold_s:
continue
pre_hit = [l.text for l in sp[i - 1][1] if _markers(l.text)]
post_hit = [(t, l) for (t, l) in ev if t1 <= t <= t1 + pw]
full_boot = any(t0 >= an - 0.001 and t1 <= an + be
for an in anchors)
in_probe = any(op - 2.0 <= t0 and t1 <= cl + 2.0
for op, cl in pwins)
if pre_hit or post_hit:
verdict = "TRUE-STALL-SUSPECT"
elif full_boot:
verdict = "BOOT-WINDOW"
elif in_probe:
verdict = "PROBE-WINDOW(no-source)"
else:
verdict = "NO-SOURCE"
evl = [_clip(x[1].text, 140) for x in post_hit[:5]]
evl += [_clip(x, 140) for x in pre_hit[:2]]
gaps.append({"session": si + 1,
"start_line": sp[i - 1][1][-1].no,
"end_line": sp[i][1][0].no,
"t_start": round(t0, 3), "t_end": round(t1, 3),
"dur_s": round(t1 - t0, 1), "verdict": verdict,
"before": _clip(sp[i - 1][1][-1].text),
"after": _clip(sp[i][1][0].text),
"evidence_lines": evl})
# v1.1 违规口径(道F task-22,2026-10-25 大窗判读 F1):
# violation := stall>0(看门狗 v1.5 按位分账后每行 = 同一任务连续缺席 >=8
# 窗的单任务真停摆,docs/子系统设计/看门狗设计说明.md §5.3)
# 或 真停滞嫌疑段 >0。MISS start 自 v1.1 起降级为**诊断行**(复合缺席边沿,
# 不代表停喂,docs/子系统设计/看门狗设计说明.md §6.1):v1.0 把运行期 MISS
# 计入违规属把判读口径收紧过头(2026-10-25 台架 70min 窗 1262 条运行期 MISS
# 经逐窗归因实为风暴复合边沿 + 传输窗抖动,非心跳丢失),改出下方归因诊断。
violations = []
if cs["stall"] > 0:
n1 = cs["stall"]
violations.append(f"task stall 行 {n1} 条(判据 = 0;v1.5 按位分账后 "
f"每行 = 同一任务 800ms 单串真停摆)")
nsus = sum(1 for g in gaps if g["verdict"] == "TRUE-STALL-SUSPECT")
if nsus:
violations.append(f"真停滞嫌疑静默段 {nsus} 处(贴边命中丢弃/故障账)")
topmin = sorted(cs["miss_minutes"].items(), key=lambda kv: -kv[1])[:5]
mdiag = {
"total": cs["miss"],
"boot_benign": cs["miss_benign"],
"runtime": cs["miss_runtime"],
"in_probe_window": cs["miss_probe_win"],
"outside_windows": cs["miss_outside"],
"runtime_top_minutes": [{"minute": mi, "count": nc} for mi, nc in topmin],
"semantics": "v1.1 诊断行:MISS start = 复合缺席边沿,不代表停喂,不参与违规",
}
return {"mode": mode, "points": len(points), "sessions": len(sessions),
"meta": meta, "counters": cs, "threshold_s": args.gap_threshold_s,
"gaps_over_threshold": gaps, "miss_diag": mdiag,
"violations": violations, "verdict_pass": not violations}
def render(rep, args):
if rep is None:
return "[ERR] 日志中无有效时戳行(既无行首相对钟,也无设备绝对钟)"
out = ["=== V5 串口日志静默/看门狗计数报告 ==="]
md, npt, nss = rep["mode"], rep["points"], rep["sessions"]
thr, be, pw = rep["threshold_s"], args.boot_exempt_s, args.post_window_s
out.append(f"时钟源={md} 活动时点={npt} 会话={nss} 静默阈值={thr:.0f}s "
f"开机豁免={be:.0f}s 贴边回看={pw:.0f}s")
m = rep["meta"]
if m["jumps"] or m["clock_backs"] or m["no_ts_head"]:
out.append(f"边界注记:向前时钟跳变(同步边界、不计静默)={m['jumps']} "
f"回拨钳制={m['clock_backs']} 无钟头行={m['no_ts_head']}")
c = rep["counters"]
out.append("")
out.append(f"[计数] task stall={c['stall']} | MISS start 总={c['miss']}"
f"(开机良性={c['miss_benign']} / 运行期={c['miss_runtime']})"
f"| 账目留痕行={c['evidence']}")
mdg = rep["miss_diag"]
tot, bb = mdg["total"], mdg["boot_benign"]
ipw, outw = mdg["in_probe_window"], mdg["outside_windows"]
out.append(f"[MISS 诊断(非违规)] 总={tot} 开机良性={bb} 探针窗内={ipw}"
f" 窗外运行期={outw}")
if mdg["runtime_top_minutes"]:
parts = []
for mm in mdg["runtime_top_minutes"]:
parts.append("第%d分x%d" % (mm["minute"], mm["count"]))
out.append(" 窗外 MISS 每分钟速率 top:" + " ".join(parts)
+ "(分钟号=所选时钟源时基的第几分钟)")
gaps = rep["gaps_over_threshold"]
out.append(f"[静默] >= {thr:.0f}s 共 {len(gaps)} 段:")
for k, g in enumerate(gaps[: args.top], 1):
out.append(f" #{k} 会话{g['session']} [{g['t_start']} -> "
f"{g['t_end']}]s {g['dur_s']}s {g['verdict']}")
out.append(f" 前 L{g['start_line']}: {g['before']}")
out.append(f" 后 L{g['end_line']}: {g['after']}")
for e in g["evidence_lines"]:
out.append(f" 证据: {e}")
if len(gaps) > args.top:
out.append(f" ...其余 {len(gaps) - args.top} 段(--top 放大)")
out.append("")
if rep["violations"]:
out.append("结论: FAIL")
for v in rep["violations"]:
out.append(" - " + v)
else:
out.append("结论: PASS(零 stall / 零真停滞嫌疑段;MISS 为诊断行、"
"NO-SOURCE 为正常空闲,均不构成违规)")
return "\n".join(out)
def selftest():
fails = []
nchk = [0]
def check(cond, name):
nchk[0] += 1
if not cond:
fails.append(name)
class A:
gap_threshold_s = 10.0
boot_exempt_s = 10.0
post_window_s = 8.0
jump_cap_s = 3600.0
top = 50
rel = "\n".join([
"[+50ms] reset cause: software (NVIC_SystemReset)",
"[+600ms] MISS start snap=0x000001E7 req=0x000001F7 t=710",
"[+2800ms] MISS start snap=0x000001E7 req=0x000001F7 t=2900",
"[+4000ms] [2026-01-01 00:00:04][APP] storage ready",
"[+9000ms] [2026-01-01 00:00:09][NET] link up",
"[+25000ms] [2026-01-01 00:00:25][UI] page refresh",
"[+26000ms] [2026-01-01 00:00:26][FTP] transfer done",
"[+30000ms] [2026-01-01 00:00:30][FTP] stor start",
"[+42000ms] [ERR][WD] CONSOLE dropped 3 lines",
"[+43000ms] [2026-01-01 00:00:43][UI] page refresh",
"[+50000ms] [ERR] [WD] task stall: mask=0x00000001 req=0x00000037 miss=8",
"[+55000ms] [2026-01-01 00:00:55][TIME_SYNC] no TIME packet in 30s",
"[+56000ms] [2026-01-01 00:00:56][NET] ftp session end",
"[+70000ms] [2026-01-01 00:01:10][NET] link check",
"[+75000ms] [INFO][LIS] sock1 auto-relisten ok (drop)",
"[+100ms] reset cause: POR",
"[+3000ms] [2026-01-01 00:02:03][APP] boot",
"[+16000ms] [2026-01-01 00:02:16][UI] idle",
]) + "\n"
lines, mode = parse_stream(rel)
check(mode == "rel", "rel 模式识别")
rep = analyze(lines, mode, A())
check(rep["sessions"] == 2, f"rel 回退+reset 切会话=2(得 {rep['sessions']})")
c = rep["counters"]
check(c["stall"] == 1, f"stall=1(得 {c['stall']})")
check(c["miss"] == 2 and c["miss_benign"] == 2 and c["miss_runtime"] == 0,
f"MISS 2/2/0(得 {c['miss']}/{c['miss_benign']}/{c['miss_runtime']})")
check(c["evidence"] == 1, f"账目留痕行=1(得 {c['evidence']})")
by = {}
for x in rep["gaps_over_threshold"]:
by[x["verdict"]] = by.get(x["verdict"], 0) + 1
check(len(rep["gaps_over_threshold"]) == 4,
f"静默段=4(得 {len(rep['gaps_over_threshold'])})")
check(by.get("TRUE-STALL-SUSPECT") == 1, f"嫌疑段=1(得 {by})")
check(by.get("PROBE-WINDOW(no-source)") == 1, f"探针窗段=1(得 {by})")
check(by.get("NO-SOURCE") == 2, f"无源段=2(得 {by})")
check(rep["verdict_pass"] is False, "含 stall 的样例应判 FAIL")
# v1.1 回归钉:零 stall + 运行期 MISS 存在 = PASS(MISS 已降级为诊断行),
# 且归因正确:探针窗内 / 窗外各自计数、每分钟速率分桶
rel2 = "\n".join([
"[+100ms] reset cause: POR",
"[+5000ms] [2026-01-01 00:00:05][NET] link up",
"[+40000ms] [2026-01-01 00:00:40][TS] no TIME packet in 30s",
"[+45000ms] MISS start snap=0x000001F6 req=0x000001F7 t=45000",
"[+47000ms] [2026-01-01 00:00:47][NET] ftp session end",
"[+50000ms] [INFO][LIS] sock1 auto-relisten ok (drop)",
"[+70000ms] MISS start snap=0x000001E7 req=0x000001F7 t=70000",
"[+72000ms] [2026-01-01 00:01:12][UI] tick",
]) + "\n"
lines3, mode3 = parse_stream(rel2)
rep3 = analyze(lines3, mode3, A())
c3, d3 = rep3["counters"], rep3["miss_diag"]
check(rep3["verdict_pass"] is True,
f"零 stall + 运行期 MISS 应判 PASS(v1.1 降级口径)(得 {rep3['violations']})")
check(d3["runtime"] == 2 and d3["total"] == 2,
f"MISS 运行期=2(得 {d3['runtime']}/{d3['total']})")
check(d3["in_probe_window"] == 1 and d3["outside_windows"] == 1,
f"探针窗内/窗外 = 1/1(得 {d3['in_probe_window']}/{d3['outside_windows']})")
check(sum(x["count"] for x in d3["runtime_top_minutes"]) == 2,
"窗外+窗内 MISS 分钟桶总量=2")
abslog = "\n".join([
"[2026-01-01 00:00:00] boot banner",
"[----/--/-- --:--:--][APP] reset cause: POR",
"[2026-01-01 00:00:05] pre-boot line",
"[2026-01-01 00:00:20] after 15s gap",
"[2026-10-25 08:00:00] rtc synced",
"[2026-10-25 08:00:15] after gap 2",
]) + "\n"
lines2, mode2 = parse_stream(abslog)
check(mode2 == "abs", "abs 模式识别")
rep2 = analyze(lines2, mode2, A())
check(rep2["meta"]["jumps"] >= 1, "RTC 前跳识别为边界(不计静默)")
check(len(rep2["gaps_over_threshold"]) == 2,
f"abs 静默段=2(得 {len(rep2['gaps_over_threshold'])})")
check(rep2["verdict_pass"] is True, "abs 样例应判 PASS")
if fails:
for f in fails:
print("SELFTEST FAIL: " + f)
return 1
print(f"SELFTEST PASS {nchk[0]}/{nchk[0]}(rel+abs 双钟源、会话切分、四类"
"判定、良性豁免、时钟跳变边界、v1.1 MISS 诊断降级与归因)")
return 0
def main(argv=None):
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
p = argparse.ArgumentParser(
prog="v5_log_gap_report.py",
description="V5 串口日志静默分析 + 看门狗计数(判据 = 控制台单写者日志"
"设计说明文末附录 + 实机测试实施指南 §9 第 8 条)")
p.add_argument("log", nargs="?", help="serial-assistant 落盘日志路径")
p.add_argument("--gap-threshold-s", type=float, default=10.0)
p.add_argument("--boot-exempt-s", type=float, default=10.0)
p.add_argument("--post-window-s", type=float, default=8.0)
p.add_argument("--jump-cap-s", type=float, default=3600.0)
p.add_argument("--top", type=int, default=20)
p.add_argument("--json", action="store_true", help="输出机读 JSON")
p.add_argument("--report-only", action="store_true",
help="只出报告:分析成功即退 0,不按判据定成败")
p.add_argument("--selftest", action="store_true", help="跑内嵌合成样例")
a = p.parse_args(argv)
if a.selftest:
return selftest()
if not a.log:
p.error("需要日志路径(或使用 --selftest)")
try:
lines, mode = load_lines(a.log)
except OSError as e:
print(f"[ERR] 读取失败: {e}")
return ENV_ERR
rep = analyze(lines, mode, a)
if rep is None:
print(render(None, a))
return ENV_ERR
if a.json:
print(json.dumps(rep, ensure_ascii=False, indent=1))
else:
print(render(rep, a))
if a.report_only:
return 0
return 0 if rep["verdict_pass"] else 1
if __name__ == "__main__":
sys.exit(main())