4. OPC UA客户端数据订阅

任务说明

本案例实现一个OPC UA客户端，连接指定的服务器，订阅一个或多个节点的数据变化，并在数据变化时实时打印新值。案例使用Python与asyncua库完成，目标是让读者掌握OPC UA订阅机制的最小可运行实现。

环境依赖

需要Python 3.8+，并安装OPC UA异步库：

pip install asyncua

完整实现代码

import asyncio
import sys
from asyncua import Client

# OPC UA服务器地址（根据实际服务器修改）
SERVER_URL = "opc.tcp://localhost:4840"
# 要订阅的节点ID列表（根据实际服务器节点修改）
NODE_IDS = [
    "ns=2;i=2",   # 示例：模拟量
    "ns=2;i=3",   # 示例：开关量
]

async def subscribe_data_change():
    """连接服务器并订阅指定节点的数据变化"""
    client = Client(SERVER_URL)
    try:
        # 建立连接
        await client.connect()
        print(f"已连接服务器: {SERVER_URL}")

        # 获取节点对象
        nodes = [client.get_node(node_id) for node_id in NODE_IDS]

        # 创建订阅
        subscription = await client.create_subscription(100, handler)  # 100ms发布间隔
        # 订阅节点数据变化
        for node in nodes:
            await subscription.subscribe_data_change(node)
            print(f"已订阅节点: {node}")

        # 保持运行，等待数据变化回调
        print("等待数据变化（Ctrl+C退出）...")
        while True:
            await asyncio.sleep(1)

    except KeyboardInterrupt:
        print("\n用户中断，退出订阅")
    except Exception as e:
        print(f"发生错误: {e}")
    finally:
        # 断开连接
        await client.disconnect()
        print("已断开连接")

class SubHandler:
    """数据变化回调处理类"""
    def datachange_notification(self, node, val, data):
        """当订阅的节点数据变化时，此方法被调用"""
        print(f"[数据变化] 节点: {node}, 新值: {val}")

# 全局回调实例
handler = SubHandler()

if __name__ == "__main__":
    try:
        asyncio.run(subscribe_data_change())
    except KeyboardInterrupt:
        sys.exit(0)

运行与验证

运行前提：需要有一个可访问的OPC UA服务器。若本地没有服务器，可使用asyncua自带的示例服务器快速验证：

# 终端1：启动一个简易OPC UA服务器（含上述节点）
python -m asyncua.server

# 终端2：运行本案例客户端
python opcua_subscriber.py

验证方法：

1. 客户端启动后，控制台输出“已连接服务器”和“已订阅节点”即表示连接与订阅成功。
2. 在服务器端修改对应节点的值（例如使用UaExpert或另一个脚本写入新值），客户端控制台应立即打印[数据变化] 节点: ... 新值: ...。
3. 若服务器端值不变，客户端不会打印任何数据变化信息，这是正常现象。

通过条件：当服务器节点值发生变化时，客户端能在1秒内打印出新值，且无异常报错，即视为案例验证通过。

关键说明

• NODE_IDS中的节点ID需与服务器实际存在的节点匹配，否则订阅时会报BadNodeIdUnknown错误。
• 订阅发布间隔设为100ms，即服务器每100ms检查一次是否有变化；实际延迟取决于服务器端采样间隔。
• 回调类SubHandler必须实现datachange_notification方法，否则无法接收通知。
• 若需订阅更多节点，直接在NODE_IDS列表中追加节点ID即可。