如何用pymodbus捕获原始字节并创建中断处理Modbus请求响应？

一、捕获原始请求字节并解析寄存器信息 #

要绕过轮询直接获取请求的关键信息，你需要自定义RTU帧处理器，在请求被处理前捕获原始字节并解析：

1. 自定义RTU帧处理器 #

继承ModbusRtuFramer并重写processIncomingPacket方法，在此阶段获取原始字节流并解析请求参数：

from pymodbus.framer.rtu_framer import ModbusRtuFramer
from pymodbus.pdu import ModbusRequest

class CustomRtuFramer(ModbusRtuFramer):
    def processIncomingPacket(self, data, callback, slave=None):
        # 打印原始请求字节
        print(f"原始请求字节: {data.hex()}")
        
        # 解析请求PDU，提取关键信息
        try:
            request = self._parseRequest(data)
            if isinstance(request, ModbusRequest):
                slave_id = request.slave_id
                func_code = request.function_code
                start_addr = request.address
                reg_count = request.count
                
                # 直接从context获取对应数据
                data_values = self.decoder.context[slave_id].getValues(func_code, start_addr, reg_count)
                print(f"解析请求: 从站ID={slave_id}, 功能码={func_code}, 起始地址={start_addr}, 寄存器数量={reg_count}, 数据={data_values}")
        except Exception as e:
            print(f"解析请求失败: {e}")
        
        # 调用父类方法继续处理请求，保证正常响应客户端
        super().processIncomingPacket(data, callback, slave)

2. 启动服务器时使用自定义帧处理器 #

修改服务器启动代码，替换framer参数为自定义类：

server = StartSerialServer(
    context, 
    framer=CustomRtuFramer, 
    identity=identity,
    port='/dev/ttyUSB0', 
    timeout=1, 
    baudrate=9600, 
    parity='N',
    bytesize=8,
    stopbits=1
)

二、实现请求/响应触发式通知（替代轮询） #

通过自定义服务器类，重写请求处理方法，实现收到请求或发送响应时的触发式操作：

1. 自定义Modbus服务器 #

继承ModbusSerialServer并重写handle_request方法：

from pymodbus.server.sync import ModbusSerialServer

class CustomModbusServer(ModbusSerialServer):
    def handle_request(self, request):
        # 收到请求时触发打印
        print(f"收到请求: {request}")
        
        # 处理请求获取响应
        response = super().handle_request(request)
        
        # 发送响应时触发打印
        if response:
            print(f"发送响应: {response}")
        return response

2. 启动自定义服务器 #

替换默认启动方式，直接实例化自定义服务器：

server = CustomModbusServer(
    context, 
    framer=CustomRtuFramer, 
    identity=identity,
    port='/dev/ttyUSB0', 
    timeout=1, 
    baudrate=9600, 
    parity='N',
    bytesize=8,
    stopbits=1
)
server.serve_forever()

三、寄存器变化的触发式检测 #

如果需要检测寄存器值变化，无需轮询，可重写SlaveContext的setValues方法，在值被修改时触发通知：

from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext

class CustomSlaveContext(ModbusSlaveContext):
    def setValues(self, fc_as_hex, address, values):
        # 记录修改前的值
        old_values = self.getValues(fc_as_hex, address, len(values))
        # 执行修改操作
        super().setValues(fc_as_hex, address, values)
        # 对比值是否变化，变化则触发通知
        if old_values != values:
            print(f"寄存器更新: 功能码={fc_as_hex}, 起始地址={address}, 旧值={old_values}, 新值={values}")

# 创建上下文时使用自定义SlaveContext
slave_context = CustomSlaveContext()
context = ModbusServerContext(slaves=slave_context, single=True)

内容的提问来源于stack exchange，提问作者Umer Farooq