pymodbus连接西门子PLC：项目应用实例
用Python玩转工业现场：pymodbus直连西门子PLC实战手记
最近在做一个边缘数据采集项目，客户用的是西门子S7-1200 PLC，但不想上SCADA系统，只想把关键工艺参数（温度、压力、运行状态）实时传到云端做可视化和预警分析。他们问我：“能不能不用OPC UA？搞个轻量点的方案。”我想了想——pymodbus + Modbus TCP，不就是为这种场景而生的吗？
于是，我花了三天时间打通了这条链路：从TIA Portal配置开始，到Python脚本稳定读取寄存器，再到数据上传InfluxDB。过程中踩了不少坑，也总结出一套可复用的经验。今天就来分享这个“零成本打通IT与OT层”的真实案例。
为什么选 pymodbus？而不是 Snap7 或 OPC UA？
先说结论：如果你要快速搭一个数据代理服务（data agent），又不想被工控软件绑架，那pymodbus是目前最合适的工具之一。
| 方案 | 开发效率 | 学习成本 | 部署灵活性 | 是否需要授权 |
|---|---|---|---|---|
| pymodbus (Modbus TCP) | ⭐⭐⭐⭐☆ | ⭐⭐ | ⭐⭐⭐⭐⭐ | 否（开源） |
| Snap7（S7协议） | ⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐ | 否（开源） |
| OPC UA Client | ⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐ | 常需许可证 |
别误会，Snap7 很强大，能直接访问S7的M区、DB块，性能也好。但它依赖C库，在树莓派或Docker容器里容易出现兼容性问题；而OPC UA虽然标准高、安全性强，但配置复杂，中间件一上，运维难度翻倍。
相比之下，pymodbus的优势太明显了：
- 纯Python，安装一条命令搞定：
pip install pymodbus - 支持异步（asyncio）、多线程、批量读写
- 可以轻松对接 Pandas、Flask、FastAPI、MQTT……整个Python生态任你调用
- 跨平台跑在Windows、Linux、树莓派甚至Jetson Nano都没问题
关键是——西门子PLC原生支持Modbus TCP服务器功能，只要固件版本够新，根本不需要额外网关！
第一步：让西门子S7-1200当Modbus从站
很多人以为西门子只认S7通信，其实不然。从STEP 7 V14开始，S7-1200/1500就可以通过指令库启用Modbus TCP Server 模式。
在TIA Portal中怎么配？
- 打开项目 → 添加“Modbus”指令块（在指令表搜索
MB_SERVER） - 将该块拖入主程序循环（OB1）
- 配置参数：
-Mode: 设置为1（TCP服务器模式）
-Port: 默认502
-MaxConnections: 最大连接数（一般设为2就够了）
-FirstSlaveReg: 映射起始地址（比如DB1.DBW0） - 下载程序并重启CPU
✅ 提示：确保PLC IP地址已正确设置（如
192.168.0.10），且与上位机在同一网段。
此时，你的PLC就已经是一个标准的Modbus服务器了，等待客户端来“问话”。
第二步：地址映射必须搞清楚！否则读出来全是错的
这是最容易翻车的地方——Modbus地址和西门子内部地址不是一一对应的！
举个例子：
| Modbus 地址类型 | 功能码 | 对应西门子区域 | 起始偏移说明 |
|---|---|---|---|
| Coils (0x) | 0x01 | Q 区（输出点） | Coil 0 → Q0.0 |
| Discrete Inputs (1x) | 0x02 | I 区（输入点） | DI 0 → I0.0 |
| Input Registers (3x) | 0x04 | AI/AQ（模拟量输入） | IR1000 → IW1000 |
| Holding Registers (4x) | 0x03 | M区、DB块、V存储器等 | HR1 → MD0 或 DB1.DBW0 |
重点来了：
假设你在PLC里有个变量存放在DB1.DBW20（即第20个字），你想通过Modbus读取它，该怎么映射？
答案是：HR 地址 = DB1起始偏移 + 字索引 / 2
比如你把DB1.DBW0映射为 HR1，则：
-DB1.DBW0→ HR1
-DB1.DBW2→ HR2
-DB1.DBW20→ HR11
⚠️ 注意：每个“寄存器”占2字节（16位），所以地址是以“字”为单位递增的。
我在实际项目中专门建了一个Excel表格来做地址对照，避免混乱：
| Modbus HR 地址 | 数据类型 | PLC 地址 | 含义 |
|---|---|---|---|
| 1 | INT | DB1.DBW0 | 温度值（℃） |
| 2 | INT | DB1.DBW2 | 压力值（kPa） |
| 3-4 | REAL | DB1.DBW4 | 流量（m³/h） |
| 5 | BOOL | DB1.DBX10.0 | 故障标志位 |
这样，代码一写就知道该读哪个位置。
第三步：Python代码怎么写？这才是核心
下面是我最终落地的精简版代码，已经用于生产环境一周无故障运行。
from pymodbus.client import ModbusTcpClient from pymodbus.payload import BinaryPayloadDecoder from pymodbus.constants import Endian import logging import time # 日志配置 logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[logging.FileHandler("plc_reader.log"), logging.StreamHandler()] ) log = logging.getLogger(__name__) # 连接参数 PLC_IP = "192.168.0.10" PORT = 502 SLAVE_ID = 1 # 必须与MB_SERVER中的从站ID一致 RETRY_TIMES = 3 TIMEOUT = 3.0 class SiemensModbusReader: def __init__(self): self.client = ModbusTcpClient(PLC_IP, port=PORT, timeout=TIMEOUT) def connect(self): """建立连接，带重试机制""" for i in range(RETRY_TIMES): try: if self.client.connect(): log.info(f"✅ 成功连接至PLC {PLC_IP}") return True else: log.warning(f"🔁 第{i+1}次连接失败，2秒后重试...") time.sleep(2) except Exception as e: log.error(f"❌ 连接异常: {e}") time.sleep(2) return False def read_sensor_data(self): """读取传感器数据：温度、压力、流量（REAL）、故障标志""" try: # 一次性读取前5个保持寄存器（HR1~HR5） result = self.client.read_holding_registers(address=0, count=5, slave=SLAVE_ID) if result.isError(): log.error(f"⚠️ Modbus错误响应: {result}") return None registers = result.registers # 返回 [val1, val2, val3, val4, val5] decoder = BinaryPayloadDecoder.fromRegisters( registers, byteorder=Endian.Big, # 字节序：大端 wordorder=Endian.Big # 双寄存器排序：高位在前 ) temp_c = decoder.decode_16bit_int() # HR1: 温度 pressure_kpa = decoder.decode_16bit_uint() # HR2: 压力 flow = decoder.decode_32bit_float() # HR3-HR4: 流量（REAL） fault_bit = (registers[4] & 0x01) == 1 # HR5: 低字节第一位表示故障 return { "temperature": temp_c, "pressure": pressure_kpa, "flow_rate": round(flow, 2), "fault": fault_bit, "timestamp": time.time() } except Exception as e: log.exception(f"📊 数据解析失败: {e}") return None def close(self): self.client.close() # 主循环 if __name__ == "__main__": reader = SiemensModbusReader() if not reader.connect(): log.critical("⛔ 所有重试均失败，程序退出") exit(1) try: while True: data = reader.read_sensor_data() if data: log.info(f"📈 获取数据 → {data}") # 此处可接入 MQTT、InfluxDB、REST API... else: log.warning("📭 未能获取有效数据") time.sleep(0.5) # 控制采样频率：2Hz except KeyboardInterrupt: log.info("👋 用户中断，安全退出") finally: reader.close()关键点解读：
address=0表示读HR1？
- 是的。pymodbus中address是从0开始计数的，所以HR1对应address=0，HR100对应address=99。字节序问题解决了吗？
- 西门子默认使用大端（Big-Endian），所以我们设置byteorder=Endian.Big和wordorder=Endian.Big。
- 如果发现浮点数读出来是乱码（比如1.2e-38），八成是字节序错了，试试交换wordorder。为什么一次读多个寄存器？
- 减少网络请求次数，提升效率。Modbus协议每次通信都有固定开销，批量读更高效。心跳和断线重连都做了吗？
- 有！连接失败会自动重试3次；
- 主循环每500ms轮询一次，相当于心跳监测；
- 异常全捕获，不会因单次错误导致程序崩溃。
实战中遇到的问题及解决方案
❌ 问题1：频繁超时，偶尔连接不上
现象：日志里时不时出现Connection timed out
排查过程：
- 用Wireshark抓包发现，TCP握手阶段没问题，但PLC回了RST
- 登录TIA Portal一看，原来PG在线调试占用了大量通信资源
解决办法：
- 关闭所有HMI连接和编程设备监控
- 在MB_SERVER块中降低扫描周期（默认100ms→200ms）
- 客户端增加连接池管理（后续升级方向）
❌ 问题2：浮点数读出来总是0.0或极大值
原因：字节顺序没对齐！
例如：PLC写入3.14（十六进制4048F5C3），如果客户端按小端解析，就会变成9.6e-39。
验证方法：
在PLC中手动给DB块赋值3.14，然后用不同组合测试解码方式：
decoder.decode_32bit_float() # Big-Big → 正确 decoder.decode_32bit_float() with wordorder=Little → 错误最后确认：必须使用 Big-Big 模式。
💡 秘籍：可在TIA Portal中勾选“Network Byte Order”，强制统一字节序。
❌ 问题3：某些寄存器读不到数据
可能原因：
- 地址越界（比如试图读HR1000，但映射只到HR50）
- 数据块未初始化（DB块没有激活或清零）
- 权限不足（某些区域禁止外部访问）
建议做法：
- 先用Modbus Poll这类工具测试通路是否正常
- 再比对地址映射表，逐项验证
设计优化建议（来自血泪经验）
不要频繁创建/销毁客户端
每次connect()都是一次TCP建连，开销大。应保持长连接，定期心跳检测。尽量批量读取，减少请求数
与其发5次单寄存器读，不如1次读10个寄存器。加缓存机制防断连黑屏
即使短暂断开，前端也能显示“最后有效值”，体验更好。限制写权限，只读最安全
除非必要，不要开放写操作。万一误写控制位，可能导致停机！网络隔离 + 防火墙规则
给PLC划分独立VLAN，仅允许边缘主机IP访问502端口。考虑未来扩展性
把地址映射做成JSON配置文件，方便后期维护：
{ "temp": {"type": "int16", "addr": 0}, "pressure": {"type": "uint16", "addr": 1}, "flow": {"type": "float32", "addr": 2} }结语：这不是玩具，是真正的工业级解决方案
当我看到第一个{"temperature": 87, "flow_rate": 12.5}被成功插入InfluxDB，并在Grafana上画出曲线时，我知道这条路走通了。
这套方案已经在两个小型水处理站部署，每天稳定采集超过10万条记录。它不追求极致性能，也不替代DCS系统，而是作为一个低成本、高灵活性的数据桥梁，把沉默的PLC变成可感知、可分析的智能节点。
如果你也在做类似项目，不妨试试这条路。无需昂贵授权，无需专用硬件，只需几行Python，就能让你的PLC开口说话。
📣 如果你正在尝试pymodbus连接西门子PLC，欢迎留言交流具体问题。我可以分享完整的地址映射模板、Docker部署脚本和报警逻辑设计。