OPC-UA 与 Python 详细使用教程

python
教程
发布日期:   2025-10-19
更新日期:   2025-10-25
文章字数:   2.3k
阅读时长:   9 分
阅读次数:  

OPC-UA 与 Python 详细使用教程

1. 什么是 OPC-UA？

OPC-UA (Open Platform Communications Unified Architecture) 是一种独立于平台、面向服务的机器对机器（M2M）通信协议。它被广泛应用于工业自动化领域，用于在不同厂商的设备、控制器和软件应用之间实现安全、可靠的数据交换。

核心概念：

  • 服务器 (Server): 通常是设备（如 PLC、DCS）或网关，它持有数据并将其暴露给网络。
  • 客户端 (Client): 通常是监控软件（如 SCADA、HMI）或数据采集程序，它连接到服务器以读取、写入或订阅数据。
  • 地址空间 (Address Space): 服务器内部所有数据的分层结构。
  • 节点 (Node): 地址空间中的一个元素，可以是对象 (Object)、变量 (Variable)、方法 (Method) 等。每个节点都有一个唯一的 NodeId。
  • 变量 (Variable): 存放实际数据的节点，例如温度、压力、开关状态等。

2. 安装 python-opcua

安装非常简单，使用 pip 即可：

pip install opcua

3. 第一部分：创建一个简单的 OPC-UA 服务器

我们将创建一个服务器，它会暴露一个模拟的温度传感器变量。

import time
from opcua import ua, Server

# 1. 创建服务器实例
server = Server()

# 2. 设置服务器端点 (Endpoint)
#    客户端将通过这个 URL 连接
url = "opc.tcp://0.0.0.0:4840/freeopcua/server/"
server.set_endpoint(url)

# 3. 设置服务器名称（可选）
server.set_server_name("Python OPC-UA 示例服务器")

# 4. 设置安全策略（这里使用最不安全的，方便测试）
#    实际应用中应使用更安全的策略
server.set_security_policy([
    ua.SecurityPolicyType.NoSecurity,
    ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt,
    ua.SecurityPolicyType.Basic256Sha256_Sign
])

# 5. 注册一个唯一的命名空间 (Namespace)
#    命名空间用于组织你自己的节点，防止与标准节点冲突
ns_name = "http://examples.freeopcua.github.io"
idx = server.register_namespace(ns_name)

# 6. 在服务器的 Objects 节点下创建一个“MyObjects”文件夹
objects = server.get_objects_node()
my_folder = objects.add_object(idx, "MyObjects")

# 7. 在“MyObjects”文件夹下创建变量
#    idx: 命名空间索引
#    "MyVariable": 节点的浏览名称 (BrowseName)
#    0: 变量的初始值
my_var = my_folder.add_variable(idx, "MyVariable", 0)
my_temp_sensor = my_folder.add_variable(idx, "TemperatureSensor", 20.0)

# 8. 使变量可被客户端写入
#    默认情况下，变量是只读的
my_var.set_writable()
my_temp_sensor.set_writable()

# 9. 启动服务器
try:
    print(f"启动服务器于 {url}")
    server.start()
    
    # 让服务器持续运行，并模拟温度变化
    temperature = 20.0
    delta = 0.5
    
    while True:
        time.sleep(2)  # 每2秒更新一次
        
        temperature += delta
        if temperature > 30 or temperature < 19:
            delta = -delta # 调转方向
            
        print(f"设置新温度值为: {temperature:.1f}")
        my_temp_sensor.set_value(temperature)

except KeyboardInterrupt:
    print("\n服务器正在停止...")
finally:
    # 10. 关闭服务器
    server.stop()
    print("服务器已关闭")

运行： 将以上代码保存为 server.py 并运行 python server.py。你现在就有了一个正在运行的 OPC-UA 服务器。

4. 第二部分：创建一个 OPC-UA 客户端（读写）

现在我们创建另一个脚本来连接服务器、读取和写入数据。

import time
from opcua import Client, ua

# 服务器的 URL，与 server.py 中设置的保持一致
url = "opc.tcp://localhost:4840/freeopcua/server/"

# 1. 创建客户端实例
client = Client(url)

