zjk
90caacf9a2
- CanBms: bcuCtrlCanId(0x10XX81FF) + buildBcuRelayCtrlData(02 00 总正 总负 FF*4) - SystemData: BcuCtrlCmd 下行队列(Web线程入队->Iterate/MOOS线程出队Notify) + 下发统计 - CCU: /api/bcu_ctrl(addr/action 或 pos/neg, 校验地址01h~36h) + Iterate 经 MOOS CAN_TX_0x* 下发 - pCanBridge 变双向透传: 订阅 CAN_TX_0x*，CanEndpoint::sendFrame(13字节CANET帧,TCP写,互斥保护) - 网页: 每BCU节点断路器闭合/断开按钮(confirm确认) + 下发统计卡片 - 测试: pccuTest 控制帧编码用例(183项)；集成测试新增假CANET服务器验证完整下行链路(闭合/断开/非法地址)
489 lines
20 KiB
Python
489 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
"""pCCU 集成测试：验证真实 UDP 链路上的完整数据流。
|
||
|
||
拓扑：
|
||
测试脚本(FC模拟) --16000--> pCCU(fc_local) [FC状态 0x0002]
|
||
测试脚本(PM模拟) --16002--> pCCU(pm_local) [PM操控 0x0001, PM参数设定 0x0002]
|
||
pCCU --16001--> 测试脚本 [FC控制 0x0001 转发]
|
||
pCCU --16003--> 测试脚本 [PM状态 0x0004, PM参数反馈 0x0003]
|
||
测试脚本(真实CCU模拟) --16004--> pCCU(rcu_local) [配电反馈 0x0005~0x0008]
|
||
pCCU --16003--> 测试脚本 [配电反馈原帧转发 pPowerManger]
|
||
测试脚本(PM模拟) --16002--> pCCU(pm_local) [配电指令 0x0001~0x0004 短帧]
|
||
pCCU --16005--> 测试脚本 [配电指令原帧转发真实CCU]
|
||
|
||
锂电池已切换为 CAN 总线（BMS 协议，经 pCanBridge 透传 CAN_0x* 消息），
|
||
不在本脚本 UDP 拓扑内：BMS 解码由 pccuTest 单元测试覆盖
|
||
（testCanBmsDecode，向量取自实车抓包）。
|
||
|
||
校验：
|
||
1. FC 状态收到并整合，pCCU 周期发送 PM 状态(0x0004, 248B)到 16003
|
||
2. PM 操控(0x0001)转发为 FC 控制(0x0001, 24B)到 16001
|
||
3. PM 参数设定(0x0002)触发参数反馈(0x0003, 12B)到 16003
|
||
4. 配电反馈(0x0005~0x0008)原帧转发到 16003（字节一致）
|
||
5. 配电指令(0x0001~0x0004 短帧)原帧转发到 16005（字节一致）；
|
||
PM 操控指令(45B, 同id)不转发到真实CCU（按帧长区分）
|
||
6. SQLite 记录了收/发原始帧（含 link='rcu'）
|
||
7. Web 页面可访问
|
||
"""
|
||
import socket
|
||
import struct
|
||
import threading
|
||
import time
|
||
import subprocess
|
||
import os
|
||
import signal
|
||
import sys
|
||
import json
|
||
import urllib.request
|
||
|
||
# ---- 配置 ----
|
||
FC_STATUS_PORT = 16000
|
||
FC_CTRL_RECV_PORT = 16001
|
||
PM_CTRL_PORT = 16002
|
||
PM_STATUS_RECV_PORT = 16003
|
||
RCU_FB_PORT = 16004 # pCCU rcu 链路本地端口（收配电反馈）
|
||
RCU_CMD_RECV_PORT = 16005 # 模拟真实CCU 接收转发的配电指令
|
||
CANET_PORT = 16006 # 模拟 USBCAN-8E-U TCP Server（捕获下行 CAN 帧）
|
||
WEB_PORT = 18081
|
||
DB_PATH = "/tmp/pccu_it_test.db"
|
||
MOOSDB_EXE = "/usr/local/bin/MOOSDB"
|
||
PCCU_EXE = "/home/zjk/project/H100PowerManger/bin/pCCU"
|
||
CANBRIDGE_EXE = "/home/zjk/project/H100PowerManger/bin/pCanBridge"
|
||
MISSION = os.path.join(os.path.dirname(os.path.abspath(__file__)), "pccu_it.moos")
|
||
|
||
# 帧头
|
||
START = bytes([0x40, 0x40])
|
||
|
||
|
||
def frame(msg_id, payload, checksum=True):
|
||
total = 6 + len(payload) + (4 if checksum else 0)
|
||
hdr = START + struct.pack("<HH", msg_id, total)
|
||
body = hdr + payload
|
||
if checksum:
|
||
cks = sum(body) & 0xFFFFFFFF
|
||
return body + struct.pack("<I", cks)
|
||
return body
|
||
|
||
|
||
def dis_frame(msg_id, payload):
|
||
"""配电协议帧：0x40 0x40 + id(2B) + length(2B) + payload + Sum8 校验"""
|
||
total = 6 + len(payload) + 1
|
||
body = START + struct.pack("<HH", msg_id, total) + payload
|
||
return body + bytes([sum(body) & 0xFF])
|
||
|
||
|
||
def fc_status_frame():
|
||
# payload 144B，仅填关键字段，其余为0
|
||
p = bytearray(144)
|
||
p[0] = 2 # 运行模式 自动
|
||
p[1] = 6 # 运行状态 运行
|
||
struct.pack_into("<H", p, 2, 0x0102) # 一级故障码
|
||
struct.pack_into("<H", p, 10, 1234) # 累积发电时间
|
||
p[12] = 0 # 故障等级
|
||
struct.pack_into("<H", p, 14, 2000) # 系统发电功率 0.01kW
|
||
p[16] = 80 # 储氢
|
||
p[17] = 70 # 液氧
|
||
struct.pack_into("<H", p, 24, 350) # 钯膜温度
|
||
struct.pack_into("<H", p, 68, 2500) # 舱室温度1
|
||
struct.pack_into("<H", p, 140, 4321) # 剩余发电量 MWh (字节147,148)
|
||
p[142] = 5 # 心跳
|
||
p[143] = 0x01 # 应急指令
|
||
return frame(0x0002, bytes(p), checksum=False)
|
||
|
||
|
||
def pm_control_frame():
|
||
p = bytearray(35)
|
||
struct.pack_into("<H", p, 0, 2026)
|
||
p[2], p[3], p[4], p[5], p[6], p[7] = 8, 26, 10, 30, 0, 0
|
||
p[8] = 2 # 模式 自动
|
||
p[9] = 2 # 操控 启动
|
||
p[10] = 120 # 输出功率 0.1kW
|
||
struct.pack_into("<h", p, 11, 123) # 纵倾姿态数据1 12.3°
|
||
struct.pack_into("<h", p, 13, -45) # 横倾姿态数据1 -4.5°
|
||
p[15] = 0x03 # 应急允许
|
||
struct.pack_into("<H", p, 16, 150) # 潜深
|
||
p[21] = 0x10 # 仪表锂电启动
|
||
p[22] = 0x20 # 动力锂电关闭
|
||
struct.pack_into("<H", p, 23, 300) # 动力电池功率
|
||
p[25] = 7 # 心跳
|
||
p[26] = 0xAA # 主机状态
|
||
return frame(0x0001, bytes(p))
|
||
|
||
|
||
def pm_paramset_frame():
|
||
p = bytearray(18)
|
||
p[0], p[1], p[2] = 75, 85, 93
|
||
p[5], p[6], p[7] = 75, 85, 93
|
||
return frame(0x0002, bytes(p))
|
||
|
||
|
||
#------------------ 配电协议帧（真实CCU桥接） ------------------
|
||
|
||
def dis_hv_fb_frame():
|
||
"""0x0005 高压母线配电反馈（disHighVolBusState 53B：19字节+17×u16）"""
|
||
p = bytearray(53)
|
||
p[0] = 0x55 # 燃料电池断路器 闭合
|
||
p[1] = 0xAA # 动力锂电池断路器 断开
|
||
struct.pack_into("<H", p, 19, 0x0FA0) # 动力汇流排电压
|
||
struct.pack_into("<H", p, 21, 123) # DC/DC5模块电流
|
||
struct.pack_into("<H", p, 25, 0x0F90) # A汇流排电压
|
||
return dis_frame(0x0005, bytes(p))
|
||
|
||
|
||
def dis_hva_fb_frame():
|
||
"""0x0006 高压母线A配电反馈（disHighAVolBusState 25B：9字节+8×u16）"""
|
||
p = bytearray(25)
|
||
p[0] = 0x55 # 艏部FT装置1/2/3断路器 闭合
|
||
struct.pack_into("<H", p, 9, 0x0F88) # 汇流排B电压
|
||
struct.pack_into("<H", p, 11, 321) # 汇流排B电流
|
||
return dis_frame(0x0006, bytes(p))
|
||
|
||
|
||
def dis_hvb_fb_frame():
|
||
"""0x0007 低压主母线配电反馈（disLowMainBusState 34B：10字节+12×u16）"""
|
||
p = bytearray(34)
|
||
p[0] = 0x55 # 锂电池组(仪表)断路器 闭合
|
||
struct.pack_into("<H", p, 10, 0x0F70) # 仪表母排电压
|
||
struct.pack_into("<H", p, 12, 654) # 艏部低压配电箱电流
|
||
return dis_frame(0x0007, bytes(p))
|
||
|
||
|
||
def dis_lv_fb_frame():
|
||
"""0x0008 仪表汇流排配电反馈（disLowBusState 29B：11字节+9×u16）"""
|
||
p = bytearray(29)
|
||
p[0] = 0x55 # UNIT1断路器 闭合
|
||
struct.pack_into("<H", p, 11, 0x0F60) # 仪表母排电压
|
||
struct.pack_into("<H", p, 13, 111) # UNIT1电流
|
||
return dis_frame(0x0008, bytes(p))
|
||
|
||
|
||
def dis_hv_cmd_frame():
|
||
"""0x0001 高压母线配电指令（disHighVolBusCmd 14B，pPowerManger 下发）"""
|
||
p = bytearray(14)
|
||
p[0] = 0x55 # 燃料电池断路器 闭合
|
||
p[1] = 0xAA # 动力锂电池断路器 断开
|
||
p[12] = 1 # DC/DC5模块启动
|
||
return dis_frame(0x0001, bytes(p))
|
||
|
||
|
||
def dis_hva_cmd_frame():
|
||
"""0x0002 高压母线A配电指令（disHighAVolBusCmd 6B）"""
|
||
p = bytearray(6)
|
||
p[0] = 0x55
|
||
return dis_frame(0x0002, bytes(p))
|
||
|
||
|
||
def dis_hvb_cmd_frame():
|
||
"""0x0003 低压主母线配电指令（disLowMainBusCmd 7B）"""
|
||
p = bytearray(7)
|
||
p[3] = 0x55 # DC-C1直流配电板断路器
|
||
return dis_frame(0x0003, bytes(p))
|
||
|
||
|
||
def dis_lv_cmd_frame():
|
||
"""0x0004 仪表汇流排配电指令（disLowBusCmd 8B）"""
|
||
p = bytearray(8)
|
||
p[0] = 0x55 # UNIT1断路器
|
||
return dis_frame(0x0004, bytes(p))
|
||
|
||
|
||
#------------------ BCU 断路器控制下行测试 ------------------
|
||
|
||
class CanetServer(threading.Thread):
|
||
"""最小 CANET TCP Server 模拟：接受 pCanBridge 连接，捕获下行 13 字节帧"""
|
||
def __init__(self, port):
|
||
super().__init__(daemon=True)
|
||
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||
self.sock.bind(("127.0.0.1", port))
|
||
self.sock.listen(2)
|
||
self.frames = []
|
||
self.lock = threading.Lock()
|
||
self.stop_flag = False
|
||
|
||
def run(self):
|
||
self.sock.settimeout(1.0)
|
||
conn = None
|
||
while not self.stop_flag:
|
||
try:
|
||
if conn is None:
|
||
conn, _ = self.sock.accept()
|
||
conn.settimeout(0.5)
|
||
buf = b""
|
||
while not self.stop_flag:
|
||
try:
|
||
d = conn.recv(4096)
|
||
except socket.timeout:
|
||
break
|
||
if not d:
|
||
conn.close()
|
||
conn = None
|
||
break
|
||
buf += d
|
||
while len(buf) >= 13:
|
||
with self.lock:
|
||
self.frames.append(buf[:13])
|
||
buf = buf[13:]
|
||
except socket.timeout:
|
||
continue
|
||
except OSError:
|
||
break
|
||
|
||
def snapshot(self):
|
||
with self.lock:
|
||
return list(self.frames)
|
||
|
||
|
||
def bcu_ctrl_expect(addr, pos, neg):
|
||
"""期望的下行 13 字节 CANET 帧：扩展帧 DLC8 + 0x10XX81FF + 02 00 pos neg FF*4"""
|
||
return bytes([0x80 | 8,
|
||
(0x10000000 | (addr << 16) | 0x81FF) >> 24 & 0xFF,
|
||
(0x10000000 | (addr << 16) | 0x81FF) >> 16 & 0xFF,
|
||
(0x10000000 | (addr << 16) | 0x81FF) >> 8 & 0xFF,
|
||
(0x10000000 | (addr << 16) | 0x81FF) & 0xFF,
|
||
0x02, 0x00, 0x01 if pos else 0x00, 0x01 if neg else 0x00,
|
||
0xFF, 0xFF, 0xFF, 0xFF])
|
||
|
||
|
||
def api_get(path):
|
||
try:
|
||
resp = urllib.request.urlopen(f"http://127.0.0.1:{WEB_PORT}{path}", timeout=3)
|
||
return json.loads(resp.read().decode())
|
||
except Exception as e:
|
||
return {"ok": False, "error": str(e)}
|
||
|
||
|
||
def recv_loop(sock, label, results, msg_ids, timeout):
|
||
"""持续接收指定 msg_id 的帧，记录到 results"""
|
||
sock.settimeout(timeout)
|
||
end = time.time() + timeout
|
||
while time.time() < end:
|
||
try:
|
||
data, addr = sock.recvfrom(2048)
|
||
except socket.timeout:
|
||
break
|
||
except OSError:
|
||
break
|
||
if len(data) < 6:
|
||
continue
|
||
mid = struct.unpack_from("<H", data, 2)[0]
|
||
if mid in msg_ids:
|
||
results.append(data)
|
||
|
||
|
||
def main():
|
||
# 清理
|
||
for f in (DB_PATH, DB_PATH + "-wal", DB_PATH + "-shm", "/tmp/pccu_it_test.log",
|
||
"/tmp/pccu_it_can.db", "/tmp/pccu_it_can.db-wal", "/tmp/pccu_it_can.db-shm"):
|
||
if os.path.exists(f):
|
||
os.remove(f)
|
||
|
||
procs = []
|
||
canet = CanetServer(CANET_PORT)
|
||
try:
|
||
# 1. 启动 MOOSDB
|
||
moosdb = subprocess.Popen([MOOSDB_EXE, "--moos", "--port", "9000"],
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
procs.append(moosdb)
|
||
time.sleep(1.0)
|
||
|
||
# 2. 启动 pCCU + pCanBridge（pCanBridge 连接模拟 USBCAN，用于下行测试）
|
||
pccu = subprocess.Popen([PCCU_EXE, MISSION],
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
|
||
procs.append(pccu)
|
||
canet.start()
|
||
pcan = subprocess.Popen([CANBRIDGE_EXE, MISSION],
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
procs.append(pcan)
|
||
time.sleep(2.5)
|
||
if pccu.poll() is not None:
|
||
print("FAIL: pCCU exited early, code", pccu.returncode)
|
||
return 1
|
||
|
||
# 3. 启动接收监听（FC 控制转发 16001，PM 状态/配电反馈 16003，
|
||
# 转发真实CCU 的配电指令 16005）
|
||
fc_ctl_recv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
fc_ctl_recv.bind(("127.0.0.1", FC_CTRL_RECV_PORT))
|
||
pm_status_recv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
pm_status_recv.bind(("127.0.0.1", PM_STATUS_RECV_PORT))
|
||
rcu_cmd_recv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
rcu_cmd_recv.bind(("127.0.0.1", RCU_CMD_RECV_PORT))
|
||
|
||
fc_ctl_results, pm_status_results, rcu_cmd_results = [], [], []
|
||
t1 = threading.Thread(target=recv_loop,
|
||
args=(fc_ctl_recv, "fc-ctl", fc_ctl_results, {0x0001}, 14))
|
||
t2 = threading.Thread(target=recv_loop,
|
||
args=(pm_status_recv, "pm-status", pm_status_results,
|
||
{0x0003, 0x0004, 0x0005, 0x0006, 0x0007, 0x0008}, 14))
|
||
t3 = threading.Thread(target=recv_loop,
|
||
args=(rcu_cmd_recv, "rcu-cmd", rcu_cmd_results,
|
||
{0x0001, 0x0002, 0x0003, 0x0004}, 14))
|
||
t1.start(); t2.start(); t3.start()
|
||
|
||
# 4. 发送 FC 状态（周期模拟）
|
||
fc_send = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
for _ in range(3):
|
||
fc_send.sendto(fc_status_frame(), ("127.0.0.1", FC_STATUS_PORT))
|
||
time.sleep(0.5)
|
||
|
||
# 5. 发送 PM 操控指令 + 参数设定
|
||
pm_send = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
pm_send.sendto(pm_control_frame(), ("127.0.0.1", PM_CTRL_PORT))
|
||
time.sleep(0.5)
|
||
pm_send.sendto(pm_paramset_frame(), ("127.0.0.1", PM_CTRL_PORT))
|
||
|
||
# 6. 发送配电反馈（真实CCU -> pCCU -> pPowerManger，应原帧转发到 16003）
|
||
rcu_fb_frames = [dis_hv_fb_frame(), dis_hva_fb_frame(),
|
||
dis_hvb_fb_frame(), dis_lv_fb_frame()]
|
||
for f in rcu_fb_frames:
|
||
pm_send.sendto(f, ("127.0.0.1", RCU_FB_PORT))
|
||
time.sleep(0.2)
|
||
|
||
# 7. 发送配电指令（pPowerManger -> pCCU -> 真实CCU，应原帧转发到 16005）
|
||
dis_cmd_frames = [dis_hv_cmd_frame(), dis_hva_cmd_frame(),
|
||
dis_hvb_cmd_frame(), dis_lv_cmd_frame()]
|
||
for f in dis_cmd_frames:
|
||
pm_send.sendto(f, ("127.0.0.1", PM_CTRL_PORT))
|
||
time.sleep(0.2)
|
||
|
||
t1.join(); t2.join(); t3.join()
|
||
|
||
ok = True
|
||
|
||
# 校验 1：PM 状态报文（248B）周期发送
|
||
pm_status = [d for d in pm_status_results if len(d) == 248]
|
||
print(f"[PM 状态报文] 收到 {len(pm_status)} 帧 (期望>=1, 每帧248B)")
|
||
if len(pm_status) >= 1:
|
||
d = pm_status[0]
|
||
total = struct.unpack_from("<H", d, 4)[0]
|
||
mode = d[6]; status = d[7]
|
||
print(f" -> length={total}, mode={mode}, status={status}, size={len(d)}")
|
||
ok = ok and (total == 248) and (mode == 2) and (status == 6)
|
||
else:
|
||
ok = False
|
||
|
||
# 校验 2：PM 操控转发为 FC 控制（24B，0827协议：姿态数据1转发，数据2字节预留恒0）
|
||
fc_ctl = [d for d in fc_ctl_results if len(d) == 24]
|
||
print(f"[FC 控制转发] 收到 {len(fc_ctl)} 帧 (期望>=1, 每帧24B)")
|
||
if fc_ctl:
|
||
d = fc_ctl[0]
|
||
cmd = d[7]; power = d[8]; hb = d[23]
|
||
pitch = struct.unpack_from("<h", d, 9)[0] # 字节10,11
|
||
roll = struct.unpack_from("<h", d, 13)[0] # 字节14,15
|
||
resv1 = struct.unpack_from("<H", d, 11)[0] # 字节12,13 预留
|
||
resv2 = struct.unpack_from("<H", d, 15)[0] # 字节16,17 预留
|
||
print(f" -> cmd={cmd}, outputPower={power}, pitch={pitch}, roll={roll}, "
|
||
f"resv1={resv1}, resv2={resv2}, heartbeat={hb}")
|
||
ok = ok and (cmd == 2) and (power == 120) and (hb == 7)
|
||
ok = ok and (pitch == 123) and (roll == -45)
|
||
ok = ok and (resv1 == 0) and (resv2 == 0)
|
||
else:
|
||
ok = False
|
||
|
||
# 校验 3：PM 参数反馈（12B）
|
||
fb = [d for d in pm_status_results if len(d) == 12]
|
||
print(f"[PM 参数反馈] 收到 {len(fb)} 帧 (期望>=1, 每帧12B)")
|
||
if fb:
|
||
flag = fb[0][6]
|
||
print(f" -> flag=0x{flag:02X}")
|
||
ok = ok and (flag == 0x10)
|
||
else:
|
||
ok = False
|
||
|
||
# 校验 4：配电反馈原帧转发到 pPowerManger（字节一致）
|
||
for i, f in enumerate(rcu_fb_frames):
|
||
got = [d for d in pm_status_results if d == f]
|
||
mid = struct.unpack_from("<H", f, 2)[0]
|
||
print(f"[配电反馈 0x{mid:04X} 转发] {'OK' if got else 'MISSING'} (字节一致)")
|
||
ok = ok and bool(got)
|
||
|
||
# 校验 5：配电指令原帧转发到真实CCU（字节一致）
|
||
for f in dis_cmd_frames:
|
||
got = [d for d in rcu_cmd_results if d == f]
|
||
mid = struct.unpack_from("<H", f, 2)[0]
|
||
print(f"[配电指令 0x{mid:04X} 转发] {'OK' if got else 'MISSING'} (字节一致)")
|
||
ok = ok and bool(got)
|
||
|
||
# 校验 6：PM 操控指令(0x0001, 45B)不得转发到真实CCU（按帧长区分）
|
||
pm_ctl_leak = [d for d in rcu_cmd_results if len(d) == 45]
|
||
print(f"[PM操控指令不桥接] 收到 {len(pm_ctl_leak)} 帧 (期望 0)")
|
||
ok = ok and not pm_ctl_leak
|
||
|
||
# 校验 7：数据库有记录（FC/PM 链路收发 + rcu 链路桥接）
|
||
time.sleep(1.0)
|
||
db_count, db_rcu = -1, -1
|
||
if os.path.exists(DB_PATH):
|
||
import sqlite3
|
||
conn = sqlite3.connect(DB_PATH)
|
||
db_count = conn.execute("SELECT COUNT(*) FROM comm_log").fetchone()[0]
|
||
db_rcu = conn.execute(
|
||
"SELECT COUNT(*) FROM comm_log WHERE link='rcu'").fetchone()[0]
|
||
conn.close()
|
||
print(f"[数据库] comm_log 记录数 = {db_count} (期望>=6), "
|
||
f"rcu 链路记录数 = {db_rcu} (期望>=8)")
|
||
ok = ok and (db_count >= 6) # 3条FC状态 + 1条PM操控 + 1条PM参数 + 周期PM状态(至少1)
|
||
ok = ok and (db_rcu >= 8) # 4条配电反馈(收) + 4条配电指令(发)
|
||
|
||
# 校验 8：BCU 断路器控制下行（网页API -> pCCU -> MOOS -> pCanBridge -> 模拟USBCAN）
|
||
# 闭合（addr=5, 总正+总负=1）
|
||
j = api_get("/api/bcu_ctrl?addr=5&pos=1&neg=1")
|
||
print(f"[断路器闭合 API] {j}")
|
||
ok = ok and j.get("ok") is True
|
||
# action 语法（addr=6, action=1 等价 pos=neg=1）
|
||
j = api_get("/api/bcu_ctrl?addr=6&action=1")
|
||
print(f"[断路器闭合 API(action)] {j}")
|
||
ok = ok and j.get("ok") is True
|
||
# 非法地址
|
||
j = api_get("/api/bcu_ctrl?addr=99&action=1")
|
||
print(f"[非法地址 API] {j}")
|
||
ok = ok and j.get("ok") is False
|
||
# 断开（addr=5, 总正+总负=0）
|
||
j = api_get("/api/bcu_ctrl?addr=5&pos=0&neg=0")
|
||
print(f"[断路器断开 API] {j}")
|
||
ok = ok and j.get("ok") is True
|
||
time.sleep(3.0)
|
||
frames = canet.snapshot()
|
||
print(f"[下行帧捕获] {len(frames)} 帧")
|
||
exp_close5 = bcu_ctrl_expect(5, 1, 1)
|
||
exp_close6 = bcu_ctrl_expect(6, 1, 1)
|
||
exp_open5 = bcu_ctrl_expect(5, 0, 0)
|
||
hit5 = any(f == exp_close5 for f in frames)
|
||
hit6 = any(f == exp_close6 for f in frames)
|
||
hit5o = any(f == exp_open5 for f in frames)
|
||
leak99 = any(f[1] == 0x63 and f[2] == 0x00 for f in frames) # addr=99 未下发
|
||
print(f"[断路器闭合下行(addr=5)] {'OK' if hit5 else 'MISSING'}")
|
||
print(f"[断路器闭合下行(addr=6)] {'OK' if hit6 else 'MISSING'}")
|
||
print(f"[断路器断开下行(addr=5)] {'OK' if hit5o else 'MISSING'}")
|
||
print(f"[非法地址未下发(addr=99)] {'OK' if not leak99 else 'LEAK'}")
|
||
ok = ok and hit5 and hit6 and hit5o and not leak99
|
||
|
||
web_ok = False
|
||
try:
|
||
resp = urllib.request.urlopen(f"http://127.0.0.1:{WEB_PORT}/", timeout=3)
|
||
html = resp.read().decode()
|
||
web_ok = ("pCCU" in html)
|
||
print(f"[Web] 页面可访问, 长度={len(html)}")
|
||
except Exception as e:
|
||
print(f"[Web] 访问失败: {e}")
|
||
ok = ok and web_ok
|
||
|
||
print("\n==== 集成测试 " + ("PASSED" if ok else "FAILED") + " ====")
|
||
return 0 if ok else 1
|
||
|
||
finally:
|
||
canet.stop_flag = True
|
||
for p in reversed(procs):
|
||
try:
|
||
p.terminate()
|
||
except Exception:
|
||
pass
|
||
time.sleep(0.5)
|
||
for p in procs:
|
||
try:
|
||
p.kill()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main()) |