227 lines
9.2 KiB
Python
227 lines
9.2 KiB
Python
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"]
|