教程
用 Python 操作 OPC UA
最快理解 OPC UA 的方式,是用代码与服务器通信。本教程使用 asyncua(免费开源的 opcua-asyncio 库)进行连接、浏览、读取、写入和订阅:这四件事占了你 90% 的工作。
先修要求:Python 3.8+,以及任意一台可连接的 OPC UA 服务器(仿真器、open62541 演示服务器,或你的 PLC)。还没有服务器?参见免费工具与测试服务器.
1. 安装
pip install asyncua
2. 连接并读取一个数值
import asyncio
from asyncua import Client
URL = "opc.tcp://localhost:4840/freeopcua/server/"
async def main():
async with Client(url=URL) as client:
# Read a node by its NodeId (namespace index + identifier)
node = client.get_node("ns=2;i=2")
value = await node.read_value()
print("Value:", value)
asyncio.run(main())
3. 浏览地址空间
async with Client(url=URL) as client:
objects = client.nodes.objects
for child in await objects.get_children():
name = await child.read_browse_name()
print(name)
4. 写入一个值
from asyncua import ua
async with Client(url=URL) as client:
node = client.get_node("ns=2;i=3")
await node.write_value(ua.DataValue(ua.Variant(42.0, ua.VariantType.Double)))
写入常常被服务器权限阻止。如果失败,请确认该节点可写且你的用户已获授权:参见故障排查.
5. 订阅数据变化
订阅把更新推送给你,而不是由你轮询:这是监视一个数值的正确方式。
class Handler:
def datachange_notification(self, node, val, data):
print("Change:", node, val)
async with Client(url=URL) as client:
handler = Handler()
sub = await client.create_subscription(500, handler) # 500 ms
node = client.get_node("ns=2;i=2")
await sub.subscribe_data_change(node)
await asyncio.sleep(30) # keep running to receive changes
安全地建立连接
对于 Sign & Encrypt,请在连接前加载你的证书和密钥:
await client.set_security_string(
"Basic256Sha256,SignAndEncrypt,client-cert.der,client-key.pem")
服务器必须信任你的证书(把它从 rejected 移到 trusted)。详见:OPC UA 安全与证书。
在 Python 里读节点只是第一步。把这股数据流变成车间上排好序、核过成本的行动,正是 Jemba 所做的事。
查看 Jemba →常见问题
我该用哪个 Python OPC UA 库?
asyncua(opcua-asyncio) 是活跃维护的异步库,也是通常的选择。较旧的同步python-opcua已因它而被废弃。我可以用 Python 搭建 OPC UA 服务器吗?
是的:asyncua 包含完整的服务器实现,用于仿真器和测试很方便。
什么是 NodeId?
节点的唯一地址,例如
ns=2;i=2(命名空间索引 2,整数标识符 2)或ns=3;s=Temperature(字符串标识符)。最后更新于 2026 年 9 月 4 日 · 教育资源。库名称为其所有者的商标。