Python Modbus 通讯攻略

Python Modbus 通讯全攻略 (基于 pymodbus)

1. 简介与准备工作

什么是 Modbus？

Modbus 是一种工业通讯协议，主要分为两种传输模式：

• Modbus TCP: 运行在以太网之上，使用 IP 地址和端口（默认 502）。
• Modbus RTU: 运行在串行通讯（如 RS-485/RS-232）之上。

为什么选择 pymodbus？

虽然像 minimalmodbus 这样的库在 RTU 方面很简单，但 pymodbus 是最全面的：

• 同时支持 Client (主站) and Server (从站)。
• 同时支持 TCP, UDP, Serial (RTU/ASCII)。
• 支持异步 (Asyncio) 和同步模式。
• 提供强大的数据载荷处理（Payload Builder/Decoder）工具，用于处理浮点数和长整型。

环境安装

确保你已经安装了 Python (建议 3.8+)。在终端执行以下命令安装库：

Bash

pip install pymodbus pyserial

(注意：pyserial 是运行 Modbus RTU 所必需的依赖)

2. 核心概念速查表

在编写代码前，你需要了解 Modbus 的功能码和数据类型：

3. 实战一：Modbus TCP Client (以太网通讯)

这是最常见的场景：Python 作为主站（Client），去读取 PLC 或远程 I/O 模块（Server）的数据。

Python

from pymodbus.client import ModbusTcpClient
import time

# 1. 配置连接参数
# 请替换为你设备的实际 IP
IP_ADDRESS = '192.168.1.10' 
PORT = 502

# 2. 创建客户端实例
client = ModbusTcpClient(IP_ADDRESS, port=PORT)

try:
    # 3. 建立连接
    connection = client.connect()
    if connection:
        print(f"成功连接到 {IP_ADDRESS}:{PORT}")

        # --- 读取操作 ---
        # 例子：读取从地址 100 开始的 10 个保持寄存器 (Function Code 03)
        # slave=1 是从站 ID (Unit ID)，在 TCP 中通常默认为 1 或 0，但也可能被设备指定
        rr = client.read_holding_registers(address=100, count=10, slave=1)

        if not rr.isError():
            print("读取成功:", rr.registers)
            # rr.registers 返回的是一个整数列表，例如 [123, 0, 45, ...]
        else:
            print("读取失败:", rr)

        # --- 写入操作 ---
        # 例子：向地址 100 写入数值 55 (Function Code 06)
        wr = client.write_register(address=100, value=55, slave=1)

        if not wr.isError():
            print("写入成功")
        else:
            print("写入失败:", wr)

    else:
        print("连接失败")

finally:
    # 4. 关闭连接
    client.close()

4. 实战二：Modbus RTU Client (串口/RS-485通讯)

场景：使用 USB 转 RS-485 适配器连接传感器。

Python

from pymodbus.client import ModbusSerialClient

# 1. 配置串口参数
# Windows下通常是 'COM3', 'COM4' 等
# Linux下通常是 '/dev/ttyUSB0' 或 '/dev/ttyS0'
COM_PORT = 'COM3' 

client = ModbusSerialClient(
    port=COM_PORT,
    baudrate=9600,        # 波特率
    bytesize=8,           # 数据位
    parity='N',           # 校验位: 'N'-None, 'E'-Even, 'O'-Odd
    stopbits=1,           # 停止位
    timeout=1             # 超时时间(秒)
)

try:
    if client.connect():
        print(f"串口 {COM_PORT} 连接成功")

        # 读取从站地址为 1 的设备的寄存器
        # 读取地址 0 开始的 2 个寄存器
        res = client.read_holding_registers(address=0, count=2, slave=1)

        if not res.isError():
            print(f"收到数据: {res.registers}")
        else:
            print(f"读取错误: {res}")
    else:
        print(f"无法打开串口 {COM_PORT}")

finally:
    client.close()

5. 高级话题：处理复杂数据类型 (Float/32-bit Int)

Modbus 寄存器本身只有 16 位（0-65535）。如果你要读取一个 32位浮点数 (Float) 或 32位整数 (Int32)，数据会被拆分成两个寄存器。

