#!/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())