emcp/backend/tests/test_modbus_client.py

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"]