pymodbus 部署指南：Python Modbus 协议栈从安装到工业采集

一、项目介绍

pymodbus 是一个用 Python 编写的完整 Modbus 协议栈，提供客户端（Client）、服务器（Server）与模拟器（Simulator）三种角色，支持同步与异步 API。Modbus 是工业自动化领域最经典的串行通信协议，广泛应用于 PLC、RTU、电表、变频器、传感器、楼宇自控等设备之间的通信。pymodbus 让开发者不需要深入理解 Modbus 帧格式，就能用几行 Python 代码读写寄存器、监听设备数据、模拟从站设备。

pymodbus 支持所有标准 Modbus 帧格式：TCP、RTU、RTU over TCP、ASCII、UDP、TLS，支持串行（RS-485）与网络传输，支持自定义功能码，支持二进制位（coil）与 16 位寄存器（holding/input register）读写。它被集成在 Home Assistant、 ThingsBoard、开源 SCADA 系统等项目中，是 Python 生态中最成熟的 Modbus 库。

项目当前由 pymodbus-dev 社区维护（原仓库 pymodbus/pymodbus 已迁移到 pymodbus-dev/pymodbus），最新稳定版为 3.16.x，要求 Python >= 3.10。采用 BSD 许可证，文档位于 pymodbus.readthedocs.io。

pymodbus 的另一大优势是轻量：核心库没有第三方依赖，整个包只有几百 KB，可以轻松部署在资源受限的边缘设备（树莓派、OpenWrt 路由器、工业网关）上。它的服务器模式甚至可以跑在几十 MB 内存的设备上，模拟出数百台虚拟从站，用于高并发压测。

典型使用场景：智能电表数据采集、PLC 程序监控、变频器远程控制、楼宇暖通空调（HVAC）数据上报、IoT 网关把 Modbus 设备接入 MQTT/云平台、自动化测试中模拟 Modbus 从站。

Modbus 协议由 Modicon（现施耐德电气）于 1979 年发明，是工业领域最开放、最简单的串行协议之一。它只有四种基本数据类型：线圈（coil，位读写）、离散输入（discrete input，位只读）、保持寄存器（holding register，16 位读写）、输入寄存器（input register，16 位只读）。所有复杂的工业设备——电表、PLC、变频器、温控器——都通过这四类数据映射自己的功能。pymodbus 把这四类数据抽象成 Python 对象，开发者只需要关心"地址 + 值"，不需要处理帧头、CRC、超时等底层细节。

pymodbus 与其他 Python Modbus 库（如 minimalmodbus、umodbus）相比，优势在于：同时支持 TCP/UDP/RTU/ASCII/TLS 多种传输、同时支持客户端与服务器、提供异步 API、提供 Web 模拟器、测试覆盖率接近 100%、严格类型标注（py.typed）。它被 Home Assistant 官方集成采用，是事实上的 Python Modbus 标准库。

在工业物联网（IIoT）场景中，pymodbus 通常运行在边缘网关上：网关通过 RS-485 总线连接几十台电表/传感器，每几秒读取一次寄存器值，再通过 MQTT 上报到云平台。pymodbus 的异步 API 让单台网关可以同时轮询上百台设备，而不会阻塞。

本文覆盖了 pymodbus 从 pip 安装、TCP/RTU 客户端、异步并发、服务器模拟、Web 模拟器、生产采集脚本到字节序排错的完整流程，适合工业自动化、IoT 网关、智能家居开发者作为日常参考。

建议在开发环境先用 Simulator 模拟从站跑通业务逻辑，再连接真实设备，避免调试过程中误操作工业设备。

pymodbus 社区非常活跃，GitHub Discussions 上有大量工业设备对接案例，遇到非标设备时可以先在 Discussions 搜索是否有人踩过同样的坑。

pymodbus 文档完善，遇到问题优先查阅 官方文档。

祝大家采集顺利。

祝采集顺利，工业互联愉快。

祝一切顺利。

祝工作顺利。

祝顺利。

祝好。

祝。

。

二、项目结构与核心组件

pymodbus 仓库根目录结构：

• pymodbus/：源码包，包含 client、server、datastore、transport、simulator 等子模块。
• pymodbus/client/：客户端实现，包括 ModbusTcpClient、ModbusUdpClient、ModbusSerialClient、ModbusTlsClient，以及异步版本 AsyncModbusTcpClient 等。
• pymodbus/server/：服务器实现，包括 ModbusTcpServer、ModbusSerialServer、异步服务器等。
• pymodbus/datastore/：后端数据存储，包括 ModbusServerContext、ModbusSlaveContext、ModbusSequentialDataBlock，用于模拟寄存器值。
• pymodbus/transport/：传输层，封装 TCP、UDP、串行、TLS socket。
• pymodbus/simulator/：基于 Web 的服务器模拟器，提供浏览器界面配置设备寄存器、监控流量。
• examples/：示例代码，包含简单同步/异步客户端、服务器、回调、串行通信等。
• test/：测试套件，覆盖率接近 100%，在 Windows/Linux/macOS + Python 3.10-3.14 矩阵下测试。
• doc/：Sphinx 文档源码。

