STM32F4-Base/test/host/soe/decode_soe.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""decode_soe.py - /log/soe.log 主机解码器 + 断言（B9 台架腿证据工具，纯标准库）。

做什么：把 FTP 下载的设备端 /log/soe.log（定长 64B 二进制记录，格式见
    App/log/soe_rec.h）逐条解码成可读行，并断言：
      1) 文件大小 = 64 的整数倍（定长不变量）；
      2) 每条记录版本 = 1 且 sum8 校验和通过（撕裂/半写检测）；
      3) seq 单调：同一 boot 段内严格 +1（seq 回落 = 设备复位/容量重建段边界，
         如实分段计数，不算失败）；
      4) 时标：IV 记录如实报告；非 IV 记录时标必须落在合理窗口
         （2020..2099，对时链已就绪的台架前提）；
      5) 类型合法（1..6）；元件事件 elm_id < --max-elm（默认 16 留余量；
         PE_COUNT 现 7，B8 过频/低频并入后 9——写死 7 会在 B8 合并后误报）。
    任一断言失败 -> 退出码 1（台架腿 FAIL 判据之一）。

用法（仓库根目录）：
    & '.venv\\Scripts\\python.exe' test/host/soe/decode_soe.py <soe.log 路径>
    可选：--expect-assert N  断言至少 N 条"元件动作"记录（台架腿用 1）

退出码：0=全部断言 PASS / 1=断言失败 / 2=参数或文件错误。

格式镜像：App/log/soe_rec.h（改格式必须同步本工具）。
创建日期：2026-10-12（B9）
"""
import argparse
import calendar
import datetime
import struct
import sys

REC_SIZE = 64
VER = 1
OFF_VER, OFF_TYPE, OFF_ELM, OFF_FLAGS = 0, 1, 2, 3
OFF_SEQ, OFF_CP56, OFF_EPOCH, OFF_VALUE, OFF_SUM = 4, 8, 15, 19, 63
FLG_IV = 0x80

TYPE_NAMES = {
    1: "ELM_ASSERT", 2: "ELM_RETURN", 3: "PLATE_OUT", 4: "PLATE_IN",
    5: "SAMPLE_BLOCK", 6: "SAMPLE_RESUME",
}


def cp56_decode(b):
    """CP56Time2a 7B -> (datetime|None, iv). ms 域 = sec*1000（B9 秒域时基）。"""
    iv = (b[2] & 0x80) != 0
    ms_min = b[0] | (b[1] << 8)
    minute = b[2] & 0x3F
    hour = b[3] & 0x1F
    day = b[4] & 0x1F
    mon = b[5] & 0x0F
    year = 2000 + (b[6] & 0x7F)
    sec, ms = divmod(ms_min, 1000)
    try:
        dt = datetime.datetime(year, mon, day, hour, minute, sec, ms * 1000)
    except ValueError:
        dt = None
    return dt, iv


def main():
    ap = argparse.ArgumentParser(description="decode /log/soe.log")
    ap.add_argument("path", help="downloaded soe.log file")
    ap.add_argument("--expect-assert", type=int, default=0,
                    help="min ELM_ASSERT records required (bench leg: 1)")
    ap.add_argument("--max-elm", type=int, default=16,
                    help="exclusive upper bound for element elm_id "
                         "(PE_COUNT: 7 pre-B8, 9 post-B8; 16 = headroom)")
    args = ap.parse_args()

    try:
        with open(args.path, "rb") as f:
            data = f.read()
    except OSError as e:
        print("ERROR: cannot read %s: %s" % (args.path, e))
        return 2

    print("soe.log: %d bytes, %d records" % (len(data), len(data) // REC_SIZE))
    fails = []
    if len(data) == 0:
        fails.append("empty file (no SOE records)")
    if len(data) % REC_SIZE != 0:
        fails.append("size %d not multiple of %d (torn tail)"
                     % (len(data), REC_SIZE))

    n_seg = 0
    prev_seq = None
    n_assert = 0
    for i in range(len(data) // REC_SIZE):
        rec = data[i * REC_SIZE:(i + 1) * REC_SIZE]
        if rec[OFF_VER] != VER:
            fails.append("rec %d: bad version %d" % (i, rec[OFF_VER]))
            continue
        s = sum(rec[:OFF_SUM]) & 0xFF
        if rec[OFF_SUM] != s:
            fails.append("rec %d: checksum mismatch (stored 0x%02X calc 0x%02X)"
                         % (i, rec[OFF_SUM], s))
            continue
        rtype = rec[OFF_TYPE]
        elm = rec[OFF_ELM]
        flags = rec[OFF_FLAGS]
        seq = struct.unpack_from("<I", rec, OFF_SEQ)[0]
        epoch = struct.unpack_from("<I", rec, OFF_EPOCH)[0]
        value = struct.unpack_from("<I", rec, OFF_VALUE)[0]
        dt, iv = cp56_decode(rec[OFF_CP56:OFF_CP56 + 7])

        if rtype not in TYPE_NAMES:
            fails.append("rec %d: bad type %d" % (i, rtype))
        if rtype in (1, 2) and elm >= args.max_elm:
            fails.append("rec %d: elm_id %d out of range (max %d)"
                         % (i, elm, args.max_elm - 1))

        # seq monotonic within a boot segment; drop = new segment (reset/rebuild)
        if prev_seq is None or seq <= prev_seq:
            n_seg += 1
        elif seq != prev_seq + 1:
            fails.append("rec %d: seq gap %d -> %d (not strictly +1)"
                         % (i, prev_seq, seq))
        prev_seq = seq

        if (flags & FLG_IV) != 0:
            if not iv:
                fails.append("rec %d: flags IV but CP56 IV clear" % i)
            ts_s = "IV"
        else:
            if iv:
                fails.append("rec %d: CP56 IV set but flags IV clear" % i)
            if dt is None or not (2020 <= dt.year <= 2099):
                fails.append("rec %d: timestamp out of window: %r" % (i, dt))
            ts_s = dt.strftime("%Y-%m-%d %H:%M:%S") if dt else "BAD"
            # CP56 = UTC 分解；epoch 字段 = UTC 秒 → 用 calendar.timegm（UTC）。
            # ⚠ 本机时区 +08 时 dt.timestamp() 会差 28800 s（首跑实证坑）
            if dt and epoch and abs(calendar.timegm(dt.utctimetuple()) - epoch) > 2:
                fails.append("rec %d: cp56/epoch cross-check off (%s vs %d)"
                             % (i, ts_s, epoch))

        if rtype == 1:
            n_assert += 1
        print("  #%03d seq=%-10d %-13s elm=%-3s val=0x%08X ts=%s"
              % (i, seq, TYPE_NAMES.get(rtype, "?%d" % rtype),
                 "-" if elm == 0xFF else str(elm), value, ts_s))

    print("segments (boot/rebuild boundaries): %d, ELM_ASSERT records: %d"
          % (n_seg, n_assert))
    if n_assert < args.expect_assert:
        fails.append("ELM_ASSERT count %d < expected %d"
                     % (n_assert, args.expect_assert))

    if fails:
        for x in fails:
            print("FAIL: %s" % x)
        print("SUMMARY: FAIL (%d)" % len(fails))
        return 1
    print("SUMMARY: PASS")
    return 0


if __name__ == "__main__":
    sys.exit(main())