在Python中集成OPC UA（开放平台通信统一架构），最成熟且推荐使用的库是 opcua-asyncio（或简称为 asyncua）。它基于异步编程模型，同时提供了同步和异步的API，代码简洁、社区活跃，测试覆盖率超过95%。

以下示例将带你从环境搭建开始，逐步实现一个基础的OPC UA服务器与客户端，并涵盖数据订阅、安全连接等进阶场景。

环境准备

核心库是 asyncua，它依赖 asyncio。

pip install asyncua

1. 实现一个最小化 OPC UA 服务器

这个示例展示了如何创建一个服务器，添加一个变量，并允许客户端读写它。

import asyncio
from asyncua import Server, ua

async def main():
    # 1. 创建服务器实例并初始化
    server = Server()
    await server.init()
    
    # 2. 设置服务器终结点（URL）和名称
    server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
    server.set_server_name("Python OPC UA 示例服务器")

    # 3. 注册一个自定义命名空间（用于组织节点）
    uri = "http://my.custom.namespace"
    idx = await server.register_namespace(uri)

    # 4. 在服务器地址空间的"对象"节点下创建我们的自定义对象和变量
    my_object = await server.nodes.objects.add_object(idx, "MyDevice")
    my_variable = await my_object.add_variable(
        idx, 
        "Temperature",   # 变量名称
        25.0,            # 初始值
        datatype=ua.VariantType.Float
    )
    # 5. 设置该变量为可写
    await my_variable.set_writable(True)

    print(f"服务器已启动，监听: {server.endpoint}")

    # 6. 启动服务器并保持运行
    async with server:
        # 使用 await asyncio.sleep(1e9) 或类似方法保持事件循环运行
        try:
            await asyncio.sleep(1e9)
        except asyncio.CancelledError:
            print("服务器正在关闭...")

if __name__ == "__main__":
    asyncio.run(main())

2. 实现一个最小化 OPC UA 客户端

客户端连接至服务器，读取并修改 Temperature 变量的值。

import asyncio
from asyncua import Client

async def main():
    # 1. 定义服务器URL（需与服务器设置一致）
    url = "opc.tcp://localhost:4840/freeopcua/server/"
    
    # 2. 创建客户端并连接到服务器
    async with Client(url=url) as client:
        print(f"已连接到服务器: {url}")

        # 3. 获取目标节点（通过节点ID或浏览路径）
        # 我们的变量位于 Objects -> MyDevice -> Temperature
        # 注意：这里的 "2" 对应我们注册的自定义命名空间索引
        node = client.get_node("ns=2;s=MyDevice/Temperature")
        
        # 4. 读取变量的当前值
        value = await node.read_value()
        print(f"当前温度值: {value} °C")

        # 5. 写入一个新值
        new_value = 26.5
        await node.write_value(ua.Variant(new_value, ua.VariantType.Float))
        print(f"温度已更新为: {new_value} °C")
        
        # 验证写入
        value_after = await node.read_value()
        print(f"验证读取: {value_after} °C")

if __name__ == "__main__":
    asyncio.run(main())

3. 进阶功能：数据变更订阅 (Subscription)

在需要实时监控变量变化（而不是轮询）时，订阅功能非常有用。这个例子展示了一个客户端订阅一个节点，并在其值改变时收到通知。

import asyncio
from asyncua import Client, Node

# 1. 定义订阅的事件处理类
class SubscriptionHandler:
    """处理来自订阅的通知"""
    def datachange_notification(self, node: Node, val, data):
        # 当节点值发生变化时，此方法会被调用
        print(f"[订阅通知] 节点: {node}, 新值: {val}")

async def main():
    url = "opc.tcp://localhost:4840/freeopcua/server/"
    async with Client(url=url) as client:
        # 获取需要订阅的节点
        node = client.get_node("ns=2;s=MyDevice/Temperature")
        
        # 2. 创建订阅：参数为发送间隔(ms)和处理器实例
        handler = SubscriptionHandler()
        subscription = await client.create_subscription(500, handler)  # 500ms间隔
        
        # 3. 订阅该节点的数据变化
        await subscription.subscribe_data_change([node])
        print("已订阅温度变化，等待更新...")
        
        # 保持程序运行以接收通知
        try:
            await asyncio.sleep(60)  # 运行60秒
        finally:
            # 4. 取消订阅并删除
            await subscription.delete()
            print("订阅已取消")

if __name__ == "__main__":
    asyncio.run(main())

4. 进阶功能：开启安全连接 (Security)

在生产环境中，通常需要开启加密和签名。这需要准备证书和私钥，并设置安全策略。

首先，你需要生成证书和私钥。可以用OpenSSL命令快速创建：

# 生成私钥
openssl genpkey -algorithm RSA -out private_key.pem
# 生成证书签名请求
openssl req -new -key private_key.pem -out certificate.csr
# 自签名证书
openssl x509 -req -days 365 -in certificate.csr -signkey private_key.pem -out certificate.pem

在服务器端加载证书并启用安全策略：

# ... 在 server.init() 之后
# 加载证书和私钥
await server.load_certificate("certificate.pem")
await server.load_private_key("private_key.pem")

# 设置安全策略：使用Basic256Sha256进行签名和加密
server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt])

在客户端配置安全连接：

# ... 在创建 Client 之后
client = Client(url="opc.tcp://192.168.1.100:4840/")
# 设置安全字符串：安全策略, 消息模式, 证书文件, 私钥文件
client.set_security_string("Basic256Sha256,SignAndEncrypt,certificate.pem,private_key.pem")
# 如果服务器需要用户名密码
client.set_user("username")
client.set_password("password")

# 然后正常连接
await client.connect()

关键点与最佳实践总结