try:
    # 2. 连接到服务器
    print(f"正在连接到 {url} ...")
    client.connect()
    print("连接成功！")

    # 3. 浏览节点 (可选, 但有助于理解结构)
    #    获取根节点
    root = client.get_root_node()
    print(f"根节点: {root}")
    print(f"根节点的子节点: {root.get_children()}")

    # 4. 获取我们感兴趣的特定节点
    #    方法a: 通过浏览路径 (Browse Path)
    #    路径: Root -> Objects (0:Objects) -> 命名空间2 (2:MyObjects) -> 变量 (2:TemperatureSensor)
    #    '2' 是我们自己注册的命名空间的索引 (idx)
    #    注意: 命名空间索引 (ns) 可能会变化，通常 0 是标准，1 是服务器内部，我们注册的从 2 开始
    temp_node = client.get_node("ns=2;s=TemperatureSensor")
    my_var_node = client.get_node("ns=2;s=MyVariable")

    #    方法b: 通过 NodeId (如果已知)
    #    node_id_str = "ns=2;i=2" # 假设你知道它的 NodeId
    #    temp_node = client.get_node(node_id_str)
    
    print("-" * 30)
    
    # 5. 读取节点的值
    current_temp = temp_node.get_value()
    print(f"读取到当前温度: {current_temp}")
    
    current_var = my_var_node.get_value()
    print(f"读取到 MyVariable: {current_var}")
    
    print("-" * 30)

    # 6. 写入节点的值
    new_value = 99.9
    print(f"正在向 MyVariable 写入新值: {new_value}")
    
    # 构造 ua.DataValue 和 ua.Variant
    data_value = ua.DataValue(ua.Variant(new_value, ua.VariantType.Double))
    my_var_node.set_value(data_value)
    
    # 验证写入
    time.sleep(1)
    read_back = my_var_node.get_value()
    print(f"回读 MyVariable 的值: {read_back}")
    
    # 尝试写入只读节点（如果 TemperatureSensor 没设置 set_writable()，这里会报错）
    try:
        temp_node.set_value(ua.DataValue(ua.Variant(100.0, ua.VariantType.Float)))
        print("成功写入温度值 (服务器端允许写入)")
    except ua.UaError as e:
        print(f"写入温度值失败 (可能是只读): {e}")


finally:
    # 7. 断开连接
    if client:
        client.disconnect()
        print("客户端已断开连接")

运行： 保持 server.py 运行，打开一个新的终端，运行 python client.py。你将看到客户端连接、读取、写入并断开连接的过程。

5. 第三部分：订阅数据变化

OPC-UA 最强大的功能之一是订阅。客户端不需要（也不应该）频繁轮询数据，而是可以告诉服务器：“当这个变量变化时，请通知我。”

我们将修改客户端代码来实现订阅。

import time
from opcua import Client

# 1. 定义一个订阅处理器类 (Handler)
class SubHandler(object):
    """
    订阅处理器，用于处理来自服务器的数据变更通知。
    """
    def datachange_notification(self, node, val, data):
        """
        当订阅的节点数据发生变化时，此方法被调用。
        """
        print(f"[订阅] 节点 {node} 的值变为: {val}")

    def event_notification(self, event):
        """
        处理事件通知 (本例中未使用)
        """
        print(f"[订阅] 收到新事件: {event}")

# --- 主程序 ---
url = "opc.tcp://localhost:4840/freeopcua/server/"
client = Client(url)

try:
    client.connect()
    print("客户端已连接")

    # 2. 获取要订阅的节点
    temp_node = client.get_node("ns=2;s=TemperatureSensor")
    print(f"将要订阅的节点: {temp_node}")

    # 3. 创建订阅处理器实例
    handler = SubHandler()

    # 4. 在客户端上创建订阅 (Subscription)
    #    参数 1000: 订阅的发布间隔 (ms)。服务器将每1000ms检查一次是否有变化。
    subscription = client.create_subscription(1000, handler)
    print("已创建订阅")

    # 5. 告诉订阅要监控哪个节点 (MonitoredItem)
    #    参数 10: 队列大小 (QueueSize)
    handle = subscription.subscribe_data_change(temp_node, queuesize=10)
    print(f"已订阅节点 {temp_node} 的数据变化")

    # 6. 让客户端保持运行以接收通知
    print("正在等待数据变化 (按 Ctrl+C 停止)...")
    while True:
        time.sleep(1)

