wandering1
efc3ed1c61
道F 批 2 支援 task-19（纯主机侧工具，零固件接触；合窗台架由 lead 执行， 本工具承担判读面）。 ① scripts/v5_log_gap_report.py：输入 serial-assistant 落盘日志，自动列 >=10s 静默段并逐段判类 NO-SOURCE / PROBE-WINDOW / BOOT-WINDOW / TRUE-STALL-SUSPECT——贴边证据 = 控制台两本账（CONSOLE dropped/tx fault） + syslog 三类 drop 的留痕行；口径唯一来源 = docs/子系统设计/ 控制台单写者日志设计说明.md 文末附录（2026-10-25 合并评估）。 task stall 计数 + MISS start 开机良性豁免内建（实机指南 §9 第 8 条）。 双钟源：采集相对钟 [+NNNms] 优先，缺行首钟退化设备绝对钟；RTC 前跳按 会话边界处理不产假静默、回拨钳制；会话切分 = 相对钟回退或 reset cause 锚点。--json 机读 / --report-only 只出报告 / --selftest 免台架自检。 退出码 0=双零达标 / 1=违规 / 2=日志不可用 / 3=用法（沿用 scripts 惯例）。 ② 用法十行落 docs/测试说明/实机测试实施指南.md §7.2（判据互链）。 【验证】py_compile 绿；--selftest 15/15 断言（双钟源、四类判定、良性豁免、 时钟边界、FAIL 判分路径全覆盖，断言即口径防漂移）；真实样例三份： B6 PROTECT_MISS = 2 条良性 MISS + 0 静默 + PASS；B9 prod = PASS； probe_stale_gate = 124 时点/2 会话/PASS；合成负样例（stall 行 + 34s/28s 静默段）RC=1 且 28s 段判 PROBE-WINDOW——与 V5 时代 ~28s 静默签名吻合。 【门禁自查】doc-links 首跑拦下一处自引用错号（开机良性豁免条实在 §9 第 8 条 而非 §4.3——§4.3 属看门狗设计说明），指南与脚本头两处一并改齐后绿。 【越界】无——写域 = 看板声明两件（脚本 + 指南 §7.2 互链行）。 【门禁】脚本类无 -unit-test 钩子义务（判定面由 --selftest 承担）。
402 lines
17 KiB
Python
402 lines
17 KiB
Python
#!/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())
|