emcp/backend/app/adapters/device_client.py
2026-06-12 10:37:57 +08:00

701 lines
26 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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,
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 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_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
EVENT_FLAG_REGISTER = 0x00E4
EVENT_HOLDING_START_REGISTER = 0x0000
UART_CONFIG_START_REGISTER = 0x0004
UART_CONFIG_REGISTER_COUNT = 16
LINE_ALARM_START_REGISTER = 0x0014
LINE_ALARM_REGISTERS_PER_LINE = 14
AI_CHANNEL_START_REGISTER = 0x004C
AO_CHANNEL_START_REGISTER = 0x006A
AI_ALARM_START_REGISTER = 0x0088
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
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 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]:
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]:
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]:
return {"send_status": "成功", "target": "ai_alarm", "items": len(payload)}
def send_system_config(self, payload: SystemConfigIn) -> Dict[str, Any]:
return {"send_status": "成功", "target": "system", "brightness": payload.brightness}
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 _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:
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 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]
if event_flag == 0:
return []
registers = self.transport.read_holding_registers(EVENT_HOLDING_START_REGISTER, 4)
signature = tuple(int(value) for value in registers)
if signature == self._last_event_signature:
return []
self._last_event_signature = signature
timestamp = int(registers[0]) | (int(registers[1]) << 16)
milliseconds = int(registers[2]) & 0xFFFF
line_and_type = int(registers[3]) & 0xFFFF
line_no = (line_and_type >> 8) & 0xFF
event_type = line_and_type & 0xFF
event_name = {
1: "PT断线",
2: "CT断线",
}.get(event_type, f"事件{event_type}")
event_time = datetime.fromtimestamp(timestamp).replace(microsecond=min(milliseconds, 999) * 1000)
line_text = "总路" if line_no == 0 else f"线路{line_no}"
return [
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}",
level="",
)
]
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]:
def build_channel_payload(items: List[Any]) -> List[int]:
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 self._bytes_to_registers(bytes(raw))
ai_registers = build_channel_payload(payload.ai_channel)
ao_registers = build_channel_payload(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}
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))
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_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},
}