from __future__ import annotations from datetime import datetime from typing import Any, Dict, List, Optional from app.adapters.device_client import CDeviceClient, MockDeviceClient from app.cache.memory_store import memory_store from app.core.config import settings from app.core.security import hash_password from app.repositories.alarm_repo import AlarmRepository from app.repositories.json_config_repo import JsonConfigRepository from app.schemas.platform import ( AlarmEvent, AiAlarmSettingIn, ChannelConfigIn, DeviceConfigIn, DevicePasswordUpdateIn, DeviceStatus, LineAlarmSettingIn, NetConfigItem, LineData, RealtimeData, SwitchControlIn, SystemConfigIn, TimeSyncConfigIn, ValueGroup, UartConfigItem, ) from app.ws.manager import ws_manager # 主站侧操作类事件代码 OPERATION_EVENT_CODE = { 1: "控制分/合", 2: "读定值失败", 3: "线路定值修改", 4: "AI报警参数修改", 5: "AI通道参数修改", 6: "AO通道参数修改", 7: "通讯参数修改", 8: "密码修改", } # 操作类事件 event_type EVENT_TYPE_OPERATION = 4 # 开机自检异常归入报警类，避免前端事件报表出现未知类型 EVENT_TYPE_SELF_CHECK = 3 SELF_CHECK_COMMUNICATION_ERROR_CODE = 99 def _generate_operation_event_content(event_code: int, line_code: int = 0, detail: str = "") -> str: """根据操作事件代码生成事件内容描述文字。 参数 ---- event_code : int 操作事件代码 (1~8) line_code : int 关联线路编号，0 表示装置级 detail : str 附加描述信息，如网卡名、端口号等 """ label = OPERATION_EVENT_CODE.get(event_code, f"未知操作事件(code={event_code})") parts = [label] if line_code > 0: parts.append(f"线路{line_code}") if detail: parts.append(detail) return "；".join(parts) def _collect_startup_status_abnormalities(status: DeviceStatus) -> List[str]: abnormal_items: List[str] = [] if status.self_check != "正常": abnormal_items.append(f"装置自检={status.self_check}") for label, value in ( ("网口1", status.net1), ("网口2", status.net2), ("网口3", status.net3), ("网口4", status.net4), ("串口1", status.uart1), ("串口2", status.uart2), ("串口3", status.uart3), ("串口4", status.uart4), ): if value != "正常": abnormal_items.append(f"{label}={value}") return abnormal_items class PlatformService: def __init__(self) -> None: self.config_repo = JsonConfigRepository() self.alarm_repo = AlarmRepository() self.device_client = MockDeviceClient() if settings.use_mock_device else CDeviceClient() def _safe_ratio(self, primary: float, secondary: float, default: float = 1.0) -> float: """安全计算变比。 secondary 接近 0 时返回默认值，避免除 0。 """ if abs(float(secondary)) < 1e-9: return default return float(primary) / float(secondary) def _get_config_transformer_ratios(self, line_no: int) -> Dict[str, float]: """从设备参数区读取 PT/CT 变比。 来源：设备线路定值回读结果中的 transformer_change[] 当前约定：category 为“PT变化/CT变化”（兼容旧的“PT变比/CT变比”），取 value 字段。 """ try: current = self.device_client.read_line_alarm_setting(line_no) except Exception: return {} transformer_list = current.get("transformer_change") if not isinstance(transformer_list, list): return {} pt_ratio: Optional[float] = None ct_ratio: Optional[float] = None for transformer in transformer_list: if not isinstance(transformer, dict): continue category = str(transformer.get("category", "")).strip() raw_value = transformer.get("value", transformer.get("limit", 1.0)) try: value = float(raw_value) except (TypeError, ValueError): value = 1.0 if value <= 0: value = 1.0 if category in {"PT变化", "PT变比"}: pt_ratio = value if category in {"CT变化", "CT变比"}: ct_ratio = value result: Dict[str, float] = {} if pt_ratio is not None: result["pt_ratio"] = pt_ratio if ct_ratio is not None: result["ct_ratio"] = ct_ratio return result def _get_line_transformer_ratios(self, line: LineData) -> Dict[str, float]: """确定当前线路使用的 PT/CT 变比。 优先级： 1) 配置文件中指定的变比（若存在）； 2) 由设备返回的 pri/sec 数据反算得到的变比（兜底）。 """ config_ratios = self._get_config_transformer_ratios(line.line_no) if config_ratios.get("pt_ratio") is not None and config_ratios.get("ct_ratio") is not None: return {"pt_ratio": float(config_ratios["pt_ratio"]), "ct_ratio": float(config_ratios["ct_ratio"])} pri = line.pri_val sec = line.sec_val pt_ratio_candidates = [ self._safe_ratio(pri.Ua, sec.Ua, 0.0), self._safe_ratio(pri.Ub, sec.Ub, 0.0), self._safe_ratio(pri.Uc, sec.Uc, 0.0), self._safe_ratio(pri.Uab, sec.Uab, 0.0), self._safe_ratio(pri.Ubc, sec.Ubc, 0.0), self._safe_ratio(pri.Uca, sec.Uca, 0.0), ] ct_ratio_candidates = [ self._safe_ratio(pri.Ia, sec.Ia, 0.0), self._safe_ratio(pri.Ib, sec.Ib, 0.0), self._safe_ratio(pri.Ic, sec.Ic, 0.0), ] pt_ratio = next((ratio for ratio in pt_ratio_candidates if ratio > 0), 1.0) ct_ratio = next((ratio for ratio in ct_ratio_candidates if ratio > 0), 1.0) if config_ratios.get("pt_ratio") is not None: pt_ratio = float(config_ratios["pt_ratio"]) if config_ratios.get("ct_ratio") is not None: ct_ratio = float(config_ratios["ct_ratio"]) return {"pt_ratio": pt_ratio, "ct_ratio": ct_ratio} def _build_primary_from_secondary(self, secondary: ValueGroup, pt_ratio: float, ct_ratio: float) -> ValueGroup: """根据二次值和变比计算一次值。 规则： - 电压一次值 = 电压二次值 * PT - 电流一次值 = 电流二次值 * CT - 功率一次值 = 功率二次值 * PT * CT - 频率一次值 = 频率二次值 - 功率因数一次值 = 功率因数二次值 """ power_ratio = pt_ratio * ct_ratio return ValueGroup( Ua=secondary.Ua * pt_ratio, Ub=secondary.Ub * pt_ratio, Uc=secondary.Uc * pt_ratio, Ia=secondary.Ia * ct_ratio, Ib=secondary.Ib * ct_ratio, Ic=secondary.Ic * ct_ratio, Pa=secondary.Pa * power_ratio, Pb=secondary.Pb * power_ratio, Pc=secondary.Pc * power_ratio, Pt=secondary.Pt * power_ratio, Qa=secondary.Qa * power_ratio, Qb=secondary.Qb * power_ratio, Qc=secondary.Qc * power_ratio, Qt=secondary.Qt * power_ratio, Sa=secondary.Sa * power_ratio, Sb=secondary.Sb * power_ratio, Sc=secondary.Sc * power_ratio, St=secondary.St * power_ratio, PFa=secondary.PFa, PFb=secondary.PFb, PFc=secondary.PFc, PFt=secondary.PFt, Uab=secondary.Uab * pt_ratio, Ubc=secondary.Ubc * pt_ratio, Uca=secondary.Uca * pt_ratio, frq=secondary.frq, ) def _round_value_group(self, group: ValueGroup) -> ValueGroup: group_data = group.model_dump() precision_map = { "Ua": 3, "Ub": 3, "Uc": 3, "Ia": 3, "Ib": 3, "Ic": 3, "Pa": 2, "Pb": 2, "Pc": 2, "Pt": 2, "Qa": 2, "Qb": 2, "Qc": 2, "Qt": 2, "Sa": 2, "Sb": 2, "Sc": 2, "St": 2, "PFa": 3, "PFb": 3, "PFc": 3, "PFt": 3, "Uab": 3, "Ubc": 3, "Uca": 3, "frq": 3, } for field_name, precision in precision_map.items(): value = group_data.get(field_name) if isinstance(value, (int, float)): group_data[field_name] = round(float(value), precision) return ValueGroup(**group_data) def _normalize_realtime_precision(self, realtime: RealtimeData) -> RealtimeData: """实时数据归一化：按配置变比计算一次值，并按界面要求控制小数位。""" line_list = [] for line in realtime.line_list: ratios = self._get_line_transformer_ratios(line) line_list.append( LineData( line_no=line.line_no, pri_val=self._round_value_group( self._build_primary_from_secondary( line.sec_val, ratios["pt_ratio"], ratios["ct_ratio"], ) ), sec_val=self._round_value_group(line.sec_val), ) ) ai_collect = { key: round(float(value), 2) if isinstance(value, (int, float)) else value for key, value in realtime.ai_collect.items() } return RealtimeData(line_list=line_list, switch=dict(realtime.switch), ai_collect=ai_collect) def get_realtime_data(self) -> RealtimeData: cached = memory_store.get_realtime() if cached is not None: normalized = self._normalize_realtime_precision(cached) memory_store.set_realtime(normalized) return normalized realtime = self.device_client.read_realtime_data() normalized = self._normalize_realtime_precision(realtime) memory_store.set_realtime(normalized) return normalized def get_device_status(self) -> DeviceStatus: cached = memory_store.get_status() if cached is not None: return cached status = self.device_client.read_device_status() memory_store.set_status(status) return status def startup_self_check(self) -> Dict[str, Any]: """开机自检：通过 03H 读取参数区，验证通讯链路。 若通讯失败，生成"开机自检（通讯链路）异常"报警事件， 并将设备状态置为"断开"，网络/串口状态清空。 返回值 ---- dict {"success": bool, "status": DeviceStatus} """ success = self.device_client.startup_self_check() if success: status = self.device_client.read_device_status() memory_store.set_status(status) abnormal_items = _collect_startup_status_abnormalities(status) if abnormal_items: alarm = AlarmEvent( event_time=datetime.now(), line_code=0, event_type=EVENT_TYPE_SELF_CHECK, event_code=SELF_CHECK_COMMUNICATION_ERROR_CODE, event_value=len(abnormal_items), content=f"开机自检状态异常：{', '.join(abnormal_items)}", ) alarm.id = self.alarm_repo.save_alarm(alarm) memory_store.push_alarm(alarm) return {"success": True, "status": status.model_dump()} # 通讯失败：生成报警事件 alarm = AlarmEvent( event_time=datetime.now(), line_code=0, event_type=EVENT_TYPE_SELF_CHECK, event_code=SELF_CHECK_COMMUNICATION_ERROR_CODE, event_value=0, content="开机自检（通讯链路）异常", ) alarm.id = self.alarm_repo.save_alarm(alarm) memory_store.push_alarm(alarm) # 设置状态为断开 status = DeviceStatus( self_check="断开", net1="断开", net2="断开", net3="断开", net4="断开", uart1="断开", uart2="断开", uart3="断开", uart4="断开", frequency_limit="未知", ct_break="未知", pt_break="未知", ) memory_store.set_status(status) return {"success": False, "status": status.model_dump()} def _default_device_config(self) -> Dict[str, Any]: return { "password": "", "hardware_version": { "board_version": "B001.001.001", "display_version": "S001.001.001", "other_version": "Y001.001.001", }, "software_version": { "display_program": "001.001.001", "communication_program": "001.001.001", "measurement_program": "001.001.001", }, "net": [ {"nic": "网卡一", "ip": "192.168.1.10", "mask": "255.255.255.0", "gateway": "192.168.1.1", "protocol": "Modbus TCP"}, {"nic": "网卡二", "ip": "192.168.1.56", "mask": "255.255.255.255", "gateway": "192.168.1.56", "protocol": "Modbus TCP"}, ], "uart": [ {"port": "COM1", "baud": 9600, "parity": "NONE", "data_bits": 8, "stop_bits": 1, "protocol": ""}, {"port": "COM2", "baud": 115200, "parity": "NONE", "data_bits": 8, "stop_bits": 1, "protocol": "Modbus RTU"}, ], } def _merge_keyed_items(self, defaults: List[Dict[str, Any]], current: Any, key_field: str) -> List[Dict[str, Any]]: merged = [dict(item) for item in defaults] if not isinstance(current, list): return merged index_map = { item.get(key_field): index for index, item in enumerate(merged) if isinstance(item, dict) and item.get(key_field) is not None } for item in current: if not isinstance(item, dict): continue key = item.get(key_field) if key in index_map: merged[index_map[key]].update(item) else: merged.append(dict(item)) return merged def _upsert_keyed_item(self, items: List[Dict[str, Any]], payload: Dict[str, Any], key_field: str) -> List[Dict[str, Any]]: key = payload.get(key_field) updated = False for index, item in enumerate(items): if item.get(key_field) == key: items[index] = payload updated = True break if not updated: items.append(payload) return items def _find_keyed_item(self, items: List[Dict[str, Any]], key_field: str, key_value: str) -> Optional[Dict[str, Any]]: for item in items: if item.get(key_field) == key_value: return item return None def _load_device_config_with_defaults(self) -> Dict[str, Any]: current = self.config_repo.read_device_config() defaults = self._default_device_config() data = { "password": current.get("password", "") if isinstance(current, dict) else "", "hardware_version": dict(defaults["hardware_version"]), "software_version": dict(defaults["software_version"]), "net": [], "uart": [], } if isinstance(current, dict): if isinstance(current.get("hardware_version"), dict): data["hardware_version"].update(current["hardware_version"]) if isinstance(current.get("software_version"), dict): data["software_version"].update(current["software_version"]) data["net"] = self._merge_keyed_items(defaults["net"], current.get("net"), "nic") data["uart"] = self._merge_keyed_items(defaults["uart"], current.get("uart"), "port") else: data["net"] = self._merge_keyed_items(defaults["net"], None, "nic") data["uart"] = self._merge_keyed_items(defaults["uart"], None, "port") return data def save_device_config(self, payload: DeviceConfigIn) -> Dict[str, Any]: data = self._load_device_config_with_defaults() payload_data = payload.model_dump() data["hardware_version"] = payload_data["hardware_version"] data["software_version"] = payload_data["software_version"] data["net"] = self._merge_keyed_items(data["net"], payload_data["net"], "nic") data["uart"] = self._merge_keyed_items(data["uart"], payload_data["uart"], "port") if payload.password.strip(): data["password"] = hash_password(payload.password) self._save_operation_event(event_code=8, detail="设备密码") if payload_data.get("net"): nics = [item.get("nic", "") for item in payload_data["net"] if isinstance(item, dict)] self._save_operation_event(event_code=7, detail=f"网卡: {', '.join(nics)}") if payload_data.get("uart"): ports = [item.get("port", "") for item in payload_data["uart"] if isinstance(item, dict)] self._save_operation_event(event_code=7, detail=f"串口: {', '.join(ports)}") path = self.config_repo.write_device_config(data) device_result = self.device_client.send_device_config(payload) return {"save_path": f"/config/{path.name}", **device_result} def save_device_password(self, payload: DevicePasswordUpdateIn) -> Dict[str, Any]: data = self._load_device_config_with_defaults() data["password"] = hash_password(payload.password) path = self.config_repo.write_device_config(data) self._save_operation_event(event_code=8, detail="设备密码") return {"save_path": f"/config/{path.name}", "target": "password", "send_status": "成功"} def get_device_config(self) -> Dict[str, Any]: data = self._load_device_config_with_defaults() # 不返回已保存的哈希密码，避免前端把哈希串直接显示到设置界面 data["password"] = "" return data def get_net_config(self, nic: str) -> Dict[str, Any]: return self.device_client.read_net_config(nic) def save_net_config(self, payload: NetConfigItem | List[NetConfigItem]) -> Dict[str, Any]: if isinstance(payload, list): nics = [item.nic for item in payload] results = [self.device_client.send_net_config(item) for item in payload] self._save_operation_event(event_code=7, detail=f"网卡: {', '.join(nics)}") return {"send_status": "成功", "target": "net_batch", "items": len(results), "results": results} result = self.device_client.send_net_config(payload) self._save_operation_event(event_code=7, detail=f"网卡: {payload.nic}") return result def get_uart_config(self, port: str) -> Dict[str, Any]: device_config = self._load_device_config_with_defaults() item = self._find_keyed_item(device_config["uart"], "port", port) if item is not None: return item return {"port": port, "baud": 9600, "parity": "NONE", "data_bits": 8, "stop_bits": 1, "protocol": ""} def save_uart_config(self, payload: UartConfigItem | List[UartConfigItem]) -> Dict[str, Any]: device_config = self._load_device_config_with_defaults() if isinstance(payload, list): results = [] ports = [] for item in payload: device_config["uart"] = self._upsert_keyed_item(device_config["uart"], item.model_dump(), "port") ports.append(item.port) results.append({"target": "uart", "port": item.port, "send_status": "成功"}) path = self.config_repo.write_device_config(device_config) self._save_operation_event(event_code=7, detail=f"串口: {', '.join(ports)}") return { "save_path": f"/config/{path.name}", "target": "uart_batch", "items": len(results), "send_status": "成功", "results": results, } device_config["uart"] = self._upsert_keyed_item(device_config["uart"], payload.model_dump(), "port") path = self.config_repo.write_device_config(device_config) self._save_operation_event(event_code=7, detail=f"串口: {payload.port}") return {"save_path": f"/config/{path.name}", "target": "uart", "port": payload.port, "send_status": "成功"} def save_channel_config(self, payload: ChannelConfigIn) -> Dict[str, Any]: device_result = self.device_client.send_channel_config(payload) if payload.ai_channel: ai_chs = [str(item.ch) for item in payload.ai_channel] self._save_operation_event(event_code=5, detail=f"通道: {', '.join(ai_chs)}") if payload.ao_channel: ao_chs = [str(item.ch) for item in payload.ao_channel] self._save_operation_event(event_code=6, detail=f"通道: {', '.join(ao_chs)}") return device_result def get_channel_config(self) -> Dict[str, Any]: return self.device_client.read_channel_config() def _default_line_alarm_setting(self, line_no: int) -> Dict[str, Any]: return { "line_no": line_no, "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": 180, "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": 1.0, "enabled": False}, {"category": "CT变化", "value": 1.0, "enabled": False}, ], } def _normalize_line_alarm_settings(self, raw: Any) -> List[Dict[str, Any]]: if isinstance(raw, list): return [item for item in raw if isinstance(item, dict)] if isinstance(raw, dict): return [raw] return [] def save_line_alarm_setting(self, payload: LineAlarmSettingIn) -> Dict[str, Any]: device_result = self.device_client.send_line_alarm_setting(payload) self._save_operation_event(event_code=3, line_code=payload.line_no) return device_result def get_line_alarm_setting(self, line_no: int = 1) -> Dict[str, Any]: merged = self._default_line_alarm_setting(line_no) try: current = self.device_client.read_line_alarm_setting(line_no) except Exception as exc: self._save_operation_event(event_code=2, line_code=line_no, detail=f"线路{line_no}：{exc}") current = None if isinstance(current, dict): merged.update(current) normalized_transformer: List[Dict[str, Any]] = [] for transformer in merged.get("transformer_change", []): if not isinstance(transformer, dict): continue category = transformer.get("category", "") enabled = bool(transformer.get("enabled", False)) raw_value = transformer.get("value", transformer.get("limit", 1.0)) try: value = float(raw_value) except (TypeError, ValueError): value = 1.0 normalized_transformer.append({"category": category, "value": value, "enabled": enabled}) merged["transformer_change"] = normalized_transformer return merged def save_ai_alarm_setting(self, payload: List[AiAlarmSettingIn]) -> Dict[str, Any]: payload_data = [item.model_dump() for item in payload] result = self.device_client.send_ai_alarm_setting(payload_data) chs = [str(item.channel_no) for item in payload] self._save_operation_event(event_code=4, detail=f"通道: {', '.join(chs)}") return result def get_ai_alarm_setting(self) -> List[Dict[str, Any]]: return self.device_client.read_ai_alarm_setting() def save_system_config(self, payload: SystemConfigIn) -> Dict[str, Any]: return self.device_client.send_system_config(payload) def get_system_config(self) -> Dict[str, Any]: return self.device_client.read_system_config() def save_time_sync_config(self, payload: TimeSyncConfigIn) -> Dict[str, Any]: return self.device_client.send_time_sync_config(payload) def get_time_sync_config(self) -> Dict[str, Any]: return self.device_client.read_time_sync_config() def list_alarms( self, page: int, size: int, line_code: Optional[int] = None, event_type: Optional[int] = None, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, ) -> List[Dict[str, Any]]: return self.alarm_repo.list_alarms( page=page, size=size, line_code=line_code, event_type=event_type, start_time=start_time, end_time=end_time, ) def _save_operation_event( self, event_code: int, line_code: int = 0, event_value: int = 0, detail: str = "", ) -> AlarmEvent: """通用操作事件生成保存方法（同步）。 将主站侧操作类事件保存到 SQLite 并推入内存缓存。 返回保存后的 AlarmEvent 对象（含数据库自增 id），调用方可进一步广播。 参数 ---- event_code : int 操作事件代码，取值 1~8，对应含义见 OPERATION_EVENT_CODE line_code : int 关联线路编号，0 表示装置级 event_value : int 事件附带数值，如分合闸通道号 detail : str 附加描述（如网卡名、端口号、参数名等） 返回值 ---- AlarmEvent 已保存的事件对象（id 已回填） """ content = _generate_operation_event_content(event_code, line_code, detail) alarm = AlarmEvent( event_time=datetime.now(), line_code=line_code, event_type=EVENT_TYPE_OPERATION, event_code=event_code, event_value=event_value, content=content, ) alarm.id = self.alarm_repo.save_alarm(alarm) memory_store.push_alarm(alarm) return alarm def switch_control(self, payload: SwitchControlIn) -> Dict[str, Any]: result = self.device_client.send_switch_control(payload) action_label = "合闸" if payload.action == 1 else "分闸" self._save_operation_event( event_code=1, event_value=payload.ch, detail=f"CH{payload.ch} {action_label}", ) return result async def poll_time_sync_once(self) -> None: time_sync = self.device_client.read_time_sync_config() await ws_manager.broadcast("time-sync", {"type": "time_sync", "data": time_sync}) async def poll_device_once(self) -> None: realtime = self._normalize_realtime_precision(self.device_client.read_realtime_data()) status = self.device_client.read_device_status() alarms = self.device_client.read_alarm_events() memory_store.set_realtime(realtime) memory_store.set_status(status) await ws_manager.broadcast("real-time", {"type": "real_time", "data": realtime.model_dump()}) await ws_manager.broadcast("status", {"type": "status", "data": status.model_dump()}) for alarm in alarms: alarm.id = self.alarm_repo.save_alarm(alarm) memory_store.push_alarm(alarm) await ws_manager.broadcast("alarm", alarm.model_dump(mode="json")) platform_service = PlatformService()