except KeyboardInterrupt:
    print("\n正在停止订阅...")
finally:
    # 7. 删除订阅 (可选, 但良好实践)
    if 'subscription' in locals():
        subscription.delete()
        print("订阅已删除")
    
    # 8. 断开连接
    client.disconnect()
    print("客户端已断开")

运行： 确保 server.py 正在运行（它会每2秒更新一次温度）。然后运行这个新的订阅客户端脚本 subscriber.py。你将看到它每2秒（或根据服务器的更新频率）打印一次 [订阅] ... 的消息。

6. 第四部分：在服务器上创建并调用方法

OPC-UA 不仅能传输数据，还能执行命令（方法）。

1. 修改 server.py：

在 server.start() 之前，添加以下代码来定义一个方法：

# --- (接第 8 步之后, server.start() 之前) ---

# 定义一个 Python 函数作为方法的实现
def multiply_by_two(parent, variant_x):
    """
    一个简单的 OPC-UA 方法，接收一个输入参数并返回其两倍。
    """
    x = variant_x.Value
    print(f"服务器方法 'multiply_by_two' 被调用，输入: {x}")
    
    result = x * 2
    
    # 返回值必须是 ua.Variant
    return [ua.Variant(result, ua.VariantType.Double)]

# 9. 在 "MyObjects" 文件夹下注册这个方法
#    输入参数定义
in_arg = ua.Argument()
in_arg.Name = "InputX"
in_arg.DataType = ua.NodeId(ua.ObjectIds.Double)
in_arg.ValueRank = -1
in_arg.ArrayDimensions = []
in_arg.Description = ua.LocalizedText("一个浮点数输入")

# 输出参数定义
out_arg = ua.Argument()
out_arg.Name = "ResultY"
out_arg.DataType = ua.NodeId(ua.ObjectIds.Double)
out_arg.ValueRank = -1
out_arg.ArrayDimensions = []
out_arg.Description = ua.LocalizedText("返回 InputX * 2")

# 将方法添加到地址空间
my_folder.add_method(
    idx,                          # 命名空间
    "MultiplyByTwo",              # 方法名称
    multiply_by_two,              # 绑定的 Python 函数
    [in_arg],                     # 输入参数列表
    [out_arg]                     # 输出参数列表
)

2. 修改 client.py（普通的读写客户端）：

在 client.disconnect() 之前，添加以下代码来调用该方法：

# --- (接第 6 步之后, client.disconnect() 之前) ---

print("-" * 30)
# 7. 调用服务器上的方法
print("正在调用服务器方法 'MultiplyByTwo'...")

# 获取方法节点
method_node = my_folder_node.get_child("2:MultiplyByTwo")

# 准备输入参数
input_value = 10.5
print(f"输入参数为: {input_value}")

# 调用方法 (父节点, 方法节点, 输入参数)
# 父节点是 'MyObjects'
my_folder_node = client.get_node("ns=2;s=MyObjects") 
result = my_folder_node.call_method(method_node, input_value)
# 或者: result = client.call_method(method_node, input_value) 

print(f"服务器返回结果: {result}")

注意： 你需要先在客户端代码中获取 my_folder_node，例如 my_folder_node = client.get_node("ns=2;s=MyObjects")。

运行： 重启服务器 server.py（现在它包含了方法）。运行 client.py，你将看到客户端成功调用了服务器上的 MultiplyByTwo 方法并获得了 21.0 的返回结果。

总结

本教程涵盖了 python-opcua 最核心的三个功能：

  1. 创建服务器 并暴露变量。
  2. 创建客户端 以读取和写入变量。
  3. 使用订阅 来实时接收数据变更通知。
  4. 使用方法 来执行远程命令。

在实际应用中，你还需要深入了解：

  • 安全性： 使用证书和用户认证来保护你的服务器。
  • 复杂数据类型： 定义和使用自定义的结构体 (Structs)。
  • 历史数据： 查询历史数据 (Historical Access)。
  • 事件 (Events)： 订阅比数据变更更复杂的事件（如警报）。
文章作者: 0xdadream
文章链接: https://0xdadream.github.io/2025/10/19/python-zhong-shi-yong-opcua/
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 0xdadream !
python
赏
评论
  目录