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(" 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}, }