OPC UA 方法调用：动态检测与转换自定义输入参数

from asyncua import Client
from asyncua.ua import ObjectIds

async def get_method_input_arguments(client, method_node):
    """
    获取 OPC UA 方法的输入参数信息。

    Args:
        client: asyncua.Client 实例。
        method_node: 方法节点的 Node 对象。

    Returns:
        一个包含 Argument 对象的列表。
    """
    input_arguments_node = await client.get_node(method_node.nodeid.to_string() + ".InputArguments")
    arguments = await input_arguments_node.read_value()
    return arguments

# 示例用法
async def main():
    async with Client("opc.tcp://your_opcua_server:4840") as client:
        # 假设 method_node 是你要调用的方法节点
        method_node = await client.get_node("ns=2;i=1234")  # 替换为你的方法节点 ID
        arguments = await get_method_input_arguments(client, method_node)
        for arg in arguments:
            print(f"参数名称: {arg.Name}, 数据类型: {arg.DataType}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

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):

   # check all type dicts   
   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

async def call_method_with_dynamic_args(client, method_node, arg_values):
    """
    使用动态参数调用 OPC UA 方法。

    Args:
        client: asyncua.Client 实例。
        method_node: 方法节点的 Node 对象。
        arg_values: 参数值的列表。

    Returns:
        方法调用的结果。
    """
    arguments = await get_method_input_arguments(client, method_node)
    args = []
    for i, arg in enumerate(arguments):
        data_type_class = get_class_from_nodeid(arg.DataType)
        if data_type_class is None:
            raise ValueError(f"不支持的数据类型: {arg.DataType}")

        # 将参数值转换为正确的类型
        try:
            converted_value = data_type_class(arg_values[i]) if data_type_class != bool else bool(arg_values[i]) # bool类型转换需要特殊处理
        except Exception as e:
            raise ValueError(f"参数转换失败: {e}")

        args.append(converted_value)

    # 调用方法
    result = await method_node.call_method(*args)
    return result

# 示例用法
async def main():
    async with Client("opc.tcp://your_opcua_server:4840") as client:
        method_node = await client.get_node("ns=2;i=1234")  # 替换为你的方法节点 ID
        arg_values = [99, 100, 199]  # 替换为你的参数值
        try:
            result = await call_method_with_dynamic_args(client, method_node, arg_values)
            print(f"方法调用结果: {result}")
        except Exception as e:
            print(f"方法调用失败: {e}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())