wandering1/STM32F4-Base
1
0
Fork 0
Code Issues 30 Pull Requests Actions Packages Projects 2 Releases Wiki Activity

test(b7): 定值持久化+软压板台架腿闭环（跨复位保持 42+8 判据）+ B6/B7 销项

Browse Source 
- @test PASS（B6 门控修复生效，protect 锚点行完好）
- 主站工具 iec104_master.py 增 --b7-pre/--b7-post/--b7-mark/--b7-check
  四模式（纯标准库、ASCII 输出可归档，中间 ST-LINK 复位）
- pre 21/21：YK12 两段 OFF → 0.05 A 低定值不动作（保护退出）；
  ON → YX15/16 立即动作（COT=3）；C_SE→C_RD 回读
- post 13/13：压板退出态跨复位保持（开机日志 global plate OUT +
  低定值仍不动作）、重启后 YK12 ON 联动在位
- mark/check 4/4+4/4：非默认定值 7.5 跨复位保持（C_RD=7.500，
  排除"回落出厂默认"假阳性）；台架留干净态（定值/压板回默认）
- 开机日志两次 [CFG] prot set loaded (/prot_set.bin)
- 证据：boot_log_data 20261012_B7_* 两份 + 清单行
- 文档：IEC104 §7 台架行/变更记录、保护规划状态行/§7 P4/B7 行尾巴、
  W1 卡 B6+B7 销项（删行 + §1 指针/§2 判据/§3 图/写范围注）、
  任务规划 README §2.0/§2.1、test/host/README
