python pymodbus模块的具体使用_Python

一、详解

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 0 \leq \text{address} \leq 65535 0≤address≤65535）。
功能码：modbus操作使用功能码（function code），例如：
- 读线圈：功能码1。
- 写单个寄存器：功能码6。
- 完整列表可参考modbus标准文档。

pymodmos库实现了这些概念，支持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支持更复杂场景：

异步模式：使用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())

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

性能优化建议：

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

6、注意事项

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

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

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

二、代码示例

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)