面向智能制造 / 工业机器人方向的开发者。本文带你用 OPC UA 打通 ROS2(Lyrical Luth)与西门子 S7-1500,从 TIA Portal 配置、Python 最小验证,到封装成 ROS2 节点,最后附一份可直接排查的坑位清单。

环境:TIA Portal V21 + S7-1500 + ROS2 Lyrical + Python asyncua。文中关键结论均标注来源。


目录

  1. 为什么打通 ROS2 和 PLC,又为什么选 OPC UA
  2. 环境准备
  3. TIA Portal 侧配置(4 步,含许可证)
  4. 安全策略:先免加密跑通,再上证书
  5. Python 端最小测试(asyncua)
  6. 封装成 ROS2 节点
  7. 双向通信(读 + 写)
  8. 常见坑位排查表(重点)
  9. S7 数据类型 ↔ OPC UA 映射表
  10. 总结

1. 为什么打通 ROS2 和 PLC,又为什么选 OPC UA

在工业机器人 / 智能制造项目中,典型分工:

┌─────────────────────────┐         ┌──────────────────────────┐
│        ROS2 (上位机)     │         │      S7-1500 (PLC)      │
│  - 视觉检测 (YOLO)       │  ────►  │  - 伺服/步进运动控制      │
│  - SLAM 导航             │  ◄────  │  - 气缸/传感器 IO         │
│  - 路径规划 (MoveIt)     │  数据交换 │  - 逻辑与安全联锁         │
│  - 决策与调度            │         │  - 现场总线 (Profinet)    │
└─────────────────────────┘         └──────────────────────────┘

ROS2 擅长感知、规划、任务决策;PLC 擅长确定性逻辑、状态机、IO 和现场联锁。具体运动插补与伺服闭环,则可能由 PLC、运动控制器、机器人控制器或驱动器承担。

注意:OPC UA 适合状态、任务命令、设定值、生产数据等上位层数据交换,不应承担伺服闭环、Safety 或硬实时周期控制。实时运动控制仍应放在 PLC、运动控制器、机器人控制器或驱动器侧。

为什么选 OPC UA,而不是 snap7

python-snap7 OPC UA
协议 legacy S7 / S7CommPlus OPC UA(IEC 62541)
S7-1500 legacy 需 PUT/GET;S7CommPlus 可原生连接 CPU 原生 OPC UA Server
跨厂商 主要面向 Siemens S7 跨厂商
安全 legacy PUT/GET 较弱;S7CommPlus 支持 TLS 证书、签名、加密
访问方式 legacy 字节偏移;S7CommPlus 可符号化访问 标准 OPC UA 地址空间 / NodeId

补充背景:python-snap7 3.x 已重写为纯 Python、不再需要 .dll/.so,并且已经加入 S7CommPlus 支持s7commplus 包,含 TLS 通道)。本文仍选择 OPC UA,是因为 OPC UA 在跨厂商互操作、符号化数据模型、证书体系和标准化集成方面更有优势,也更适合作为长期的生产级方案。[来源:python-snap7 官方文档]


2. 环境准备

项目 说明
PLC S7-1500,固件 V2.0+(OPC UA server 的最低要求,TIA V14+)[来源:西门子官方「找答案」]
编程软件 TIA Portal V21
上位机 Ubuntu 26.04 + ROS2 Lyrical Luth(2026-05 发布,LTS 至 2031)[来源:ROS 官方文档]
网络 PLC 与上位机同一网段,TCP 4840 端口互通

3. TIA Portal 侧配置(4 步)

3.1 激活 OPC UA 服务器

右键 CPU → 属性 → OPC UA → 服务器 → 勾选「激活 OPC UA 服务器」,端口保持默认 4840

[来源:西门子官方「找答案」- S7-1500 CPU 启用 OPC UA 服务器功能的具体步骤]

3.2 配置运行系统许可证(很多人漏)

