如何使用pymodbus构造带指定前缀的自定义请求并读取设备响应

问题根因 #

你当前代码无法收到响应核心是三个实现错误：

• 结构体打包规则错误：struct.pack中>H代表2字节无符号整数，你传入的三个33（对应ASCII感叹号0x21）会被打包为00 21 00 21 00 21，和要求的连续三个单字节21 21 21前缀完全不符。
• 未绕过默认Modbus TCP帧封装：pymodbus默认的TCP framer会自动在请求前加7字节标准MBAP头，你构造的自定义报文会被包裹在MBAP头之后，实际发出去的报文开头不是三个感叹号，设备无法识别。
• 冗余的RTU参数干扰：你定义的_rtu_frame_size是RTU串行模式专用的帧长度校验参数，TCP传输下该参数无效，还会导致响应帧长度计算错误，无法正常解析返回内容。

正确实现步骤 #

1. 自定义TCP Framer #

重写帧处理逻辑，跳过默认MBAP头的自动添加，直接发送构造的完整报文，同时识别以三个0x21开头的响应帧：

from pymodbus.framer import ModbusFramer
import struct

class CustomPrefixFramer(ModbusFramer):
    def __init__(self, decoder, client=None):
        super().__init__(decoder, client)
        self.prefix = b'!!!'  # 固定三字节前缀

    def buildPacket(self, message):
        # 直接返回编码后的完整报文，不添加额外MBAP头
        return message.encode()

    def processIncomingPacket(self, data, callback, *args, **kwargs):
        # 校验前缀匹配
        if not data.startswith(self.prefix):
            return
        # 跳过前缀，按设备协议规则计算响应总长度
        payload = data[3:]
        if len(payload) < 3:
            return
        # 功能码后第一个字节为数据段长度，据此判断是否接收完整
        data_len = payload[2]
        if len(payload) < 2 + data_len:
            return
        # 解析完整响应
        resp = self.decoder.decode(payload)
        callback(resp)

2. 修正自定义请求/响应类 #

调整打包格式为单字节前缀，移除无效RTU参数，同时定义匹配的响应类解析时间数据：

from pymodbus.pdu import ModbusRequest, ModbusResponse
import struct

class CustomBSCRACResponse(ModbusResponse):
    function_code = 127
    def __init__(self, datetime_data=None, **kwargs):
        super().__init__(**kwargs)
        self.datetime_data = datetime_data  # 存储解析出的日期时间

    def decode(self, data):
        # 跳过长度字段，按设备协议解析返回的日期时间字节，可根据实际字节顺序调整
        time_payload = data[1:]
        self.datetime_data = {
            "year": time_payload[0] + 2000,  # 按常见设备存储规则补全年份基数
            "month": time_payload[1],
            "day": time_payload[2],
            "hour": time_payload[3],
            "minute": time_payload[4],
            "second": time_payload[5]
        }

class CustomBSCRACRequest(ModbusRequest):
    function_code = 127
    command = 0x88  # 对应十进制136
    flags = 0x40    # 对应十进制64
    dest_task = 0x00

    def __init__(self, address=1, **kwargs):
        super().__init__(**kwargs)
        self.address = address
        self.data_len = 7  # 对应示例报文中的0x07长度字段

    def encode(self):
        # 打包结果完全匹配示例报文：21 21 21 01 7F 07 40 01 00 88 00 00
        return struct.pack(
            ">BBBBBBBHH",
            0x21, 0x21, 0x21,  # 三个感叹号前缀
            self.address,      # 从站地址 0x01
            self.function_code,# 功能码 0x7F
            self.data_len,     # 后续数据长度 0x07
            self.flags,        # 标志位 0x40
            0x01,              # 示例中的序号字段
            self.command,      # 命令字 0x0088
            self.dest_task     # 保留字段 0x0000
        )

3. 客户端初始化绑定自定义逻辑 #

创建Modbus TCP客户端时，指定使用自定义framer，同时注册自定义请求/响应类，避免被默认解析逻辑拦截：

from pymodbus.client import ModbusTcpClient

# 初始化客户端，绑定自定义framer
client = ModbusTcpClient(
    host="你的设备IP",
    port=502,
    framer=CustomPrefixFramer
)
# 注册自定义PDU映射
client.register(CustomBSCRACRequest, CustomBSCRACResponse)

client.connect()
# 发送自定义请求
resp = client.execute(CustomBSCRACRequest(address=1))
if not resp.isError():
    print("设备返回时间：", resp.datetime_data)
client.close()

调试提示 #

首次调试如果仍无响应，可以先抓包确认实际发出的报文是否和示例21 21 21 01 7F 07 40 01 00 88 00 00完全一致，再根据设备实际返回的帧结构调整framer中的长度计算、响应解析逻辑即可。

内容的提问来源于stack exchange，提问作者Juan Navarro