如何用asyncua(Python)检查OPC UA服务器中特定节点是否存在

核心逻辑 #

针对客户端离线后服务器可能保留或重置节点的场景，实现以下流程：

• 连接OPC UA服务器
• 检查指定名称的节点是否存在
• 存在则直接复用，不存在则创建对应对象容器与变量节点
• 向节点写入数据

完整实现代码 #

import asyncio
from asyncua import Client
from asyncua.ua import ObjectIds, Variant, VariantType

async def check_or_create_node(server_url: str, parent_node_path: str, node_name: str):
    async with Client(url=server_url) as client:
        # 解析父节点路径，获取父节点对象
        parent_node = await client.nodes.root.get_child(parent_node_path.split(":"))
        
        # 拼接目标节点的完整路径
        target_node_full_path = parent_node_path.split(":") + [node_name]
        try:
            # 尝试获取目标节点
            target_node = await client.nodes.root.get_child(target_node_full_path)
            print(f"节点 {node_name} 已存在，直接使用")
        except Exception:
            # 节点不存在，先创建业务对象容器（可选，根据业务需求调整）
            business_obj = await parent_node.add_object(ObjectIds.BaseObjectType, f"{node_name}_Container")
            # 在对象容器下创建可读写的变量节点
            target_node = await business_obj.add_variable(
                ObjectIds.BaseDataVariableType, 
                node_name, 
                Variant(0, VariantType.Int32)  # 初始值与数据类型
            )
            # 设置变量节点为可写状态
            await target_node.set_writable()
            print(f"节点 {node_name} 不存在，已完成对象容器与变量节点的创建")
        
        # 写入业务数据（这里以Int32类型的100为例）
        await target_node.write_value(Variant(100, VariantType.Int32))
        print(f"成功向节点 {node_name} 写入数据: 100")
        
        return target_node

if __name__ == "__main__":
    # 配置服务器地址、父节点路径与目标节点名称
    OPCUA_SERVER_URL = "opc.tcp://localhost:4840/freeopcua/server/"
    PARENT_NODE_PATH = "0:Objects"  # 默认的Objects节点，通常所有业务节点都挂载在此
    TARGET_NODE_NAME = "ProductionData"
    
    asyncio.run(check_or_create_node(OPCUA_SERVER_URL, PARENT_NODE_PATH, TARGET_NODE_NAME))

关键细节说明 #

• 连接管理：使用async with Client自动处理连接的建立与关闭，避免手动管理连接生命周期的疏漏
• 节点存在性判断：通过get_child方法尝试获取节点，捕获异常即可判断节点不存在（无需额外的查询API）
• 节点创建规则：先创建对象节点作为业务容器，再在其下创建变量节点，符合OPC UA的节点组织规范，便于后续扩展
• 可写设置：新创建的变量节点默认只读，必须调用set_writable()才能写入数据
• 适配业务场景：
  • 服务器保留节点时：客户端重新上线会直接复用原有节点，无需重复创建
  • 服务器重置节点时：客户端会自动重新创建节点并写入初始数据

内容的提问来源于stack exchange，提问作者Samuel