CPU 属性 → 运行系统许可证 → OPC UA,按 CPU 型号选择对应的许可证类型:

许可证 典型适用 CPU
Small(小型) 1511 / 1512 / 1513、ET 200SP CPU
Medium(中型) 1515 / 1516、软控制器 1507
Large(大型) 1517 / 1518

许可证类型跟 CPU 型号/性能等级绑定,不是按变量数量自由选的。实际要求以 TIA Portal 中 CPU 属性 → 运行系统许可证 → OPC UA 显示的内容为准。[来源:西门子官方「找答案」]

3.3 暴露变量(关键)

OPC UA 走符号名访问,不需要像 snap7 那样关「优化块访问」。你要做的是:在 DB 的声明表里,给需要暴露的变量启用读写属性:

属性列 含义
Accessible from HMI/OPC UA OPC UA 是否可以访问该变量
Writable from HMI/OPC UA OPC UA 是否允许写入该变量

需要客户端读取的变量 → 勾 Accessible from HMI/OPC UA;需要客户端写入的变量 → 再勾 Writable from HMI/OPC UA。[来源:西门子官方文档 - OPC UA 变量访问权限定义]

3.4 编译并下载

配置完成后编译 + 下载到 CPU,并把 CPU 切到 RUN


4. 安全策略:先免加密跑通,再上证书

4.1 快速打通:SecurityPolicy = None

调试阶段可临时启用 SecurityPolicy = None,让 OPC UA 消息通道不签名、不加密。

⚠️ 但要注意:SecurityPolicy 和用户认证是两个独立维度None 只代表"消息通道不签名不加密",认证仍然可能是用户名/密码;如果 CPU 未启用 Anonymous/Guest 访问,客户端仍需要用户名和密码登录。

西门子官方说明:「No Security」默认是关闭的,且可与用户名/密码认证同时使用。测试时需在 TIA 里手动勾选 No Security 策略。[来源:西门子官方手册 - OPC UA 服务器通信 / 安全策略]

4.2 生产:证书 + 加密

生产环境用 Basic256Sha256 + 签名加密,客户端证书要导入 PLC 的「受信客户端」列表。Python 端 asyncua 设置证书的方式见第 7 节。


5. Python 端最小测试(asyncua)

先不碰 ROS2,用裸 Python 把链路打通。

5.1 安装

pip install asyncua

asyncua 是纯 Python 的 OPC UA 客户端库,基于 asyncio,跨平台无原生依赖,是 ROS2 侧连接 S7-1500 的主流选择。[来源:asyncua 官方 + 多个实战教程]

5.2 读一个变量

import asyncio
from asyncua import Client

async def main():
    url = "opc.tcp://192.168.0.1:4840"   # 改成你的 PLC IP
    async with Client(url=url) as client:
        # 本文测试 CPU 中该变量位于 ns=3。
        # 不同项目/服务器配置下 namespace index 可能不同,
        # 请用 UaExpert Browse 或 client.get_namespace_array() 确认实际 NodeId。
        node = client.get_node('ns=3;s="DB_Data"."Temperature"')
        value = await node.read_value()
        print(f"Temperature = {value}")

asyncio.run(main())

[来源:controlbyte.tech - Python OPC UA: Read Data from Siemens S7-1500]

5.3 写一个变量

import asyncio
from asyncua import Client, ua

async def main():
    async with Client("opc.tcp://192.168.0.1:4840") as client:
        node = client.get_node('ns=3;s="DB_Data"."SetSpeed"')
        # S7 Real 对应 OPC UA Float;LReal 对应 Double
        await node.write_value(ua.Variant(500.0, ua.VariantType.Float))

asyncio.run(main())

写之前确认:变量在 TIA 里勾了 Writable from HMI/OPC UA,且客户端以有写权限的用户登录(不是 Guest)。网络写入的值只能当"设定值"用,必须由 PLC 程序做校验、限幅、联锁后再执行,绝不能当安全输入直接用。[来源:controlbyte.tech]


