999 lines
39 KiB
Python
999 lines
39 KiB
Python
from __future__ import annotations
|
||
|
||
import struct
|
||
from abc import ABC, abstractmethod
|
||
from datetime import datetime
|
||
from random import Random
|
||
from typing import Any, Dict, List, Optional, Sequence
|
||
|
||
from app.core.config import settings
|
||
|
||
from app.schemas.platform import (
|
||
AlarmEvent,
|
||
ChannelConfigIn,
|
||
DeviceConfigIn,
|
||
DeviceStatus,
|
||
LineAlarmSettingIn,
|
||
LineData,
|
||
RealtimeData,
|
||
SystemConfigIn,
|
||
SwitchControlIn,
|
||
TimeSyncConfigIn,
|
||
ValueGroup,
|
||
)
|
||
|
||
|
||
class DeviceClient(ABC):
|
||
@abstractmethod
|
||
def read_realtime_data(self) -> RealtimeData:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def read_device_status(self) -> DeviceStatus:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def read_alarm_events(self) -> List[AlarmEvent]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def read_channel_config(self) -> Dict[str, Any]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def read_line_alarm_setting(self, line_no: int) -> Dict[str, Any]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def read_ai_alarm_setting(self) -> List[Dict[str, Any]]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def read_system_config(self) -> Dict[str, Any]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def read_time_sync_config(self) -> Dict[str, str]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def send_device_config(self, payload: DeviceConfigIn) -> Dict[str, Any]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def send_channel_config(self, payload: ChannelConfigIn) -> Dict[str, Any]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def send_line_alarm_setting(self, payload: LineAlarmSettingIn) -> Dict[str, Any]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def send_ai_alarm_setting(self, payload: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def send_system_config(self, payload: SystemConfigIn) -> Dict[str, Any]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def send_time_sync_config(self, payload: TimeSyncConfigIn) -> Dict[str, Any]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def send_switch_control(self, payload: SwitchControlIn) -> Dict[str, Any]:
|
||
raise NotImplementedError
|
||
|
||
|
||
VALUE_GROUP_FIELDS = [
|
||
"Ua",
|
||
"Ub",
|
||
"Uc",
|
||
"Ia",
|
||
"Ib",
|
||
"Ic",
|
||
"Pa",
|
||
"Pb",
|
||
"Pc",
|
||
"Pt",
|
||
"Qa",
|
||
"Qb",
|
||
"Qc",
|
||
"Qt",
|
||
"Sa",
|
||
"Sb",
|
||
"Sc",
|
||
"St",
|
||
"PFa",
|
||
"PFb",
|
||
"PFc",
|
||
"PFt",
|
||
"Uab",
|
||
"Ubc",
|
||
"Uca",
|
||
"frq",
|
||
]
|
||
LINE_START_REGISTERS = [0x0000, 0x0034, 0x0068, 0x009C]
|
||
LINE_REGISTER_COUNT = 0x34
|
||
AI_INPUT_START_REGISTER = 0x00D0
|
||
AI_INPUT_COUNT = 8
|
||
DI_INPUT_REGISTER = 0x00E3
|
||
DO_INPUT_REGISTER = 0x00E4
|
||
EVENT_FLAG_REGISTER = 0x00E5
|
||
EVENT_INPUT_START_REGISTER = 0x0100
|
||
EVENT_SUCCESS_REPLY_REGISTER = 0x0100
|
||
UART_CONFIG_START_REGISTER = 0x0004
|
||
UART_CONFIG_REGISTER_COUNT = 16
|
||
LINE_ALARM_START_REGISTER = 0x0014
|
||
LINE_ALARM_REGISTERS_PER_LINE = 16
|
||
AI_CHANNEL_START_REGISTER = 0x0054
|
||
AO_CHANNEL_START_REGISTER = 0x0072
|
||
AI_ALARM_START_REGISTER = 0x0090
|
||
ANALOG_BLOCK_REGISTER_COUNT = 30
|
||
DO_STATE_REGISTER = 0x00A6
|
||
TIME_SYNC_START_REGISTER = 0x0000
|
||
|
||
|
||
def _clamp_byte(value: Any) -> int:
|
||
try:
|
||
number = int(round(float(value)))
|
||
except (TypeError, ValueError):
|
||
number = 0
|
||
return max(0, min(255, number))
|
||
|
||
|
||
def _normalize_text(value: str) -> str:
|
||
return value.strip().upper().replace(" ", "")
|
||
|
||
|
||
def _serial_parity_flag(value: str) -> str:
|
||
normalized = _normalize_text(value)
|
||
return {"NONE": "N", "N": "N", "ODD": "O", "O": "O", "EVEN": "E", "E": "E"}.get(normalized, "N")
|
||
|
||
|
||
class ModbusTransport(ABC):
|
||
@abstractmethod
|
||
def read_input_registers(self, address: int, count: int) -> List[int]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def read_holding_registers(self, address: int, count: int) -> List[int]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def read_discrete_inputs(self, address: int, count: int) -> List[bool]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def read_coils(self, address: int, count: int) -> List[bool]:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def write_registers(self, address: int, values: Sequence[int]) -> None:
|
||
raise NotImplementedError
|
||
|
||
@abstractmethod
|
||
def write_coil(self, address: int, value: bool) -> None:
|
||
raise NotImplementedError
|
||
|
||
|
||
class PymodbusTransport(ModbusTransport):
|
||
def __init__(self) -> None:
|
||
self._client: Any = None
|
||
|
||
def _build_client(self) -> Any:
|
||
try:
|
||
from pymodbus.client import ModbusSerialClient, ModbusTcpClient
|
||
except ImportError as exc:
|
||
raise RuntimeError("未安装 pymodbus，无法启用 Modbus 采集") from exc
|
||
|
||
transport = settings.modbus_transport.strip().lower()
|
||
if transport == "tcp":
|
||
return ModbusTcpClient(
|
||
host=settings.modbus_tcp_host,
|
||
port=settings.modbus_tcp_port,
|
||
timeout=settings.modbus_timeout_seconds,
|
||
)
|
||
|
||
try:
|
||
from pymodbus import FramerType
|
||
|
||
return ModbusSerialClient(
|
||
port=settings.modbus_serial_device,
|
||
framer=FramerType.RTU,
|
||
baudrate=settings.modbus_serial_baud,
|
||
parity=_serial_parity_flag(settings.modbus_serial_parity),
|
||
stopbits=settings.modbus_serial_stop_bits,
|
||
bytesize=settings.modbus_serial_data_bits,
|
||
timeout=settings.modbus_timeout_seconds,
|
||
)
|
||
except Exception:
|
||
return ModbusSerialClient(
|
||
method="rtu",
|
||
port=settings.modbus_serial_device,
|
||
baudrate=settings.modbus_serial_baud,
|
||
parity=_serial_parity_flag(settings.modbus_serial_parity),
|
||
stopbits=settings.modbus_serial_stop_bits,
|
||
bytesize=settings.modbus_serial_data_bits,
|
||
timeout=settings.modbus_timeout_seconds,
|
||
)
|
||
|
||
def _ensure_client(self) -> Any:
|
||
if self._client is None:
|
||
self._client = self._build_client()
|
||
connect = getattr(self._client, "connect", None)
|
||
if callable(connect):
|
||
result = connect()
|
||
if result is False:
|
||
raise RuntimeError("Modbus 连接失败")
|
||
return self._client
|
||
|
||
def _call(self, method_name: str, *args: Any, **kwargs: Any) -> Any:
|
||
client = self._ensure_client()
|
||
method = getattr(client, method_name)
|
||
try:
|
||
return method(*args, slave=settings.modbus_slave_id, **kwargs)
|
||
except TypeError:
|
||
return method(*args, unit=settings.modbus_slave_id, **kwargs)
|
||
|
||
def _check_response(self, response: Any, action: str) -> Any:
|
||
if response is None:
|
||
raise RuntimeError(f"Modbus {action} 返回为空")
|
||
is_error = getattr(response, "isError", None)
|
||
if callable(is_error) and response.isError():
|
||
raise RuntimeError(f"Modbus {action} 返回错误: {response}")
|
||
return response
|
||
|
||
def read_input_registers(self, address: int, count: int) -> List[int]:
|
||
response = self._check_response(
|
||
self._call("read_input_registers", address, count),
|
||
f"读输入寄存器[{hex(address)}:{count}]",
|
||
)
|
||
return list(getattr(response, "registers", []) or [])
|
||
|
||
def read_holding_registers(self, address: int, count: int) -> List[int]:
|
||
response = self._check_response(
|
||
self._call("read_holding_registers", address, count),
|
||
f"读保持寄存器[{hex(address)}:{count}]",
|
||
)
|
||
return list(getattr(response, "registers", []) or [])
|
||
|
||
def read_discrete_inputs(self, address: int, count: int) -> List[bool]:
|
||
response = self._check_response(
|
||
self._call("read_discrete_inputs", address, count),
|
||
f"读离散输入[{hex(address)}:{count}]",
|
||
)
|
||
return [bool(value) for value in (getattr(response, "bits", []) or [])[:count]]
|
||
|
||
def read_coils(self, address: int, count: int) -> List[bool]:
|
||
response = self._check_response(
|
||
self._call("read_coils", address, count),
|
||
f"读线圈[{hex(address)}:{count}]",
|
||
)
|
||
return [bool(value) for value in (getattr(response, "bits", []) or [])[:count]]
|
||
|
||
def write_registers(self, address: int, values: Sequence[int]) -> None:
|
||
self._check_response(
|
||
self._call("write_registers", address, list(values)),
|
||
f"写保持寄存器[{hex(address)}:{len(values)}]",
|
||
)
|
||
|
||
def write_coil(self, address: int, value: bool) -> None:
|
||
self._check_response(
|
||
self._call("write_coil", address, bool(value)),
|
||
f"写线圈[{hex(address)}]",
|
||
)
|
||
|
||
|
||
class MockDeviceClient(DeviceClient):
|
||
def __init__(self) -> None:
|
||
self._random = Random(3568)
|
||
self._tick = 0
|
||
self._channel_config = {
|
||
"ai_channel": [
|
||
{"ch": 1, "singal_type": "4-20mA", "line_no": 1, "type": "UA", "limit_low": 0, "limit_high": 20}
|
||
],
|
||
"ao_channel": [
|
||
{"ch": 1, "singal_type": "1~5V", "line_no": 2, "type": "Q", "limit_low": 0, "limit_high": 20}
|
||
],
|
||
}
|
||
self._line_alarm_settings = {
|
||
1: {
|
||
"line_no": 1,
|
||
"over_limit_alarm": [
|
||
{"category": "电压", "limit": 180, "delay": 180, "output_node": "开出1", "enabled": True},
|
||
{"category": "电流", "limit": 180, "delay": 180, "output_node": "开出1", "enabled": True},
|
||
{"category": "差流", "limit": 180, "delay": 180, "output_node": "开出1", "enabled": False},
|
||
{"category": "功率", "limit": 180, "delay": 180, "output_node": "开出1", "enabled": False},
|
||
{"category": "频率", "limit": 50, "delay": 180, "output_node": "开出1", "enabled": False},
|
||
],
|
||
"fault_alarm": [
|
||
{"category": "PT断线", "limit": 0, "delay": 180, "output_node": "开出1", "enabled": True},
|
||
{"category": "CT断线", "limit": 0, "delay": 180, "output_node": "开出2", "enabled": False},
|
||
],
|
||
"transformer_change": [
|
||
{"category": "PT变化", "value": 100.0, "enabled": True},
|
||
{"category": "CT变化", "value": 50.0, "enabled": True},
|
||
],
|
||
}
|
||
}
|
||
self._ai_alarm_setting = [
|
||
{
|
||
"channel_no": 1,
|
||
"singal_type": "4-20mA",
|
||
"limit_low": 0,
|
||
"limit_high": 20,
|
||
"delay": 180,
|
||
"output_node": "开出1",
|
||
"enabled": True,
|
||
}
|
||
]
|
||
self._system_config = {"time_sync": "2026-05-25 12:34:56", "brightness": 80, "screen_saver": 60}
|
||
|
||
def _value_group(self, base_u: float, base_i: float, base_p: float) -> ValueGroup:
|
||
delta = self._tick % 5
|
||
return ValueGroup(
|
||
Ua=base_u + delta,
|
||
Ub=base_u + 2 + delta,
|
||
Uc=base_u - 1 + delta,
|
||
Ia=base_i + 0.1 * delta,
|
||
Ib=base_i - 0.1 + 0.1 * delta,
|
||
Ic=base_i + 0.2 + 0.1 * delta,
|
||
Pa=base_p + delta,
|
||
Pb=base_p - 2 + delta,
|
||
Pc=base_p + 3 + delta,
|
||
Pt=base_p * 3 + delta,
|
||
Qa=12 + delta,
|
||
Qb=11 + delta,
|
||
Qc=13 + delta,
|
||
Qt=36 + delta,
|
||
Sa=base_p + 15 + delta,
|
||
Sb=base_p + 10 + delta,
|
||
Sc=base_p + 16 + delta,
|
||
St=base_p * 3 + 44 + delta,
|
||
PFa=0.98,
|
||
PFb=0.97,
|
||
PFc=0.99,
|
||
PFt=0.98,
|
||
Uab=base_u * 1.73,
|
||
Ubc=base_u * 1.72,
|
||
Uca=base_u * 1.73,
|
||
frq=50.0,
|
||
)
|
||
|
||
def read_realtime_data(self) -> RealtimeData:
|
||
self._tick += 1
|
||
line_list = [
|
||
LineData(line_no=1, pri_val=self._value_group(6000, 150, 820), sec_val=self._value_group(57.6, 1.2, 68)),
|
||
LineData(line_no=2, pri_val=self._value_group(5990, 148, 810), sec_val=self._value_group(57.2, 1.1, 66)),
|
||
LineData(line_no=3, pri_val=self._value_group(6010, 151, 830), sec_val=self._value_group(58.1, 1.3, 70)),
|
||
LineData(line_no=4, pri_val=self._value_group(6020, 149, 825), sec_val=self._value_group(58.4, 1.2, 69)),
|
||
]
|
||
switch = {f"di{i}": int((i + self._tick) % 2 == 0) for i in range(1, 13)}
|
||
switch.update({f"do{i}": int((i + self._tick + 1) % 2 == 0) for i in range(1, 13)})
|
||
ai_collect = {f"ai{i}": round(1 + self._random.random() * 5 + (self._tick % 3) * 0.1, 2) for i in range(1, 13)}
|
||
return RealtimeData(line_list=line_list, switch=switch, ai_collect=ai_collect)
|
||
|
||
def read_device_status(self) -> DeviceStatus:
|
||
return DeviceStatus(
|
||
self_check="正常",
|
||
net1="正常",
|
||
net2="正常",
|
||
uart1="正常",
|
||
uart2="正常" if self._tick % 4 else "断开",
|
||
)
|
||
|
||
def read_alarm_events(self) -> List[AlarmEvent]:
|
||
if self._tick % 6 != 0:
|
||
return []
|
||
return [
|
||
AlarmEvent(
|
||
alarm_type="line_alarm",
|
||
time=datetime.now(),
|
||
no=str((self._tick // 6) % 4 + 1),
|
||
type="电压",
|
||
content="模拟告警：线路电压越限",
|
||
level="中",
|
||
)
|
||
]
|
||
|
||
def read_channel_config(self) -> Dict[str, Any]:
|
||
return {
|
||
"ai_channel": [dict(item) for item in self._channel_config["ai_channel"]],
|
||
"ao_channel": [dict(item) for item in self._channel_config["ao_channel"]],
|
||
}
|
||
|
||
def read_line_alarm_setting(self, line_no: int) -> Dict[str, Any]:
|
||
if line_no in self._line_alarm_settings:
|
||
current = self._line_alarm_settings[line_no]
|
||
return {
|
||
"line_no": current["line_no"],
|
||
"over_limit_alarm": [dict(item) for item in current["over_limit_alarm"]],
|
||
"fault_alarm": [dict(item) for item in current["fault_alarm"]],
|
||
"transformer_change": [dict(item) for item in current["transformer_change"]],
|
||
}
|
||
return {
|
||
"line_no": line_no,
|
||
"over_limit_alarm": [],
|
||
"fault_alarm": [],
|
||
"transformer_change": [],
|
||
}
|
||
|
||
def read_ai_alarm_setting(self) -> List[Dict[str, Any]]:
|
||
return [dict(item) for item in self._ai_alarm_setting]
|
||
|
||
def read_system_config(self) -> Dict[str, Any]:
|
||
return dict(self._system_config)
|
||
|
||
def read_time_sync_config(self) -> Dict[str, str]:
|
||
return {"time_sync": str(self._system_config.get("time_sync", ""))}
|
||
|
||
def send_device_config(self, payload: DeviceConfigIn) -> Dict[str, Any]:
|
||
return {"send_status": "成功", "target": "device", "items": len(payload.net) + len(payload.uart)}
|
||
|
||
def send_channel_config(self, payload: ChannelConfigIn) -> Dict[str, Any]:
|
||
self._channel_config = payload.model_dump()
|
||
return {
|
||
"send_status": "成功",
|
||
"target": "channel",
|
||
"items": len(payload.ai_channel) + len(payload.ao_channel),
|
||
}
|
||
|
||
def send_line_alarm_setting(self, payload: LineAlarmSettingIn) -> Dict[str, Any]:
|
||
self._line_alarm_settings[payload.line_no] = payload.model_dump()
|
||
return {"send_status": "成功", "target": "line_alarm", "line_no": payload.line_no}
|
||
|
||
def send_ai_alarm_setting(self, payload: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
self._ai_alarm_setting = [dict(item) for item in payload]
|
||
return {"send_status": "成功", "target": "ai_alarm", "items": len(payload)}
|
||
|
||
def send_system_config(self, payload: SystemConfigIn) -> Dict[str, Any]:
|
||
self._system_config.update(payload.model_dump())
|
||
return {"send_status": "成功", "target": "system", "brightness": payload.brightness}
|
||
|
||
def send_time_sync_config(self, payload: TimeSyncConfigIn) -> Dict[str, Any]:
|
||
self._system_config["time_sync"] = payload.time_sync
|
||
return {"send_status": "成功", "target": "time_sync", "time_sync": payload.time_sync}
|
||
|
||
def send_switch_control(self, payload: SwitchControlIn) -> Dict[str, Any]:
|
||
action_text = "合" if payload.action == 1 else "分"
|
||
return {"control_status": f"执行成功: 开关{payload.ch}{action_text}"}
|
||
|
||
|
||
class CDeviceClient(DeviceClient):
|
||
"""基于 Modbus 协议的真实设备客户端。"""
|
||
|
||
def __init__(self, transport: Optional[ModbusTransport] = None) -> None:
|
||
self.transport = transport or PymodbusTransport()
|
||
self._do_state = {index: 0 for index in range(1, 13)}
|
||
self._last_event_signature: Optional[tuple[int, int, int, int]] = None
|
||
|
||
def _registers_to_float(self, registers: Sequence[int]) -> float:
|
||
ordered = list(registers)
|
||
if settings.modbus_float_word_order.strip().lower() == "little":
|
||
ordered.reverse()
|
||
|
||
raw = bytearray()
|
||
for register in ordered:
|
||
high = (int(register) >> 8) & 0xFF
|
||
low = int(register) & 0xFF
|
||
if settings.modbus_float_byte_order.strip().lower() == "little":
|
||
raw.extend([low, high])
|
||
else:
|
||
raw.extend([high, low])
|
||
return float(struct.unpack(">f", bytes(raw))[0])
|
||
|
||
def _float_to_registers(self, value: float) -> List[int]:
|
||
raw = struct.pack(">f", float(value))
|
||
words = [bytearray(raw[0:2]), bytearray(raw[2:4])]
|
||
if settings.modbus_float_byte_order.strip().lower() == "little":
|
||
words = [bytearray(reversed(word)) for word in words]
|
||
registers = [(word[0] << 8) | word[1] for word in words]
|
||
if settings.modbus_float_word_order.strip().lower() == "little":
|
||
registers.reverse()
|
||
return registers
|
||
|
||
def _bytes_to_registers(self, payload: bytes) -> List[int]:
|
||
if len(payload) % 2:
|
||
payload += b"\x00"
|
||
return [(payload[index] << 8) | payload[index + 1] for index in range(0, len(payload), 2)]
|
||
|
||
def _registers_to_bytes(self, registers: Sequence[int]) -> bytes:
|
||
raw = bytearray()
|
||
for register in registers:
|
||
raw.extend([(int(register) >> 8) & 0xFF, int(register) & 0xFF])
|
||
return bytes(raw)
|
||
|
||
def _build_value_group(self, registers: Sequence[int]) -> ValueGroup:
|
||
values: Dict[str, float] = {}
|
||
for index, field_name in enumerate(VALUE_GROUP_FIELDS):
|
||
start = index * 2
|
||
values[field_name] = self._registers_to_float(registers[start : start + 2])
|
||
return ValueGroup(**values)
|
||
|
||
def _read_line_data(self, line_no: int, start_register: int) -> LineData:
|
||
registers = self.transport.read_input_registers(start_register, LINE_REGISTER_COUNT)
|
||
value_group = self._build_value_group(registers)
|
||
return LineData(line_no=line_no, pri_val=value_group, sec_val=value_group)
|
||
|
||
def _read_ai_collect(self) -> Dict[str, float]:
|
||
registers = self.transport.read_input_registers(AI_INPUT_START_REGISTER, AI_INPUT_COUNT * 2)
|
||
values = {
|
||
f"ai{index + 1}": self._registers_to_float(registers[index * 2 : index * 2 + 2])
|
||
for index in range(AI_INPUT_COUNT)
|
||
}
|
||
for index in range(AI_INPUT_COUNT + 1, 13):
|
||
values[f"ai{index}"] = 0.0
|
||
return values
|
||
|
||
def _read_di_state(self) -> Dict[str, int]:
|
||
try:
|
||
bits = self.transport.read_discrete_inputs(0x0000, 12)
|
||
return {f"di{index + 1}": int(bits[index]) for index in range(min(12, len(bits)))}
|
||
except Exception:
|
||
register = self.transport.read_input_registers(DI_INPUT_REGISTER, 1)[0]
|
||
return {f"di{index}": int(bool(register & (1 << (index - 1)))) for index in range(1, 13)}
|
||
|
||
def _read_do_state(self) -> Dict[str, int]:
|
||
try:
|
||
register = self.transport.read_input_registers(DO_INPUT_REGISTER, 1)[0]
|
||
for index in range(1, 13):
|
||
self._do_state[index] = int(bool(register & (1 << (index - 1))))
|
||
except Exception:
|
||
try:
|
||
bits = self.transport.read_coils(0x0000, 12)
|
||
if bits:
|
||
for index, bit in enumerate(bits[:12], start=1):
|
||
self._do_state[index] = int(bool(bit))
|
||
except Exception:
|
||
pass
|
||
return {f"do{index}": self._do_state.get(index, 0) for index in range(1, 13)}
|
||
|
||
def _build_status(self) -> DeviceStatus:
|
||
transport = settings.modbus_transport.strip().lower()
|
||
if transport == "tcp":
|
||
return DeviceStatus(self_check="正常", net1="正常", net2="正常", uart1="断开", uart2="断开")
|
||
return DeviceStatus(self_check="正常", net1="断开", net2="断开", uart1="正常", uart2="正常")
|
||
|
||
def _category_map(self, category: str) -> str:
|
||
normalized = category.strip().upper()
|
||
return {
|
||
"UA": "UA",
|
||
"UB": "UB",
|
||
"UC": "UC",
|
||
"UAB": "UAB",
|
||
"UBC": "UBC",
|
||
"UCA": "UCA",
|
||
"P": "P",
|
||
"Q": "Q",
|
||
"F": "F",
|
||
"FRQ": "F",
|
||
"频率": "F",
|
||
"功率": "P",
|
||
}.get(normalized, normalized)
|
||
|
||
def _signal_type_code(self, signal_type: str) -> int:
|
||
normalized = _normalize_text(signal_type)
|
||
return 0 if normalized in {"4-20MA", "4~20MA", "4-20MA"} else 1
|
||
|
||
def _measure_type_code(self, measure_type: str) -> int:
|
||
normalized = self._category_map(measure_type)
|
||
return {
|
||
"UA": 0,
|
||
"UB": 1,
|
||
"UC": 2,
|
||
"UAB": 3,
|
||
"UBC": 4,
|
||
"UCA": 5,
|
||
"P": 6,
|
||
"Q": 7,
|
||
"F": 8,
|
||
}.get(normalized, 0)
|
||
|
||
def _output_node_code(self, output_node: str) -> int:
|
||
digits = "".join(character for character in output_node if character.isdigit())
|
||
return _clamp_byte(digits or 0)
|
||
|
||
def _parity_code(self, parity: str) -> int:
|
||
normalized = _normalize_text(parity)
|
||
return {"NONE": 0, "N": 0, "ODD": 1, "O": 1, "EVEN": 2, "E": 2}.get(normalized, 0)
|
||
|
||
def _data_bits_code(self, data_bits: int) -> int:
|
||
return 0 if int(data_bits) == 8 else _clamp_byte(data_bits)
|
||
|
||
def _stop_bits_code(self, stop_bits: int) -> int:
|
||
return 1 if int(stop_bits) >= 2 else 0
|
||
|
||
def _protocol_code(self, protocol: str) -> int:
|
||
return 1 if "MODBUS" in _normalize_text(protocol) else 0
|
||
|
||
def _line_rule_payload(self, limit: Any, delay: Any, output_node: str, enabled: bool) -> bytes:
|
||
return bytes(
|
||
[
|
||
_clamp_byte(limit),
|
||
_clamp_byte(delay),
|
||
self._output_node_code(output_node),
|
||
1 if enabled else 0,
|
||
]
|
||
)
|
||
|
||
def _channel_payload_bytes(self, items: List[Any]) -> bytes:
|
||
raw = bytearray()
|
||
channel_map = {int(item.ch): item for item in items}
|
||
for channel_no in range(1, 13):
|
||
item = channel_map.get(channel_no)
|
||
if item is None:
|
||
raw.extend(b"\x00" * 5)
|
||
continue
|
||
raw.extend(
|
||
bytes(
|
||
[
|
||
self._signal_type_code(item.singal_type),
|
||
_clamp_byte(item.line_no),
|
||
self._measure_type_code(item.type),
|
||
_clamp_byte(item.limit_high),
|
||
_clamp_byte(item.limit_low),
|
||
]
|
||
)
|
||
)
|
||
return bytes(raw)
|
||
|
||
def _decode_signal_type(self, code: int) -> str:
|
||
return "4-20mA" if int(code) == 0 else "1~5V"
|
||
|
||
def _decode_measure_type(self, code: int) -> str:
|
||
return {
|
||
0: "UA",
|
||
1: "UB",
|
||
2: "UC",
|
||
3: "UAB",
|
||
4: "UBC",
|
||
5: "UCA",
|
||
6: "P",
|
||
7: "Q",
|
||
8: "F",
|
||
}.get(int(code), "UA")
|
||
|
||
def _decode_output_node(self, code: int) -> str:
|
||
return "" if int(code) <= 0 else f"开出{int(code)}"
|
||
|
||
def _decode_channel_items(self, registers: Sequence[int]) -> List[Dict[str, Any]]:
|
||
payload = self._registers_to_bytes(registers)
|
||
items: List[Dict[str, Any]] = []
|
||
for channel_no in range(1, 13):
|
||
start = (channel_no - 1) * 5
|
||
chunk = payload[start : start + 5]
|
||
if len(chunk) < 5:
|
||
chunk = chunk + b"\x00" * (5 - len(chunk))
|
||
items.append(
|
||
{
|
||
"ch": channel_no,
|
||
"singal_type": self._decode_signal_type(chunk[0]),
|
||
"line_no": int(chunk[1]),
|
||
"type": self._decode_measure_type(chunk[2]),
|
||
"limit_low": float(chunk[4]),
|
||
"limit_high": float(chunk[3]),
|
||
}
|
||
)
|
||
return items
|
||
|
||
def _decode_ai_alarm_items(self, registers: Sequence[int]) -> List[Dict[str, Any]]:
|
||
payload = self._registers_to_bytes(registers)
|
||
items: List[Dict[str, Any]] = []
|
||
for channel_no in range(1, 13):
|
||
start = (channel_no - 1) * 5
|
||
chunk = payload[start : start + 5]
|
||
if len(chunk) < 5:
|
||
chunk = chunk + b"\x00" * (5 - len(chunk))
|
||
items.append(
|
||
{
|
||
"channel_no": channel_no,
|
||
"singal_type": "4-20mA",
|
||
"limit_low": float(chunk[1]),
|
||
"limit_high": float(chunk[0]),
|
||
"delay": int(chunk[2]),
|
||
"output_node": self._decode_output_node(chunk[3]),
|
||
"enabled": bool(chunk[4]),
|
||
}
|
||
)
|
||
return items
|
||
|
||
def _encode_transformer_value(self, value: Any) -> int:
|
||
try:
|
||
number = int(round(float(value)))
|
||
except (TypeError, ValueError):
|
||
number = 1
|
||
return max(1, min(0xFFFF, number))
|
||
|
||
def _decode_transformer_items(self, payload: bytes) -> List[Dict[str, Any]]:
|
||
pt_ratio = int.from_bytes(payload[28:30], "big", signed=False) if len(payload) >= 30 else 1
|
||
ct_ratio = int.from_bytes(payload[30:32], "big", signed=False) if len(payload) >= 32 else 1
|
||
return [
|
||
{"category": "PT变化", "value": float(pt_ratio or 1), "enabled": bool(pt_ratio)},
|
||
{"category": "CT变化", "value": float(ct_ratio or 1), "enabled": bool(ct_ratio)},
|
||
]
|
||
|
||
def _decode_line_alarm_setting(self, line_no: int, registers: Sequence[int]) -> Dict[str, Any]:
|
||
payload = self._registers_to_bytes(registers)
|
||
categories = ["电压", "电流", "差流", "功率", "频率", "PT断线", "CT断线"]
|
||
over_limit_alarm: List[Dict[str, Any]] = []
|
||
fault_alarm: List[Dict[str, Any]] = []
|
||
for index, category in enumerate(categories):
|
||
start = index * 4
|
||
chunk = payload[start : start + 4]
|
||
if len(chunk) < 4:
|
||
chunk = chunk + b"\x00" * (4 - len(chunk))
|
||
rule = {
|
||
"category": category,
|
||
"limit": float(chunk[0]),
|
||
"delay": int(chunk[1]),
|
||
"output_node": self._decode_output_node(chunk[2]),
|
||
"enabled": bool(chunk[3]),
|
||
}
|
||
if category in {"PT断线", "CT断线"}:
|
||
fault_alarm.append(rule)
|
||
else:
|
||
over_limit_alarm.append(rule)
|
||
return {
|
||
"line_no": line_no,
|
||
"over_limit_alarm": over_limit_alarm,
|
||
"fault_alarm": fault_alarm,
|
||
"transformer_change": self._decode_transformer_items(payload),
|
||
}
|
||
|
||
def _read_line_alarm_registers(self, line_no: int) -> List[int]:
|
||
start_register = LINE_ALARM_START_REGISTER + max(0, line_no - 1) * LINE_ALARM_REGISTERS_PER_LINE
|
||
return self.transport.read_holding_registers(start_register, LINE_ALARM_REGISTERS_PER_LINE)
|
||
|
||
def read_realtime_data(self) -> RealtimeData:
|
||
line_list = [
|
||
self._read_line_data(line_no=index + 1, start_register=start_register)
|
||
for index, start_register in enumerate(LINE_START_REGISTERS)
|
||
]
|
||
switch = self._read_di_state()
|
||
switch.update(self._read_do_state())
|
||
return RealtimeData(line_list=line_list, switch=switch, ai_collect=self._read_ai_collect())
|
||
|
||
def read_device_status(self) -> DeviceStatus:
|
||
self.transport.read_input_registers(EVENT_FLAG_REGISTER, 1)
|
||
return self._build_status()
|
||
|
||
def read_alarm_events(self) -> List[AlarmEvent]:
|
||
event_flag = self.transport.read_input_registers(EVENT_FLAG_REGISTER, 1)[0]
|
||
event_count = int(event_flag)
|
||
if event_count <= 0:
|
||
return []
|
||
|
||
registers = self.transport.read_input_registers(EVENT_INPUT_START_REGISTER, event_count * 5)
|
||
alarms: List[AlarmEvent] = []
|
||
for index in range(event_count):
|
||
start = index * 5
|
||
event_registers = registers[start : start + 5]
|
||
if len(event_registers) < 5:
|
||
continue
|
||
signature = tuple(int(value) for value in event_registers[:4])
|
||
if signature == self._last_event_signature:
|
||
continue
|
||
self._last_event_signature = signature
|
||
|
||
timestamp = int(event_registers[0]) | (int(event_registers[1]) << 16)
|
||
milliseconds = int(event_registers[2]) & 0xFFFF
|
||
encoded = int(event_registers[3]) & 0xFFFF
|
||
action_value = int(event_registers[4]) & 0xFFFF
|
||
line_no = (encoded >> 12) & 0x0F
|
||
event_type = (encoded >> 8) & 0x0F
|
||
event_code = encoded & 0xFF
|
||
|
||
event_name = self._decode_event_name(event_type, event_code)
|
||
event_time = datetime.fromtimestamp(timestamp).replace(microsecond=min(milliseconds, 999) * 1000)
|
||
line_text = "装置" if line_no == 0 else f"线路{line_no}"
|
||
alarms.append(
|
||
AlarmEvent(
|
||
alarm_type="modbus_event",
|
||
time=event_time,
|
||
no="" if line_no == 0 else str(line_no),
|
||
type=event_name,
|
||
content=f"{line_text}{event_name}，动作值={action_value}",
|
||
level=self._decode_event_level(event_type),
|
||
)
|
||
)
|
||
|
||
if alarms:
|
||
# 协议文档将“读事件成功回复”描述为 03 读取保持寄存器，但该流程需要携带成功个数；
|
||
# 这里通过写保持寄存器把已成功读取的事件个数回告给设备。
|
||
self.transport.write_registers(EVENT_SUCCESS_REPLY_REGISTER, [len(alarms)])
|
||
return alarms
|
||
|
||
def _decode_event_name(self, event_type: int, event_code: int) -> str:
|
||
if event_type == 1:
|
||
return {1: "PT断线", 2: "CT断线"}.get(event_code, f"保护事件{event_code}")
|
||
if event_type == 2:
|
||
return f"动作事件{event_code}"
|
||
if event_type == 3:
|
||
return {1: "过压", 2: "过流", 3: "频率越限"}.get(event_code, f"报警事件{event_code}")
|
||
if event_type == 4:
|
||
return f"操作事件{event_code}"
|
||
if event_type == 5:
|
||
return f"AI越限{event_code}"
|
||
return f"事件{event_type}-{event_code}"
|
||
|
||
def _decode_event_level(self, event_type: int) -> str:
|
||
return {1: "严重", 2: "中", 3: "中", 4: "一般", 5: "一般"}.get(event_type, "一般")
|
||
|
||
def read_channel_config(self) -> Dict[str, Any]:
|
||
ai_registers = self.transport.read_holding_registers(AI_CHANNEL_START_REGISTER, ANALOG_BLOCK_REGISTER_COUNT)
|
||
ao_registers = self.transport.read_holding_registers(AO_CHANNEL_START_REGISTER, ANALOG_BLOCK_REGISTER_COUNT)
|
||
return {
|
||
"ai_channel": self._decode_channel_items(ai_registers),
|
||
"ao_channel": self._decode_channel_items(ao_registers),
|
||
}
|
||
|
||
def read_line_alarm_setting(self, line_no: int) -> Dict[str, Any]:
|
||
return self._decode_line_alarm_setting(line_no, self._read_line_alarm_registers(line_no))
|
||
|
||
def read_ai_alarm_setting(self) -> List[Dict[str, Any]]:
|
||
registers = self.transport.read_holding_registers(AI_ALARM_START_REGISTER, ANALOG_BLOCK_REGISTER_COUNT)
|
||
return self._decode_ai_alarm_items(registers)
|
||
|
||
def read_system_config(self) -> Dict[str, Any]:
|
||
registers = self.transport.read_holding_registers(TIME_SYNC_START_REGISTER, 2)
|
||
timestamp = int(registers[0]) | (int(registers[1]) << 16)
|
||
return {
|
||
"time_sync": datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S"),
|
||
"brightness": SystemConfigIn().brightness,
|
||
"screen_saver": SystemConfigIn().screen_saver,
|
||
}
|
||
|
||
def read_time_sync_config(self) -> Dict[str, str]:
|
||
registers = self.transport.read_holding_registers(TIME_SYNC_START_REGISTER, 2)
|
||
timestamp = int(registers[0]) | (int(registers[1]) << 16)
|
||
return {"time_sync": datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S")}
|
||
|
||
def send_device_config(self, payload: DeviceConfigIn) -> Dict[str, Any]:
|
||
uart_map = {item.port.upper(): item for item in payload.uart}
|
||
raw = bytearray()
|
||
for port_name in ("COM1", "COM2", "COM3", "COM4"):
|
||
item = uart_map.get(port_name)
|
||
baud = item.baud if item is not None else 0
|
||
parity = self._parity_code(item.parity) if item is not None else 0
|
||
data_bits = self._data_bits_code(item.data_bits) if item is not None else 0
|
||
stop_bits = self._stop_bits_code(item.stop_bits) if item is not None else 0
|
||
protocol = self._protocol_code(item.protocol) if item is not None else 0
|
||
raw.extend(struct.pack("<I4B", int(baud), parity, data_bits, stop_bits, protocol))
|
||
|
||
registers = self._bytes_to_registers(bytes(raw))
|
||
self.transport.write_registers(UART_CONFIG_START_REGISTER, registers)
|
||
return {
|
||
"send_status": "成功",
|
||
"target": "device",
|
||
"modbus_written": {"uart_registers": len(registers)},
|
||
"ignored_fields": ["password", "hardware_version", "software_version", "net"],
|
||
}
|
||
|
||
def send_channel_config(self, payload: ChannelConfigIn) -> Dict[str, Any]:
|
||
ai_registers = self._bytes_to_registers(self._channel_payload_bytes(payload.ai_channel))
|
||
ao_registers = self._bytes_to_registers(self._channel_payload_bytes(payload.ao_channel))
|
||
self.transport.write_registers(AI_CHANNEL_START_REGISTER, ai_registers)
|
||
self.transport.write_registers(AO_CHANNEL_START_REGISTER, ao_registers)
|
||
return {
|
||
"send_status": "成功",
|
||
"target": "channel",
|
||
"modbus_written": {
|
||
"ai_registers": len(ai_registers),
|
||
"ao_registers": len(ao_registers),
|
||
},
|
||
}
|
||
|
||
def send_line_alarm_setting(self, payload: LineAlarmSettingIn) -> Dict[str, Any]:
|
||
over_limit_map = {rule.category: rule for rule in payload.over_limit_alarm}
|
||
fault_map = {rule.category: rule for rule in payload.fault_alarm}
|
||
transformer_map = {item.category: item for item in payload.transformer_change}
|
||
|
||
ordered_rules = [
|
||
over_limit_map.get("电压"),
|
||
over_limit_map.get("电流"),
|
||
over_limit_map.get("差流"),
|
||
over_limit_map.get("功率"),
|
||
over_limit_map.get("频率"),
|
||
fault_map.get("PT断线"),
|
||
fault_map.get("CT断线"),
|
||
]
|
||
|
||
raw = bytearray()
|
||
for rule in ordered_rules:
|
||
if rule is None:
|
||
raw.extend(b"\x00" * 4)
|
||
continue
|
||
raw.extend(self._line_rule_payload(rule.limit, rule.delay, rule.output_node, rule.enabled))
|
||
|
||
pt_item = transformer_map.get("PT变化") or transformer_map.get("PT变比")
|
||
ct_item = transformer_map.get("CT变化") or transformer_map.get("CT变比")
|
||
raw.extend(self._encode_transformer_value(getattr(pt_item, "value", 1.0)).to_bytes(2, "big"))
|
||
raw.extend(self._encode_transformer_value(getattr(ct_item, "value", 1.0)).to_bytes(2, "big"))
|
||
|
||
start_register = LINE_ALARM_START_REGISTER + max(0, payload.line_no - 1) * LINE_ALARM_REGISTERS_PER_LINE
|
||
registers = self._bytes_to_registers(bytes(raw))
|
||
self.transport.write_registers(start_register, registers)
|
||
return {
|
||
"send_status": "成功",
|
||
"target": "line_alarm",
|
||
"line_no": payload.line_no,
|
||
"modbus_written": {"start_register": start_register, "registers": len(registers)},
|
||
}
|
||
|
||
def send_ai_alarm_setting(self, payload: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
raw = bytearray()
|
||
item_map = {int(item.get("channel_no", 0)): item for item in payload}
|
||
for channel_no in range(1, 13):
|
||
item = item_map.get(channel_no)
|
||
if item is None:
|
||
raw.extend(b"\x00" * 5)
|
||
continue
|
||
raw.extend(
|
||
bytes(
|
||
[
|
||
_clamp_byte(item.get("limit_high", 0)),
|
||
_clamp_byte(item.get("limit_low", 0)),
|
||
_clamp_byte(item.get("delay", 0)),
|
||
self._output_node_code(str(item.get("output_node", ""))),
|
||
1 if bool(item.get("enabled", False)) else 0,
|
||
]
|
||
)
|
||
)
|
||
|
||
registers = self._bytes_to_registers(bytes(raw))
|
||
self.transport.write_registers(AI_ALARM_START_REGISTER, registers)
|
||
return {
|
||
"send_status": "成功",
|
||
"target": "ai_alarm",
|
||
"items": len(payload),
|
||
"modbus_written": {"registers": len(registers)},
|
||
}
|
||
|
||
def send_system_config(self, payload: SystemConfigIn) -> Dict[str, Any]:
|
||
time_text = str(payload.time_sync).strip()
|
||
if time_text and time_text.lower() not in {"auto", "manual"}:
|
||
timestamp = int(datetime.fromisoformat(time_text).timestamp())
|
||
else:
|
||
timestamp = int(datetime.now().timestamp())
|
||
|
||
registers = [timestamp & 0xFFFF, (timestamp >> 16) & 0xFFFF]
|
||
self.transport.write_registers(TIME_SYNC_START_REGISTER, registers)
|
||
return {
|
||
"send_status": "成功",
|
||
"target": "system",
|
||
"brightness": payload.brightness,
|
||
"screen_saver": payload.screen_saver,
|
||
"modbus_written": {"time_registers": len(registers)},
|
||
"ignored_fields": ["brightness", "screen_saver"],
|
||
}
|
||
|
||
def send_time_sync_config(self, payload: TimeSyncConfigIn) -> Dict[str, Any]:
|
||
timestamp = int(datetime.fromisoformat(payload.time_sync.strip()).timestamp())
|
||
registers = [timestamp & 0xFFFF, (timestamp >> 16) & 0xFFFF]
|
||
self.transport.write_registers(TIME_SYNC_START_REGISTER, registers)
|
||
return {
|
||
"send_status": "成功",
|
||
"target": "time_sync",
|
||
"time_sync": datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S"),
|
||
"modbus_written": {"time_registers": len(registers)},
|
||
}
|
||
|
||
def send_switch_control(self, payload: SwitchControlIn) -> Dict[str, Any]:
|
||
coil_address = max(0, payload.ch - 1)
|
||
coil_value = bool(payload.action)
|
||
self.transport.write_coil(coil_address, coil_value)
|
||
self._do_state[payload.ch] = int(coil_value)
|
||
|
||
bitmask = 0
|
||
for index, state in self._do_state.items():
|
||
if state:
|
||
bitmask |= 1 << (index - 1)
|
||
self.transport.write_registers(DO_STATE_REGISTER, [bitmask & 0xFFFF])
|
||
|
||
action_text = "合" if payload.action == 1 else "分"
|
||
return {
|
||
"control_status": f"执行成功: 开关{payload.ch}{action_text}",
|
||
"modbus_written": {"coil_address": coil_address, "state_register": DO_STATE_REGISTER},
|
||
}
|