简介pymodbus 是 Python 中实现 Modbus 通信协议的完整协议栈支持 TCP、UDP、RTU 与 ASCII 四种传输模式并允许自定义数据格式适合从事工业自动化、PLC 数据采集与设备通信的开发者及运维人员使用。资源包共 306 个文件以 167 个 py 源码与 61 个 rst 文档为主辅以 txt、js、c、makefile、java、md 等类型涵盖协议实现、示例脚本、测试用例与说明文档压缩包约 1.17MB目录结构清晰便于按模块查阅。目前已有 1119 人学习下载。内容围绕 Modbus 协议介绍、库的安装与基本用法、客户端与服务器搭建、自定义数据存储类、功能代码与数据类型、错误处理与日志调试等展开读者可据此掌握寄存器读写、浮点数据转换与通信排错思路快速构建自动化控制与数据采集系统。1. 从一次现场调试说起为什么我最后选了 pymodbus去年冬天某公司的一条产线改造项目卡在最后联调阶段。PLC 侧数据死活读不上来现场工程师换了三根串口线、两轮终端电阻示波器都架上了波形干净得很可上位机就是拿不到寄存器值。后来把抓包数据拉出来一看问题出在报文格式上——对方设备用的是非标准的寄存器映射和自定义功能码而当时用的调试工具只认标准 Modbus 帧压根没法改。那天晚上我回去翻了半天最后用 pymodbus 写了个几十行的脚本把自定义格式拼进去十分钟跑通。这就是 pymodbus 的价值所在它不只是一个「能连 Modbus」的库而是一个允许你拆开协议栈、自己拼帧、自己定义解析规则的完整 Python 实现。标准 Modbus TCP/RTU/ASCII 它全支持但真正让它在现场站住脚的是那套可扩展的 framer、decoder 和 payload 体系。你遇到非标设备、私有寄存器、混合协议网关标准工具趴窝的时候它能让你从协议层往下改。这篇文章面向的是手里有真实设备、需要落地通信的从业者。不管你是第一次接触 Modbus 的新手还是被非标协议折磨过的老手下面会从环境搭建一路讲到自定义格式的实现和排错把我在现场踩过的坑都摊开说。2. 环境搭建与最小可运行示例先把链路跑通再谈自定义2.1 安装与版本选择pymodbus 的版本迭代比较快2.x 和 3.x 在 API 上有明显差异。3.x 重构了 client 和 server 的接口把同步和异步彻底分开framer 的注册方式也变了。如果你维护的是老项目2.x 的ModbusClient和ModbusServer还能用但新项目建议直接上 3.x因为自定义格式的扩展点在新版里更清晰。安装本身没什么玄学# 建议在虚拟环境里装避免和系统 Python 冲突 python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 安装 pymodbus指定 3.x 大版本 pip install pymodbus3.0,4.0 # 串口通信需要 pyserialTCP 不需要但建议一起装 pip install pyserial装完之后验证一下import pymodbus print(pymodbus.__version__)版本号打印出来是 3.x 就对了。这里有个细节pymodbus 3.x 把pymodbus.client.sync改成了pymodbus.client很多网上抄来的老代码会在这里报ModuleNotFoundError遇到别慌是版本对不上。2.2 一个能跑通的 TCP 客户端先别急着上自定义把标准链路跑通确认环境、网络、设备三件事都没问题。下面这段代码读保持寄存器是最常见的场景from pymodbus.client import ModbusTcpClient # 连接设备端口默认 502 client ModbusTcpClient(192.168.1.100, port502) client.connect() # 读保持寄存器从地址 0 开始读 10 个 # slave1 是从站地址3.x 里参数名是 slave rr client.read_holding_registers(address0, count10, slave1) if rr.isError(): print(f读取失败: {rr}) else: print(f寄存器值: {rr.registers}) client.close()逻辑说明ModbusTcpClient封装了 socket 连接和 MBAP 头的处理read_holding_registers对应功能码 0x03。address是起始寄存器地址count是数量slave是从站 ID。返回的rr是一个响应对象正常时.registers是列表异常时.isError()为 True。参数上最容易翻车的是地址偏移。Modbus 协议里寄存器地址从 0 开始但很多设备手册写的是 40001 这种「PLC 地址」。40001 对应的是保持寄存器区实际协议地址是 0。如果你按手册的 40001 去读得先减 40001。这个坑我见过太多次现场对不上地址八成是这里。2.3 串口 RTU 的配置要点RTU 比 TCP 多几个物理层参数配错了就是收不到数据from pymodbus.client import ModbusSerialClient client ModbusSerialClient( port/dev/ttyUSB0, # Windows 下是 COM3 这种 baudrate9600, # 波特率必须和设备一致 bytesize8, # 数据位通常 8 parityN, # 校验位N 无校验 / E 偶 / O 奇 stopbits1, # 停止位通常 1 timeout1 # 超时秒数现场建议 1~3 ) client.connect() rr client.read_input_registers(address0, count5, slave2) print(rr.registers if not rr.isError() else rr) client.close()baudrate、parity、stopbits这三个必须和从站设备完全一致差一个都通不了。timeout在长距离 RS485 上要适当放大我一般设 2 秒太短会误判超时。read_input_registers对应功能码 0x04和保持寄存器的 0x03 是两个不同的地址空间别混。提示先用标准功能把链路跑通确认能读到数据再动自定义格式。否则出了问题你分不清是协议改错了还是线没接好。3. 拆开协议栈framer、decoder 与自定义格式的落点3.1 pymodbus 的分层结构要改自定义格式得先知道 pymodbus 把一帧报文拆成了哪几层。从下往上大致是transportsocket 或串口→ framer帧边界识别→ decoder字节流解析成 PDU→ payloadPDU 转成具体请求/响应对象。framer 负责从连续的字节流里切出一帧。TCP 靠 MBAP 头里的长度字段RTU 靠 3.5 字符间隔和 CRC。decoder 负责把切出来的帧解析成功能码加数据。payload 则是把数据映射成ReadHoldingRegistersRequest这类对象。自定义格式的改动点通常落在两处一是 framer当你的设备帧头帧尾不是标准 MBAP 或 RTU 格式时二是 payload当功能码或数据字段布局非标时。搞清这个分层改起来才不会把整个库搅乱。3.2 自定义 framer 的实现假设某设备用的是一种私有 TCP 帧2 字节魔数0xAA 0xBB开头2 字节长度然后是 PDU没有 MBAP 头。标准 framer 认不出这种帧得自己写一个from pymodbus.framer import FramerBase from pymodbus.exceptions import ModbusIOException class CustomFramer(FramerBase): 自定义帧AA BB 2字节长度 PDU def __init__(self, clientNone): super().__init__(client) self.buffer b def processIncomingPacket(self, data, callback, **kwargs): # 把新数据追加到缓冲区 self.buffer data while len(self.buffer) 4: # 检查魔数 if self.buffer[0:2] ! b\xAA\xBB: # 魔数不对丢弃一个字节继续找 self.buffer self.buffer[1:] continue # 读长度字段大端 length int.from_bytes(self.buffer[2:4], big) frame_len 4 length if len(self.buffer) frame_len: break # 帧还没收全等下次 frame self.buffer[:frame_len] self.buffer self.buffer[frame_len:] # 去掉 4 字节头把 PDU 交给上层 callback(frame[4:]) def buildPacket(self, message): # 发送时加自定义头 pdu message.encode() header b\xAA\xBB len(pdu).to_bytes(2, big) return header pdu逻辑说明processIncomingPacket是收包入口pymodbus 会把 transport 收到的原始字节喂进来。这里维护一个缓冲区先找魔数再按长度字段判断帧是否完整完整了才回调上层。buildPacket是发包时调用负责加自定义头。参数上要注意length字段的含义——它指的是 PDU 长度还是整帧长度必须和设备手册对齐。上面代码里length是 PDU 长度所以整帧是4 length。如果手册写的是整帧长度这里就得改成frame_len length。这个歧义是自定义格式里最常见的翻车点。3.3 把自定义 framer 挂到 client 上写好了 framer得让 client 用它。3.x 里通过framer参数传入from pymodbus.client import ModbusTcpClient client ModbusTcpClient( 192.168.1.100, port502, framerCustomFramer # 传入类不是实例 ) client.connect() rr client.read_holding_registers(address0, count10, slave1) print(rr.registers if not rr.isError() else rr) client.close()注意framer传的是类本身pymodbus 内部会实例化。如果你传了实例某些版本会报参数错误。挂上之后收发都走你的自定义帧上层read_holding_registers的调用方式不变这就是分层的好处——改底层不影响业务代码。3.4 自定义 payload 与功能码如果设备的功能码也是非标的比如用 0x41 读数据那就得自定义 payload。pymodbus 的 payload 体系基于ModbusRequest和ModbusResponsefrom pymodbus.pdu import ModbusRequest, ModbusResponse from pymodbus.register_read_message import ReadRegistersResponseBase class CustomReadRequest(ModbusRequest): function_code 0x41 # 自定义功能码 def __init__(self, address, count, slave1): super().__init__(slave) self.address address self.count count def encode(self): # 请求 PDU功能码 起始地址(2) 数量(2) return (self.function_code.to_bytes(1, big) self.address.to_bytes(2, big) self.count.to_bytes(2, big)) def decode(self, data): self.address int.from_bytes(data[0:2], big) self.count int.from_bytes(data[2:4], big) class CustomReadResponse(ReadRegistersResponseBase): function_code 0x41 def decode(self, data): # 响应 PDU功能码 字节数 寄存器数据 byte_count data[0] self.registers [ int.from_bytes(data[1i:3i], big) for i in range(0, byte_count, 2) ]逻辑说明encode负责把请求对象拼成字节decode负责把收到的字节还原成对象。function_code是类属性pymodbus 靠它做路由。响应继承ReadRegistersResponseBase是为了复用寄存器列表的处理逻辑。参数上address和count的字节序必须和设备一致Modbus 标准是大端但私有协议里小端也不少见。byte_count是响应里数据段的字节数等于寄存器数乘 2。如果设备返回的字节数不对registers列表长度就会错读出来的值全是乱的。4. 避坑与排查那些让我加班到凌晨的细节4.1 现象连接成功但读不到数据返回 IllegalFunction原因功能码不被从站支持或者你用的功能码和设备的寄存器区不匹配。比如设备只支持 0x03 读保持寄存器你用了 0x04 读输入寄存器。解决先查设备手册确认支持的功能码再用标准 client 逐个试。如果标准功能码都不行说明设备用的是私有功能码得走自定义 payload 那条路。4.2 现象RTU 下偶尔能通大部分时候超时原因多半是串口参数或时序问题。RS485 半双工下发送和接收切换需要时间如果timeout太短或者从站响应慢就会丢帧。另外终端电阻没接、线太长、波特率太高都会导致误码。解决把timeout放大到 2~3 秒波特率降到 9600 试试检查 120 欧终端电阻是否接在总线两端。如果用的是 USB 转 485换一个质量好点的转换器劣质芯片在 115200 下丢包很严重。4.3 现象自定义 framer 挂上后收到的数据全是乱码原因帧长度字段的含义搞错了或者字节序不对。比如手册里长度是整帧长度你按 PDU 长度解析就会多切或少切字节后续全乱。解决抓一段原始字节流手动按手册算一遍。用print(data.hex())把收到的原始数据打出来对照手册逐字节核对。这一步没有捷径只能硬对。4.4 现象多从站轮询时响应串台原因TCP 网关下多个从站的响应可能交错到达如果 client 没有正确匹配请求和响应就会把 A 从站的数据当成 B 的。解决确保每个请求带正确的slave地址并且用同步方式逐个请求不要并发。pymodbus 的同步 client 本身是串行的但如果你自己开了多线程共享一个 client就会出这个问题。每个线程用独立 client或者加锁。4.5 现象升级到 3.x 后老代码全报错原因3.x 改了 APIpymodbus.client.sync没了ModbusClient拆成了ModbusTcpClient和ModbusSerialClientunit参数改成了slave。解决按 3.x 的接口重写导入和调用。如果项目大改不动就锁死在 2.xpip install pymodbus3.0。但新项目别这么干2.x 迟早要淘汰。注意每次改完 framer 或 payload先用一个已知正确的标准设备验证库本身没被改坏再上非标设备。否则你分不清是库的问题还是设备的问题。5. 进阶技巧用单元测试锁住自定义格式别再靠现场试自定义格式最怕的是「这次通了下次换个设备又不行」。我现在的习惯是每写一个自定义 framer 或 payload先写单元测试把字节级的编解码锁死现场只做联调不做调试。5.1 用 pytest 验证 framer 的切帧逻辑import pytest from your_module import CustomFramer def test_framer_split(): framer CustomFramer() received [] # 模拟分两次到达的半帧 framer.processIncomingPacket(b\xAA\xBB\x00\x04\x41, received.append) assert received [] # 帧不完整不应回调 # 补齐剩余字节 framer.processIncomingPacket(b\x00\x00\x00\x02, received.append) assert len(received) 1 assert received[0] b\x41\x00\x00\x00\x02 def test_framer_bad_magic(): framer CustomFramer() received [] # 前面有垃圾字节应该能跳过找到魔数 framer.processIncomingPacket(b\x00\xAA\xBB\x00\x02\x41\x00, received.append) assert len(received) 1逻辑说明第一个测试验证半帧不会提前回调第二个测试验证垃圾字节能被跳过。这两个场景在现场太常见了——TCP 粘包、串口噪声都会导致半帧或脏数据。测试里把callback换成一个收集列表就能断言切出来的帧内容。参数上processIncomingPacket的第二个参数是回调函数测试时传received.append就行。注意断言的是去掉自定义头之后的 PDU和 framer 的设计保持一致。5.2 用已知报文做回归把现场抓到的真实报文存成 fixture每次改代码跑一遍# conftest.py 或测试文件里 KNOWN_FRAMES [ # (原始字节, 期望的 PDU) (bytes.fromhex(aabb00044100000002), b\x41\x00\x00\x00\x02), (bytes.fromhex(aabb000641000000020001), b\x41\x00\x00\x00\x02\x00\x01), ] pytest.mark.parametrize(raw,expected, KNOWN_FRAMES) def test_known_frames(raw, expected): framer CustomFramer() received [] framer.processIncomingPacket(raw, received.append) assert received[0] expected这样每次重构 framer只要跑一遍测试就知道有没有破坏已有格式。现场联调的时间能省一大半。5.3 几个我固定会检查的参数检查项常见值翻车后果字节序大端 / 小端寄存器值高低字节颠倒长度字段含义PDU 长度 / 整帧长度切帧错位全乱功能码标准 0x03/0x04 / 私有IllegalFunction地址偏移0 基 / 40001 基读错寄存器超时1~3 秒误判超时从站地址1~247无响应或串台这张表我贴在工位上每次新设备联调前过一遍能挡掉八成低级错误。从那以后我每次写自定义 framer都先把单元测试和已知报文 fixture 建好再上设备。现场只做验证不做调试。希望帮到你。本文还有配套的精品资源点击获取