import socket import struct import threading import time from udpMsg import ( msg_disHighVolBusFbMsg, msg_disHighAVolBusFbMsg, msg_disLowMainBusFbMsg, msg_disLowBusFbMsg, msg_disHVBCmdMsg, msg_disHVBACmdMsg, msg_disLVMBCmdMsg, msg_disLVBCmdMsg ) class PowerManagerSimulator: def __init__(self, host='127.0.0.1', port=5001): self.host = host self.port = port self.running = False self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.bind((host, port)) # 初始化消息对象 self.high_vol_fb = msg_disHighVolBusFbMsg() self.high_a_vol_fb = msg_disHighAVolBusFbMsg() self.low_main_fb = msg_disLowMainBusFbMsg() self.low_bus_fb = msg_disLowBusFbMsg() # 模拟数据初始化 self._init_sim_data() def _init_sim_data(self): """初始化模拟数据""" # 高压母线反馈数据 - 测试所有成员 self.high_vol_fb.fuelCellCircuitBreaker = 0x55 self.high_vol_fb.powerLithiumBatteryCircuitBreaker = 0x55 self.high_vol_fb.propulsionMotorCircuitBreaker = 0x55 self.high_vol_fb.lithiumBatteryGroupInstrumentCircuitBreaker = 0x55 self.high_vol_fb.dcDc5ModuleCircuitBreaker = 0x55 self.high_vol_fb.bowHighVoltageDistributionBoxCircuitBreaker = 0x55 self.high_vol_fb.sternFTDevice45CircuitBreaker = 0x55 self.high_vol_fb.sternRudderSwitch1CircuitBreaker = 0x55 self.high_vol_fb.sternRudderSwitch2CircuitBreaker = 0x55 self.high_vol_fb.fbReservedCircuitBreaker = 0x55 self.high_vol_fb.tyzReservedCircuitBreaker = 0x55 self.high_vol_fb.reservedCircuitBreaker = 0x55 self.high_vol_fb.dcDcFaultWord1 = 0xFF # 测试故障模式 self.high_vol_fb.dcDcFaultWord2 = 0xAA self.high_vol_fb.powerBusInsulationStatus = 0x55 self.high_vol_fb.aBusInsulationStatus = 0x55 self.high_vol_fb.meterPowerLossSignal = 0 self.high_vol_fb.emergencyPowerLossSignal = 0 self.high_vol_fb.dcdcTemperature = 0x08 # 正常值 self.high_vol_fb.powerBusVoltage = 600 # 正常值 self.high_vol_fb.dcDc5ModuleCurrent = 50 self.high_vol_fb.powerBusCurrent = 100 self.high_vol_fb.busbarAVoltage = 600 self.high_vol_fb.propulsionMotorControlBoxCurrent = 80 self.high_vol_fb.busbarACurrent = 50 self.high_vol_fb.lithiumBatteryGroupMeterCurrent = 30 self.high_vol_fb.bowHighVoltageDistributionBoxCurrent = 20 self.high_vol_fb.sternFTDevice45Current = 15 self.high_vol_fb.tyzReservedSwitchCurrent = 0 self.high_vol_fb.sternRudderSwitch1Current = 10 self.high_vol_fb.reservedCurrent1 = 0 self.high_vol_fb.sternRudderSwitch2Current = 10 self.high_vol_fb.reservedCurrent2 = 0 self.high_vol_fb.fbReservedSwitchCurrent = 0 self.high_vol_fb.reservedCurrent3 = 0 self.high_vol_fb.coolWaterPressure = 100 # 高压A母线反馈数据 - 测试所有成员 self.high_a_vol_fb.bowFTDevice123CircuitBreaker = 0x55 self.high_a_vol_fb.actuatorCircuitBreaker = 0x55 self.high_a_vol_fb.mastSteeringGearControlBoxCircuitBreaker = 0x55 self.high_a_vol_fb.xczCircuitBreaker = 0x55 self.high_a_vol_fb.bowRudderControlBoxCircuitBreaker = 0x55 self.high_a_vol_fb.openWaterCoverStartCylinderCircuitBreaker = 0x55 self.high_a_vol_fb.instrumentPowerFailureSignal = 0x55 self.high_a_vol_fb.emergencyPowerFailureSignal = 0x55 self.high_a_vol_fb.waterIngressionAlarm = 0x55 self.high_a_vol_fb.busbarBVoltage = 600 self.high_a_vol_fb.busbarBCurrent = 30 self.high_a_vol_fb.bowFTDevice123Current = 20 self.high_a_vol_fb.actuatorCurrent = 15 self.high_a_vol_fb.mastSteeringGearControlBoxCurrent = 10 self.high_a_vol_fb.xczCurrent = 5 self.high_a_vol_fb.bowRudderControlBoxCurrent = 8 self.high_a_vol_fb.openWaterCoverStartCylinderCurrent = 0 # 仪表主母线反馈数据 - 测试所有成员 self.low_main_fb.lithiumBatteryGroupInstrumentCircuitBreaker = 0x55 self.low_main_fb.bowLowVoltageDistributionBoxCircuitBreaker = 0x55 self.low_main_fb.unit4InstrumentDC48VCircuitBreaker = 0x55 self.low_main_fb.dcC1DCDistributionPanelCircuitBreaker = 0x55 self.low_main_fb.reservedCircuitBreaker1 = 0x55 self.low_main_fb.reservedCircuitBreaker2 = 0x55 self.low_main_fb.emergencyLithiumBatteryGroup2CircuitBreaker = 0x55 self.low_main_fb.instrumentPowerFailureSignal = 0x55 self.low_main_fb.emergencyPowerFailureSignal = 0x55 self.low_main_fb.instrumentBusbarInsulationLow = 0x55 self.low_main_fb.instrumentBusbarVoltage = 48 self.low_main_fb.bowLowVoltageDistributionBoxCurrent = 10 self.low_main_fb.dcC1InstrumentDC48VCurrent = 5 self.low_main_fb.reservedCurrent1 = 0 self.low_main_fb.emergencyLithiumBatteryGroup2Current = 0 self.low_main_fb.lithiumBatteryGroupInstrumentCurrent = 20 self.low_main_fb.unit4InstrumentDC48VCurrent = 5 self.low_main_fb.reservedCurrent2 = 0 self.low_main_fb.emergency2BusbarVoltage = 48 self.low_main_fb.compositeEnergyManagementSystemEmergencyDC48VCurrent = 2 self.low_main_fb.fuelCellSecuritySystemEmergencyDC48VCurrent = 2 self.low_main_fb.plcControlPowerCurrent = 1 # 仪表汇流排反馈数据 - 测试所有成员 self.low_bus_fb.unit1CircuitBreaker = 1 self.low_bus_fb.unit2CircuitBreaker = 1 self.low_bus_fb.unit3CircuitBreaker = 1 self.low_bus_fb.unit5CircuitBreaker = 1 self.low_bus_fb.bowPZDeviceCircuitBreaker = 1 self.low_bus_fb.reservedCircuitBreaker1 = 0 self.low_bus_fb.reservedCircuitBreaker2 = 0 self.low_bus_fb.emergencyLithiumBatteryGroup1CircuitBreaker = 0 self.low_bus_fb.powerFailureSignal = 0 self.low_bus_fb.waterIngressionAlarm = 0 self.low_bus_fb.instrumentBusbarVoltage = 48 self.low_bus_fb.instrumentBusbarCurrent = 500 self.low_bus_fb.unit1Current = 100 self.low_bus_fb.unit2Current = 100 self.low_bus_fb.unit3Current = 100 self.low_bus_fb.unit5Current = 100 self.low_bus_fb.bowPZDeviceCurrent = 50 self.low_bus_fb.emergencyLithiumBatteryGroup1Current = 0 self.low_bus_fb.emergency1BusbarVoltage = 48 def start(self): """启动模拟器""" self.running = True # 启动发送线程 self.send_thread = threading.Thread(target=self._send_loop) self.send_thread.daemon = True self.send_thread.start() # 启动接收线程 self.recv_thread = threading.Thread(target=self._recv_loop) self.recv_thread.daemon = True self.recv_thread.start() def stop(self): """停止模拟器""" self.running = False self.sock.close() def _send_loop(self): """周期性发送反馈消息""" while self.running: try: # 发送高压母线反馈 data = self.high_vol_fb.pack() self.sock.sendto(data, (self.host, 7000)) # 发送高压A母线反馈 data = self.high_a_vol_fb.pack() self.sock.sendto(data, (self.host, 7000)) # 发送仪表主母线反馈 data = self.low_main_fb.pack() self.sock.sendto(data, (self.host, 7000)) # 发送仪表汇流排反馈 data = self.low_bus_fb.pack() self.sock.sendto(data, (self.host, 7000)) time.sleep(1) # 每秒发送一次 except Exception as e: print(f"发送消息出错: {e}") break def _recv_loop(self): """接收并处理命令消息""" while self.running: try: data, addr = self.sock.recvfrom(1024) if len(data) < 6: # 最小消息长度检查 continue msg_id = struct.unpack_from('