Python PyModbus 模块详解

环境要求：

• Python 3.6+。
• 对于 Modbus RTU，需物理串口或虚拟串口工具（如 pyserial）。
• 测试工具推荐：使用 Modbus 模拟软件（如 Modbus Poll）验证通信。

4、使用示例

以下示例展示常见场景，代码基于同步模式（异步模式类似，使用 asyncio）。

示例 1: Modbus TCP 客户端（读取保持寄存器）
作为客户端，连接到服务器并读取数据。

from pymodbus.client import ModbusTcpClient

# 连接到服务器（假设 IP: 127.0.0.1, 端口: 502）
client = ModbusTcpClient('127.0.0.1', port=502)
client.connect()

# 读取保持寄存器（地址 0，数量 2 个寄存器）
result = client.read_holding_registers(address=0, count=2, unit=1)
# unit 为从站 ID
if not result.isError():
    print("读取成功:", result.registers)
    # 输出寄存器值列表，如 [100, 200]
else:
    print("错误:", result)

client.close() # 关闭连接

示例 2: Modbus TCP 服务器（模拟从站）
创建服务器并存储数据。

from pymodbus.server import StartTcpServer
from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext
from pymodbus.datastore import ModbusSequentialDataBlock

# 初始化数据存储：设置保持寄存器（地址 0-9，初始值 0）
store = ModbusSlaveContext(
    hr=ModbusSequentialDataBlock(0, [0]*10) # hr 表示保持寄存器
)
context = ModbusServerContext(slaves=store, single=True)
# single 模式为单从站

# 启动服务器（端口 502）
StartTcpServer(context=context, address=("0.0.0.0", 502))

示例 3: Modbus RTU 通信（使用串口）
适用于串行设备。

from pymodbus.client import ModbusSerialClient

# 配置串口（端口 COM1, 波特率 9600）
client = ModbusSerialClient(method='rtu', port='COM1', baudrate=9600)
client.connect()

# 写单个线圈（地址 0，值 True）
client.write_coil(address=0, value=True, unit=1)
client.close()

5、高级特性

pymodbus 支持更复杂场景：

• 自定义数据处理器：重写 ModbusRequestHandler 处理特定逻辑。
• 安全扩展：支持 TLS 加密（Modbus TCP）或自定义认证。
• 诊断工具：内置日志记录（import logging），便于调试通信问题。

异步模式：使用 AsyncModbusTcpClient 结合 asyncio，提高并发性能。

import asyncio
from pymodbus.client import AsyncModbusTcpClient

async def main():
    client = AsyncModbusTcpClient('127.0.0.1', port=502)
    await client.connect()
    result = await client.read_holding_registers(0, 2)
    print(result.registers)
    await client.close()

asyncio.run(main())

性能优化建议：

• 使用连接池管理客户端实例。
• 批量读写减少请求次数（如 read_holding_registers 中 count 参数）。
• 在资源受限设备上，启用精简模式（禁用不必要功能）。

6、注意事项

• 协议限制：
  • Modbus RTU 要求严格时序（如 3.5 字符间隔）。
  • 地址溢出：确保地址在有效范围（如超出 65535 会报错）。
• 常见问题：
  • 连接超时：检查网络或串口配置。
  • 数据不一致：验证字节序（pymodbus 默认大端序）。
  • 性能瓶颈：在高频通信中，优先使用异步模式。
• 资源：
  • 官方文档：pymodbus GitHub
  • 社区支持：通过 GitHub Issues 获取帮助。

错误处理：pymodbus 可能抛出异常（如 ModbusIOException），需捕获处理：

try:
    result = client.read_coils(0, 1)
except Exception as e:
    print("通信错误:", e)

二、代码示例

1、modbusTCP client

from time import sleep
from pymodbus.client import ModbusTcpClient

client = ModbusTcpClient('127.0.0.1')
client.connect()
while True:
    res = client.read_holding_registers(slave=1, address=0, count=10)
    print(res.registers)
    sleep(0.5)

2、modbusRTU client

from time import sleep
from pymodbus.client import ModbusSerialClient

client = ModbusSerialClient(method='rtu', port='COM1', baudrate=9600, timeout=1)
connection = client.connect()
while True:
    res = client.read_holding_registers(slave=1, address=0, count=10)
    print(res.registers)
    sleep(0.5)