ROS2 Lyrical 与西门子 S7-1500 通信实战:OPC UA 从 TIA 配置到 ROS2 Bridge
面向智能制造 / 工业机器人方向的开发者。本文带你用 OPC UA 打通 ROS2(Lyrical Luth)与西门子 S7-1500,从 TIA Portal 配置、Python 最小验证,到封装成 ROS2 节点,最后附一份可直接排查的坑位清单。
环境:TIA Portal V21 + S7-1500 + ROS2 Lyrical + Python
asyncua。文中关键结论均标注来源。
目录
- 为什么打通 ROS2 和 PLC,又为什么选 OPC UA
- 环境准备
- TIA Portal 侧配置(4 步,含许可证)
- 安全策略:先免加密跑通,再上证书
- Python 端最小测试(asyncua)
- 封装成 ROS2 节点
- 双向通信(读 + 写)
- 常见坑位排查表(重点)
- S7 数据类型 ↔ OPC UA 映射表
- 总结
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 调用
注:
AsyncNode属experimentalAPI,接口可能微调。上面的线程方案不依赖 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、变量名都是示例,替换成你自己的实际配置即可。
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐


所有评论(0)