from __future__ import annotations from datetime import datetime from typing import Dict, List, Sequence, Tuple from app.adapters.device_client import ( AI_ALARM_START_REGISTER, AI_CHANNEL_START_REGISTER, AI_INPUT_COUNT, AI_INPUT_START_REGISTER, AO_CHANNEL_START_REGISTER, CDeviceClient, DI_INPUT_REGISTER, DO_STATE_REGISTER, EVENT_FLAG_REGISTER, EVENT_HOLDING_START_REGISTER, LINE_ALARM_REGISTERS_PER_LINE, LINE_ALARM_START_REGISTER, LINE_START_REGISTERS, ModbusTransport, TIME_SYNC_START_REGISTER, UART_CONFIG_REGISTER_COUNT, UART_CONFIG_START_REGISTER, VALUE_GROUP_FIELDS, ) from app.schemas.platform import ( AiAlarmSettingIn, AlarmRule, ChannelConfigIn, ChannelItem, DeviceConfigIn, LineAlarmSettingIn, NetConfigItem, SoftwareVersion, HardwareVersion, SwitchControlIn, SystemConfigIn, UartConfigItem, ) class FakeTransport(ModbusTransport): def __init__(self) -> None: self.input_registers: Dict[int, int] = {} self.holding_registers: Dict[int, int] = {} self.discrete_inputs: Dict[int, bool] = {} self.coils: Dict[int, bool] = {} self.written_registers: List[Tuple[int, List[int]]] = [] self.written_coils: List[Tuple[int, bool]] = [] def set_input_block(self, address: int, values: Sequence[int]) -> None: for offset, value in enumerate(values): self.input_registers[address + offset] = int(value) def set_holding_block(self, address: int, values: Sequence[int]) -> None: for offset, value in enumerate(values): self.holding_registers[address + offset] = int(value) def set_discrete_block(self, address: int, values: Sequence[bool]) -> None: for offset, value in enumerate(values): self.discrete_inputs[address + offset] = bool(value) def set_coil_block(self, address: int, values: Sequence[bool]) -> None: for offset, value in enumerate(values): self.coils[address + offset] = bool(value) def read_input_registers(self, address: int, count: int) -> List[int]: return [self.input_registers.get(address + offset, 0) for offset in range(count)] def read_holding_registers(self, address: int, count: int) -> List[int]: return [self.holding_registers.get(address + offset, 0) for offset in range(count)] def read_discrete_inputs(self, address: int, count: int) -> List[bool]: return [self.discrete_inputs.get(address + offset, False) for offset in range(count)] def read_coils(self, address: int, count: int) -> List[bool]: return [self.coils.get(address + offset, False) for offset in range(count)] def write_registers(self, address: int, values: Sequence[int]) -> None: block = [int(value) for value in values] self.written_registers.append((address, block)) for offset, value in enumerate(block): self.holding_registers[address + offset] = value def write_coil(self, address: int, value: bool) -> None: self.written_coils.append((address, bool(value))) self.coils[address] = bool(value) def test_modbus_client_reads_upstream_data() -> None: transport = FakeTransport() client = CDeviceClient(transport=transport) for line_index, start_register in enumerate(LINE_START_REGISTERS, start=1): values: List[int] = [] base = float(line_index * 100) for field_offset, _field_name in enumerate(VALUE_GROUP_FIELDS, start=1): values.extend(client._float_to_registers(base + field_offset)) transport.set_input_block(start_register, values) ai_values: List[int] = [] for index in range(AI_INPUT_COUNT): ai_values.extend(client._float_to_registers(1.25 + index)) transport.set_input_block(AI_INPUT_START_REGISTER, ai_values) transport.set_discrete_block(0x0000, [True, False, True, False, True, False, True, False, True, False, True, False]) transport.set_coil_block(0x0000, [False, True, False, True, False, True, False, True, False, True, False, True]) transport.set_input_block(DI_INPUT_REGISTER, [0x0555]) realtime = client.read_realtime_data() assert len(realtime.line_list) == 4 assert realtime.line_list[0].sec_val.Ua == 101.0 assert realtime.line_list[1].sec_val.Pt == 210.0 assert realtime.line_list[3].sec_val.frq == 426.0 assert realtime.ai_collect["ai1"] == 1.25 assert realtime.ai_collect["ai8"] == 8.25 assert realtime.ai_collect["ai9"] == 0.0 assert realtime.switch["di1"] == 1 assert realtime.switch["di2"] == 0 assert realtime.switch["do1"] == 0 assert realtime.switch["do2"] == 1 def test_modbus_client_reads_alarm_event_once() -> None: transport = FakeTransport() client = CDeviceClient(transport=transport) timestamp = int(datetime(2026, 5, 25, 12, 0, 0).timestamp()) transport.set_input_block(EVENT_FLAG_REGISTER, [1]) transport.set_holding_block( EVENT_HOLDING_START_REGISTER, [ timestamp & 0xFFFF, (timestamp >> 16) & 0xFFFF, 123, (2 << 8) | 1, ], ) alarms = client.read_alarm_events() duplicate = client.read_alarm_events() assert len(alarms) == 1 assert alarms[0].no == "2" assert alarms[0].type == "PT断线" assert "线路2PT断线" in alarms[0].content assert duplicate == [] def test_modbus_client_writes_downstream_registers() -> None: transport = FakeTransport() client = CDeviceClient(transport=transport) device_payload = DeviceConfigIn( password="", hardware_version=HardwareVersion(), software_version=SoftwareVersion(), net=[NetConfigItem(nic="网卡一", ip="192.168.1.10", mask="255.255.255.0", gateway="192.168.1.1", protocol="Modbus TCP")], uart=[ UartConfigItem(port="COM1", baud=9600, parity="NONE", data_bits=8, stop_bits=1, protocol="Modbus RTU"), UartConfigItem(port="COM2", baud=115200, parity="EVEN", data_bits=8, stop_bits=2, protocol="Modbus RTU"), ], ) channel_payload = ChannelConfigIn( ai_channel=[ChannelItem(ch=1, singal_type="4-20mA", line_no=1, type="UA", limit_low=0, limit_high=20)], ao_channel=[ChannelItem(ch=2, singal_type="1~5V", line_no=2, type="Q", limit_low=1, limit_high=8)], ) line_alarm_payload = LineAlarmSettingIn( line_no=2, over_limit_alarm=[ AlarmRule(category="电压", limit=180, delay=5, output_node="开出1", enabled=True), AlarmRule(category="电流", limit=200, delay=6, output_node="开出2", enabled=False), AlarmRule(category="差流", limit=20, delay=7, output_node="开出3", enabled=True), AlarmRule(category="功率", limit=30, delay=8, output_node="开出4", enabled=False), AlarmRule(category="频率", limit=50, delay=9, output_node="开出5", enabled=True), ], fault_alarm=[ AlarmRule(category="PT断线", delay=10, output_node="开出6", enabled=True), AlarmRule(category="CT断线", delay=11, output_node="开出7", enabled=False), ], ) ai_alarm_payload = [ AiAlarmSettingIn( channel_no=1, singal_type="4-20mA", limit_low=0, limit_high=20, delay=3, output_node="开出1", enabled=True, ).model_dump() ] system_payload = SystemConfigIn(time_sync="2026-05-25 12:34:56", brightness=80, screen_saver=60) switch_payload = SwitchControlIn(ch=3, action=1) device_result = client.send_device_config(device_payload) channel_result = client.send_channel_config(channel_payload) line_result = client.send_line_alarm_setting(line_alarm_payload) ai_alarm_result = client.send_ai_alarm_setting(ai_alarm_payload) system_result = client.send_system_config(system_payload) control_result = client.send_switch_control(switch_payload) register_map = {address: values for address, values in transport.written_registers} assert UART_CONFIG_START_REGISTER in register_map assert len(register_map[UART_CONFIG_START_REGISTER]) == UART_CONFIG_REGISTER_COUNT assert AI_CHANNEL_START_REGISTER in register_map assert len(register_map[AI_CHANNEL_START_REGISTER]) == 30 assert AO_CHANNEL_START_REGISTER in register_map assert len(register_map[AO_CHANNEL_START_REGISTER]) == 30 assert LINE_ALARM_START_REGISTER + LINE_ALARM_REGISTERS_PER_LINE in register_map assert len(register_map[LINE_ALARM_START_REGISTER + LINE_ALARM_REGISTERS_PER_LINE]) == 14 assert AI_ALARM_START_REGISTER in register_map assert len(register_map[AI_ALARM_START_REGISTER]) == 30 assert TIME_SYNC_START_REGISTER in register_map assert len(register_map[TIME_SYNC_START_REGISTER]) == 2 assert DO_STATE_REGISTER in register_map assert register_map[DO_STATE_REGISTER][0] == 0b100 assert transport.written_coils == [(2, True)] assert device_result["send_status"] == "成功" assert channel_result["modbus_written"]["ai_registers"] == 30 assert line_result["line_no"] == 2 assert ai_alarm_result["items"] == 1 assert system_result["send_status"] == "成功" assert "执行成功" in control_result["control_status"]