#!/usr/bin/env python3 """pCCU 演示模拟器：模拟燃料电池(FC) 与 控制主机(PM) 两端。 - 每 1s 向 pCCU 的 fc_local 端口发送 FC 状态帧(0x0002, 150B)，数据带波动 - 每 5s 向 pCCU 的 pm_local 端口发送 PM 操控指令(0x0001) 与参数设定(0x0002) - 监听 pCCU 的 fc_remote / pm_remote 端口，打印 pCCU 转发的 FC 控制与 PM 状态反馈 用法: python3 pccu_sim.py [fc_status_port] [pm_cmd_port] [fc_ctl_listen] [pm_status_listen] 默认端口与 test/pccu/pccu_it.moos 一致。 """ import socket import struct import threading import time import math import sys # 默认端口（与 pccu_it.moos 一致） FC_STATUS_PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 16000 PM_CMD_PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 16002 FC_CTL_LISTEN = int(sys.argv[3]) if len(sys.argv) > 3 else 16001 PM_STAT_LISTEN = int(sys.argv[4]) if len(sys.argv) > 4 else 16003 START = bytes([0x40, 0x40]) def frame(msg_id, payload, checksum=True): total = 6 + len(payload) + (4 if checksum else 0) body = START + struct.pack(" 启动(5) -> 运行(6) self.status_timer += dt if self.fc_status == 4 and self.status_timer > 3: self.fc_status = 5; self.status_timer = 0 elif self.fc_status == 5 and self.status_timer > 3: self.fc_status = 6; self.status_timer = 0 return self.build() class PmSim: """控制主机：周期发送操控指令，监听 pCCU 状态反馈""" def __init__(self, cmd_port, stat_listen): self.cmd_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.cmd_port = cmd_port self.stat_listen = stat_listen self.power = 100 self.heartbeat = 0 def control_frame(self): p = bytearray(35) now = time.localtime() struct.pack_into("120->140 循环 def listen_loop(port, label, msg_ids, name_of): """监听 pCCU 输出端口，打印收到的帧关键字段""" sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: sock.bind(("127.0.0.1", port)) except OSError as e: print(f"[{label}] 监听端口 {port} 失败: {e}") return sock.settimeout(1.0) print(f"[{label}] 监听 {label} 输出端口 {port}") while True: try: data, addr = sock.recvfrom(2048) except socket.timeout: continue except OSError: break if len(data) < 6: continue mid = struct.unpack_from("= 20: # FC 控制转发 cmd = data[7]; power = data[8]; hb = data[19] print(f"[{label}] ← FC控制转发 0x0001: cmd={cmd} power={power} 心跳={hb} ({len(data)}B)") elif mid == 0x0004 and len(data) >= 248: # PM 状态报文 mode = data[6]; status = data[7]; hb = data[242] genp = struct.unpack_from("= 12: # 参数反馈 flag = data[6] print(f"[{label}] ← 参数反馈 0x0003: flag=0x{flag:02X} ({len(data)}B)") def main(): print("=" * 64) print(f"pCCU 演示模拟器启动") print(f" FC 状态 -> pCCU:{FC_STATUS_PORT} (0x0002 150B)") print(f" PM 指令 -> pCCU:{PM_CMD_PORT} (0x0001 45B / 0x0002 28B)") print(f" 监听 FC 控制转发 于 {FC_CTL_LISTEN}") print(f" 监听 PM 状态报文 于 {PM_STAT_LISTEN}") print("=" * 64) fc = FcSim(FC_STATUS_PORT) pm = PmSim(PM_CMD_PORT, PM_STAT_LISTEN) threading.Thread(target=listen_loop, args=(FC_CTL_LISTEN, "FC", {0x0001}, None), daemon=True).start() threading.Thread(target=listen_loop, args=(PM_STAT_LISTEN, "PM", {0x0004, 0x0003}, None), daemon=True).start() last_cmd_t = 0 last_print_t = 0 try: while True: # FC 状态 1Hz f = fc.tick(1.0) fc.sock.sendto(f, ("127.0.0.1", FC_STATUS_PORT)) # PM 指令每 5s if time.time() - last_cmd_t >= 5.0: pm.send_cmd() pm.cmd_sock.sendto(pm.param_frame(), ("127.0.0.1", PM_CMD_PORT)) print(f"[PM] → 发送操控指令 power={pm.power}, 参数设定") last_cmd_t = time.time() # 每秒打印一行 FC 发送摘要 if time.time() - last_print_t >= 5.0: last_print_t = time.time() print(f"[FC] → 状态: status={fc.fc_status} 心跳={fc.heartbeat} " f"储氢={fc.hydrogen:.1f}% 液氧={fc.lo2:.1f}% 发电时间={fc.gen_time:.1f}h") time.sleep(1.0) except KeyboardInterrupt: print("\n模拟器退出") if __name__ == "__main__": main()