6. 封装成 ROS2 节点

asyncua 是 asyncio 异步的,rclpy 是同步的。用后台线程跑 asyncio 事件循环 + run_coroutine_threadsafe 桥接:

import threading
import asyncio
import rclpy
from rclpy.node import Node
from std_msgs.msg import Float64
from asyncua import Client

class PlcOpcuaBridge(Node):
    def __init__(self):
        super().__init__('plc_opcua_bridge')

        self.pub = self.create_publisher(Float64, '/plc/temperature', 10)

        self.loop = asyncio.new_event_loop()
        self.client = None
        self.node = None

        # 后台线程跑事件循环
        self.thread = threading.Thread(target=self._run, daemon=True)
        self.thread.start()

        # 0.5s 定时读取并发布
        self.create_timer(0.5, self.publish_cycle)

    def _run(self):
        asyncio.set_event_loop(self.loop)
        self.loop.run_until_complete(self._connect())
        self.loop.run_forever()   # 关键:connect 完成后保持 loop 持续运行

    async def _connect(self):
        self.client = Client("opc.tcp://192.168.0.1:4840")
        await self.client.connect()
        self.node = self.client.get_node('ns=3;s="DB_Data"."Temperature"')
        self.get_logger().info('OPC UA connected')

    def publish_cycle(self):
        if self.node is None:
            return
        # 从主线程安全地调用异步读取。
        # 注意:同步等待会阻塞 ROS executor;这里作为 Demo 可接受,
        # 高频生产节点建议改用异步完成回调 / AsyncNode / 独立通信线程。
        future = asyncio.run_coroutine_threadsafe(self.node.read_value(), self.loop)
        value = future.result(timeout=1.0)
        self.pub.publish(Float64(data=float(value)))

    async def _disconnect(self):
        if self.client is not None:
            await self.client.disconnect()

    def shutdown(self):
        if self.client is not None:
            future = asyncio.run_coroutine_threadsafe(self._disconnect(), self.loop)
            try:
                future.result(timeout=2.0)
            except Exception:
                pass
        self.loop.call_soon_threadsafe(self.loop.stop)

def main(args=None):
    rclpy.init(args=args)
    node = PlcOpcuaBridge()
    try:
        rclpy.spin(node)
    finally:
        node.shutdown()
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

Lyrical 加分项:AsyncNode

ROS2 Lyrical 新增了 rclpy.experimental.AsyncNode,节点内部自带 asyncio 事件循环,回调里可以直接 await,省掉上面的线程桥接。[来源:ROS 官方 Lyrical Release Notes]

from rclpy.experimental import AsyncNode
import asyncio

class MyNode(AsyncNode):
    async def _cb(self):
        await self.get_clock().sleep(1.0)
        # 这里可以直接 await 你的 asyncua 调用

注:AsyncNodeexperimental API,接口可能微调。上面的线程方案不依赖 Lyrical 的 experimental AsyncNode API,兼容性更广;Lyrical 用户也可以进一步尝试 AsyncNode。


7. 双向通信(读 + 写)

上行(读 PLC → topic)已在第 6 节。下行(topic → 写 PLC)加一个订阅即可:

from asyncua import ua

# 在 __init__ 里加订阅
self.sub = self.create_subscription(Float64, '/plc/set_speed', self.on_cmd, 10)

def on_cmd(self, msg):
    if self.node is None:
        return
    async def _write():
        n = self.client.get_node('ns=3;s="DB_Data"."SetSpeed"')
        await n.write_value(ua.Variant(float(msg.data), ua.VariantType.Float))
    future = asyncio.run_coroutine_threadsafe(_write(), self.loop)
    future.result(timeout=1.0)

生产环境加证书

from asyncua import ua
from asyncua.crypto.security_policies import SecurityPolicyBasic256Sha256

