3、Modbus协议实战:使用Python的modbus_tk库实现Modbus RTU主站、使用pymodbus库实现Modbus TCP从站
本节我们将通过两个具体的Python库,分别实现Modbus RTU主站和Modbus TCP从站。这是协议转换实战中最核心的环节:一端从物理串口读取RTU数据,另一端通过以太网提供TCP服务。
3.1 环境准备与库安装
在开始编码前,请确保已安装以下Python库:
pip install modbus_tk pymodbus pyserial
- modbus_tk:轻量级Modbus协议库,支持RTU/ASCII/TCP,适合主站开发。
- pymodbus:功能全面的Modbus库,支持异步/同步模式,适合从站开发。
- pyserial:modbus_tk的RTU模式依赖此库进行串口通信。
3.2 使用modbus_tk实现Modbus RTU主站
以下代码实现一个RTU主站,读取从站地址为1的保持寄存器(功能码03)。
import serial
import modbus_tk
import modbus_tk.defines as cst
from modbus_tk import modbus_rtu
# 配置串口参数
serial_port = serial.Serial(
port='COM3', # Windows下串口号,Linux下如'/dev/ttyUSB0'
baudrate=9600,
bytesize=8,
parity='N',
stopbits=1,
xonxoff=False
)
# 创建RTU主站
master = modbus_rtu.RtuMaster(serial_port)
master.set_timeout(3.0) # 设置超时时间(秒)
master.set_verbose(True) # 开启调试日志
try:
# 读取从站ID=1,起始地址0,读取10个保持寄存器
data = master.execute(
slave=1,
function_code=cst.READ_HOLDING_REGISTERS,
starting_address=0,
quantity_of_x=10,
output_value=0
)
print("读取到的寄存器值:", data)
# 写入单个寄存器(功能码06)
master.execute(
slave=1,
function_code=cst.WRITE_SINGLE_REGISTER,
starting_address=5,
output_value=1234
)
print("写入成功")
except modbus_tk.modbus.ModbusError as e:
print(f"Modbus错误: {e}")
except Exception as e:
print(f"通信异常: {e}")
finally:
master.close()
serial_port.close()
关键点说明:
RtuMaster需要传入一个已打开的serial.Serial对象。execute方法的function_code使用cst模块中的常量,如READ_HOLDING_REGISTERS、WRITE_SINGLE_REGISTER。- 对于写操作,
output_value参数传入要写入的值;对于读操作,该参数可设为0。 - 务必在
finally块中关闭主站和串口,释放资源。
3.3 使用pymodbus实现Modbus TCP从站
以下代码创建一个TCP从站,监听502端口,提供保持寄存器数据供主站读写。
from pymodbus.server import StartTcpServer
from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext
from pymodbus.device import ModbusDeviceIdentification
import logging
# 配置日志(可选)
logging.basicConfig()
logging.getLogger('pymodbus').setLevel(logging.DEBUG)
# 创建数据存储上下文
# 参数顺序:di, co, hr, ir (离散输入、线圈、保持寄存器、输入寄存器)
# 这里初始化100个保持寄存器,默认值为0
store = ModbusSlaveContext(
di=None, # 离散输入,不初始化
co=None, # 线圈,不初始化
hr={i: 0 for i in range(100)}, # 保持寄存器0-99,初始值0
ir=None # 输入寄存器,不初始化
)
# 创建服务器上下文(单从站模式)
context = ModbusServerContext(slaves=store, single=True)
# 可选:设置设备标识
identity = ModbusDeviceIdentification()
identity.VendorName = 'OPC UA Expert'
identity.ProductCode = 'M2O'
identity.VendorUrl = 'http://example.com'
identity.ProductName = 'Modbus TCP Slave'
identity.ModelName = 'Simulator'
identity.MajorMinorRevision = '1.0'
# 启动TCP服务器,监听所有网卡的502端口
if __name__ == "__main__":
print("Modbus TCP 从站启动,监听端口 502...")
StartTcpServer(
context=context,
identity=identity,
address=("0.0.0.0", 502) # 监听所有IP
)
关键点说明:
ModbusSlaveContext的四个参数对应四种数据区,传入字典可初始化数据。本例只初始化了保持寄存器(hr)。single=True表示所有从站ID(1-247)共享同一个数据存储,适合简单模拟场景。- 默认端口502需要管理员/root权限。在Windows上需以管理员身份运行;Linux下使用
sudo。 - 若端口被占用,可更换为其他端口(如 1502),但需确保主站连接时指定相同端口。
3.4 实战验证:主站连接从站
将上述两个程序分别运行:
- 先启动TCP从站(监听502端口)。
- 修改RTU主站代码,将串口通信改为TCP通信(仅用于验证,实际场景中RTU主站应连接串口设备)。
若需在单台电脑上测试,可将RTU主站改为TCP主站,连接本机从站:
# 使用modbus_tk的TCP主站替代RTU主站进行测试
from modbus_tk import modbus_tcp
master = modbus_tcp.TcpMaster(host="127.0.0.1", port=502)
master.set_timeout(3.0)
# 后续读取/写入操作与RTU主站相同
data = master.execute(1, cst.READ_HOLDING_REGISTERS, 0, 10)
print(data)
常见问题排查:
| 现象 | 可能原因 | 解决方案 |
|---|---|---|
| RTU主站超时 | 串口参数不匹配、从站地址错误 | 核对波特率、数据位、校验位;确认从站ID |
| TCP从站启动失败 | 端口502被占用、权限不足 | 使用 netstat -ano 检查端口;以管理员运行 |
| 读取数据全为0 | 从站未初始化数据、地址越界 | 检查从站数据存储初始化范围;确认起始地址 |
| 写入不生效 | 写入地址为只读区域 | 保持寄存器可读写,线圈可读写;离散输入和输入寄存器只读 |
3.5 本节小结
- 使用 modbus_tk 的
RtuMaster可快速实现RTU主站,需配合pyserial管理串口。 - 使用 pymodbus 的
StartTcpServer可搭建TCP从站,通过ModbusSlaveContext管理数据区。 - 两种库的API风格不同,但都遵循Modbus协议标准,功能码和数据地址映射一致。
- 在协议转换网关中,RTU主站负责采集现场设备,TCP从站负责将数据暴露给上位机或OPC UA服务器。