"""内置模拟器：产生四路母线反馈，支持配电指令回显。 设计：模拟器状态(反馈消息 + 最近指令)独立维护；周期性调用 update() 返回需要注入解析管线的反馈消息列表。配电指令通过 apply_command() 回写对应断路器状态，从而与真实设备的收发链路完全一致。 """ import random from . import protocol from .protocol import (MsgDisHighVolBusFb, MsgDisHighAVolBusFb, MsgDisLowMainBusFb, MsgDisLowBusFb, BREAKER_ON) class Simulator: # 指令消息 ID -> 对应反馈消息 ID CMD_TO_FB = {0x0001: 0x0005, 0x0002: 0x0006, 0x0003: 0x0007, 0x0004: 0x0008} def __init__(self): self.fb = { 0x0005: MsgDisHighVolBusFb(), 0x0006: MsgDisHighAVolBusFb(), 0x0007: MsgDisLowMainBusFb(), 0x0008: MsgDisLowBusFb(), } # 最近一次配电指令（用于回显） self.last_cmd = { 0x0001: protocol.MsgDisHVBCmd(), 0x0002: protocol.MsgDisHVBACmd(), 0x0003: protocol.MsgDisLVMBCmd(), 0x0004: protocol.MsgDisLVBCmd(), } self.fault_animation = True self.noise = True self._fault_counter = 0 self.reset() def reset(self): self._init_high_vol() self._init_high_a_vol() self._init_low_main() self._init_low_bus() # 复位指令状态 for cmd in self.last_cmd.values(): for name in cmd.field_names(): setattr(cmd, name, 0) # ---- 初始值（与 test.py 一致） --------------------------------- def _init_high_vol(self): m = self.fb[0x0005] for name in m.field_names(): setattr(m, name, 0) for name in ("fuelCellCircuitBreaker", "powerLithiumBatteryCircuitBreaker", "propulsionMotorCircuitBreaker", "lithiumBatteryGroupInstrumentCircuitBreaker", "dcDc5ModuleCircuitBreaker", "bowHighVoltageDistributionBoxCircuitBreaker", "sternFTDevice45CircuitBreaker", "sternRudderSwitch1CircuitBreaker", "sternRudderSwitch2CircuitBreaker", "fbReservedCircuitBreaker", "tyzReservedCircuitBreaker", "reservedCircuitBreaker"): setattr(m, name, BREAKER_ON) m.dcDcFaultWord1 = 0x01 m.dcDcFaultWord2 = 0x01 m.powerBusInsulationStatus = BREAKER_ON m.aBusInsulationStatus = BREAKER_ON m.meterPowerLossSignal = BREAKER_ON m.emergencyPowerLossSignal = BREAKER_ON m.dcdcTemperature = 0x08 m.powerBusVoltage = 6000 m.dcDc5ModuleCurrent = 500 m.powerBusCurrent = 1000 m.busbarAVoltage = 6000 m.propulsionMotorControlBoxCurrent = 800 m.busbarACurrent = 500 m.lithiumBatteryGroupMeterCurrent = 300 m.bowHighVoltageDistributionBoxCurrent = 200 m.sternFTDevice45Current = 150 m.tyzReservedSwitchCurrent = 0 m.sternRudderSwitch1Current = 100 m.reservedCurrent1 = 1000 m.sternRudderSwitch2Current = 100 m.reservedCurrent2 = 1000 m.fbReservedSwitchCurrent = 10000 m.reservedCurrent3 = 1000 m.coolWaterPressure = 1000 def _init_high_a_vol(self): m = self.fb[0x0006] for name in m.field_names(): setattr(m, name, 0) for name in ("bowFTDevice123CircuitBreaker", "actuatorCircuitBreaker", "mastSteeringGearControlBoxCircuitBreaker", "xczCircuitBreaker", "bowRudderControlBoxCircuitBreaker", "openWaterCoverStartCylinderCircuitBreaker"): setattr(m, name, BREAKER_ON) m.instrumentPowerFailureSignal = BREAKER_ON m.emergencyPowerFailureSignal = BREAKER_ON m.waterIngressionAlarm = BREAKER_ON m.busbarBVoltage = 6000 m.busbarBCurrent = 300 m.bowFTDevice123Current = 200 m.actuatorCurrent = 150 m.mastSteeringGearControlBoxCurrent = 100 m.xczCurrent = 50 m.bowRudderControlBoxCurrent = 80 m.openWaterCoverStartCylinderCurrent = 0 def _init_low_main(self): m = self.fb[0x0007] for name in m.field_names(): setattr(m, name, 0) m.lithiumBatteryGroupInstrumentCircuitBreaker = BREAKER_ON m.bowLowVoltageDistributionBoxCircuitBreaker = BREAKER_ON m.unit4InstrumentDC48VCircuitBreaker = BREAKER_ON m.dcC1DCDistributionPanelCircuitBreaker = BREAKER_ON m.reservedCircuitBreaker1 = BREAKER_ON m.reservedCircuitBreaker2 = BREAKER_ON m.emergencyLithiumBatteryGroup2CircuitBreaker = BREAKER_ON m.instrumentPowerFailureSignal = BREAKER_ON m.emergencyPowerFailureSignal = BREAKER_ON m.instrumentBusbarInsulationLow = BREAKER_ON m.instrumentBusbarVoltage = 480 m.bowLowVoltageDistributionBoxCurrent = 100 m.dcC1InstrumentDC48VCurrent = 50 m.reservedCurrent1 = 1100 m.emergencyLithiumBatteryGroup2Current = 1000 m.lithiumBatteryGroupInstrumentCurrent = 200 m.unit4InstrumentDC48VCurrent = 50 m.reservedCurrent2 = 1000 m.emergency2BusbarVoltage = 480 m.compositeEnergyManagementSystemEmergencyDC48VCurrent = 20 m.fuelCellSecuritySystemEmergencyDC48VCurrent = 20 m.plcControlPowerCurrent = 10 def _init_low_bus(self): m = self.fb[0x0008] for name in m.field_names(): setattr(m, name, 0) for name in ("unit1CircuitBreaker", "unit2CircuitBreaker", "unit3CircuitBreaker", "unit5CircuitBreaker", "bowPZDeviceCircuitBreaker", "reservedCircuitBreaker1", "reservedCircuitBreaker2", "emergencyLithiumBatteryGroup1CircuitBreaker"): setattr(m, name, BREAKER_ON) m.instrumentPowerFailureSignal = BREAKER_ON m.emergencyPowerFailureSignal = BREAKER_ON m.waterIngressionAlarm = BREAKER_ON m.instrumentBusbarVoltage = 480 m.instrumentBusbarCurrent = 5000 m.unit1Current = 1000 m.unit2Current = 1000 m.unit3Current = 1000 m.unit5Current = 1000 m.bowPZDeviceCurrent = 500 m.emergencyLithiumBatteryGroup1Current = 1000 m.emergency1BusbarVoltage = 480 # ---- 对外接口 ------------------------------------------------ def apply_command(self, cmd): """收到配电指令后回写对应母线反馈的断路器状态。""" fb = self.fb.get(self.CMD_TO_FB.get(cmd.MSG_ID)) if fb is None: return self.last_cmd[cmd.MSG_ID] = cmd for name in cmd.field_names(): if hasattr(fb, name): setattr(fb, name, getattr(cmd, name)) def _jitter(self, fb): """给电流/电压叠加轻微随机波动，模拟真实读数。""" for name in fb.field_names(): if name.endswith("Current") or name.endswith("Voltage") or name.endswith("Pressure"): v = getattr(fb, name) if v > 0: setattr(fb, name, max(0, v + random.randint(-int(v * 0.05), int(v * 0.05)))) def _step_fault(self): m = self.fb[0x0005] w1 = ((m.dcDcFaultWord1 << 1) | (m.dcDcFaultWord1 >> 7)) & 0xFF w2 = ((m.dcDcFaultWord2 >> 1) | (m.dcDcFaultWord2 << 7)) & 0xFF m.dcDcFaultWord1 = w1 m.dcDcFaultWord2 = w2 def update(self, interval=1.0): """推进一帧模拟。返回需要注入解析管线的反馈消息列表。""" if self.fault_animation: self._step_fault() if self.noise: for fb in self.fb.values(): self._jitter(fb) return list(self.fb.values())