你需要处理 字节序 (Endianness) 和 字序 (Word Order)。pymodbus 提供了 BinaryPayloadDecoder 和 BinaryPayloadBuilder 来完美解决这个问题。

读取浮点数示例 (解码)

假设设备在地址 200 存放了一个 Float (占2个寄存器)。

Python

from pymodbus.constants import Endian
from pymodbus.payload import BinaryPayloadDecoder
from pymodbus.client import ModbusTcpClient

client = ModbusTcpClient('192.168.1.10')
client.connect()

# 1. 读取原始寄存器 (Float 需要读取 2 个寄存器 = 32位)
rr = client.read_holding_registers(address=200, count=2, slave=1)

if not rr.isError():
    # 2. 使用解码器处理数据
    # byteorder 和 wordorder 取决于你的 PLC 厂商
    # Big Endian (大端) 是最常见的网络标准
    decoder = BinaryPayloadDecoder.fromRegisters(
        rr.registers, 
        byteorder=Endian.Big, 
        wordorder=Endian.Big
    )

    # 3. 解析为 32位 浮点数
    decoded_value = decoder.decode_32bit_float()
    print(f"解码后的温度值: {decoded_value:.2f}")

client.close()

写入浮点数示例 (编码)

Python

from pymodbus.constants import Endian
from pymodbus.payload import BinaryPayloadBuilder

# 1. 创建构建器
builder = BinaryPayloadBuilder(byteorder=Endian.Big, wordorder=Endian.Big)

# 2. 添加数据
desired_value = 123.45
builder.add_32bit_float(desired_value)

# 3. 获取 payload (这是准备写入的 16位 寄存器列表)
payload = builder.to_registers()
print(f"转换后的寄存器值: {payload}") 
# 输出类似: [17260, 52429]

# 4. 写入设备
# skip_encode=True 告诉 pymodbus 我们已经把数据处理好了，直接发
client.write_registers(address=200, values=payload, slave=1, skip_encode=True)

6. 常见陷阱与调试指南

1. 地址偏移 (Off-by-one Error): - 现象: 你想读地址 40001，但读到了错误的数据。 - 原因: Modbus 协议地址通常从 0 开始，但 PLC 文档通常从 1 开始。 - 解决: 如果 PLC 文档写地址是 40001 (Holding Register 1)，你在代码中通常需要填 address=0。如果不对，尝试 address=1。
2. 字节序/字序混乱: - 现象: 读取到的数字非常巨大或极小，或者完全不对（如 0.0）。 - 解决: 调整 BinaryPayloadDecoder 中的 byteorder 和 wordorder。常见的组合有 Big/Big (Modicon), Little/Little (Intel), 或 Big/Little (Swap)。
3. 连接超时: - 在创建 Client 时增加 timeout 参数。 - 如果是 RTU，检查波特率、校验位是否与设备完全一致。
4. Unit ID (Slave ID): - 对于 Modbus TCP，很多设备忽略 Unit ID（默认为0或255），但通过网关连接 RTU 设备时，Unit ID 必须 正确，否则网关不知道把指令转发给哪个子设备。

7. 快速构建一个 Modbus 模拟器 (Server)

如果你手头没有 PLC，可以用 Python 电脑变成一个 Modbus 服务器来测试你的 Client 代码。

Python

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

def run_server():
    # 1. 定义存储区
    # 创建 0-100 的地址空间，初始值为 0
    store = ModbusSlaveContext(
        di=ModbusSequentialDataBlock(0, [0]*100), # 离散输入
        co=ModbusSequentialDataBlock(0, [0]*100), # 线圈
        hr=ModbusSequentialDataBlock(0, [0]*100), # 保持寄存器
        ir=ModbusSequentialDataBlock(0, [0]*100)  # 输入寄存器
    )

    # single=True 表示只模拟一个从站 ID
    context = ModbusServerContext(slaves=store, single=True)

    print("启动 Modbus TCP 服务器在 localhost:5020...")
    # 2. 启动服务 (注意：端口使用 5020，因为 502 通常需要管理员权限)
    StartTcpServer(context=context, address=("localhost", 5020))

if __name__ == "__main__":
    run_server()