Uni-Lab-OS/unilabos/device_comms/opcua_client/node/uniopcua.py

# coding=utf-8
from enum import Enum
from abc import ABC, abstractmethod
from typing import Tuple, Union, Optional, Any, List

from opcua import Client, Node, ua
from opcua.ua import NodeId, NodeClass, VariantType


class DataType(Enum):
    BOOLEAN = VariantType.Boolean
    SBYTE = VariantType.SByte
    BYTE = VariantType.Byte
    INT16 = VariantType.Int16
    UINT16 = VariantType.UInt16
    INT32 = VariantType.Int32
    UINT32 = VariantType.UInt32
    INT64 = VariantType.Int64
    UINT64 = VariantType.UInt64
    FLOAT = VariantType.Float
    DOUBLE = VariantType.Double
    STRING = VariantType.String
    DATETIME = VariantType.DateTime
    BYTESTRING = VariantType.ByteString


class NodeType(Enum):
    VARIABLE = NodeClass.Variable
    OBJECT = NodeClass.Object
    METHOD = NodeClass.Method
    OBJECTTYPE = NodeClass.ObjectType
    VARIABLETYPE = NodeClass.VariableType
    REFERENCETYPE = NodeClass.ReferenceType
    DATATYPE = NodeClass.DataType
    VIEW = NodeClass.View


class Base(ABC):
    def __init__(self, client: Client, name: str, node_id: str, typ: NodeType, data_type: DataType):
        self._node_id: str = node_id
        self._client = client
        self._name = name
        self._type = typ
        self._data_type = data_type
        self._node: Optional[Node] = None

    def _get_node(self) -> Node:
        if self._node is None:
            try:
                # 尝试多种 NodeId 字符串格式解析，兼容不同服务器/库的输出
                # 可能的格式示例: 'ns=2;i=1234', 'ns=2;s=SomeString',
                # 'StringNodeId(ns=4;s=OPC|变量名)', 'NumericNodeId(ns=2;i=1234)' 等
                import re

                nid = self._node_id
                # 如果已经是 NodeId/Node 对象（库用户可能传入），直接使用
                try:
                    from opcua.ua import NodeId as UaNodeId
                    if isinstance(nid, UaNodeId):
                        self._node = self._client.get_node(nid)
                        return self._node
                except Exception:
                    # 若导入或类型判断失败，则继续下一步
                    pass

                # 直接以字符串形式处理
                if isinstance(nid, str):
                    nid = nid.strip()

                    # 处理包含类名的格式，如 'StringNodeId(ns=4;s=...)' 或 'NumericNodeId(ns=2;i=...)'
                    # 提取括号内的内容
                    match_wrapped = re.match(r'(String|Numeric|Byte|Guid|TwoByteNode|FourByteNode)NodeId\((.*)\)', nid)
                    if match_wrapped:
                        # 提取括号内的实际 node_id 字符串
                        nid = match_wrapped.group(2).strip()

                    # 常见短格式 'ns=2;i=1234' 或 'ns=2;s=SomeString'
                    if re.match(r'^ns=\d+;[is]=', nid):
                        self._node = self._client.get_node(nid)
                    else:
                        # 尝试提取 ns 和 i 或 s
                        # 对于字符串标识符，可能包含特殊字符，使用非贪婪匹配
                        m_num = re.search(r'ns=(\d+);i=(\d+)', nid)
                        m_str = re.search(r'ns=(\d+);s=(.+?)(?:\)|$)', nid)
                        if m_num:
                            ns = int(m_num.group(1))
                            identifier = int(m_num.group(2))
                            node_id = NodeId(identifier, ns)
                            self._node = self._client.get_node(node_id)
                        elif m_str:
                            ns = int(m_str.group(1))
                            identifier = m_str.group(2).strip()
                            # 对于字符串标识符，直接使用字符串格式
                            node_id_str = f"ns={ns};s={identifier}"
                            self._node = self._client.get_node(node_id_str)
                        else:
                            # 回退：尝试直接传入字符串（有些实现接受其它格式）
                            try:
                                self._node = self._client.get_node(self._node_id)
                            except Exception as e:
                                # 输出更详细的错误信息供调试
                                print(f"获取节点失败(尝试直接字符串): {self._node_id}, 错误: {e}")
                                raise
                else:
                    # 非字符串，尝试直接使用
                    self._node = self._client.get_node(self._node_id)
            except Exception as e:
                print(f"获取节点失败: {self._node_id}, 错误: {e}")
                # 添加额外提示，帮助定位 BadNodeIdUnknown 问题
                print("提示: 请确认该 node_id 是否来自当前连接的服务器地址空间，" \
                      "以及 CSV/配置中名称与服务器 BrowseName 是否匹配。")
                raise
        return self._node

    @abstractmethod
    def read(self) -> Tuple[Any, bool]:
        """读取节点值，返回(值, 是否出错)"""
        pass

    @abstractmethod
    def write(self, value: Any) -> bool:
        """写入节点值，返回是否出错"""
        pass

    @property
    def type(self) -> NodeType:
        return self._type

    @property
    def node_id(self) -> str:
        return self._node_id

    @property
    def name(self) -> str:
        return self._name