- 验证：doc-links 全绿（141 文档）、9 个文件字节卫生干净
This commit is contained in:
Wanderingss
2026-10-12 16:54:23 +08:00
parent dafcf8098b
commit 6215bb6f53
9 changed files with 476 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/host/README.md
+1 -1
View File
@@ -22,7 +22,7 @@
| `ftp_speed_test.py` | FTP 传输速度测试 |
| `ftp_capture.py` | FTP 流量抓取 |
| `ftp_regression.py` | FTP 回归套件 |
| `iec104_master.py` | **IEC 60870-5-104 主站客户端（B6）**：连设备 :2404 跑 STARTDT → 总召（actcon/四组背景扫描/actterm）→ 经 C_SE 两段写临时压低 OC1 定值确定性触发 YX16 变位（COT=3）→ 恢复 → YK11 两段式 → C_RD 读回；纯标准库，输出纯 ASCII 可直接重定向归档为台架证据（判据 exit 0/1/2） |
| `iec104_master.py` | **IEC 60870-5-104 主站客户端（B6/B7）**：连设备 :2404 跑 STARTDT → 总召（actcon/四组背景扫描/actterm）→ 经 C_SE 两段写临时压低 OC1 定值确定性触发 YX16 变位（COT=3）→ 恢复 → YK11 两段式 → C_RD 读回；**B7 持久化模式 `--b7-pre/--b7-post/--b7-mark/--b7-check`**（中间 ST-LINK 复位）：YK12 两段投退→保护退出/联动、定值与压板态跨复位保持（含非默认 7.5 往返）；纯标准库，输出纯 ASCII 可直接重定向归档为台架证据（判据 exit 0/1/2） |
### 时间 / 时延
test/host/iec104_master.py
+182 -2
View File
@@ -29,6 +29,19 @@ round-trip log as bench evidence).
Usage: python iec104_master.py [--ip 192.168.1.100] [--port 2404]
[--addr 0x8858] [--low 0.05]
[--restore 10.0]
B7 bench leg modes (setpoint persistence + global soft plate, task card
docs/任务规划/W1-采集保护收尾.md B7; firmware side App/cfg/prot_set_store.c):
--b7-pre : before reset: mark setpoint 7.5 (C_SE+C_RD), YK12 OFF (two-step)
-> low setpoint 0.05 must NOT trip YX16 (plate OUT disables
protection); YK12 ON -> YX16 trips (linkage); restore 10.0;
leave plate OUT (persistence proof), wait for dirty->save.
--b7-post: after reset: C_RD 10.0 (setpoint persisted); low setpoint 0.05
still must NOT trip (plate OUT persisted); YK12 ON -> YX16
trips (linkage after reboot); restore 10.0, leave plate IN.
--b7-mark / --b7-check [--mark 7.5]: write a distinctive NON-default
setpoint (save), reset, read it back (literal "setpoint
hold" proof, distinguishable from factory default), restore.
"""
import argparse
import collections
@@ -263,6 +276,145 @@ def actcon_of(type_id, want_pn=None):
and (want_pn is None or e[1][2] == want_pn))
def spont(ioa, val):
return lambda e: (e[0] == "yx" and e[1][0] == ioa
and e[1][1] == val and e[1][3] == COT_SPONT)
def gi_baseline(m, ck, tag):
"""General interrogation; wait for actterm (baseline YX state loaded)."""
m.send_i(asdu(T_C_IC_NA_1, 1, COT_ACT, 0, ioa=0, payload=bytes([RQT_GROUP1])),
"I C_IC act (%s)" % tag)
ck.add("gi_actterm_%s" % tag, m.wait_event(
lambda e: (e[0] == "asdu" and e[1][0] == T_C_IC_NA_1
and e[1][1] == COT_ACTTERM), 15.0) is not None)
def se_write(m, ck, addr, val, tag):
"""Two-step C_SE_NC_1 write (select then execute), both must actcon."""
pl = struct.pack("<f", val)
m.send_i(asdu(T_C_SE_NC_1, 1, COT_ACT, 0, ioa=addr,
payload=pl + bytes([0x80])),
"I C_SE select 0x%X = %.3f (%s)" % (addr, val, tag))
ck.add("se_sel_%s" % tag, m.wait_event(actcon_of(T_C_SE_NC_1, False),
5.0) is not None)
m.send_i(asdu(T_C_SE_NC_1, 1, COT_ACT, 0, ioa=addr,
payload=pl + bytes([0x00])),
"I C_SE exec 0x%X = %.3f (%s)" % (addr, val, tag))
ck.add("se_exe_%s" % tag, m.wait_event(actcon_of(T_C_SE_NC_1, False),
5.0) is not None)
def rd_set(m, ck, addr, want, name):
m.send_i(asdu(T_C_RD_NA_1, 1, COT_REQ, 0, ioa=addr), "I C_RD 0x%X" % addr)
e = m.wait_event(lambda ev: (ev[0] == "yc" and ev[1][0] == addr
and ev[1][3] == COT_REQ), 5.0)
ck.add(name, e is not None and abs(e[1][1] - want) < 0.1,
"read=%.3f want=%.3f" % (e[1][1], want) if e else "no response")
def yk_two_step(m, ck, ioa, qdc_sel, qdc_exe, name):
m.send_i(asdu(T_C_DC_NA_1, 1, COT_ACT, 0, ioa=ioa,
payload=bytes([qdc_sel])),
"I C_DC 0x%X select QDC=%02X" % (ioa, qdc_sel))
ck.add(name + "_sel", m.wait_event(actcon_of(T_C_DC_NA_1, False),
5.0) is not None)
m.send_i(asdu(T_C_DC_NA_1, 1, COT_ACT, 0, ioa=ioa,
payload=bytes([qdc_exe])),
"I C_DC 0x%X execute QDC=%02X" % (ioa, qdc_exe))
ck.add(name + "_exe", m.wait_event(actcon_of(T_C_DC_NA_1, False),
5.0) is not None)
def yx16_off_confirmed(m, ck, name):
"""YX16 back to OFF: spontaneous COT=3, else second GI background scan."""
m.events.clear()
if m.wait_event(spont(16, 0), 5.0) is not None:
ck.add(name, True, "via spontaneous COT=3")
return
log("no spont OFF within 5s -> second GI background scan")
gi_baseline(m, ck, "off-scan")
ck.add(name, m.yx.get(16) == (0, False), "via GI scan, yx16=%r"
% (m.yx.get(16),))
def b7_pre(m, ck, args):
"""B7 phase 1 (before reset): plate linkage + mark setpoint for save."""
addr = int(args.addr, 0)
gi_baseline(m, ck, "pre")
v16 = m.yx.get(16)
ck.add("baseline_yx16_off", v16 is not None and v16 == (0, False),
"yx16=%r" % (v16,))
se_write(m, ck, addr, 7.5, "mark")
rd_set(m, ck, addr, 7.5, "rd_mark_7.5")
yk_two_step(m, ck, 0x600B, 0x81, 0x01, "yk12_off") # plate OUT
se_write(m, ck, addr, args.low, "low") # 0.05 A
m.events.clear()
ck.add("plate_out_no_trip", m.wait_event(spont(16, 1), 3.0) is None,
"YX16 must stay OFF while plate OUT (protection disabled)")
rd_set(m, ck, addr, args.low, "rd_low_0.05") # setpoint still writable
yk_two_step(m, ck, 0x600B, 0x82, 0x02, "yk12_on") # plate IN -> linkage
m.events.clear()
ck.add("spont_yx16_ON", m.wait_event(spont(16, 1), 5.0) is not None)
ck.add("spont_yx15_ON", m.wait_event(spont(15, 1), 3.0) is not None)
se_write(m, ck, addr, args.restore, "restore") # back to 10.0
yx16_off_confirmed(m, ck, "yx16_off_after_restore")
yk_two_step(m, ck, 0x600B, 0x81, 0x01, "yk12_off_final") # leave OUT
log("wait 2s: dirty -> SYS_EVT_PROT_SET_SAVE -> /prot_set.bin")
time.sleep(2.0)
def b7_post(m, ck, args):
"""B7 phase 2 (after reset): persistence + linkage after reboot."""
addr = int(args.addr, 0)
gi_baseline(m, ck, "post")
v16 = m.yx.get(16)
ck.add("baseline_yx16_off", v16 is not None and v16 == (0, False),
"yx16=%r (plate OUT after reboot -> protection disabled)" % (v16,))
rd_set(m, ck, addr, args.restore, "persist_rd_10.0") # setpoint persisted
se_write(m, ck, addr, args.low, "low")
m.events.clear()
ck.add("plate_out_persisted_no_trip",
m.wait_event(spont(16, 1), 3.0) is None,
"plate OUT state survived the reset")
yk_two_step(m, ck, 0x600B, 0x82, 0x02, "yk12_on") # linkage after reboot
m.events.clear()
ck.add("spont_yx16_ON_after_reboot",
m.wait_event(spont(16, 1), 5.0) is not None)
se_write(m, ck, addr, args.restore, "restore")
yx16_off_confirmed(m, ck, "yx16_off_final")
log("wait 1s: final dirty -> save, leave plate IN")
time.sleep(1.0)
def b7_mark(m, ck, args):
"""B7 distinctive-value mark: write non-default setpoint, let it save."""
addr = int(args.addr, 0)
se_write(m, ck, addr, args.mark, "distinct")
rd_set(m, ck, addr, args.mark, "rd_distinct")
log("wait 1.5s: dirty -> save (/prot_set.bin)")
time.sleep(1.5)
def b7_check(m, ck, args):
"""B7 distinctive-value check after reset: must read back non-default,
then restore default so the bench is left in a clean state."""
addr = int(args.addr, 0)
rd_set(m, ck, addr, args.mark, "persist_distinct_after_reset")
se_write(m, ck, addr, args.restore, "restore")
log("wait 1s: restore saved, bench left at defaults")
time.sleep(1.0)
def main():
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--ip", default="192.168.1.100")
@@ -273,12 +425,24 @@ def main():
help="temporary low OC1 setting (A) to force pickup")
ap.add_argument("--restore", type=float, default=10.0,
help="OC1 default setting to restore (A)")
ap.add_argument("--b7-pre", action="store_true",
help="B7 phase 1 (before reset) instead of B6 flow")
ap.add_argument("--b7-post", action="store_true",
help="B7 phase 2 (after reset) instead of B6 flow")
ap.add_argument("--b7-mark", action="store_true",
help="B7 write distinctive non-default setpoint (pre-reset)")
ap.add_argument("--b7-check", action="store_true",
help="B7 verify distinctive setpoint survived reset, restore")
ap.add_argument("--mark", type=float, default=7.5,
help="distinctive setpoint value (A), must be non-default")
args = ap.parse_args()
addr = int(args.addr, 0)
b7_mode = ("pre" if args.b7_pre else "post" if args.b7_post else
"mark" if args.b7_mark else "check" if args.b7_check else None)
ck = Checks()
log("iec104 master client -> %s:%d (COV trigger: write 0x%X %.2f then "
"restore %.2f)" % (args.ip, args.port, addr, args.low, args.restore))
log("iec104 master client -> %s:%d mode=%s"
% (args.ip, args.port, b7_mode or "b6"))
try:
m = Master(args.ip, args.port)
@@ -293,6 +457,22 @@ def main():
ck.add("startdt_con", m.wait_event(
lambda e: e[0] == "startdt_con", 5.0) is not None)
if b7_mode is not None:
if b7_mode == "pre":
b7_pre(m, ck, args)
elif b7_mode == "post":
b7_post(m, ck, args)
elif b7_mode == "mark":
b7_mark(m, ck, args)
else:
b7_check(m, ck, args)
m.close()
rc = 0 if all(ok for _, ok, _ in ck.items) else 1
n_ok = sum(1 for _, ok, _ in ck.items)
log("SUMMARY: %d/%d checks PASS -> %s"
% (n_ok, len(ck.items), "PASS" if rc == 0 else "FAIL"))
return rc
# 2. first general interrogation
m.send_i(asdu(T_C_IC_NA_1, 1, COT_ACT, 0, ioa=0,
payload=bytes([RQT_GROUP1])),