串口助手补残行封顶与会话边界复位、截头按行对齐、open 保持 dtr/rts；判定加命中行/顺序/耗时/JSON 与产物清理

2026-10-10 03:17:16 +08:00
parent fccf99bc90
commit 61a6061950
8 changed files with 410 additions and 46 deletions
@@ -8,12 +8,15 @@
      标记——芯片里跑的必须就是这份构建（--no-fw-check 关闭）
  3. wait-idle：RX 静默 >= settle-ms 收尾；首字节超过 --first-wait-ms 未到 → no_data
     （快速失败）；整体 max-wait-ms 超时
-  4. 从日志提取 [RX] 载荷行（剥掉 "[RX] <ISO时间戳> " 前缀）
+  4. 从日志提取 [RX] 载荷行（剥掉 "[RX] <ISO时间戳> " 前缀），并给每行加 `[+Nms]` 相对
+     偏移（daemon 毫秒时基，本次第一行 = +0ms；--raw-log 出纯载荷）。MCU 侧时间戳只有
+     秒级、开机头几行还没时钟，量不了耗时——这个前缀就是耗时口径
  4.5 烧录回读校验：MDK-ARM\\reset.bat cmp（= ST-LINK_CLI -CmpFile 全片逐字节比对，
      `No difference found` 才算真写进去；--no-verify-flash 关闭）。
      **必须排在采集之后**：实测 -CmpFile 连接时会复位目标（板子当场重打一遍开机日志）
  5. verify_log.py 标记校验（含启动锚点计数：`reset cause:` 出现 0 次或 >N 次 = FAIL，
-      N 由 --max-boot-count 给，默认 1；--expect-reset-cause 可对第一个锚点做原因白名单）
+      N 由 --max-boot-count 给，默认 1；--expect-reset-cause 可对第一个锚点做原因白名单；
+      --require-order 查标记先后顺序；--done-marker + --max-boot-ms 量开机耗时）
  6. （仅当 --require-selftest-pass）自检记录联合判定（S7）：
     --selftest-log 提供本地副本则直接用它，否则从设备 FTP 抓取 /log/selftest.log
     （lftpd :21 匿名）；判定：文件缺失 / 0 字节 = 无故障 → PASS，有 64B 记录 = FAIL
@@ -41,6 +44,9 @@
      [--fw-hex <构建产物 hex>] [--no-fw-check]
      [--no-verify-flash] [--readback-timeout 120]
      [--max-boot-count 1] [--expect-reset-cause 'software;power-on']
+      [--require-order 'App init OK>Network Init>net init: OK']
+      [--done-marker 'net init: OK'] [--max-boot-ms 0]
+      [--raw-log] [--keep-out 20]

  --port 缺省取 .opencode/serial_config.json 的 portName（单一配置源）

@@ -166,6 +172,26 @@ def _ftp_quit(ftp):
        pass


