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_INPUT_REGISTER, DO_STATE_REGISTER, EVENT_FLAG_REGISTER, EVENT_INPUT_START_REGISTER, EVENT_SUCCESS_REPLY_REGISTER, STATUS_WORD_00E6_REGISTER, STATUS_WORD_00E7_REGISTER, NET_CONFIG_REGISTER_COUNT, NET_CONFIG_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, TimeSyncConfigIn, 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_input_block(DO_INPUT_REGISTER, [0x0AAA]) 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_input_block( EVENT_INPUT_START_REGISTER, [ timestamp & 0xFFFF, (timestamp >> 16) & 0xFFFF, 123, (2 << 12) | (1 << 8) | 1, 456, ], ) alarms = client.read_alarm_events() duplicate = client.read_alarm_events() assert len(alarms) == 1 assert alarms[0].line_code == 2 assert alarms[0].event_type == 1 assert alarms[0].event_code == 1 assert alarms[0].event_value == 456 assert "线路2PT断线" in alarms[0].content assert duplicate == [] assert transport.written_registers[-1] == (EVENT_SUCCESS_REPLY_REGISTER, [1]) def test_modbus_client_reads_alarm_events_until_event_flag_clears() -> None: class CyclingTransport(FakeTransport): def __init__(self, batches: List[List[int]]) -> None: super().__init__() self._batches = [list(batch) for batch in batches] def read_input_registers(self, address: int, count: int) -> List[int]: if address == EVENT_FLAG_REGISTER and count == 1: return [len(self._batches[0]) // 5 if self._batches else 0] if address == EVENT_INPUT_START_REGISTER: return list(self._batches[0])[:count] if self._batches else [0] * count return super().read_input_registers(address, count) def write_registers(self, address: int, values: Sequence[int]) -> None: super().write_registers(address, values) if address == EVENT_SUCCESS_REPLY_REGISTER and self._batches: self._batches.pop(0) timestamp1 = int(datetime(2026, 5, 25, 12, 0, 0).timestamp()) timestamp2 = int(datetime(2026, 5, 25, 12, 0, 1).timestamp()) transport = CyclingTransport( batches=[ [ timestamp1 & 0xFFFF, (timestamp1 >> 16) & 0xFFFF, 111, (1 << 12) | (3 << 8) | 1, 230, ], [ timestamp2 & 0xFFFF, (timestamp2 >> 16) & 0xFFFF, 222, (2 << 12) | (2 << 8) | 7, 1, ], ] ) client = CDeviceClient(transport=transport) alarms = client.read_alarm_events() assert len(alarms) == 2 assert alarms[0].line_code == 1 assert alarms[0].event_type == 3 assert alarms[1].line_code == 2 assert alarms[1].event_type == 2 assert transport.written_registers[-2:] == [ (EVENT_SUCCESS_REPLY_REGISTER, [1]), (EVENT_SUCCESS_REPLY_REGISTER, [1]), ] assert client.read_alarm_events() == [] def test_modbus_client_reads_status_words() -> None: transport = FakeTransport() client = CDeviceClient(transport=transport) transport.set_input_block(EVENT_FLAG_REGISTER, [3]) transport.set_input_block(STATUS_WORD_00E6_REGISTER, [0x8A05]) transport.set_input_block(STATUS_WORD_00E7_REGISTER, [0x000E]) status = client.read_device_status() assert status.event_count == 3 assert status.self_check == "异常" assert status.net1 == "正常" assert status.net2 == "断开" assert status.net3 == "正常" assert status.net4 == "断开" assert status.uart1 == "断开" assert status.uart2 == "正常" assert status.uart3 == "断开" assert status.uart4 == "正常" assert status.frequency_limit == "越限" assert status.ct_break == "断线" assert status.pt_break == "断线" assert status.status_word_00e6 == 0x8A05 assert status.status_word_00e7 == 0x000E def test_modbus_client_reads_and_writes_downstream_registers() -> None: transport = FakeTransport() client = CDeviceClient(transport=transport) ai_channel_payload = client._bytes_to_registers( bytes( [ 0, 1, 0, 20, 0, 1, 2, 7, 8, 1, ] + [0] * 50 ) ) ao_channel_payload = client._bytes_to_registers( bytes( [ 1, 1, 6, 10, 2, ] + [0] * 55 ) ) line_alarm_payload_bytes = bytes( [ 180, 5, 1, 1, 200, 6, 2, 0, 20, 7, 3, 1, 30, 8, 4, 0, 50, 9, 5, 1, 0, 10, 6, 1, 0, 11, 7, 0, ] ) + (10).to_bytes(2, "big") + (20).to_bytes(2, "big") ai_alarm_payload_bytes = bytes( [ 20, 0, 3, 1, 1, ] + [0] * 55 ) net_payload_bytes = bytes( [ 192, 168, 1, 10, 255, 255, 255, 0, 192, 168, 1, 1, 1, 0, 10, 0, 0, 2, 255, 255, 255, 0, 10, 0, 0, 1, 1, 0, ] + [0] * 28 ) timestamp = int(datetime(2026, 5, 25, 12, 34, 56).timestamp()) transport.set_holding_block(AI_CHANNEL_START_REGISTER, ai_channel_payload) transport.set_holding_block(AO_CHANNEL_START_REGISTER, ao_channel_payload) transport.set_holding_block( LINE_ALARM_START_REGISTER + LINE_ALARM_REGISTERS_PER_LINE, client._bytes_to_registers(line_alarm_payload_bytes), ) transport.set_holding_block(AI_ALARM_START_REGISTER, client._bytes_to_registers(ai_alarm_payload_bytes)) transport.set_holding_block(NET_CONFIG_START_REGISTER, client._bytes_to_registers(net_payload_bytes)) transport.set_holding_block(TIME_SYNC_START_REGISTER, [timestamp & 0xFFFF, (timestamp >> 16) & 0xFFFF]) 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) read_channel = client.read_channel_config() read_line_alarm = client.read_line_alarm_setting(2) read_ai_alarm = client.read_ai_alarm_setting() read_system = client.read_system_config() read_time_sync = client.read_time_sync_config() read_net = client.read_net_config("网卡二") device_result = client.send_device_config(device_payload) net_result = client.send_net_config(NetConfigItem(nic="网卡一", ip="172.16.1.10", mask="255.255.0.0", gateway="172.16.0.1", protocol="Modbus TCP")) 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) time_sync_result = client.send_time_sync_config(TimeSyncConfigIn(time_sync="2026-05-25 13:45:00")) 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 read_channel["ai_channel"][0]["singal_type"] == "4-20mA" assert read_channel["ao_channel"][0]["type"] == "P" assert read_line_alarm["line_no"] == 2 assert read_line_alarm["transformer_change"][0]["value"] == 10.0 assert read_line_alarm["transformer_change"][1]["value"] == 20.0 assert read_ai_alarm[0]["channel_no"] == 1 assert read_ai_alarm[0]["delay"] == 3 assert read_system["time_sync"] == "2026-05-25 12:34:56" assert read_time_sync["time_sync"] == "2026-05-25 12:34:56" assert read_net["nic"] == "网卡二" assert read_net["ip"] == "10.0.0.2" assert read_net["gateway"] == "10.0.0.1" 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]) == 16 assert AI_ALARM_START_REGISTER in register_map assert len(register_map[AI_ALARM_START_REGISTER]) == 30 assert NET_CONFIG_START_REGISTER in register_map assert len(register_map[NET_CONFIG_START_REGISTER]) == NET_CONFIG_REGISTER_COUNT 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 net_result["target"] == "net" assert net_result["nic"] == "网卡一" 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 "brightness" in system_result["ignored_fields"] assert time_sync_result["target"] == "time_sync" assert time_sync_result["time_sync"] == "2026-05-25 13:45:00" assert "执行成功" in control_result["control_status"]