wandering1
8c28f3bfdd
- AGENTS.md：钩子 40→42、主机单测钩子 16→18（adc-cal/prot-dim）、 覆盖率总闸 28→35 套、layer-deps 基线 6→3（记录 reader 注入兑现） - IEC104上行设计说明.md：遥测读数改工程量口径 + 死区额定 0.5% 重定 已落地（卡 6）；§8 第 4 条改写、变更记录追加行 - 采集与算法层设计说明.md：状态行/红线③/幅值门三处由「标定未实现」 更新为「折算已落地、精度复核待卡 12 注入」 - App/log/soe_rec.h 类型注释表补 7/8 两行（lane-a 口误 6/7 已勘误， 真值按枚举链 SAMPLE_RESUME=6 连续计数） - test/host/soe/decode_soe.py：TYPE_NAMES + docstring 补 7/8—— 否则台架 FTP 下载校验会把标定闭锁记录判 bad type 误红（隐藏消费边）； soe 套件 32/32 复跑绿
161 lines
6.1 KiB
Python
161 lines
6.1 KiB
Python
#!/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..8）；元件事件 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",
|
||
7: "CAL_BLOCK", 8: "CAL_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())
|