wandering1
4cf2df4770
与 2fd8a864 同一件事，上一笔的 git add 只覆盖了文档与部分脚本，这三只漏在暂存区外：帮助文本与
两处注释里的内部阶段代号（自检台账判定、用例追溯校验）已换成读得到、查得着的说法，判据未变。
自查：三文件 noBOM / LF / U+FFFD=0；id-check --strict 与 doc-links --all 均 rc=0；工作树至此为空。
757 lines
41 KiB
Python
757 lines
41 KiB
Python
#!/usr/bin/env python3
|
||
"""verify_log.py - 对抓取的 MCU 调试日志做标记校验（纯标准库，2026-09-20 替代 verify_log.ps1）。
|
||
|
||
用于 @test 闭环：烧录后从串口助手日志提取的开机日志交给本脚本 grep 标记，
|
||
输出结构化判定（格式与旧 PowerShell 版逐字一致，下游文档/习惯不受影响）。
|
||
|
||
用法:
|
||
& '.venv\\Scripts\\python.exe' .dsh/skills/serial-boot-test/scripts/verify_log.py --log-path <file>
|
||
[--required 'marker1;marker2']
|
||
[--forbidden 'FAIL;ERROR;HardFault;assert;[ERR]']
|
||
[--require-suite-summary 'Storage Tests;GD5F2GQ5UE Tests']
|
||
[--require-selftest-pass] [--selftest-log <selftest.log 本地副本>]
|
||
[--show-tail-lines 40]
|
||
[--max-boot-count 1] [--boot-marker 'reset cause:']
|
||
[--expect-reset-cause 'software;power-on']
|
||
[--require-order 'App init OK>Network Init>net init: OK']
|
||
[--done-marker 'net init: OK'] [--max-boot-ms 8000] [--json]
|
||
|
||
退出码（2026-10-27 task-32 起把 2 讲清楚；@test 那层的码不变）:
|
||
0 = PASS（所有 Required 标记找到，Forbidden 零命中，汇总行要求满足）
|
||
1 = FAIL（**设备侧真有坏消息**：缺 Required / 命中 Forbidden / 汇总行不满足 / 顺序或耗时超限）
|
||
2 = **测量没成立**（日志不存在或不可读；或开了 --max-boot-count 却零启动锚点 = 采集器没开、
|
||
日志被清、或采集没从复位开始）——这**不是**一次设备故障，按码聚合的脚本必须把它和 1
|
||
分开算，否则"采集断了"会被统计成"又崩了一次"。verdict 字段给 MEASUREMENT_BROKEN，
|
||
JSON 里 measurement.* 带三计数明细（payload_lines / boot_anchor_count / exact_hits /
|
||
loose_neighbours）。注意优先级：**零锚点但同时有 Forbidden 命中或自检故障记录时仍判 1**，
|
||
测量问题最多在旁边加一行 WARN——绝不能把真崩溃降级成"没数据"。
|
||
|
||
默认 Required 标记对应 STM32F4-Base 启动序列（见 App/app/app_main.c + App/task/net_task.c），
|
||
按测试用例覆盖，例如:
|
||
--required 'App init OK;net init: OK;PHY: 100M FULL'
|
||
--forbidden 'FAIL;ERROR;HardFault;assert;[ERR]'
|
||
|
||
语义约定（与旧 PowerShell 版对齐）:
|
||
- Required：大小写敏感子串匹配（全文）
|
||
- Forbidden：大小写**不敏感**子串匹配、逐行计数（旧版 -like 语义；
|
||
如 Forbidden 'FAIL' 会命中 "write /log/run.log fail (7)" 这类行）
|
||
- 固件错误级日志前缀是 `[ERR]`（`DBG_ERROR`，形如 `[时间][ERR][TAG] 消息`），
|
||
与 `ERROR` 子串**不匹配**——默认 Forbidden 因此显式包含 `[ERR]`，驱动/模块
|
||
错误行不再漏检（2026-09-21 修复）。测试套件中**预期**出现 `[ERR]` 的用例
|
||
（如 GD5F DMA 守卫负路径、FatFs 非法名校验）须用 --forbidden 覆盖默认值。
|
||
- 读取编码：先按 UTF-8，解码失败回退系统 locale 编码（旧版 Get-Content 行为）
|
||
- 控制台输出重配置为 UTF-8（errors=replace）：任何 Windows 控制台代码页
|
||
都只影响终端显示，不会导致本脚本崩溃
|
||
- 汇总行（2026-10-01 加）：纯 Required 子串匹配有个盲区——套件启动横幅
|
||
命中即判 PASS，**套件是否跑到汇总行、断言数是否为 0 完全不可见**（套件中途
|
||
挂死/复位、或 Phase 宏全关时会被误判为通过）。`--require-suite-summary`
|
||
要求日志中出现 `=== <名称>: N/M PASSED (failed=0) ===` 且 **N > 0**。
|
||
- 自检记录联合判定（2026-10 实机窗口；判据 2026-10-22 改为"只看本次新增"）：
|
||
`--require-selftest-pass` 对 /log/selftest.log 的本地副本（定长 64B 记录，格式见
|
||
App/log/sys_log.h）判定。自检仅故障写盘（v1 不记通过项）：**文件缺失 / 0 字节 = 无故障
|
||
→ PASS**；存在记录 = FAIL（退出码 1），逐条打印时间戳 + 描述。
|
||
⚠ 该文件是 512 条环形**累计历史**，所以本函数收到的"有记录"必须由调用方先做差集：
|
||
run_boot_test.py 在烧录前取基线、启动后只把"基线之后新增"的记录写成临时副本交给本
|
||
函数（`--selftest-scope this-run`，默认）。**抓取失败/无数据源/无法证明无新增一律由
|
||
调用方判 FAIL**——旧实现把这些都当"无记录 → PASS"，是假绿（详见
|
||
docs/测试说明/自动化测试流程.md 的自检结果判定节）。
|
||
- 启动锚点计数（2026-10-10）：`--max-boot-count N`（run_boot_test.py 默认传 1）统计
|
||
`reset cause:` 锚点出现次数——**0 次** = 采集没从复位开始（标记全中也可能只是上一轮
|
||
的残留日志）；**>N 次** = 板子在采集窗口内自己复位过（IWDG 咬狗 / Fault 后
|
||
`NVIC_SystemReset` / 掉电重启），此前这类运行只要不命中 Forbidden 就判 PASS。
|
||
`--expect-reset-cause` 对第一个锚点做原因白名单（大小写不敏感子串），空 = 只报告不判。
|
||
- Forbidden 命中行（2026-10-10）：命中时逐行打印**行号 + 内容**（最多 6 行，超出报剩余数）。
|
||
旧实现只说 `forbidden pattern: [ERR] (3 line(s))`，人还得自己再 grep 一遍才知道是哪几行
|
||
——而"是哪几行"恰恰是定位要的东西。
|
||
- 标记顺序（`--require-order 'a>b>c'`）：按**首次出现的行序**校验启动步骤先后。标记全中
|
||
但顺序颠倒 = 某步在另一步之后补做（屏障失序、重试后补跑），旧实现完全看不见。
|
||
- 开机耗时（`--done-marker` + `--max-boot-ms`）：依赖行首 `[+Nms]` 前缀（run_boot_test.py
|
||
2026-10-10 起按 daemon 毫秒时基给每行加，MCU 侧时间戳只有秒级且开机头几行根本没有）。
|
||
量"第一行 → done-marker"的真实耗时，抓"能开机但慢了几秒"（mount 重试 / PHY 重协商）。
|
||
日志里没有该前缀 → 只 WARN 跳过，不假判 FAIL。
|
||
- `--json`：同一套判定以 JSON 输出（供 CI / 上层脚本消费；退出码语义不变）。
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import locale
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
REQUIRED_DEFAULT = "App init OK;Network Init;net init: OK"
|
||
FORBIDDEN_DEFAULT = "FAIL;ERROR;HardFault;assert;[ERR]"
|
||
|
||
# Forbidden 命中时最多打印几条命中行（够定位又不刷屏；剩余条数会报出来）
|
||
HIT_LINES_MAX = 6
|
||
|
||
# 行首 `[+1234ms] `：run_boot_test.py 按 daemon 毫秒时基给每行加的相对偏移
|
||
# （MCU 侧时间戳只有秒级，且开机头几行时钟还没起 → 量不了耗时）
|
||
_TS_OFF_RE = re.compile(r"^\[\+(\d+)ms\]\s")
|
||
|
||
# 每轮启动的锚点行（App/app/app_main.c：`reset cause: <原因> (CSR=0x…)`）。
|
||
# 一次采集里出现 >1 次 = 板子在采集窗口内自己复位过（IWDG 咬狗 / Fault 后
|
||
# NVIC_SystemReset / 掉电重启），此前只有 Forbidden 命中时才被发现，
|
||
# "跑完 PASS 但中途重启过"完全不可见（2026-10-10 加）。
|
||
BOOT_MARKER_DEFAULT = "reset cause:"
|
||
|
||
# 套件汇总行：`=== Storage Tests: 3/3 PASSED (failed=0) ===`
|
||
# 名称可含空格；N/M 为十进制；failed 允许前导 0。宽松容忍前后空白与
|
||
# 日志时间戳前缀（`[2026-10-01 12:00:00.000]`）——故用 search 而非 match。
|
||
_SUMMARY_RE = re.compile(
|
||
r"===\s*(?P<name>[^:=]+?)\s*:\s*(?P<passed>\d+)\s*/\s*(?P<total>\d+)\s+"
|
||
r"PASSED\s*\(\s*failed\s*=\s*(?P<failed>\d+)\s*\)\s*==="
|
||
)
|
||
|
||
|
||
def _split_list(spec):
|
||
"""';' 分隔列表 → 去空白、去空项。"""
|
||
return [p.strip() for p in spec.split(";") if p.strip()]
|
||
|
||
|
||
def _read_lines(path):
|
||
raw = Path(path).read_bytes()
|
||
for enc in ("utf-8", locale.getpreferredencoding(False)):
|
||
try:
|
||
return raw.decode(enc).splitlines()
|
||
except UnicodeDecodeError:
|
||
continue
|
||
return raw.decode("utf-8", errors="replace").splitlines()
|
||
|
||
|
||
# --- whole-line exact matching (2026-10-26, task-36; absorbed from bench-runner scratch) ---
|
||
# Why this exists (measured, not theoretical): -Required is a SUBSTRING check, and the
|
||
# console-interleave defect kept the substring while destroying the line, e.g. the
|
||
# headless victim line "26-10-26 08:37:07][NAND_FTL] Initializing NAND FTL..." still
|
||
# contains the marker "Initializing NAND FTL" -> the old check called that boot a PASS.
|
||
# Any marker whose line can lose its head is therefore NOT verifiable by substring alone.
|
||
# --exact anchors the WHOLE payload (^ ... $) so one spliced byte breaks the match, and it
|
||
# also reports lines containing the pattern's longest literal run that do NOT match fully:
|
||
# that is the splicing signature and it must never stay silent. Length pinning is OPT-IN
|
||
# ("@@LEN=<n>"): byte lengths belong to one image, pinning them by default gives a false
|
||
# red on innocent commits (measured uiTask stack swing 4116..4188 on ONE image).
|
||
EXACT_LEN_SEP = "@@LEN="
|
||
_DAEMON_RX = re.compile(r"^\[RX\] \S+ (.*)$")
|
||
_RUNBT_RX = re.compile(r"^\[\+\d+ms\] (.*)$")
|
||
|
||
|
||
def _parse_exact(spec):
|
||
"""'REGEX@@LEN=42' -> (regex, 42); no suffix -> (regex, None)."""
|
||
if EXACT_LEN_SEP in spec:
|
||
pat, _, ln = spec.rpartition(EXACT_LEN_SEP)
|
||
return pat, int(ln.strip())
|
||
return spec, None
|
||
|
||
|
||
def _payload_of(line):
|
||
"""Drop the collector prefix so ^...$ applies to the firmware line itself.
|
||
|
||
Handles serial-assistant daemon lines ('[RX] <iso> ...'), run_boot_test boot logs
|
||
('[+Nms] ...') and bare payload lines. Archived evidence files may carry a leading
|
||
'#' provenance header; those are dropped unless include_comments is set (otherwise a
|
||
header that merely MENTIONS [ERR] would trip Forbidden / pollute exact counts).
|
||
"""
|
||
for rx in (_DAEMON_RX, _RUNBT_RX):
|
||
m = rx.match(line)
|
||
if m:
|
||
return m.group(1)
|
||
return line
|
||
|
||
|
||
def _longest_literal(pattern):
|
||
"""Longest run of non-metacharacters in a regex -> a greppable neighbour probe."""
|
||
parts = re.split(r"\\.|\\\[|\]|\(|\)|\{|\}|\||\*|\+|\?|\^|\$|\.|\[:", pattern)
|
||
parts = [x for x in parts if x and not x.startswith("\\")]
|
||
return max(parts, key=len) if parts else ""
|
||
|
||
|
||
def exact_check(payloads, spec, allow_residual=0, head_char=""):
|
||
"""(ok, detail, neighbours, residuals) for one --exact spec. See block note above."""
|
||
pat, want_len = _parse_exact(spec)
|
||
rx = re.compile("^" + pat + "$")
|
||
hits, residuals, consumed = [], [], set()
|
||
for idx, pl in enumerate(payloads):
|
||
if rx.match(pl):
|
||
hits.append(pl)
|
||
consumed.add(idx)
|
||
continue
|
||
if allow_residual > 0:
|
||
for k in range(1, allow_residual + 1):
|
||
if len(pl) > k and pl[k] == "[" and rx.match(pl[k:]):
|
||
hits.append(pl[k:])
|
||
residuals.append((k, pl[:k]))
|
||
consumed.add(idx)
|
||
break
|
||
key = _longest_literal(pat)
|
||
# a line counted (directly or after residual realignment) is never also a neighbour
|
||
neigh = [pl for i2, pl in enumerate(payloads) if key and key in pl and i2 not in consumed]
|
||
detail = "hits={} lens={} loose_neighbours={}".format(
|
||
len(hits), [len(h) for h in hits], len(neigh))
|
||
if residuals:
|
||
detail += " residual_head_bytes={}".format([(k, repr(v)) for k, v in residuals])
|
||
ok = len(hits) >= 1 and not neigh
|
||
if want_len is not None:
|
||
ok = ok and all(len(h) == want_len for h in hits)
|
||
detail += " want_len={}".format(want_len)
|
||
if head_char:
|
||
bad = [h for h in hits if not h.startswith(head_char)]
|
||
ok = ok and not bad
|
||
detail += " head_char={!r} bad={}".format(head_char, len(bad))
|
||
return ok, detail, neigh[:3], residuals[:3]
|
||
|
||
|
||
def pair_check(payloads, first_pat, then_pat):
|
||
"""Exactly one FIRST line with exactly one THEN line after it (cross-reset pairing).
|
||
|
||
Crash-dump legs need this: the fault line must pair with the NEXT boot's reset cause,
|
||
and a second following cause would mean an unlogged or repeated reset.
|
||
"""
|
||
frx, trx = re.compile(first_pat), re.compile(then_pat)
|
||
fi = [i for i, pl in enumerate(payloads) if frx.search(pl)]
|
||
ti = [i for i, pl in enumerate(payloads) if trx.search(pl)]
|
||
if len(fi) != 1:
|
||
return False, "first matches={} (want exactly 1), then matches={}".format(len(fi), len(ti))
|
||
after = [i for i in ti if i > fi[0]]
|
||
if len(after) != 1:
|
||
return False, "first at L{} but following THEN count={} at {} (want exactly 1)".format(
|
||
fi[0] + 1, len(after), [x + 1 for x in after])
|
||
return True, "first L{} -> only following THEN at L{} (gap {} line(s))".format(
|
||
fi[0] + 1, after[0] + 1, after[0] - fi[0])
|
||
|
||
|
||
def _reset_cause_of(line, marker):
|
||
"""从 `reset cause: software (NVIC_SystemReset) (CSR=0x14000000)` 取原因文本。"""
|
||
i = line.find(marker)
|
||
if i < 0:
|
||
return ""
|
||
tail = line[i + len(marker):].strip()
|
||
j = tail.find(" (CSR=")
|
||
return tail[:j] if j > 0 else tail
|
||
|
||
|
||
# --- measurement-vs-device classification (task-32; bench-runner night rule) ---
|
||
# Three counters decide which kind of "nothing found" we are looking at, and they must not
|
||
# be conflated: an all-zero read is a BROKEN MEASUREMENT, not a healthy device.
|
||
# hits=0, loose_neighbours=0, boot anchors=0 -> measurement broke (wrong file, closed
|
||
# serial port, dead collector). Nothing was observed -> no verdict is meaningful.
|
||
# hits=0, loose_neighbours>=1 -> the line exists but is damaged (head
|
||
# eaten / bytes spliced in) -> that is a real device-side defect, --exact's purpose.
|
||
# hits=0, loose_neighbours=0, anchors>=1 -> the boot DID happen; the marker is
|
||
# genuinely absent -> allowed to read as a functional miss.
|
||
# Reported as its own exit code (2, the existing "measurement/environment did not happen"
|
||
# family) so aggregation by exit code never counts a dead collector as a device failure.
|
||
def classify_measurement(n_lines, boot_count, boot_check_on, exact_checked, hits, neighbours):
|
||
"""(broken, reason). Only judges when a boot anchor check was actually requested."""
|
||
if not boot_check_on:
|
||
return False, "anchor check off (max_boot_count=0) -> cannot classify"
|
||
if n_lines == 0:
|
||
return True, "zero payload lines (collector closed / log truncated / wrong file)"
|
||
if boot_count == 0:
|
||
return True, ("log readable (%d line(s)) but ZERO boot anchor -> capture did not "
|
||
"start at a reset" % n_lines)
|
||
if exact_checked and hits == 0 and neighbours == 0:
|
||
return False, "anchors present: exact markers genuinely absent (functional miss, not measurement)"
|
||
return False, "measurement ok (hits=%d neighbours=%d anchors=%d)" % (hits, neighbours, boot_count)
|
||
|
||
|
||
def verify(log_path, required=REQUIRED_DEFAULT, forbidden=FORBIDDEN_DEFAULT,
|
||
show_tail=40, out=print, suite_summary="",
|
||
selftest_pass=False, selftest_log="",
|
||
boot_marker=BOOT_MARKER_DEFAULT, max_boot_count=0,
|
||
expect_reset_cause="", require_order="", done_marker="",
|
||
max_boot_ms=0, as_json=False, exact=None, pairs=None,
|
||
allow_residual=0, head_char="", include_comments=False):
|
||
"""标记校验核心（可被 run_boot_test.py 复用）。返回退出码 0/1/2。
|
||
|
||
suite_summary: ';' 分隔的套件名列表。非空时，要求日志中每个套件都出现
|
||
`=== <名称>: N/M PASSED (failed=0) ===` 且 **N > 0**。
|
||
selftest_pass: True 时对 selftest_log（/log/selftest.log 本地副本）联合判定：
|
||
文件缺失 / 0 字节 = 无故障（PASS）；存在任何 64B 记录 = FAIL（S7）。
|
||
selftest_log: 自检记录文件本地路径（空串 = 按文件缺失处理）。
|
||
boot_marker / max_boot_count（2026-10-10）：统计启动锚点出现次数。
|
||
max_boot_count>0 时启用：0 次 = 采集没从复位开始；>N 次 = 板子
|
||
在采集窗口内自己复位过（IWDG / Fault 后 NVIC_SystemReset）。
|
||
expect_reset_cause: ';' 分隔的原因子串白名单，只对**第一个**锚点生效（大小写不敏感）。
|
||
空 = 只报告不判定。
|
||
require_order: '>' 分隔的标记序列（2026-10-10），按**首次出现的行序**校验先后；
|
||
顺序颠倒 = 某步在另一步之后补做（屏障失序 / 重试补跑）。空 = 不查。
|
||
done_marker / max_boot_ms（2026-10-10）：量"第一行 → done_marker"的毫秒耗时
|
||
（靠行首 `[+Nms]` 前缀）；超过 max_boot_ms 判 FAIL。
|
||
日志无该前缀 → WARN 跳过，不假判。
|
||
as_json: True = 不打印文本判定，最后输出一个 JSON 对象（退出码不变）。
|
||
"""
|
||
res = {"log_path": str(log_path), "lines": 0, "bytes": 0, "required_missing": [],
|
||
"forbidden_hits": [], "boot": None, "suites_bad": [], "selftest": None,
|
||
"order_bad": [], "timing": None, "tail": [], "verdict": None, "exit": None,
|
||
"exact_bad": [], "exact_ok": [], "pair_bad": [], "pair_ok": []}
|
||
emit = out
|
||
|
||
def finish(code):
|
||
res["exit"] = code
|
||
res["verdict"] = {0: "PASS", 1: "FAIL", 2: "ERROR"}[code]
|
||
if as_json:
|
||
emit(json.dumps(res, ensure_ascii=False, indent=2))
|
||
return code
|
||
|
||
if as_json:
|
||
out = lambda *a, **k: None # 文本判定静默，结果走 JSON
|
||
|
||
p = Path(log_path)
|
||
if not p.is_file():
|
||
out("VERIFY_ERROR: log file not found: {}".format(log_path))
|
||
return finish(2)
|
||
try:
|
||
lines = _read_lines(log_path)
|
||
except OSError:
|
||
out("VERIFY_ERROR: cannot read log file: {}".format(log_path))
|
||
return finish(2)
|
||
if not include_comments:
|
||
# archived evidence files carry an ASCII provenance header; a header that merely
|
||
# MENTIONS [ERR] must not trip Forbidden (tool-level false-red guard, task-36)
|
||
lines = [l for l in lines if not l.lstrip().startswith("#")]
|
||
|
||
full = "\n".join(lines)
|
||
req_list = _split_list(required)
|
||
forb_list = _split_list(forbidden)
|
||
summary_list = _split_list(suite_summary)
|
||
|
||
out("=== Log verdict: {} ===".format(log_path))
|
||
out("Lines: {}, bytes: {}".format(len(lines), p.stat().st_size))
|
||
res["lines"] = len(lines)
|
||
res["bytes"] = p.stat().st_size
|
||
|
||
missing = []
|
||
for m in req_list:
|
||
if m in full:
|
||
out("[OK] required marker found: {}".format(m))
|
||
else:
|
||
out("[MISS] required marker NOT found: {}".format(m))
|
||
missing.append(m)
|
||
res["required_missing"] = list(missing)
|
||
|
||
hits = []
|
||
for f in forb_list:
|
||
fl = f.lower()
|
||
matched = [(i + 1, ln) for i, ln in enumerate(lines) if fl in ln.lower()]
|
||
if matched:
|
||
out("[HIT] forbidden pattern: {} ({} line(s))".format(f, len(matched)))
|
||
# 命中行必须打出来（2026-10-10）：旧实现只报 "(3 line(s))"，人还得自己
|
||
# 再 grep 一遍才知道是哪几行——而"哪几行"正是定位所需的那一步
|
||
for ln_no, ln in matched[:HIT_LINES_MAX]:
|
||
out(" L{}: {}".format(ln_no, ln.strip()[:200]))
|
||
if len(matched) > HIT_LINES_MAX:
|
||
out(" ... {} more".format(len(matched) - HIT_LINES_MAX))
|
||
hits.append(f)
|
||
res["forbidden_hits"].append({
|
||
"pattern": f, "count": len(matched),
|
||
"lines": [{"no": n, "text": t.strip()[:200]}
|
||
for n, t in matched[:HIT_LINES_MAX]]})
|
||
else:
|
||
out("[OK] forbidden pattern absent: {}".format(f))
|
||
|
||
# ---- whole-line exact checks + cross-reset pairing (task-36) ----
|
||
payloads = [_payload_of(l) for l in lines]
|
||
res["payload_lines"] = len(payloads)
|
||
exact_bad = []
|
||
for spec in (exact or []):
|
||
pat, _want = _parse_exact(spec)
|
||
ok_x, detail, neigh, residuals = exact_check(payloads, spec, allow_residual, head_char)
|
||
if ok_x:
|
||
out("[OK] exact line: {} {}".format(pat[:64], detail))
|
||
res["exact_ok"].append({"pattern": pat, "detail": detail})
|
||
else:
|
||
out("[FAIL] exact line: {} {}".format(pat[:64], detail))
|
||
for nb in neigh:
|
||
out(" inexact neighbour: {}".format(nb.strip()[:170]))
|
||
exact_bad.append(pat)
|
||
res["exact_bad"].append({"pattern": pat, "detail": detail,
|
||
"neighbours": [n.strip()[:170] for n in neigh]})
|
||
for k, v in residuals:
|
||
out("[NOTE] {} residual byte(s) before the line head, realigned and NOT a "
|
||
"failure: {}".format(k, repr(v)))
|
||
pair_bad = []
|
||
for f_pat, t_pat in (pairs or []):
|
||
ok_p, detail = pair_check(payloads, f_pat, t_pat)
|
||
if ok_p:
|
||
out("[OK] pair: {} -> {}".format(f_pat[:44], t_pat[:44]))
|
||
res["pair_ok"].append({"first": f_pat, "then": t_pat, "detail": detail})
|
||
else:
|
||
out("[FAIL] pair: {} -> {} {}".format(f_pat[:44], t_pat[:44], detail))
|
||
pair_bad.append({"first": f_pat, "then": t_pat, "detail": detail})
|
||
res["pair_bad"].append(detail)
|
||
|
||
# 启动锚点计数（2026-10-10）：抓"跑完但中途复位过"和"采集没从复位开始"
|
||
boot_issue = False
|
||
if max_boot_count > 0:
|
||
banner_lines = [ln for ln in lines if boot_marker in ln]
|
||
n_boot = len(banner_lines)
|
||
causes = [_reset_cause_of(ln, boot_marker) for ln in banner_lines]
|
||
if n_boot == 0:
|
||
out("[FAIL] no boot banner (marker '{}'): capture did not start at a reset".format(
|
||
boot_marker))
|
||
boot_issue = True
|
||
elif n_boot > max_boot_count:
|
||
out("[FAIL] boot banner x{} > allowed {} -> board reset itself during capture".format(
|
||
n_boot, max_boot_count))
|
||
for ln in banner_lines[:6]:
|
||
out(" {}".format(ln.strip()))
|
||
boot_issue = True
|
||
else:
|
||
out("[OK] boot banner x{} (marker '{}')".format(n_boot, boot_marker))
|
||
if causes:
|
||
out("[INFO] reset cause(s): {}".format("; ".join(causes)))
|
||
exp = _split_list(expect_reset_cause)
|
||
if exp:
|
||
got = causes[0].lower()
|
||
if any(e.lower() in got for e in exp):
|
||
out("[OK] reset cause '{}' matches expected {}".format(
|
||
causes[0], expect_reset_cause))
|
||
else:
|
||
out("[FAIL] reset cause '{}' not in expected: {}".format(
|
||
causes[0], expect_reset_cause))
|
||
boot_issue = True
|
||
res["boot"] = {"marker": boot_marker, "count": n_boot, "allowed": max_boot_count,
|
||
"causes": causes, "issue": boot_issue}
|
||
|
||
# 标记顺序（2026-10-10）：标记全中但**顺序颠倒** = 某步在另一步之后补做
|
||
# （屏障失序 / 失败重试补跑）。旧实现只看子串存在性，这类失序完全看不见。
|
||
order_issue = False
|
||
if require_order:
|
||
seq = [s.strip() for s in require_order.split(">") if s.strip()]
|
||
pos = []
|
||
for m in seq:
|
||
idx = next((i for i, ln in enumerate(lines) if m in ln), -1)
|
||
pos.append((m, idx))
|
||
absent = [m for m, i in pos if i < 0]
|
||
if absent:
|
||
out("[FAIL] require-order: marker(s) not found: {}".format("; ".join(absent)))
|
||
order_issue = True
|
||
else:
|
||
bad = [(pos[k][0], pos[k + 1][0]) for k in range(len(pos) - 1)
|
||
if pos[k][1] > pos[k + 1][1]]
|
||
if bad:
|
||
out("[FAIL] marker order violated (expected {})".format(require_order))
|
||
for a, b in bad:
|
||
out(" '{}' first appeared AFTER '{}'".format(a, b))
|
||
for m, i in pos:
|
||
out(" L{}: {}".format(i + 1, m))
|
||
order_issue = True
|
||
else:
|
||
out("[OK] marker order ok: {}".format(require_order))
|
||
res["order_bad"] = [m for m, i in pos if i < 0] + [
|
||
"{}>{}".format(a, b) for a, b in
|
||
[(pos[k][0], pos[k + 1][0]) for k in range(len(pos) - 1) if pos[k][1] > pos[k + 1][1]]]
|
||
|
||
# 开机耗时（2026-10-10）：行首 [+Nms] = daemon 毫秒时基的相对偏移
|
||
# （MCU 侧时间戳只有秒级，且开机头几行时钟还没起，量不出耗时）。
|
||
timing = None
|
||
timing_issue = False
|
||
offs = []
|
||
for i, ln in enumerate(lines):
|
||
m = _TS_OFF_RE.match(ln)
|
||
if m:
|
||
offs.append((i, int(m.group(1))))
|
||
if offs:
|
||
timing = {"ts_lines": len(offs), "first_ms": offs[0][1], "last_ms": offs[-1][1],
|
||
"span_ms": offs[-1][1] - offs[0][1]}
|
||
out("[INFO] capture span: +{}ms .. +{}ms ({} timestamped line(s))".format(
|
||
timing["first_ms"], timing["last_ms"], timing["ts_lines"]))
|
||
if done_marker:
|
||
di = next((i for i, ln in enumerate(lines) if done_marker in ln), -1)
|
||
if di < 0:
|
||
out("[WARN] done marker '{}' not found -> boot time not measured".format(
|
||
done_marker))
|
||
else:
|
||
doff = next((off for i, off in offs if i >= di), None)
|
||
if doff is None:
|
||
out("[WARN] done marker line has no [+Nms] prefix -> not measured")
|
||
else:
|
||
boot_ms = doff - offs[0][1]
|
||
timing["boot_ms"] = boot_ms
|
||
timing["done_marker"] = done_marker
|
||
out("[INFO] boot time to '{}': {} ms".format(done_marker, boot_ms))
|
||
if max_boot_ms > 0 and boot_ms > max_boot_ms:
|
||
out("[FAIL] boot time {} ms > limit {} ms".format(boot_ms, max_boot_ms))
|
||
timing_issue = True
|
||
elif done_marker or max_boot_ms > 0:
|
||
out("[WARN] no [+Nms] line prefix in this log -> timing checks skipped "
|
||
"(run_boot_test.py adds it; --raw-log turns it off)")
|
||
res["timing"] = timing
|
||
|
||
# 套件汇总行解析：抓"没跑到汇总行 / 断言数为 0"的盲区
|
||
bad_summary = []
|
||
if summary_list:
|
||
parsed = {}
|
||
for ln in lines:
|
||
m = _SUMMARY_RE.search(ln)
|
||
if m:
|
||
parsed.setdefault(m.group("name").strip(), []).append(m)
|
||
for name in summary_list:
|
||
cands = parsed.get(name)
|
||
if not cands:
|
||
out("[MISS] suite summary NOT found: '=== {}: N/M PASSED (failed=0) ==='".format(name))
|
||
bad_summary.append(name)
|
||
continue
|
||
ok = False
|
||
for m in cands:
|
||
passed = int(m.group("passed"))
|
||
total = int(m.group("total"))
|
||
failed = int(m.group("failed"))
|
||
if failed != 0:
|
||
out("[FAIL] suite summary reports failures: {} -> {}/{} PASSED (failed={})".format(
|
||
name, passed, total, failed))
|
||
continue
|
||
if total == 0:
|
||
out("[FAIL] suite summary total is 0 (no assertion ran): '{}'".format(name))
|
||
continue
|
||
if passed != total:
|
||
out("[FAIL] suite summary inconsistent: {} -> {}/{} PASSED".format(
|
||
name, passed, total))
|
||
continue
|
||
out("[OK] suite summary ok: {} -> {}/{} PASSED".format(name, passed, total))
|
||
ok = True
|
||
if not ok:
|
||
bad_summary.append(name)
|
||
res["suites_bad"] = list(bad_summary)
|
||
|
||
# 自检记录联合判定（S7）：/log/selftest.log 定长 64B 记录（格式见 App/log/sys_log.h），
|
||
# 仅故障写盘：文件缺失 / 0 字节 = 无故障 → PASS；存在任何一条记录 → FAIL
|
||
selftest_fail = False
|
||
if selftest_pass:
|
||
sp = Path(selftest_log) if selftest_log else None
|
||
if sp is None or not sp.is_file() or sp.stat().st_size == 0:
|
||
out("[OK] selftest: no fault record (file missing or empty)")
|
||
res["selftest"] = {"path": str(sp) if sp else None, "faults": 0, "records": []}
|
||
else:
|
||
raw = sp.read_bytes()
|
||
n = len(raw) // 64
|
||
if len(raw) % 64:
|
||
out("[WARN] selftest log has {} trailing incomplete bytes (ignored)".format(len(raw) % 64))
|
||
recs = []
|
||
for i in range(n):
|
||
rec = raw[i * 64:(i + 1) * 64]
|
||
ts = rec[0:19].decode("ascii", errors="replace")
|
||
desc = rec[31:59].split(b"\0", 1)[0].decode("utf-8", errors="replace")
|
||
recs.append({"no": i, "ts": ts, "desc": desc})
|
||
res["selftest"] = {"path": str(sp), "faults": n, "records": recs}
|
||
if n > 0:
|
||
selftest_fail = True
|
||
out("[FAIL] selftest: {} fault record(s) in {}".format(n, selftest_log))
|
||
for r in recs:
|
||
out(" #{}: {} {}".format(r["no"], r["ts"], r["desc"]))
|
||
else:
|
||
out("[OK] selftest: no fault record (0 complete records)")
|
||
|
||
if show_tail > 0 and lines:
|
||
res["tail"] = lines[-show_tail:]
|
||
out("")
|
||
out("=== Last {} line(s) of log ===".format(show_tail))
|
||
for ln in lines[-show_tail:]:
|
||
out(ln)
|
||
|
||
# ---- measurement vs device (task-32) ----
|
||
n_exact_checked = len(exact or [])
|
||
n_exact_hits = sum(1 for e in (res.get("exact_ok") or []) if "hits=0" not in e["detail"])
|
||
n_neighbours = 0
|
||
for e in (res.get("exact_ok") or []) + (res.get("exact_bad") or []):
|
||
m = re.search(r"loose_neighbours=(\d+)", e.get("detail", ""))
|
||
if m:
|
||
n_neighbours += int(m.group(1))
|
||
b_on = max_boot_count > 0
|
||
b_cnt = (res.get("boot") or {}).get("count", 0)
|
||
broken, m_reason = classify_measurement(len(payloads), b_cnt, b_on,
|
||
n_exact_checked > 0, n_exact_hits, n_neighbours)
|
||
res["measurement"] = {"broken": broken, "reason": m_reason, "payload_lines": len(payloads),
|
||
"boot_anchor_count": b_cnt, "anchor_check_on": b_on,
|
||
"exact_checked": n_exact_checked, "exact_hits": n_exact_hits,
|
||
"loose_neighbours": n_neighbours}
|
||
# A broken capture must never mask real device-side bad news: if the log carries
|
||
# Forbidden hits or selftest fault records, that evidence stands and we report FAIL(1)
|
||
# with the measurement problem noted on top of it (worst outcome = a real crash quietly
|
||
# re-labelled as "no data").
|
||
if broken and (hits or selftest_fail):
|
||
out("")
|
||
out("[WARN] capture also looks incomplete: " + m_reason)
|
||
out(" but device-side bad news is present -> verdict stays FAIL, not MEASUREMENT-BROKEN")
|
||
broken = False
|
||
if broken:
|
||
out("")
|
||
out("[MEASUREMENT-BROKEN] " + m_reason)
|
||
out(" no verdict is meaningful here -- fix the measurement first (open the port,")
|
||
out(" assert the collector is alive, capture from a reset). Exit 2, NOT 1:")
|
||
out(" a dead collector must not be counted as one more device failure.")
|
||
return finish(2)
|
||
|
||
out("")
|
||
if (missing or hits or bad_summary or selftest_fail or boot_issue
|
||
or order_issue or timing_issue or exact_bad or pair_bad):
|
||
out("VERDICT: FAIL")
|
||
return finish(1)
|
||
out("VERDICT: PASS")
|
||
return finish(0)
|
||
|
||
|
||
def selftest(out=print):
|
||
"""Prove the exact/pair machinery actually discriminates (exit 0 = all behaviours ok).
|
||
|
||
Written because a PASS that cannot fail is worthless: case 2/3 are the real captured
|
||
shapes of the 2026-10-26 console-interleave defect, and case 9 shows that the legacy
|
||
substring check calls the SAME defective line a PASS -- that is the blind spot --exact
|
||
exists to close.
|
||
"""
|
||
T = "[2026-10-26 08:00:00]"
|
||
prot_re = (r"\[\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\]\[PROT\] uncalibrated -> protection "
|
||
r"blocked \(UA/UN/IA VALID not all set\)")
|
||
nand_re = r"\[\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\]\[NAND_FTL\] Initializing NAND FTL\.\.\."
|
||
prot_ok = T + "[PROT] uncalibrated -> protection blocked (UA/UN/IA VALID not all set)"
|
||
prot_spliced = T + "[PROT] uncalibrated -> protection [blocked (UA/UN/IA VALI2D not all s0et)"
|
||
nand_ok = T + "[NAND_FTL] Initializing NAND FTL..."
|
||
nand_headless = "26-10-26 08:37:07][NAND_FTL] Initializing NAND FTL..."
|
||
mem_re = r"\[ERR\] MEMFAULT PC=\S+ LR=\S+ CFSR=\S+ HFSR=\S+"
|
||
mem_ok = "[ERR] MEMFAULT PC=0x08004a12 LR=0x08004a3f CFSR=0x00008200 HFSR=0x40000000"
|
||
rc_ok = T + "[APP] reset cause: software (NVIC_SystemReset) (CSR=0x14000000)"
|
||
cases = []
|
||
|
||
def chk(name, got, want):
|
||
cases.append((name, got == want, "got={!r} want={!r}".format(got, want)))
|
||
|
||
chk("clean exact", exact_check([prot_ok], prot_re)[0], True)
|
||
ok_s, _d_s, neigh_s, _r_s = exact_check([prot_spliced], prot_re)
|
||
chk("spliced line rejected", ok_s, False)
|
||
chk("spliced line fails even with broken probe key", len(neigh_s), 0)
|
||
ok_n, _d_n, neigh_n, _r_n = exact_check([nand_headless], nand_re)
|
||
chk("headless line rejected", ok_n, False)
|
||
chk("headless line reported as neighbour", len(neigh_n) >= 1, True)
|
||
chk("residual 1 rejected when not allowed", exact_check(["#" + nand_ok], nand_re)[0], False)
|
||
ok_r, det_r, _n_r, res_r = exact_check(["#" + nand_ok], nand_re, allow_residual=1)
|
||
chk("residual 1 accepted when allowed", ok_r and len(res_r) == 1, True)
|
||
chk("residual 2 still rejected", exact_check(["##" + nand_ok], nand_re, allow_residual=1)[0], False)
|
||
chk("head_char enforced", exact_check([nand_ok], nand_re, 0, "[")[0], True)
|
||
chk("head_char catches wrong head", exact_check([nand_headless], r"26-10-26.*NAND_FTL.*", 0, "[")[0], False)
|
||
chk("len pin ok", exact_check([nand_ok], nand_re + EXACT_LEN_SEP + str(len(nand_ok)))[0], True)
|
||
chk("len pin mismatch fails", exact_check([nand_ok], nand_re + EXACT_LEN_SEP + "9")[0], False)
|
||
chk("pair good", pair_check([mem_ok, rc_ok], "MEMFAULT", "reset cause:")[0], True)
|
||
chk("pair two following causes fails",
|
||
pair_check([mem_ok, rc_ok, rc_ok], "MEMFAULT", "reset cause:")[0], False)
|
||
chk("pair wrong order fails",
|
||
pair_check([rc_ok, mem_ok], "MEMFAULT", "reset cause:")[0], False)
|
||
# blind spot: legacy substring would PASS on this defective line
|
||
full = "\n".join([prot_ok, nand_headless])
|
||
legacy_present = "Initializing NAND FTL" in full # what --required sees
|
||
exact_ok_here = exact_check([prot_ok, nand_headless], nand_re)[0]
|
||
chk("blind spot proven: legacy substring PASSES where --exact FAILS",
|
||
legacy_present and not exact_ok_here, True)
|
||
|
||
# measurement-vs-device classification (task-32): the three-counter rule IS this function,
|
||
# so it is pinned here -- an all-zero read must not be reportable as a healthy device.
|
||
chk("measure: good capture -> not broken", classify_measurement(137, 1, True, True, 1, 0)[0], False)
|
||
chk("measure: zero payload lines -> broken", classify_measurement(0, 0, True, True, 0, 0)[0], True)
|
||
chk("measure: readable but no boot anchor -> broken",
|
||
classify_measurement(1, 0, True, True, 0, 0)[0], True)
|
||
chk("measure: anchor check off -> abstain (non-boot logs stay usable)",
|
||
classify_measurement(0, 0, False, True, 0, 0)[0], False)
|
||
chk("measure: damaged lines are device-side, not 'no data'",
|
||
classify_measurement(120, 1, True, True, 0, 3)[0], False)
|
||
chk("measure: anchors present + no neighbour = functional miss",
|
||
classify_measurement(120, 1, True, True, 0, 0)[0], False)
|
||
|
||
bad = 0
|
||
for name, ok, detail in cases:
|
||
out("{} {}".format("[OK] " if ok else "[FAIL]", name))
|
||
if not ok:
|
||
out(" {}".format(detail))
|
||
bad += 1
|
||
out("SELFTEST: {} case(s), {} failure(s) -> {}".format(
|
||
len(cases), bad, "PASS" if bad == 0 else "FAIL"))
|
||
return 0 if bad == 0 else 1
|
||
|
||
|
||
def main(argv=None):
|
||
# 控制台输出固定 UTF-8：Windows 代码页（GBK/UTF-8 终端）均不会让本脚本崩溃
|
||
for stream in (sys.stdout, sys.stderr):
|
||
try:
|
||
stream.reconfigure(encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
ap = argparse.ArgumentParser(description="MCU debug log marker check (pure stdlib)")
|
||
ap.add_argument("--log-path", required=True, help="boot log file to check")
|
||
ap.add_argument("--required", default=REQUIRED_DEFAULT,
|
||
help="required markers, ';' separated (case-sensitive)")
|
||
ap.add_argument("--forbidden", default=FORBIDDEN_DEFAULT,
|
||
help="forbidden patterns, ';' separated (case-insensitive)")
|
||
ap.add_argument("--require-suite-summary", default="",
|
||
help="suite names that MUST appear as '=== <name>: N/M PASSED (failed=0) ===' "
|
||
"with N>0, ';' separated. Empty = off")
|
||
ap.add_argument("--show-tail-lines", type=int, default=40,
|
||
help="show last N log lines (0 = off)")
|
||
ap.add_argument("--require-selftest-pass", action="store_true",
|
||
help="judge /log/selftest.log local copy (S7): "
|
||
"missing/empty = PASS, any 64B record = FAIL")
|
||
ap.add_argument("--selftest-log", default="",
|
||
help="local copy path of /log/selftest.log (with --require-selftest-pass)")
|
||
ap.add_argument("--boot-marker", default=BOOT_MARKER_DEFAULT,
|
||
help="per-boot banner substring (default '{}')".format(BOOT_MARKER_DEFAULT))
|
||
ap.add_argument("--max-boot-count", type=int, default=0,
|
||
help="fail if the boot banner appears 0 times or more than N times "
|
||
"(0 = check off). run_boot_test.py passes 1 by default")
|
||
ap.add_argument("--expect-reset-cause", default="",
|
||
help="';'-separated substrings the FIRST reset cause must match "
|
||
"(case-insensitive). Empty = report only, no verdict")
|
||
ap.add_argument("--require-order", default="",
|
||
help="'>'-separated markers that must appear in that line order "
|
||
"(e.g. 'App init OK>Network Init>net init: OK'). Empty = off")
|
||
ap.add_argument("--done-marker", default="",
|
||
help="marker whose line marks boot completion, for --max-boot-ms")
|
||
ap.add_argument("--max-boot-ms", type=int, default=0,
|
||
help="fail if time from first line to --done-marker exceeds N ms "
|
||
"(needs '[+Nms]' line prefix; 0 = off)")
|
||
ap.add_argument("--exact", action="append", default=[],
|
||
help="repeatable regex that must match a WHOLE firmware line "
|
||
"(collector prefix stripped, anchored ^...$); a spliced byte or a "
|
||
"missing line head fails. Optional suffix '@@LEN=<n>' pins the byte "
|
||
"length -- opt-in only, lengths belong to one image. Use --exact "
|
||
"instead of --required for any marker whose line can lose its head.")
|
||
ap.add_argument("--allow-residual", type=int, default=0,
|
||
help="accept up to N stray bytes before the line head on an --exact hit "
|
||
"(crash-dump legs: 1 shift-register leftover is known; recorded, "
|
||
"not a failure). 0 = strict (default)")
|
||
ap.add_argument("--head-char", default="",
|
||
help="every --exact hit must start with this char right after the "
|
||
"optional residual prefix (crash-dump legs: '[')")
|
||
ap.add_argument("--pair", action="append", nargs=2, metavar=("FIRST", "THEN"), default=[],
|
||
help="repeatable pair: exactly one FIRST line, and exactly one THEN line "
|
||
"after it (cross-reset pairing, e.g. 'MEMFAULT' 'reset cause:')")
|
||
ap.add_argument("--include-comments", action="store_true",
|
||
help="do NOT skip '#' lines (archived evidence files carry an ASCII "
|
||
"provenance header; skipping it is the default so a header that "
|
||
"merely mentions [ERR] cannot trip --forbidden)")
|
||
ap.add_argument("--selftest", action="store_true",
|
||
help="run the built-in discrimination cases and exit (0 = all behave "
|
||
"as specified); needs no log file and no hardware")
|
||
ap.add_argument("--json", action="store_true",
|
||
help="emit one JSON object instead of the text verdict (exit code unchanged)")
|
||
args = ap.parse_args(argv)
|
||
|
||
if args.selftest:
|
||
return selftest()
|
||
|
||
return verify(args.log_path, args.required, args.forbidden,
|
||
args.show_tail_lines, suite_summary=args.require_suite_summary,
|
||
selftest_pass=args.require_selftest_pass,
|
||
selftest_log=args.selftest_log,
|
||
boot_marker=args.boot_marker,
|
||
max_boot_count=args.max_boot_count,
|
||
expect_reset_cause=args.expect_reset_cause,
|
||
require_order=args.require_order,
|
||
done_marker=args.done_marker,
|
||
max_boot_ms=args.max_boot_ms,
|
||
as_json=args.json,
|
||
exact=args.exact, pairs=args.pair,
|
||
allow_residual=args.allow_residual, head_char=args.head_char,
|
||
include_comments=args.include_comments)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|