如何通过一次调用访问多个OPC-UA节点?报错排查与方案咨询
问题解决与优化方案
一、修复当前报错的步骤
1. 修正客户端节点路径错误
客户端代码中错误地尝试访问0:MyObject节点,但服务器端并未创建该节点——所有变量和方法都直接挂载在Objects节点下。修改客户端获取方法节点的代码:
# 替换原sum_method获取逻辑 sum_method = objects_node.get_child(["http://example.opcua.server:sum"])
或者通过遍历子节点查找:
children = objects_node.get_children() for child in children: if child.get_browse_name().Name == "sum": sum_method = child break
2. 修正服务器端方法的变量作用域
服务器端的sum_method直接引用外部变量node1和node2,需要声明全局变量避免未定义错误:
def sum_method(parent): global node1, node2 # 添加全局变量声明 val1 = node1.get_value() val2 = node2.get_value() return [ua.Variant(val1 + val2, ua.VariantType.Double)]
二、更优方案:原生批量读取节点值
OPC-UA规范内置了批量读取多节点值的功能,比自定义方法更高效、更符合标准,无需额外开发方法。
服务器端(无需改动原代码)
保持原服务器代码不变,正常暴露Node1、Node2即可。
客户端批量读取示例
from opcua import Client if __name__ == "__main__": client = Client("opc.tcp://localhost:4840/freeopcua/server/") try: client.connect() print("Client connected") objects_node = client.get_objects_node() # 获取目标节点引用 node1 = objects_node.get_child(["http://example.opcua.server:Node1"]) node2 = objects_node.get_child(["http://example.opcua.server:Node2"]) # 单次调用读取多个节点值 values = client.get_values([node1, node2]) sum_val = values[0] + values[1] print(f"Node1值: {values[0]}, Node2值: {values[1]}, 总和: {sum_val}") finally: client.disconnect() print("Client disconnected")
三、自定义方法返回多节点值的示例(若需业务定制)
如果确实需要通过自定义方法返回一组节点值(如文件夹下所有节点),可以修改服务器端方法返回数组类型的Variant:
服务器端代码
from opcua import ua, Server def get_multiple_nodes(parent): global node1, node2 val1 = node1.get_value() val2 = node2.get_value() # 返回Double类型数组 return [ua.Variant([val1, val2], ua.VariantType.Double)] if __name__ == "__main__": server = Server() server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/") uri = "http://example.opcua.server" idx = server.register_namespace(uri) objects = server.get_objects_node() node1 = objects.add_variable(idx, "Node1", 5.5) node2 = objects.add_variable(idx, "Node2", 10.5) node1.set_writable() node2.set_writable() # 添加返回多节点值的方法 objects.add_method(idx, "GetMultipleNodes", get_multiple_nodes, [], [ua.VariantType.Double]) server.start() try: input("Server is running. Press Enter to stop...") finally: server.stop()
客户端调用示例
from opcua import Client if __name__ == "__main__": client = Client("opc.tcp://localhost:4840/freeopcua/server/") try: client.connect() print("Client connected") objects_node = client.get_objects_node() get_nodes_method = objects_node.get_child(["http://example.opcua.server:GetMultipleNodes"]) # 调用方法获取多节点值列表 result = get_nodes_method.call() print(f"获取的节点值列表: {result}") finally: client.disconnect() print("Client disconnected")
内容的提问来源于stack exchange,提问作者thzu
相关产品推荐
相关产品推荐