+def _prune_out_dir(out_dir, keep):
+    """只保留最近 N 份产物（`boot_log_*.txt` / `selftest_log_*.log`）。
+
+    这些文件按 `%Y%m%d_%H%M%S` 命名、每次 @test 各留一份，旧实现从不清理——
+    本机实测 out 目录已堆到 31 份 boot_log + 7 份 selftest_log（2026-10-10）。
+    文件名含时间戳 → 按名字排序即按时间排序。
+    """
+    if keep <= 0:
+        return 0
+    removed = 0
+    for pat in ("boot_log_*.txt", "selftest_log_*.log"):
+        for p in sorted(out_dir.glob(pat), key=lambda q: q.name)[:-keep]:
+            try:
+                p.unlink()
+                removed += 1
+            except OSError:
+                pass
+    return removed
+
+
 def _fetch_selftest_log(device_ip, out_dir, stamp):
    """从设备 FTP 抓取 /log/selftest.log（lftpd :21 匿名，根目录 = 卷根）。

@@ -258,6 +284,20 @@ def main(argv=None):
    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 "
+                         "(catches out-of-order boot steps); empty = off")
+    ap.add_argument("--done-marker", default="net init: OK",
+                    help="marker whose line marks boot completion (for --max-boot-ms); "
+                         "empty = don't measure")
+    ap.add_argument("--max-boot-ms", type=int, default=0,
+                    help="fail if time from first line to --done-marker exceeds N ms "
+                         "(0 = measure and report only)")
+    ap.add_argument("--raw-log", action="store_true",
+                    help="write payload-only boot log (no '[+Nms]' relative-time prefix)")
+    ap.add_argument("--keep-out", type=int, default=20,
+                    help="keep the newest N boot_log_*/selftest_log_* files in --out-dir "
+                         "(0 = keep everything)")
    args = ap.parse_args(argv)

    # 串口助手 CLI 用当前解释器（AGENTS.md 统一入口 = 仓库 .venv python）
@@ -289,6 +329,10 @@ def main(argv=None):

    out_dir = Path(args.out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
+    pruned = _prune_out_dir(out_dir, args.keep_out)
+    if pruned:
+        print("OUT: pruned {} old artifact file(s) in {} (keep={})".format(
+            pruned, out_dir, args.keep_out))

    stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    boot_txt = out_dir / "boot_log_{}.txt".format(stamp)
@@ -371,21 +415,44 @@ def main(argv=None):
        return 2

    # 5. 提取 [RX] 载荷行（剥掉 "[RX] <ISO时间戳> " 前缀）
+    #    默认给每行加 `[+Nms]` 相对偏移（daemon 毫秒时基，本次采集第一行 = +0ms）。
+    #    为什么加：MCU 侧时间戳只有**秒级**，且开机头几行时钟还没起（`----/--/--`），
+    #    根本量不了耗时；有了它 verify_log 才能查"能开机但慢了几秒"（--max-boot-ms）。
+    #    --raw-log 恢复旧格式（纯载荷）。
    data, out = tool("log", "--tail", "0", "--timeout", "15")
    if data is None:
        print("RUN_TEST_ERROR: log read failed: {}".format(out))
        return 2
    rx = []
+    t0 = None
    for l in data.get("lines", []):
-        if isinstance(l, str) and l.startswith("[RX] "):
-            rest = l[5:]
-            sp = rest.find(" ")
-            rx.append(rest[sp + 1:] if sp >= 0 else rest)
+        if not (isinstance(l, str) and l.startswith("[RX] ")):
+            continue
+        rest = l[5:]
+        sp = rest.find(" ")
+        ts_txt = rest[:sp] if sp >= 0 else ""
+        payload = rest[sp + 1:] if sp >= 0 else rest
+        if args.raw_log:
+            rx.append(payload)
+            continue
+        ts = None
+        if ts_txt:
+            try:
+                ts = datetime.fromisoformat(ts_txt)
+            except ValueError:
+                ts = None
+        if ts is None:
+            rx.append(payload)          # 时间戳读不出来就不加前缀（不猜）
+            continue
+        if t0 is None:
+            t0 = ts
+        rx.append("[+{}ms] {}".format(int((ts - t0).total_seconds() * 1000), payload))
    if not rx:
        print("RUN_TEST_ERROR: no [RX] lines in log (device silent?)")
        return 2
    boot_txt.write_bytes(("\r\n".join(rx) + "\r\n").encode("utf-8"))
-    print("Boot log: {}  (lines={}  gen={})".format(boot_txt, len(rx), data.get("gen")))
+    print("Boot log: {}  (lines={}  gen={}  ts={})".format(
+        boot_txt, len(rx), data.get("gen"), "raw" if args.raw_log else "+ms"))

    # 5.6 烧录回读校验（AGENTS.md 铁律：`No difference found` 才算真写进去）。
    #     必须排在**采集之后**：实测 -CmpFile 连接时会复位目标（板子当场重打一遍开机
@@ -422,7 +489,10 @@ def main(argv=None):
                selftest_pass=args.require_selftest_pass,
                selftest_log=selftest_local,
                max_boot_count=args.max_boot_count,
-                expect_reset_cause=args.expect_reset_cause)
+                expect_reset_cause=args.expect_reset_cause,
+                require_order=args.require_order,
+                done_marker=args.done_marker,
+                max_boot_ms=args.max_boot_ms)
    print("RUN_TEST_RC={}".format(rc))
    return rc

@@ -13,6 +13,8 @@
      [--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]

 退出码（与旧版一致）:
  0 = PASS（所有 Required 标记找到，Forbidden 零命中，汇总行要求满足）
@@ -49,9 +51,20 @@
    的残留日志）；**>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
@@ -60,6 +73,13 @@ 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 命中时才被发现，
@@ -104,7 +124,8 @@ 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=""):
+           expect_reset_cause="", require_order="", done_marker="",
+           max_boot_ms=0, as_json=False):
    """标记校验核心（可被 run_boot_test.py 复用）。返回退出码 0/1/2。

    suite_summary: ';' 分隔的套件名列表。非空时，要求日志中每个套件都出现