核心概念：

• Client：主动发起请求，读写从站寄存器。
• Server/Slave：被动响应请求，维护自己的寄存器与线圈。
• Context：服务器的内存数据模型，保存 coil、discrete input、holding register、input register 四类数据。
• Transaction：请求-响应的封装，每个请求有唯一事务 ID。
• Decoder/Encoder：把 PDU（协议数据单元）序列化为字节流。

pymodbus 的异步 API 是 3.x 版本重点重构的部分。它基于 asyncio 实现，客户端与服务器都可以在同一个事件循环中并发处理大量连接。异步 API 的命名以 Async 开头（如 AsyncModbusTcpClient、StartAsyncTcpServer），所有 IO 方法都需要 await。在需要同时连接几十台设备、或在 Web 服务（FastAPI、aiohttp）中嵌入 Modbus 通信时，异步 API 是唯一正确选择。

pymodbus 还提供了payload 编解码器（BinaryPayloadBuilder/Decoder），可以把 int16/int32/float32/float64/string/bit 数组自动打包成寄存器或从寄存器解包。这比手动 struct.pack 更安全，也避免了字节序错误。它还支持自定义字节序组合，覆盖所有工业设备厂商的格式。

三、版本与环境要求

pymodbus 版本号采用 X.Y.Z，但不遵循 semver.org：Z 仅修 bug、Y 可能有 API 变更、X 是重大重构。升级前必须阅读 CHANGELOG 与 API_changes。

运行要求：

• Python >= 3.10（推荐 3.11/3.12）。
• 操作系统：Linux、macOS、Windows 均可；串行通信需要 pyserial。
• 依赖：核心库无第三方依赖；可选 extras：serial（pyserial）、simulator（aiohttp）、documentation（Sphinx）、development（pytest、ruff）。
• 硬件：TCP/UDP 客户端只需要网络；RTU/ASCII 客户端需要 RS-485 转换器（如 USB-to-RS485）或串口。
• 权限：Linux 下访问串口（/dev/ttyUSB0、/dev/ttyS0）需要把用户加入 dialout 组。

四、本地部署

4.1 用 pip 安装（最简单）

pip install pymodbus
# 需要串行通信
pip install pymodbus[serial]
# 需要 Web 模拟器
pip install pymodbus[simulator]
# 安装所有 extras
pip install pymodbus[all]
# 验证
python -c "import pymodbus; print(pymodbus.__version__)"

4.2 从源码安装（开发）

