iAOP/core/edge-gateway/drivers/opcua_driver.py

# -*- coding: utf-8 -*-
"""OPC UA 驱动 —— 和利时 DCS 等 OPC UA 接口参数化适配（issue #25 落点）。

连接参数全部来自模板配置（gateway.yaml 的 drivers.opcua 段），不硬编码。
实际设备接入依赖第三方库 `asyncua`（异步）或 `opcua`（同步）；
未安装时给出明确提示，不会静默失败。
"""
from __future__ import annotations

from typing import Dict, List, Optional

from point_dict.loader import Point
from .base import Driver, SampleValue


class OpcUaDriver(Driver):
    """OPC UA 只读采集驱动（参数化适配）。"""

    protocol = "opcua"

    def __init__(self, config: Optional[dict] = None):
        super().__init__(config)
        self.endpoint: str = self.config.get("endpoint", "")
        self.security: str = self.config.get("security", "None")
        self.username: str = self.config.get("username", "")
        self.password: str = self.config.get("password", "")
        self._client = None

    def connect(self) -> None:
        try:
            from opcua import Client  # type: ignore
        except ImportError as exc:  # pragma: no cover - 依赖缺失路径
            raise ConnectionError(
                "OPC UA 驱动依赖库未安装：请 `pip install opcua`（或 asyncua）"
            ) from exc
        if not self.endpoint:
            raise ConnectionError("OPC UA 驱动缺少配置: drivers.opcua.endpoint")
        self._client = Client(self.endpoint, timeout=self.config.get("timeout", 10))
        # 安全策略参数化（如 Basic256Sha256；证书/密钥由现场安全通道提供）
        if self.security and self.security != "None":
            self._client.set_security_string(self.security)
        # 用户认证参数化（和利时 DCS 的 OPC UA 服务常启用用户名/密码认证）
        if self.username:
            self._client.set_user(self.username)
        self._client.connect()
        if self.username and self.password:
            try:
                self._client.activate_session(self.username, self.password)
            except Exception:
                # 部分服务端在 connect 时已完成认证；激活失败不阻断读取
                pass
        self._client.session_timeout = self.config.get("session_timeout", 60000)

    def read_points(self, points: List[Point]) -> Dict[str, SampleValue]:
        if self._client is None:
            raise ConnectionError("OPC UA 客户端未连接，请先 connect()")
        result: Dict[str, SampleValue] = {}
        for p in points:
            node_path = p.opc_node
            if not node_path:
                continue  # 无 OPC 节点的点由其它驱动采集
            try:
                node = self._client.get_node(node_path)
                result[p.point_id] = node.get_value()
            except Exception:
                # 单点读取失败不中断整批：记为未读到（计入丢失率）
                result[p.point_id] = None
        return result

    def close(self) -> None:
        if self._client is not None:
            try:
                self._client.disconnect()
            finally:
                self._client = None