python-opcua服务器开发实战：从基础节点创建到历史数据存储完整教程

蔡丛锟

509人浏览 · 2026-02-20 02:45:49

蔡丛锟 · 2026-02-20 02:45:49 发布

python-opcua服务器开发实战：从基础节点创建到历史数据存储完整教程

【免费下载链接】python-opcua LGPL Pure Python OPC-UA Client and Server 项目地址: https://gitcode.com/gh_mirrors/py/python-opcua

OPC UA（Open Platform Communications Unified Architecture）是工业自动化领域的重要通信标准，而python-opcua作为LGPL协议下的纯Python实现，为开发者提供了构建OPC UA服务器和客户端的强大工具。本文将带你从零开始，掌握使用python-opcua开发功能完善的OPC UA服务器，包括基础节点创建、数据读写和历史数据存储的完整流程。

一、环境准备：快速搭建开发环境

1.1 安装python-opcua

使用pip命令即可快速安装python-opcua：

pip install python-opcua

1.2 获取项目源码

如需查看完整示例和源码，可克隆项目仓库：

git clone https://gitcode.com/gh_mirrors/py/python-opcua

项目核心代码位于opcua/目录，包含服务器、客户端、加密等模块，示例程序则集中在examples/目录。

二、极简服务器：5分钟启动你的第一个OPC UA服务器

2.1 最小化服务器实现

python-opcua提供了极其简洁的服务器启动方式，创建server-minimal.py：

from opcua import Server

server = Server()
server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")

# 设置服务器名称
server.set_server_name("FreeOpcUa Example Server")

# 启动服务器
server.start()
print("Server started at {}".format(server.endpoint))

# 保持运行
try:
    while True:
        time.sleep(1)
finally:
    # 停止服务器
    server.stop()

2.2 核心代码解析

服务器初始化：通过Server()创建实例，set_endpoint()定义连接地址
启动与停止：start()和stop()方法控制服务器生命周期
端点URL格式：遵循opc.tcp://<host>:<port>/<path>标准格式

运行该脚本后，即可通过OPC UA客户端连接到opc.tcp://localhost:4840/freeopcua/server/地址。

三、节点管理：构建你的OPC UA信息模型

3.1 地址空间基础

OPC UA服务器的核心是信息模型，通过节点组织数据。python-opcua提供了直观的节点创建接口：

# 获取根节点
root = server.get_root_node()
objects = server.get_objects_node()

# 创建自定义对象节点
myobj = objects.add_object("ns=1;s=MyObject", "MyObject")

# 添加变量节点
myvar = myobj.add_variable("ns=1;s=MyVariable", "MyVariable", 6.7)
myvar.set_writable()  # 允许客户端写入

3.2 常用节点类型

对象节点（Object）：用于组织其他节点，如opcua/common/node.py中定义的Node类
变量节点（Variable）：存储实际数据，支持多种数据类型
方法节点（Method）：提供远程调用功能

3.3 节点属性配置

可通过set_attribute()方法配置节点属性：

from opcua import ua

# 设置变量数据类型为浮点数
myvar.set_attribute(ua.AttributeIds.DataType, ua.Variant(ua.NodeId(ua.ObjectIds.Double), ua.VariantType.NodeId))

四、历史数据存储：实现数据的持久化记录

4.1 启用历史数据服务

python-opcua通过HistorySQL模块支持历史数据存储，需先安装依赖：

pip install sqlalchemy

4.2 配置历史数据存储

在服务器中启用历史数据记录：

from opcua.server.history_sql import HistorySQL

# 初始化历史数据库
history = HistorySQL("sqlite:///history.db")
server.iserver.history_manager.set_storage(history)

# 为变量启用历史记录
myvar = myobj.add_variable("ns=1;s=HistVariable", "HistVariable", 0)
myvar.set_writable()
server.historize_node_data_change(myvar, period=None, count=100)  # 记录所有变化

4.3 历史数据查询

客户端可通过read_history_data()方法查询历史数据：

# 客户端代码示例
from opcua import Client

client = Client("opc.tcp://localhost:4840/freeopcua/server/")
client.connect()

node = client.get_node("ns=1;s=HistVariable")
data = node.read_history_data(ua.ReadRawHistoryDataDetails(
    StartTime=datetime.datetime(2023, 1, 1),
    EndTime=datetime.datetime.now(),
    NumValuesPerNode=1000
))

五、高级功能：提升服务器实用性

5.1 事件处理

通过订阅机制实现事件通知：

# 服务器端：创建事件类型
et = server.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [
    ('MyNumericProperty', ua.VariantType.Float),
    ('MyStringProperty', ua.VariantType.String)
])

# 客户端：订阅事件
handler = SubHandler()
sub = client.create_subscription(500, handler)
handle = sub.subscribe_events()

5.2 安全配置

启用加密通信需生成证书，项目提供了便捷脚本examples/generate_certificate.sh：

cd examples
./generate_certificate.sh

在服务器中加载证书：

server.load_certificate("certificate.der")
server.load_private_key("private_key.pem")

六、实战案例：完整服务器实现

综合以上功能，我们可以构建一个包含数据采集、历史存储和事件通知的完整服务器，参考examples/server-example.py：

# 完整示例代码结构
from opcua import Server
from opcua.server.history_sql import HistorySQL
import time

server = Server()
server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
server.set_server_name("My OPC UA Server")

# 设置命名空间
uri = "http://example.com/opcua"
idx = server.register_namespace(uri)

# 初始化历史存储
history = HistorySQL("sqlite:///my_history.db")
server.iserver.history_manager.set_storage(history)

# 创建节点结构
objects = server.get_objects_node()
myobj = objects.add_object(idx, "MyObject")
myvar = myobj.add_variable(idx, "MyVariable", 0.0)
myvar.set_writable()
server.historize_node_data_change(myvar, period=None, count=1000)

# 启动服务器
server.start()
print("Server started")

try:
    count = 0
    while True:
        time.sleep(1)
        count += 0.1
        myvar.set_value(count)
finally:
    server.stop()