wandering1
4c05b58f3a
道F task-22（F1，大窗判读三票之首）commit 1/2。违规口径改造： violation := task stall>0 或 真停滞嫌疑静默段>0；MISS start 自 v1.1 起整体降级 为诊断行（看门狗 v1.5 按位分账后它只是复合缺席边沿，不代表停喂，设计说明 docs/子系统设计/看门狗设计说明.md §6.1）。v1.0 把运行期 MISS 计入违规是我在 task-19 把判读口径收紧过头——2026-10-25 台架 70min 窗 1262 条运行期 MISS 经 逐窗归因实为重建风暴复合边沿（456 条贴探针窗）+ FTP 传输窗抖动（632 条）， 非心跳丢失，予以纠正（当轮 FAIL 判定的误伤面即此）。 【新增输出】MISS 归因诊断（人读 + --json miss_diag）：开机良性 / 探针窗内 / 窗外运行期 三分类 + 窗外每分钟速率 top5；counters 同步扩字段，旧字段 （miss_runtime 等）保留不动，既有 JSON 消费方零破坏。stall 行文本不变， 语义注释更新 = v1.5 起每行都是「同一任务 800ms 单串」真猎物。 【自检】--selftest 15→18 断言：新增 v1.1 回归钉 = 零 stall + 运行期 MISS 样例 必须 PASS，且归因计数（探针窗内 1 / 窗外 1）与分钟分桶正确；PASS 文案与新 口径同步（改口径先改文档再改脚本的漂移闸语义保持）。 【文档】实机测试实施指南 §7.2 判读工具块重写（v1.1 口径 + 降级依据 + stall 语义指针）；N-2 销项条件同段由「stall 与 MISS 双零」改为「stall 与真停滞嫌疑 段双零」。开机豁免（§9 第 8 条）保留、仅作用于 MISS 诊断分类。 【顺手修（越界申报）】scripts/check_size_gate.py 无参/非法参直调在 print_help 崩（argparse 对 help 做 %-插值，「92% 红线」裸 % + 中文 = ValueError；lead 派 单附注确认）：help 改 92%%，加注释防复发；--ci 功能语义零改动。非本票写域， 系 lead 明示「要不要顺手修由你定」。 【票内待办（commit 2/2）】boot_log_data 的 gz 摘录：原定素材（16:18-16:48 megawin 段）**已不可得**——serial-assistant state 日志被后续 @test 窗 clear 轮转，全盘（含 Temp/ci-artifacts）无副本，已报 lead。替代落点 = 本轮 F2/F3 验收两轮 30min 判据轮（最终主干 1a624b1，效果取证素材）各出一只 gz 入库 + README 索引，语义上是「v1.1 判据下的达标基线」比旧误判窗更有归档价值。 【门禁】--all-files 48 全绿（本树补跑一次 @build 落地 .map 后 mem-numbers/ budget 自然绿，数字 224860/24640/1332/135100 与凭证同码一致；固件源码零 改动）；uvprojx UV4 抖动未入库。
472 lines
21 KiB
Python
472 lines
21 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
|
||
属固有交错；且自 v1.1 起 MISS start 整体降级为**诊断行**（复合缺席边沿，不代表
|
||
停喂，v2-on-ui 看门狗 v1.5 按位分账语义）——违规判定只看 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 为诊断行不参与判定，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())
|