await client.set_security(
    SecurityPolicyBasic256Sha256,
    certificate="client_cert.pem",
    private_key="client_key.pem",
    server_certificate="server_cert.der",
    mode=ua.MessageSecurityMode.SignAndEncrypt,
)

客户端证书用 OpenSSL 生成,导出 .der 后导入 TIA 的证书管理器并加入受信列表。[来源:controlbyte.tech]


8. 常见坑位排查表(重点)

现象 / 报错 原因 解决 来源
BadTcpEndpointUrlInvalid URL 用了 http://,或 4840 不通 opc.tcp://;放行 TCP 4840 controlbyte.tech
4840 连接被拒 OPC UA 没激活 / 没编译下载 / PLC 没 RUN / 防火墙 逐项检查 3.1、3.4 blanpa.github.io
BadSecurityChecksFailed 客户端与服务端安全策略不一致 测试用 None;生产两端都用 Basic256Sha256 并交换证书 blanpa.github.io
BadNodeIdUnknown DB 变量没勾「可访问」/ ns 错 / 字符串打错 勾 3.3;用 UaExpert 核对 ns;引号是 NodeId 一部分且大小写敏感 controlbyte.tech / blanpa.github.io
变量读不到(明明存在) DB 变量没启用 Accessible from HMI/OPC UA,或 NodeId / Namespace 错误 勾选属性;核对 NodeId 与 Namespace industrialmonitordirect
连接运行一段时间后断开 网络抖动 / NAT / 防火墙超时 / Server 主动关闭 Session 检查日志与网络链路,实现断线检测与重连;不要单纯靠改 session_timeout 掩盖问题 controlbyte.tech
写失败 BadUserAccessDenied 变量没勾 Writable from HMI/OPC UA,或 Guest 无写权限 勾「可写」;用有权限用户登录 industrialmonitordirect
Bad_TypeMismatch 数据类型对不上(如 BOOL 声明成 INT) 对齐变量类型,见第 9 节映射表 industrialmonitordirect
采样太密导致随机读取失败 采样率低于 PLC 最小采样间隔,缓冲区溢出 调低采样率 / 提高服务器缓存 tsight.io
浮点数精度异常(42.70000076293945) S7 Real 只有 ~7 位有效数字 需要精度用 LReal(对应 Double) blanpa.github.io

9. S7 数据类型 ↔ OPC UA 映射表

排查 Bad_TypeMismatch 和精度问题时对着看:

S7 类型 大小 OPC UA 类型 NodeId DataType 备注
Bool 1 bit Boolean i=1 直接映射
Byte 8 bit Byte i=3 无符号 0-255
Word 16 bit UInt16 i=5 无符号
DWord 32 bit UInt32 i=7 无符号
Int 16 bit Int16 i=4 有符号
DInt 32 bit Int32 i=6 有符号
LInt 64 bit Int64 i=8 仅 S7-1500
Real 32 bit Float i=10 IEEE 754 单精度
LReal 64 bit Double i=11 IEEE 754 双精度
String 变长 String i=12 S7 String 最多 254 字符

[来源:blanpa.github.io - Siemens S7 + OPC-UA 教程]


10. 总结

  • OPC UA 是 S7-1500 新固件上最稳的路线:原生支持、走符号名、免去 snap7 的 PUT/GET 和绝对地址烦恼。
  • TIA 侧三步别漏:激活服务器 → 配许可证 → 给变量勾 Accessible/Writable from HMI/OPC UA
  • 最快跑通路径:安全策略 None + asyncua 读一个变量,先确认链路,再上 ROS2 封装和证书。
  • NodeId 是常见坑之一:不要假定 namespace index 永远是 3,以 UaExpert 或服务器 NamespaceArray 实际结果为准;引号是 NodeId 的一部分、大小写敏感。

文中 IP 192.168.0.1、DB 名 DB_Data、变量名都是示例,替换成你自己的实际配置即可。

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