Python PyModbus 模块详解

一、详解

pymodbus 是一个基于 Python 的开源库，用于实现 Modbus 通信协议。Modbus 是一种广泛应用于工业自动化领域的串行通信协议，支持主从架构（客户端 - 服务器模型）。pymodbus 库提供了完整的 Modbus 客户端（主站）和服务器（从站）功能，支持多种传输方式，包括 Modbus RTU（串行通信）和 Modbus TCP（以太网通信）。

1、基础概念

Modbus 协议定义了一套标准的数据模型和通信规则：

数据模型：使用寄存器存储数据，分为：
- 线圈（Coils）：1 位读写寄存器，用于布尔值（如开关状态）。
- 离散输入（Discrete Inputs）：1 位只读寄存器。
- 保持寄存器（Holding Registers）：16 位读写寄存器，用于整数值（如传感器读数）。
- 输入寄存器（Input Registers）：16 位只读寄存器。
地址范围：每个寄存器类型有独立地址空间，范围从 0 到 65535（即 0 ≤ address ≤ 65535）。
功能码：Modbus 操作使用功能码（Function Code），例如：
- 读线圈：功能码 1。
- 写单个寄存器：功能码 6。
- 完整列表可参考 Modbus 标准文档。

pymodbus 库实现了这些概念，支持 Python 3.6 及以上版本，并兼容多种操作系统（Windows、Linux、macOS）。

2、核心功能

pymodbus 提供模块化设计，主要组件包括：

客户端（Client）：作为主站发起请求，支持同步和异步模式。
- 支持 Modbus TCP 和 RTU 传输。
- 提供读写 API，如 read_coils、write_register。
服务器（Server）：作为从站响应请求，可模拟设备行为。
- 支持数据存储上下文（如 ModbusServerContext）。
- 内置线程或异步事件处理。
协议处理：
- 自动处理 Modbus PDU（协议数据单元）格式。
- 支持异常响应（如非法地址时返回错误码）。
扩展性：允许自定义数据处理器和传输层（如通过串口或 Socket）。

关键优势：

轻量级：纯 Python 实现，无需外部依赖。
灵活：支持同步（pymodbus.client）和异步（pymodbus.async）编程。
兼容性：遵循 Modbus 标准，可与 PLC、HMI 等工业设备互操作。

3、安装与设置

安装 pymodbus 简单，通过 pip 命令即可：

pip install pymodbus

安装后，导入所需模块：

from pymodbus.client import ModbusTcpClient # TCP 客户端
from pymodbus.server import StartTcpServer # TCP 服务器
from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext # 数据存储

环境要求：

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)

Python PyModbus 模块详解