diff --git a/docs/测试说明/实机测试实施指南.md b/docs/测试说明/实机测试实施指南.md index 618afea8..ba49e8fe 100644 --- a/docs/测试说明/实机测试实施指南.md +++ b/docs/测试说明/实机测试实施指南.md @@ -278,6 +278,15 @@ $cli = if ($env:DTU_STLINK_CLI) { $env:DTU_STLINK_CLI } else { (Get-Command ST-L 再现静默先按三本账(`dbg_console_dropped/_tx_fault/_tx_timeout`)+ syslog 丢弃 账对照定性,再决定是否开卡;销项条件 = N-17 修复版 V5 复跑(e02_probe + FTP ≥30 min)运行期 stall 与非开机 `MISS start` 双零。 +- **判读工具(2026-10-25 起随单交付)**: + `'.venv/Scripts/python.exe' scripts/v5_log_gap_report.py ` + ——自动列 `>10s` 静默段(`--gap-threshold-s` 可调),每段带判定(NO-SOURCE / + PROBE-WINDOW / BOOT-WINDOW / TRUE-STALL-SUSPECT)与贴边证据行;内置 + `task stall` 计数与开机良性豁免(§9 第 8 条口径);退出码 0=双零达标 / + 1=违规 / 2=日志无有效时戳;`--json` 机读、`--report-only` 只出报告、 + `--selftest` 内嵌合成样例自检(无台架可跑)。口径唯一来源 = + `../子系统设计/控制台单写者日志设计说明.md` 文末附录 + §9 第 8 条——改口径 + 先改文档再改脚本(自检断言会拦漂移)。 **执行步骤**:长跑(§4.2 V5)期间周期性打点 + 对照 `CH395F` 无关的 SD2506 侧日志判别卡点。 diff --git a/scripts/v5_log_gap_report.py b/scripts/v5_log_gap_report.py new file mode 100644 index 00000000..7c9ba422 --- /dev/null +++ b/scripts/v5_log_gap_report.py @@ -0,0 +1,401 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +长跑复跑(V5 类)串口日志静默分析与看门狗计数工具(主机侧,纯标准库,不碰固件)。 + +判据口径出处(两处;改口径先改文档、再同步本脚本,自检断言会拦漂移): +- 「无源 vs 真停滞」判别:docs/子系统设计/控制台单写者日志设计说明.md 文末附录 + (N-2 × S8 合并评估,2026-10-25)——静默段贴边(前一行 / 结束后一窗)命中 + 「丢弃账 / 故障账」留痕行 = 真停滞嫌疑;零命中 = 无源(系统安静,正常形态)。 +- 开机良性豁免:docs/测试说明/实机测试实施指南.md §9 第 8 条——开机屏障期 + (reset cause / armed / netTask: started 等锚点后一窗)偶见的单抽样 MISS start + 属固有交错,不计运行期违规;连续 task stall 行才是真停摆。 + +输入契约(serial-assistant 落盘日志 = 纯文本完整行;两种时钟源自动择一): +- 首选采集相对钟:行首 [+NNNms](@test/采集管线相对时戳;相对钟回退 >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、零真停滞嫌疑段);1=有违规; + 2=文件不可读/无有效时戳行;3=用法错误;--report-only 分析成功即退 0。 + +创建日期:2026-10-25(道F 批 2 支援 task-19:探针窗心跳缺口修复后的长跑复跑 +判读工具;台架执行归 lead,本工具只做主机侧判读)。 +""" + +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, + "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 + 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}) + violations = [] + if cs["stall"] > 0: + n1 = cs["stall"] + violations.append(f"task stall 行 {n1} 条(判据 = 0;连续 stall 行 = 真停摆)") + if cs["miss_runtime"] > 0: + n2 = cs["miss_runtime"] + violations.append(f"运行期(非开机豁免窗)MISS start {n2} 条") + nsus = sum(1 for g in gaps if g["verdict"] == "TRUE-STALL-SUSPECT") + if nsus: + violations.append(f"真停滞嫌疑静默段 {nsus} 处(贴边命中丢弃/故障账)") + return {"mode": mode, "points": len(points), "sessions": len(sessions), + "meta": meta, "counters": cs, "threshold_s": args.gap_threshold_s, + "gaps_over_threshold": gaps, "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']}") + 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 = [] + + def check(cond, name): + 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") + 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("SELFTEST PASS 15/15(rel+abs 双钟源、会话切分、四类判定、" + "良性豁免、时钟跳变边界)") + 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())