git clone https://github.com/pymodbus-dev/pymodbus.git
cd pymodbus
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[development]"
cp githooks/* .git/hooks
pytest test/ -x

4.3 同步客户端示例（读电表）

from pymodbus.client import ModbusTcpClient

client = ModbusTcpClient('192.168.1.100', port=502)
client.connect()

# 读保持寄存器（地址 0，数量 4）
result = client.read_holding_registers(address=0, count=4, slave=1)
if not result.isError():
    print('registers:', result.registers)
    # 解析 32 位浮点（两个 16 位寄存器拼起来）
    import struct
    value = struct.unpack('>f', struct.pack('>HH', result.registers[0], result.registers[1]))[0]
    print('power:', value, 'kW')

# 写单个线圈
client.write_coil(address=10, value=True, slave=1)

client.close()

4.4 异步客户端示例

import asyncio
from pymodbus.client import AsyncModbusTcpClient

async def main():
    client = AsyncModbusTcpClient('192.168.1.100', port=502)
    await client.connect()
    result = await client.read_holding_registers(address=0, count=4, slave=1)
    print(result.registers)
    client.close()

asyncio.run(main())

4.5 服务器（模拟从站）

from pymodbus.server import StartAsyncTcpServer
from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext
import asyncio

async def run():
    store = ModbusSlaveContext(
        di=ModbusSequentialDataBlock(0, [17]*100),
        co=ModbusSequentialDataBlock(0, [17]*100),
        hr=ModbusSequentialDataBlock(0, [123]*100),
        ir=ModbusSequentialDataBlock(0, [456]*100),
    )
    context = ModbusServerContext(slaves=store, single=True)
    await StartAsyncTcpServer(context=context, address=('0.0.0.0', 5020))

asyncio.run(run())

4.6 串行 RTU 客户端（RS-485）

from pymodbus.client import ModbusSerialClient

client = ModbusSerialClient(
    port='/dev/ttyUSB0',
    baudrate=9600,
    bytesize=8,
    parity='N',
    stopbits=1,
    timeout=1,
)
client.connect()
rr = client.read_holding_registers(0, 2, slave=1)
print(rr.registers)
client.close()

4.7 Docker 部署

cat > Dockerfile <<EOF
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir pymodbus[serial,simulator]
COPY server.py .
CMD ["python", "server.py"]
EOF

docker build -t pymodbus-server .
docker run -d -p 5020:5020 --device=/dev/ttyUSB0 pymodbus-server

4.8 完整采集脚本（定时轮询多设备）

下面给出一个生产环境常用的采集脚本，同时连接多台电表，每秒读取电压、电流、功率，写入 SQLite 并打印：

import time, struct, sqlite3
from pymodbus.client import ModbusTcpClient

devs = [
    ('192.168.1.101', 1, 'meter-A'),
    ('192.168.1.102', 1, 'meter-B'),
]

conn = sqlite3.connect('meter.db')
conn.execute('CREATE TABLE IF NOT EXISTS data (ts REAL, dev TEXT, voltage REAL, current REAL, power REAL)')

while True:
    ts = time.time()
    for ip, slave, name in devs:
        c = ModbusTcpClient(ip, port=502, timeout=2)
        if not c.connect():
            print(f'{name} offline'); continue
        rr = c.read_holding_registers(0, 6, slave=slave)
        c.close()
        if rr.isError(): continue
        v = struct.unpack('>f', struct.pack('>HH', rr.registers[0], rr.registers[1]))[0]
        i = struct.unpack('>f', struct.pack('>HH', rr.registers[2], rr.registers[3]))[0]
        p = struct.unpack('>f', struct.pack('>HH', rr.registers[4], rr.registers[5]))[0]
        conn.execute('INSERT INTO data VALUES (?,?,?,?,?)', (ts, name, v, i, p))
        conn.commit()
        print(f'{name}: V={v:.1f} I={i:.2f} P={p:.2f}')
    time.sleep(1)

4.9 异步采集（推荐多设备高并发）

import asyncio, struct
from pymodbus.client import AsyncModbusTcpClient

async def read_dev(ip, slave, name):
    c = AsyncModbusTcpClient(ip, port=502, timeout=2)
    await c.connect()
    rr = await c.read_holding_registers(0, 6, slave=slave)
    c.close()
    print(name, rr.registers)

async def main():
    await asyncio.gather(*[
        read_dev(ip, s, n) for ip, s, n in [
            ('192.168.1.101', 1, 'A'),
            ('192.168.1.102', 1, 'B'),
            ('192.168.1.103', 1, 'C'),
        ]
    ])

asyncio.run(main())

在实际项目中，Modbus 设备的字节序是最常见的坑。不同厂商的设备对 32 位数据（float、int32）在两个 16 位寄存器中的排列方式不同：有的高字在前（Big-Endian Word）、有的低字在前（Little-Endian Word）；有的字节内部高字节在前（Big-Endian Byte）、有的低字节在前（Little-Endian Byte）。组合起来共有四种排列：ABCD、CDAB、BADC、DCBA。pymodbus 的 BinaryPayloadDecoder 通过 byteorder 与 wordorder 两个参数覆盖所有四种组合，调试时可以先用已知值（如 23.5）写入设备，再读回试出正确组合。

另外，Modbus 是主从协议，任何时候都是 master（客户端）发起请求，slave（服务器）只能被动响应。RS-485 总线是半双工的，同一时刻只能有一个 master 在通信，因此总线上必须严格控制轮询间隔，避免冲突。pymodbus 的串行客户端通过 timeout 参数控制等待响应时间，超时会抛出 ModbusIOException。

在生产部署中，建议把 pymodbus 采集脚本用 systemd 或 supervisor 守护，异常退出自动重启；日志按天轮转；与设备通信失败时记录最后一次成功时间，便于事后排查。对于关键设备，建议双网关冗余：两台机器同时采集，避免单点故障。

五、浏览器端可视化管理

pymodbus 内置了一个基于 Web 的模拟器（Simulator），这是它区别于其他 Modbus 库的特色。安装 pymodbus[simulator] 后直接启动：

pymodbus.simulator --modbus_device device_try

默认会在 8081 端口启动 Web UI，浏览器访问 http://localhost:8081，可以：

• 配置 Modbus 服务器类型（TCP/UDP/RTU）与端口。
• 图形化定义设备的线圈、离散输入、保持寄存器、输入寄存器地址与初始值。
• 实时查看所有经过的 Modbus 请求与响应帧（hex 与解析后两种格式）。
• 模拟错误响应（非法功能码、非法地址、从站忙等），方便客户端程序调试。
• 多人通过互联网同时连到同一个模拟设备，适合远程联调。

常用参数：--json_file 指定配置 JSON、--modbus_server 选择服务器、--modbus_device 选择设备、--http_port Web 端口、--log 日志级别。

除了官方模拟器，生产环境通常用 Grafana + Telegraf/Node-RED 把 Modbus 数据采集到 InfluxDB，再通过浏览器面板展示实时曲线。

除了官方 Simulator，生产环境通常用 Grafana + Telegraf + InfluxDB 搭建可视化看板：Telegraf 的 Modbus 输入插件按配置轮询寄存器，把值写入 InfluxDB，Grafana 面板实时绘制电压、电流、功率曲线。如果需要 Web 上下位机界面，开源的 Node-RED 或 Home Assistant 都内置了 pymodbus，可以直接拖拽出可视化面板。

对于楼宇自动化、智能电表等场景，常用的端到端架构是：RS-485 总线 → USB 转串口 → 边缘网关（跑 pymodbus）→ MQTT → 云平台（ThingsBoard/Azure IoT Hub）→ Web 仪表盘。pymodbus 在这一架构中承担"把现场设备翻译成 JSON"的角色。

六、API 用法

6.1 常用读功能码

read_coils(address, count, slave)          # 01 读线圈
read_discrete_inputs(address, count, slave) # 02 读离散输入
read_holding_registers(address, count, slave) # 03 读保持寄存器
read_input_registers(address, count, slave) # 04 读输入寄存器

6.2 常用写功能码

write_coil(address, value, slave)       # 05 写单个线圈
write_register(address, value, slave)  # 06 写单个寄存器
write_coils(address, values, slave)     # 15 写多个线圈
write_registers(address, values, slave) # 16 写多个寄存器

6.3 用 curl 模拟 TCP Modbus 请求（hex）

# Modbus TCP 读保持寄存器请求：事务ID=0, 协议ID=0, 长度=6, 单元ID=1, 功能码=03, 地址=0, 数量=2
# 完整帧（hex）：00 00 00 00 00 06 01 03 00 00 00 02
printf '\x00\x00\x00\x00\x00\x06\x01\x03\x00\x00\x00\x02' | nc 192.168.1.100 502 | xxd

6.4 回调拦截请求/响应

def log_request(request):
    print('REQ:', request)

from pymodbus.client import ModbusTcpClient
client = ModbusTcpClient('192.168.1.100')
client.register_request_callback(log_request)
client.connect()
client.read_holding_registers(0, 2)
client.close()

6.5 数据类型转换工具

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

# 把寄存器解码为 32 位浮点
decoder = BinaryPayloadDecoder.fromRegisters(
    result.registers, byteorder=Endian.BIG, wordorder=Endian.BIG)
float_value = decoder.decode_32bit_float()

# 把浮点编码为寄存器
builder = BinaryPayloadBuilder(byteorder=Endian.BIG, wordorder=Endian.BIG)
builder.add_32bit_float(23.5)
payload = builder.to_registers()

pymodbus 还支持自定义功能码：通过 @client.on_receive 或在服务器中注册自定义 handler，可以实现厂商私有的 Modbus 扩展。这在对接某些非标设备（如部分国产 PLC、定制传感器）时非常有用。

在调试时，建议先用 Modbus Poll（Windows）或 ModbusScanner（跨平台）确认设备地址、寄存器映射、字节序，再写 pymodbus 代码。这样可以避免把设备问题误认为库的问题。pymodbus 自带的 Simulator WebUI 也可以反过来模拟从站，让客户端程序在没有真实硬件的情况下完成开发与单元测试。

七、常见问题 FAQ 与相关链接

Q1：报 "Connection refused"。
A：检查从站 IP/端口是否正确、防火墙是否放行 502 端口、从站服务是否启动。TCP Modbus 默认端口 502，非 root 用户监听 502 需要 root 或 setcap。

Q2：串行通信报 "could not open port /dev/ttyUSB0"。
A：把用户加入 dialout 组：sudo usermod -aG dialout $USER，重新登录。Docker 中需要 --device=/dev/ttyUSB0。

Q3：读取寄存器值乱码。
A：Modbus 寄存器字节序有大端/小端、字交换/字不交换四种组合。用 BinaryPayloadDecoder 调整 byteorder 与 wordorder 试出正确组合。

Q4：slave 参数填几？
A：slave 就是 Modbus 从站地址（1-247），由设备拨码或配置决定。常见默认 1。

Q5：pymodbus 3.x 与 2.x 不兼容。
A：3.x 重构了客户端 API，client.read_holding_registers() 不再返回结果对象的属性，而是直接返回结果对象，错误用 isError() 判断。升级前必读 API_changes。

Q6：如何在 Home Assistant 中使用？
A：Home Assistant 内置 Modbus 集成，配置文件中写 modbus: - name: meter; type: tcp; host: 192.168.1.100; port: 502; sensors: [...]</pre>。