柏虎资源网

专注编程学习，Python、Java、C++ 教程、案例及资源

2025年05月21日

基于Python语言的PLC通信技术OPC UA客户端应用程序开发指南

引言

OPC UA（OPC Unified Architecture）是一种机器对机器通信的机器可读接口，它提供了开放的、跨平台的、安全的标准，用于工业自动化领域的数据交换。随着Python在工业自动化领域的广泛应用，基于Python开发OPC UA客户端应用程序变得越来越重要。本报告将全面介绍如何基于Python语言自行开发OPC UA客户端应用程序，包括必要的库、核心概念、API使用以及实际应用案例。

OPC UA基本概念与信息模型

OPC UA概述

OPC UA（OPC Unified Architecture）是一种开放的、跨平台的标准，用于工业自动化领域的数据交换。它替代了以前依赖于特定平台的OPC Classic标准，如COM/DCOM，采用基于纯二进制TCP/IP或SOAP的实现方式[16]。OPC UA定义了信息模型、消息模型和通信模型，确保系统之间的互操作性[20]。

OPC UA信息模型

OPC UA的信息模型基于节点（Node）的概念，这些节点构成一个网络或结构化图，称为OPC UA地址空间。地址空间由节点和引用（References）组成[19](
https://blog.csdn.net/h_sn9999/article/details/108561416)(
https://zhuanlan.zhihu.com/p/430243728)]。 节点可以根据不同的用途归属于不同的节点类别（NodeClass），每个节点包含属性（Attributes）和引用（References）。OPC UA建模的基本概念就是节点以及节点之间的引用关系[22]。

Python OPC UA客户端开发库

Python OPC-UA库概述

Python OPC-UA是一个纯Python实现的OPC UA客户端和服务器库，遵循LGPL-3.0开源协议[23]。它提供了两种接口：

  • 低级别接口：用于发送和接收OPC UA定义的所有结构
  • 高级别接口：允许开发者用几行代码就能编写客户端或服务器程序 该库在PyPI上可用，可以通过pip进行安装，项目标识为"opcua"[2]。

客户端类概述

Python OPC-UA库提供了Client类作为高阶客户端实现，使连接OPC UA服务器和浏览地址空间变得简单。这个类尝试暴露尽可能多的功能，但如果需要更多的灵活性，可以使用UaClient对象，它提供了原始的OPC UA服务接口[6]。 Client类的主要功能包括：

  • 连接到OPC UA服务器
  • 浏览地址空间
  • 读取和写入节点值
  • 订阅数据变化和事件
  • 导出和导入XML节点定义
  • 加载服务器的自定义枚举和类型定义

库的主要功能

Python OPC-UA库提供了全面的OPC UA功能实现，包括：

  1. 客户端功能：
  2. 连接到OPC UA服务器
  3. 浏览和操作地址空间
  4. 读取和写入节点值
  5. 订阅数据变化和事件
  6. 管理会话和安全通道
  7. 服务器功能：
  8. 创建自定义服务器
  9. 定义和管理地址空间
  10. 处理客户端连接和请求
  11. 安全功能：
  12. 支持多种安全策略
  13. 用户认证
  14. 证书管理
  15. 高级功能：
  16. 自定义类型定义
  17. 历史数据读写
  18. 方法调用
  19. 事件处理

开发环境配置

安装Python OPC-UA库

Python OPC-UA库可以通过pip安装：

pip install opcua

对于需要使用加密功能的场景，可能还需要安装额外的依赖项：

pip install opcua[cryptography]

开发环境要求

Python OPC-UA库支持Python 3.6及更高版本。开发环境应满足以下要求：

  1. 安装Python 3.6+
  2. 安装必要的依赖库
  3. 安装Python开发工具（如IDE、文本编辑器等）

证书和密钥准备

如果需要使用加密连接，还需要准备证书和私钥文件。Python OPC-UA提供了生成证书的示例脚本：

./generate_certificate.sh

这将生成以下文件：

  • private-key-example.pem：私钥文件
  • certificate-example.der：证书文件

基本OPC UA客户端开发

客户端连接和断开

连接到OPC UA服务器是客户端开发的第一步。Python OPC-UA提供了简单易用的API来实现这一功能。

from opcua import Client
# 创建客户端实例
client = Client("opc.tcp://localhost:4840/freeopcua/server/")
try:
    # 连接到OPC UA服务器
    client.connect()
    
    # 执行其他操作...
    
finally:
    # 确保断开连接
    client.disconnect()

也可以使用with语句简化连接管理：

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 连接到服务器，执行操作...
    pass

浏览地址空间

OPC UA服务器维护一个称为地址空间的节点结构化图。客户端可以通过Client类提供的方法来浏览这个地址空间。

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 获取根节点
    root = client.get_root_node()
    print("Root node is:", root)
    
    # 获取对象节点
    objects = client.get_objects_node()
    print("Objects node is:", objects)
    
    # 获取根节点的子节点
    print("Children of root are:", root.get_children())

读取和写入节点值

读取和写入节点值是OPC UA客户端的主要功能之一。Python OPC-UA提供了简单的方法来实现这一功能。

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 获取特定节点（方式1：使用NodeId对象）
    # var = client.get_node(ua.NodeId(1002, 2))
    
    # 获取特定节点（方式2：使用字符串表示）
    # var = client.get_node("ns=3;i=2002")
    
    # 通过BrowsePath获取节点
    myvar = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable"])
    print("myvar is:", myvar)
    
    # 读取节点值（方式1：返回DataValue对象）
    # value = var.get_data_value()
    
    # 读取节点值（方式2：返回Python内置类型）
    value = myvar.get_value()
    print("Value of myvar is:", value)
    
    # 写入节点值（方式1：显式指定数据类型）
    # var.set_value(ua.Variant([23], ua.VariantType.Int64))
    
    # 写入节点值（方式2：隐式确定数据类型）
    myvar.set_value(3.9)

批量读取和写入

为了提高效率，Python OPC-UA还支持批量读取和写入多个节点的值：

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 假设node1、node2、node3是已经获取的节点对象
    values = client.get_values([node1, node2, node3])
    print("Values of multiple nodes:", values)
    
    # 批量写入节点值
    client.set_values([node1, node2, node3], [value1, value2, value3])

OPC UA订阅机制

订阅数据变化

OPC UA订阅允许客户端注册对特定数据变化或事件的兴趣。当服务器检测到变化时，会自动通知已订阅的客户端。

from opcua import Client
import time
class SubHandler(object):
    """
    订阅处理程序。接收来自服务器的事件通知
    data_change和event方法会直接从接收线程调用
    不要在这些方法中执行昂贵、缓慢或网络操作
    如果需要执行这样的操作，请启动另一个线程
    """
    def datachange_notification(self, node, val, data):
        print("Python: New data change event", node, val)
    
    def event_notification(self, event):
        print("Python: New event", event)
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 创建订阅处理程序实例
    handler = SubHandler()
    
    # 创建订阅，500ms发布周期
    sub = client.create_subscription(500, handler)
    
    # 订阅特定节点的数据变化
    myvar = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable"])
    handle = sub.subscribe_data_change(myvar)
    
    # 订阅服务器事件
    sub.subscribe_events()
    
    # 保持订阅一段时间
    time.sleep(5)
    
    # 取消订阅
    sub.unsubscribe(handle)
    
    # 删除订阅
    sub.delete()

事件订阅

OPC UA客户端可以订阅服务器事件，以便在特定事件发生时收到通知。

from opcua import Client
import time
class EventSubHandler(object):
    def event_notification(self, event):
        print("Python: New event", event)
with Client("opc.ua://localhost:4840/freeopcua/server/") as client:
    handler = EventSubHandler()
    
    # 创建订阅，1000ms发布周期
    sub = client.create_subscription(1000, handler)
    
    # 订阅所有事件
    sub.subscribe_events()
    
    # 或者订阅特定类型的事件
    # sub.subscribe_events(eventtype=ua.ObjectIds.AuditUpdateEvent)
    
    # 保持订阅一段时间
    time.sleep(5)
    
    # 取消订阅
    sub.unsubscribe()
    
    # 删除订阅
    sub.delete()

调用OPC UA方法

OPC UA允许客户端调用服务器上的方法。Python OPC-UA提供了简单的方法调用API。

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 获取要调用方法的对象
    obj = root.get_child(["0:Objects", "2:MyObject"])
    
    # 调用方法multiply，参数为3和"klk"
    res = obj.call_method("2:multiply", 3, "klk")
    print("method result is:", res)

OPC UA安全配置

用户名和密码认证

OPC UA支持多种安全机制，包括用户名和密码认证。

from opcua import Client
# 方法1：在连接URL中指定用户名和密码
# client = Client("opc.tcp://username:password@localhost:4840/freeopcua/server/")
    
# 方法2：连接后设置用户名和密码
client = Client("opc.tcp://localhost:4840/freeopcua/server/")
client.set_user("username")
client.set_password("password")
client.connect()

证书认证

OPC UA还支持基于证书的安全认证，提供了更高的安全性。

from opcua import Client
import os
# 证书路径
cert_path = os.path.join(os.path.dirname(__file__), "certificate-example.der")
private_key_path = os.path.join(os.path.dirname(__file__), "private-key-example.pem")
# 设置安全模式
client.set_security(
    policy="Basic256Sha256",
    certificate_path=cert_path,
    private_key_path=private_key_path,
    mode=ua.MessageSecurityMode.SignAndEncrypt
)
# 或者使用简写方式
# client.set_security_string("Basic256Sha256,SignAndEncrypt,certificate-example.der,private-key-example.pem")
client.connect()

OPC UA历史数据读写

读取历史数据

OPC UA标准指定了历史数据的读写，主要包括：

from opcua import Client
import datetime
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 获取节点
    myvar = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable"])
    
    # 读取历史数据
    # 从10分钟前开始，到当前时间
    start = datetime.datetime.now() - datetime.timedelta(minutes=10)
    end = datetime.datetime.now()
    
    # 读取所有数据
    data = myvar.read_historical_data(start, end, ua.HistogramDataEncoding.Count)
    
    # 或者按时间间隔读取
    # data = myvar.read_historical_data(start, end, ua.HistogramDataEncoding.Count, period=ua.Duration(60000))
    
    # 打印读取结果
    print("Historical data:", data)

写入历史数据

某些OPC UA服务器支持客户端写入历史数据的功能：

from opcua import Client
import datetime
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 获取节点
    myvar = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable"])
    
    # 创建历史数据值对象
    values = [
        ua.DataValue(
            value=ua.Variant(23, ua.VariantType.Int32),
            source_timestamp=datetime.datetime.now() - datetime.timedelta(minutes=10)
        ),
        ua.DataValue(
            value=ua.Variant(42, ua.VariantType.Int32),
            source_timestamp=datetime.datetime.now() - datetime.timedelta(minutes=5)
        )
    ]
    
    # 写入历史数据
    myvar.write_historical_data(values)

OPC UA XML导入导出

导入XML节点定义

OPC UA允许客户端导入XML定义的节点，这在需要动态添加节点时非常有用。

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 从XML文件导入节点
    client.import_xml("path/to/nodeset.xml")
    
    # 或者从XML字符串导入
    # xml_content = "<UANodeSet>...</UANodeSet>"
    # client.import_xml(xmlstring=xml_content)

导出节点为XML

客户端也可以将服务器上的节点导出为XML格式，便于保存和重用。

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 获取要导出的节点
    nodes = [root, objects]
    
    # 导出节点到XML文件
    client.export_xml(nodes, "exported_nodes.xml")

OPC UA自定义类型

加载自定义枚举

OPC UA服务器可能定义了自定义枚举类型，客户端可以通过以下方式加载这些枚举：

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 加载服务器上的自定义枚举
    client.load_enums()
    
    # 现在可以在ua模块中使用这些枚举
    # 例如：ua.MyCustomEnum.Value1

加载自定义结构和扩展对象

OPC UA还支持自定义结构和扩展对象，Python OPC-UA可以动态加载这些定义：

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 加载服务器上的自定义类型定义
    client.load_type_definitions()
    
    # 现在可以在ua模块中使用这些自定义结构
    # 例如：ua.MyCustomStructure(field1=1, field2="value")

高级OPC UA客户端功能

注册和取消注册节点

注册节点可以提高读写操作的效率，特别是当服务器支持这一功能时：

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 获取节点
    node1 = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable1"])
    node2 = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable2"])
    
    # 注册节点
    client.register_nodes([node1, node2])
    
    # 注册后的节点ID会改变，原始ID保存在node.basenodeid中
    
    # 取消注册节点
    # client.unregister_nodes([node1, node2])

会话保持和超时管理

OPC UA定义了会话和安全通道的超时机制。Python OPC-UA提供了管理这些超时的API：

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/", timeout=10) as client:
    # 连接到服务器
    client.connect()
    
    # 打开安全通道
    client.open_secure_channel()
    
    # 创建会话
    client.create_session()
    
    # 保持会话活动
    client.keep_alive()
    
    # 重新协商订阅
    # client.reconciliate_subscriptions()

实际应用案例

连接到Kepware OPC UA服务器

以下是一个连接到Kepware OPC UA服务器的示例：

from opcua import Client
# 连接到Kepware OPC UA服务器
client = Client("opc.tcp://localhost:53530/KEPWARE OPC UA SERVER/")
client.set_user("username")
client.set_password("password")
client.connect()

连接到Prosys OPC UA服务器

以下是一个连接到Prosys OPC UA服务器的示例：

from opcua import Client
# 连接到Prosys OPC UA服务器
client = Client("opc.tcp://localhost:53530/OPC1")
client.connect()

使用加密连接

以下是一个使用加密连接的示例：

from opcua import Client
import os
# 证书路径
cert_path = os.path.join(os.path.dirname(__file__), "certificate.der")
private_key_path = os.path.join(os.path.dirname(__file__), "private-key.pem")
# 创建客户端
client = Client("opc.tcp://localhost:4840/freeopcua/server/")
# 设置安全模式
client.set_security(
    policy="Basic256Sha256",
    certificate_path=cert_path,
    private_key_path=private_key_path,
    mode=ua.MessageSecurityMode.SignAndEncrypt
)
# 或者使用简写方式
# client.set_security_string("Basic256Sha256,SignAndEncrypt,certificate.der,private-key.pem")
# 连接到服务器
client.connect()

性能优化和最佳实践

批量操作

为了提高性能，应该尽可能使用批量操作来读取或写入多个节点的值：

from opcua import Client
with Client("opc.tcp://localhost:4840/freeopcua/server/") as client:
    # 获取多个节点
    node1 = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable1"])
    node2 = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable2"])
    node3 = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable3"])
    
    # 批量读取节点值
    values = client.get_values([node1, node2, node3])
    
    # 批量写入节点值
    client.set_values([node1, node2, node3], [value1, value2, value3])

订阅优化

对于订阅操作，应该注意以下几点：

  1. 选择合适的发布周期
  2. 避免在通知处理程序中执行耗时操作
  3. 及时清理不再需要的订阅
from opcua import Client
import time
import threading
class SubHandler(object):
    def __init__(self):
        self.lock = threading.Lock()
        self.values = []
    
    def datachange_notification(self, node, val, data):
        # 在通知处理程序中避免执行耗时操作
        with self.lock:
            self.values.append((node, val, data))
    
    def process_values(self):
        with self.lock:
            values = self.values.copy()
            self.values.clear()
            return values
# 创建订阅处理程序
handler = SubHandler()
# 创建客户端和订阅
client = Client("opc.tcp://localhost:4840/freeopcua/server/")
sub = client.create_subscription(500, handler)
try:
    client.connect()
    
    # 订阅特定节点
    myvar = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable"])
    sub.subscribe_data_change(myvar)
    
    # 启动单独的线程来处理值
    while True:
        time.sleep(1)
        values = handler.process_values()
        if values:
            print("Processed values:", values)
finally:
    # 清理资源
    sub.unsubscribe()
    sub.delete()
    client.disconnect()

错误处理

在实际应用中，应该实现适当的错误处理机制：

from opcua import Client, ua
try:
    client = Client("opc.tcp://localhost:4840/freeopcua/server/")
    client.connect()
    
    # 可能会抛出异常的代码
    try:
        # 获取节点
        myvar = root.get_child(["0:Objects", "2:MyObject", "2:MyVariable"])
        
        # 读取节点值
        value = myvar.get_value()
        print("Value:", value)
    except ua.UaError as e:
        print("OPC UA error:", e)
    except Exception as e:
        print("General error:", e)
finally:
    # 确保断开连接
    if client and client.is_connected():
        client.disconnect()

结论

本报告详细介绍了如何基于Python语言自行开发OPC UA客户端应用程序。我们从OPC UA的基本概念和信息模型开始，然后介绍了Python OPC-UA库的主要功能和使用方法，包括客户端连接、地址空间浏览、数据读写、订阅机制、方法调用、安全配置以及历史数据操作等。 通过实际示例代码，我们展示了如何使用Python OPC-UA库实现各种OPC UA客户端功能。同时，我们也讨论了性能优化和最佳实践，帮助开发者创建高效、可靠的OPC UA客户端应用程序。 随着工业自动化的不断发展，OPC UA作为跨平台、开放的标准，在工业数据交换中发挥着越来越重要的作用。Python凭借其简洁、灵活的语法和丰富的生态系统，为OPC UA客户端开发提供了强大的工具。Python OPC-UA库作为纯Python实现的OPC UA客户端和服务器库，为Python开发者提供了开发OPC UA应用程序的便利工具。

标签：程序编程 

作者:baihuw , 分类:技术文章 , 浏览:8 , 评论:0

发表评论:

控制面板
您好，欢迎到访网站！
  查看权限
网站分类
最新留言