232 lines
10 KiB
Python
232 lines
10 KiB
Python
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=6000):
|
|
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 = 100
|
|
self.high_vol_fb.sternRudderSwitch2Current = 10
|
|
self.high_vol_fb.reservedCurrent2 = 100
|
|
self.high_vol_fb.fbReservedSwitchCurrent = 1000
|
|
self.high_vol_fb.reservedCurrent3 = 100
|
|
self.high_vol_fb.coolWaterPressure = 100
|
|
|
|
# 高压A母线反馈数据 - 测试所有成员
|
|
self.high_a_vol_fb.bowFTDevice123CircuitBreaker = 0x5A
|
|
self.high_a_vol_fb.actuatorCircuitBreaker = 0x5A
|
|
self.high_a_vol_fb.mastSteeringGearControlBoxCircuitBreaker = 0x5A
|
|
self.high_a_vol_fb.xczCircuitBreaker = 0x5A
|
|
self.high_a_vol_fb.bowRudderControlBoxCircuitBreaker = 0x5A
|
|
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 = 0x5A
|
|
self.low_main_fb.bowLowVoltageDistributionBoxCircuitBreaker = 0x55
|
|
self.low_main_fb.unit4InstrumentDC48VCircuitBreaker = 0x55
|
|
self.low_main_fb.dcC1DCDistributionPanelCircuitBreaker = 0x5A
|
|
self.low_main_fb.reservedCircuitBreaker1 = 0x55
|
|
self.low_main_fb.reservedCircuitBreaker2 = 0x55
|
|
self.low_main_fb.emergencyLithiumBatteryGroup2CircuitBreaker = 0x5A
|
|
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 = 110
|
|
self.low_main_fb.emergencyLithiumBatteryGroup2Current = 100
|
|
self.low_main_fb.lithiumBatteryGroupInstrumentCurrent = 20
|
|
self.low_main_fb.unit4InstrumentDC48VCurrent = 5
|
|
self.low_main_fb.reservedCurrent2 = 100
|
|
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 = 0x55
|
|
self.low_bus_fb.unit2CircuitBreaker = 0x55
|
|
self.low_bus_fb.unit3CircuitBreaker = 0x55
|
|
self.low_bus_fb.unit5CircuitBreaker = 0x55
|
|
self.low_bus_fb.bowPZDeviceCircuitBreaker = 0x55
|
|
self.low_bus_fb.reservedCircuitBreaker1 = 0x55
|
|
self.low_bus_fb.reservedCircuitBreaker2 = 0x55
|
|
self.low_bus_fb.emergencyLithiumBatteryGroup1CircuitBreaker = 0x55
|
|
self.low_bus_fb.instrumentPowerFailureSignal = 0x5A
|
|
self.low_bus_fb.emergencyPowerFailureSignal = 0x5A
|
|
self.low_bus_fb.waterIngressionAlarm = 0x55
|
|
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 = 100
|
|
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('<H', data, 2)[0]
|
|
|
|
# 根据消息ID处理不同类型的命令
|
|
if msg_id == 0x0001: # 动力母线配电指令
|
|
cmd = msg_disHVBCmdMsg()
|
|
if cmd.unpack(data):
|
|
print("收到动力母线配电指令")
|
|
# 这里可以添加命令处理逻辑
|
|
|
|
elif msg_id == 0x0002: # 动力母线A配电指令
|
|
cmd = msg_disHVBACmdMsg()
|
|
if cmd.unpack(data):
|
|
print("收到动力母线A配电指令")
|
|
|
|
elif msg_id == 0x0003: # 仪表主母线配电指令
|
|
cmd = msg_disLVMBCmdMsg()
|
|
if cmd.unpack(data):
|
|
print("收到仪表主母线配电指令")
|
|
|
|
elif msg_id == 0x0004: # 仪表汇流排配电指令
|
|
cmd = msg_disLVBCmdMsg()
|
|
if cmd.unpack(data):
|
|
print("收到仪表汇流排配电指令")
|
|
|
|
except Exception as e:
|
|
print(f"接收消息出错: {e}")
|
|
break
|
|
|
|
def update_sim_data(self):
|
|
"""更新模拟数据，可以扩展实现更复杂的模拟逻辑"""
|
|
# 这里可以添加数据变化的逻辑
|
|
pass
|
|
|
|
if __name__ == "__main__":
|
|
simulator = PowerManagerSimulator()
|
|
try:
|
|
simulator.start()
|
|
print("PowerManager模拟器已启动，按Ctrl+C停止...")
|
|
while True:
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
simulator.stop()
|
|
print("PowerManager模拟器已停止") |