本文将深入探讨如何在使用 OPC UA 客户端（例如 asyncua）调用自定义方法时，解决动态检测和转换输入参数数据类型的问题。核心在于理解如何利用 OPC UA 协议提供的元数据信息，动态地获取方法参数的类型，并将其转换为客户端可用的 Python 类，从而实现灵活且类型安全的方法调用。
在 OPC UA 中，方法节点的 "0:InputArguments" 属性包含了方法参数的详细信息，包括参数名称、数据类型等。通过读取这个属性的值，我们可以获得一个 Argument 对象的列表，每个对象描述一个输入参数。
以下代码展示了如何从 Argument 对象中提取数据类型，并将其转换为相应的 Python 类：
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使用方法：
- 读取 "0:InputArguments" 属性： 使用 read_value() 方法读取目标方法的 "0:InputArguments" 属性。这将返回一个 Argument 对象的列表。
- 提取数据类型 NodeId： 遍历 Argument 列表，对于每个参数，提取其 dataType 属性，该属性是一个 NodeId 对象，表示参数的数据类型。
- 转换为 Python 类： 使用 get_class_from_nodeid() 函数将 NodeId 转换为对应的 Python 类。
- 创建参数实例： 使用转换后的 Python 类，根据实际的参数值创建参数实例。
- 调用方法： 将创建的参数实例作为参数传递给 node.call_method() 方法。
示例：
假设我们有一个名为 MyMethod 的方法，它接受一个自定义数据类型 MyDataType 和一个整数作为输入参数。
# 假设已经连接到 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}")注意事项：
- get_class_from_nodeid() 函数需要维护一个包含所有可能的数据类型 NodeId 到 Python 类映射的字典。确保该字典包含了你所使用的自定义数据类型。
- 对于复杂的数据类型（例如结构体），可能需要根据数据类型的定义，手动创建相应的 Python 类。
- 错误处理至关重要。确保捕获可能发生的异常，例如未找到数据类型对应的 Python 类，或者方法调用失败等。
总结：
通过读取方法的 "0:InputArguments" 属性，并使用 get_class_from_nodeid() 函数将数据类型 NodeId 转换为 Python 类，我们可以实现动态地检测和转换 OPC UA 方法的输入参数。这种方法提高了客户端的灵活性和可扩展性，使其能够处理各种自定义数据类型，并调用不同的方法。 重要的是要维护一个完整的数据类型映射字典，并进行适当的错误处理，以确保程序的稳定性和可靠性。
