使用 OPC UA 检测和转换自定义方法输入参数

from asyncua.ua import ObjectIds
from asyncua import ua

base_type_list = {
    ua.NodeId(ObjectIds.Null): None,
    ua.NodeId(ObjectIds.Boolean): bool,
    ua.NodeId(ObjectIds.SByte): ua.SByte,
    ua.NodeId(ObjectIds.Byte): ua.Byte,
    ua.NodeId(ObjectIds.Int16): ua.Int16,
    ua.NodeId(ObjectIds.UInt16): ua.UInt16,
    ua.NodeId(ObjectIds.Int32): ua.Int32,
    ua.NodeId(ObjectIds.UInt32): ua.UInt32,
    ua.NodeId(ObjectIds.Int64): ua.Int64,
    ua.NodeId(ObjectIds.UInt64): ua.UInt64,
    ua.NodeId(ObjectIds.Float): ua.Float,
    ua.NodeId(ObjectIds.Double): ua.Double,
    ua.NodeId(ObjectIds.String): ua.String,
    ua.NodeId(ObjectIds.DateTime): ua.DateTime,
    ua.NodeId(ObjectIds.Guid): ua.Guid,
    ua.NodeId(ObjectIds.ByteString): ua.ByteString,
    ua.NodeId(ObjectIds.XmlElement): ua.XmlElement,
    ua.NodeId(ObjectIds.NodeId): ua.NodeId,
    ua.NodeId(ObjectIds.ExpandedNodeId): ua.ExpandedNodeId,
    ua.NodeId(ObjectIds.StatusCode): ua.StatusCode,
    ua.NodeId(ObjectIds.QualifiedName): ua.QualifiedName,
    ua.NodeId(ObjectIds.LocalizedText): ua.LocalizedText,
    ua.NodeId(ObjectIds.Structure): ua.ExtensionObject,
    ua.NodeId(ObjectIds.DataValue): ua.DataValue,
    ua.NodeId(ObjectIds.BaseVariableType): ua.Variant,
    ua.NodeId(ObjectIds.DiagnosticInfo): ua.DiagnosticInfo,
}

def get_class_from_nodeid(datatype_nodeid):
   """
   从 NodeId 获取对应的 Python 类。

   Args:
       datatype_nodeid (ua.NodeId): 数据类型的 NodeId。

   Returns:
       type: 对应的 Python 类，如果找不到则返回 None。
   """
   # 检查所有类型字典
   sources = [base_type_list, ua.basetype_by_datatype, ua.extension_objects_by_datatype, ua.enums_by_datatype]
   cls = None
   for src in sources:
       cls = src.get(datatype_nodeid, None)
       if cls is not None:
          return cls
   return cls

# 假设已经连接到 OPC UA 服务器并获取了 MyMethod 节点的句柄 method_node

# 读取 "0:InputArguments" 属性
input_arguments = await method_node.read_value()

# 遍历参数
args = []
for arg in input_arguments:
    data_type_nodeid = arg.dataType
    cls = get_class_from_nodeid(data_type_nodeid)

    if cls is not None:
        # 假设第一个参数是 MyDataType，第二个参数是整数
        if arg == input_arguments[0]:
            # 创建 MyDataType 实例 (假设 MyDataType 有 ID, Value1, Value2 属性)
            param_value = cls(ID=99, Value1=100, Value2=199)
        else:
            param_value = 10 # 第二个参数是整数
        args.append(param_value)
    else:
        print(f"未找到数据类型 {data_type_nodeid} 对应的 Python 类")

# 调用方法
try:
    result = await method_node.call_method(*args)
    print(f"方法调用结果: {result}")
except Exception as e:
    print(f"方法调用失败: {e}")