@@ -117,16 +138,37 @@ def verify(log_path, required=REQUIRED_DEFAULT, forbidden=FORBIDDEN_DEFAULT,
                   在采集窗口内自己复位过（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}
+    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 2
+        return finish(2)
    try:
        lines = _read_lines(log_path)
    except OSError:
        out("VERIFY_ERROR: cannot read log file: {}".format(log_path))
-        return 2
+        return finish(2)

    full = "\n".join(lines)
    req_list = _split_list(required)
@@ -135,6 +177,8 @@ def verify(log_path, required=REQUIRED_DEFAULT, forbidden=FORBIDDEN_DEFAULT,

    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:
@@ -143,14 +187,25 @@ def verify(log_path, required=REQUIRED_DEFAULT, forbidden=FORBIDDEN_DEFAULT,
        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()
-        n = sum(1 for ln in lines if fl in ln.lower())
-        if n > 0:
-            out("[HIT]  forbidden pattern: {} ({} line(s))".format(f, n))
+        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))

@@ -184,6 +239,73 @@ def verify(log_path, required=REQUIRED_DEFAULT, forbidden=FORBIDDEN_DEFAULT,
                    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

    # 套件汇总行解析（T-2）：抓"没跑到汇总行 / 断言数为 0"的盲区
    bad_summary = []
@@ -219,6 +341,7 @@ def verify(log_path, required=REQUIRED_DEFAULT, forbidden=FORBIDDEN_DEFAULT,
                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
@@ -227,34 +350,41 @@ def verify(log_path, required=REQUIRED_DEFAULT, forbidden=FORBIDDEN_DEFAULT,
        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 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")
-                    out("       #{}: {}  {}".format(i, ts, desc))
+                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)

    out("")
-    if missing or hits or bad_summary or selftest_fail or boot_issue:
+    if (missing or hits or bad_summary or selftest_fail or boot_issue
+            or order_issue or timing_issue):
        out("VERDICT: FAIL")
-        return 1
+        return finish(1)
    out("VERDICT: PASS")
-    return 0
+    return finish(0)


 def main(argv=None):
@@ -289,6 +419,16 @@ def main(argv=None):
    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("--json", action="store_true",
+                    help="emit one JSON object instead of the text verdict (exit code unchanged)")
    args = ap.parse_args(argv)

    return verify(args.log_path, args.required, args.forbidden,
@@ -297,7 +437,11 @@ def main(argv=None):
                  selftest_log=args.selftest_log,
                  boot_marker=args.boot_marker,
                  max_boot_count=args.max_boot_count,
-                  expect_reset_cause=args.expect_reset_cause)
+                  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)


 if __name__ == "__main__":