用Python和pymodbus库模拟Modbus RTU主从通信（附完整代码与数据帧解析）

用Python和pymodbus库模拟Modbus RTU主从通信（附完整代码与数据帧解析）

在工业自动化领域，Modbus RTU协议因其简单可靠的特点，成为连接PLC、传感器和上位机系统的通用语言。本文将带您从零开始，使用Python的pymodbus库构建完整的Modbus RTU通信测试环境，通过代码实例深入解析协议底层的数据帧结构。

1. 环境搭建与基础配置

1.1 虚拟串口工具配置

在开始编码前，我们需要模拟物理串口环境。Windows平台推荐使用com0com虚拟串口工具，Linux/Mac可使用socat命令创建虚拟串口对：

# Linux/Mac虚拟串口创建 socat -d -d pty,raw,echo=0 pty,raw,echo=0

安装Python环境依赖：

pip install pymodbus==3.0.0 serial

注意：pymodbus 3.x版本重构了异步IO支持，本文示例基于同步API以保证兼容性

1.2 协议基础参数设置

Modbus RTU的关键参数需要主从设备严格一致：

2. 构建Modbus从站模拟器

2.1 初始化从站服务

from pymodbus.server.sync import StartSerialServer from pymodbus.datastore import ModbusSequentialDataBlock from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext def run_slave(): store = ModbusSlaveContext( di=ModbusSequentialDataBlock(0, [0]*100), # 离散输入 co=ModbusSequentialDataBlock(0, [0]*100), # 线圈 hr=ModbusSequentialDataBlock(0, [0]*100), # 保持寄存器 ir=ModbusSequentialDataBlock(0, [0]*100)) # 输入寄存器 context = ModbusServerContext(slaves=store, single=True) StartSerialServer( context, port='/dev/ttyS1', # 虚拟串口设备 baudrate=19200, timeout=1)

2.2 数据帧结构验证

当主站发送读取保持寄存器请求（功能码0x03）时，从站响应的原始数据帧可通过Wireshark捕获分析：

响应帧示例： 01 03 02 00 0A 45 CD

• 01：从站地址
• 03：功能码
• 02：数据字节数
• 00 0A：寄存器值（十进制10）
• 45 CD：CRC校验码

3. 实现Modbus主站功能

3.1 同步读取操作实例

from pymodbus.client.sync import ModbusSerialClient client = ModbusSerialClient( method='rtu', port='/dev/ttyS0', baudrate=19200, timeout=1) # 读取保持寄存器 result = client.read_holding_registers( address=0, # 起始地址 count=5, # 寄存器数量 unit=1) # 从站地址 print("寄存器值:", result.registers)

3.2 数据帧构造原理

底层请求帧的构建过程：

def build_read_frame(slave_id, function_code, address, count): # 构造PDU pdu = bytes([function_code]) + \ address.to_bytes(2, 'big') + \ count.to_bytes(2, 'big') # 添加CRC校验 crc = compute_crc(bytes([slave_id]) + pdu) return bytes([slave_id]) + pdu + crc.to_bytes(2, 'little') # 示例：构造读取保持寄存器请求 frame = build_read_frame( slave_id=1, function_code=0x03, address=0, count=5) print("原始请求帧:", frame.hex())

4. 高级功能实现与调试

4.1 批量写入操作

# 写入多个保持寄存器（功能码0x10） values = [10, 20, 30, 40] result = client.write_registers( address=0, values=values, unit=1) if result.isError(): print("写入失败:", result) else: print("写入成功")

对应的请求帧解析：

01 10 00 00 00 04 08 00 0A 00 14 00 1E 00 28 42 1F

• 00 00：起始地址
• 00 04：寄存器数量
• 08：数据字节长度
• 后续8字节为4个寄存器的值

4.2 异常处理机制

Modbus协议定义的异常响应格式：

捕获异常响应的代码实现：

try: response = client.read_coils(0, 10, unit=1) if response.isError(): print("Modbus异常码:", response.exception_code) except Exception as e: print("通信错误:", str(e))

5. 协议分析工具链

5.1 使用Wireshark解码Modbus RTU

配置步骤：

1. 安装Wireshark并启用串口捕获
2. 添加Modbus RTU解析器
3. 过滤语法：modbus || modbusadu

典型数据包分析：

Frame 1 (Request): 0000 01 03 00 00 00 02 44 09 ......D. Frame 2 (Response): 0000 01 03 04 00 0a 00 14 f5 33 ......ó3

5.2 自定义数据监控工具

import matplotlib.pyplot as plt from collections import deque class ModbusMonitor: def __init__(self, max_points=50): self.data = deque(maxlen=max_points) def update(self, values): self.data.extend(values) plt.clf() plt.plot(self.data) plt.pause(0.01) # 使用示例 monitor = ModbusMonitor() while True: result = client.read_holding_registers(0, 5, unit=1) monitor.update(result.registers)

6. 性能优化实践

6.1 多寄存器读取优化

# 批量读取策略对比 single_read_time = timeit.timeit( lambda: client.read_holding_registers(0, 1, unit=1), number=100) batch_read_time = timeit.timeit( lambda: client.read_holding_registers(0, 10, unit=1), number=100) print(f"单点读取耗时: {single_read_time:.3f}s") print(f"批量读取耗时: {batch_read_time:.3f}s")

6.2 连接池管理

from pymodbus.client.sync import ModbusConnectionPool pool = ModbusConnectionPool( client_class=ModbusSerialClient, max_size=5, **client_params) def safe_read(address): with pool.acquire() as conn: return conn.read_holding_registers(address, 1)

在实际项目中，通过合理设置超时时间和重试机制，可以将通信成功率提升至99.9%以上。一个常见的经验是，对于19200波特率的网络，单个请求的超时时间不应低于300ms。