class Variable(Base):
    def __init__(self, client: Client, name: str, node_id: str, data_type: DataType):
        super().__init__(client, name, node_id, NodeType.VARIABLE, data_type)

    def read(self) -> Tuple[Any, bool]:
        try:
            value = self._get_node().get_value()
            return value, False
        except Exception as e:
            print(f"读取变量 {self._name} 失败: {e}")
            return None, True

    def write(self, value: Any) -> bool:
        try:
            # 如果声明了数据类型，则尝试转换并使用对应的 Variant 写入
            coerced = value
            try:
                if self._data_type is not None:
                    # 基于声明的数据类型做简单类型转换
                    dt = self._data_type
                    if dt in (DataType.SBYTE, DataType.BYTE, DataType.INT16, DataType.UINT16,
                              DataType.INT32, DataType.UINT32, DataType.INT64, DataType.UINT64):
                        # 数值类型 -> int
                        if isinstance(value, str):
                            coerced = int(value)
                        else:
                            coerced = int(value)
                    elif dt in (DataType.FLOAT, DataType.DOUBLE):
                        if isinstance(value, str):
                            coerced = float(value)
                        else:
                            coerced = float(value)
                    elif dt == DataType.BOOLEAN:
                        if isinstance(value, str):
                            v = value.strip().lower()
                            if v in ("true", "1", "yes", "on"):
                                coerced = True
                            elif v in ("false", "0", "no", "off"):
                                coerced = False
                            else:
                                coerced = bool(value)
                        else:
                            coerced = bool(value)
                    elif dt == DataType.STRING or dt == DataType.BYTESTRING or dt == DataType.DATETIME:
                        coerced = str(value)

                    # 使用 ua.Variant 明确指定 VariantType
                    try:
                        variant = ua.Variant(coerced, dt.value)
                        self._get_node().set_value(variant)
                    except Exception:
                        # 回退：有些 set_value 实现接受 (value, variant_type)
                        try:
                            self._get_node().set_value(coerced, dt.value)
                        except Exception:
                            # 最后回退到直接写入（保持兼容性）
                            self._get_node().set_value(coerced)
                else:
                    # 未声明数据类型，直接写入
                    self._get_node().set_value(value)
            except Exception:
                # 若在转换或按数据类型写入失败，尝试直接写入原始值并让上层捕获错误
                self._get_node().set_value(value)

            return False
        except Exception as e:
            print(f"写入变量 {self._name} 失败: {e}")
            return True


class Method(Base):
    def __init__(self, client: Client, name: str, node_id: str, parent_node_id: str, data_type: DataType):
        super().__init__(client, name, node_id, NodeType.METHOD, data_type)
        self._parent_node_id = parent_node_id
        self._parent_node = None

    def _get_parent_node(self) -> Node:
        if self._parent_node is None:
            try:
                # 处理父节点ID，使用与_get_node相同的解析逻辑
                import re

                nid = self._parent_node_id

                # 如果已经是 NodeId 对象，直接使用
                try:
                    from opcua.ua import NodeId as UaNodeId
                    if isinstance(nid, UaNodeId):
                        self._parent_node = self._client.get_node(nid)
                        return self._parent_node
                except Exception:
                    pass

                # 字符串处理
                if isinstance(nid, str):
                    nid = nid.strip()

                    # 处理包含类名的格式
                    match_wrapped = re.match(r'(String|Numeric|Byte|Guid|TwoByteNode|FourByteNode)NodeId\((.*)\)', nid)
                    if match_wrapped:
                        nid = match_wrapped.group(2).strip()

                    # 常见短格式
                    if re.match(r'^ns=\d+;[is]=', nid):
                        self._parent_node = self._client.get_node(nid)
                    else:
                        # 提取 ns 和 i 或 s
                        m_num = re.search(r'ns=(\d+);i=(\d+)', nid)
                        m_str = re.search(r'ns=(\d+);s=(.+?)(?:\)|$)', nid)
                        if m_num:
                            ns = int(m_num.group(1))
                            identifier = int(m_num.group(2))
                            node_id = NodeId(identifier, ns)
                            self._parent_node = self._client.get_node(node_id)
                        elif m_str:
                            ns = int(m_str.group(1))
                            identifier = m_str.group(2).strip()
                            node_id_str = f"ns={ns};s={identifier}"
                            self._parent_node = self._client.get_node(node_id_str)
                        else:
                            # 回退
                            self._parent_node = self._client.get_node(self._parent_node_id)
                else:
                    self._parent_node = self._client.get_node(self._parent_node_id)
            except Exception as e:
                print(f"获取父节点失败: {self._parent_node_id}, 错误: {e}")
                raise
        return self._parent_node

    def read(self) -> Tuple[Any, bool]:
        """方法节点不支持读取操作"""
        return None, True

    def write(self, value: Any) -> bool:
        """方法节点不支持写入操作"""
        return True

    def call(self, *args) -> Tuple[Any, bool]:
        """调用方法，返回(返回值, 是否出错)"""
        try:
            result = self._get_parent_node().call_method(self._get_node(), *args)
            return result, False
        except Exception as e:
            print(f"调用方法 {self._name} 失败: {e}")
            return None, True


class Object(Base):
    def __init__(self, client: Client, name: str, node_id: str):
        super().__init__(client, name, node_id, NodeType.OBJECT, None)

    def read(self) -> Tuple[Any, bool]:
        """对象节点不支持直接读取操作"""
        return None, True

    def write(self, value: Any) -> bool:
        """对象节点不支持直接写入操作"""
        return True

    def get_children(self) -> Tuple[List[Node], bool]:
        """获取子节点列表，返回(子节点列表, 是否出错)"""
        try:
            children = self._get_node().get_children()
            return children, False
        except Exception as e:
            print(f"获取对象 {self._name} 的子节点失败: {e}")
            return [], True