基于Pymodbus的Modbus服务器：写入指令地址值持久化问题求助

一、区分外部写入请求与内部线程的寄存器更新 #

不要通过重写ModbusSequentialDatablock.setValues捕获写入操作——该方法会同时响应内部线程调用和外部Modbus请求，无法区分来源。正确思路是拦截外部Modbus写入请求的处理流程：

1. 自定义RTU请求处理器，继承ModbusRtuRequestHandler，重写对应功能码的处理方法，在方法内完成持久化后再执行实际写入：

from pymodbus.server import ModbusRtuRequestHandler

class CustomRtuRequestHandler(ModbusRtuRequestHandler):
    def handle_write_single_coil(self, request):
        # 持久化功能码05的写入数据
        with open("persist_data.txt", "a") as f:
            f.write(f"FC05: addr={request.address}, value={request.value}\n")
        return super().handle_write_single_coil(request)

    def handle_write_single_register(self, request):
        # 持久化功能码06的写入数据
        with open("persist_data.txt", "a") as f:
            f.write(f"FC06: addr={request.address}, value={request.value}\n")
        return super().handle_write_single_register(request)

    def handle_write_multiple_coils(self, request):
        # 持久化功能码15的写入数据
        with open("persist_data.txt", "a") as f:
            f.write(f"FC15: start_addr={request.address}, values={request.values}\n")
        return super().handle_write_multiple_coils(request)

    def handle_write_multiple_registers(self, request):
        # 持久化功能码16的写入数据
        with open("persist_data.txt", "a") as f:
            f.write(f"FC16: start_addr={request.address}, values={request.values}\n")
        return super().handle_write_multiple_registers(request)

2. 启动RTU服务器时指定该自定义处理器：

from pymodbus.server import StartSerialServer
from pymodbus.datastore import ModbusSlaveContext, ModbusSequentialDatablock, ModbusServerContext
from pymodbus.transaction import ModbusRtuFramer

# 初始化数据存储
store = ModbusSlaveContext(
    co=ModbusSequentialDatablock(0, [0]*100),
    hr=ModbusSequentialDatablock(0, [0]*100),
    di=None, ir=None
)
context = ModbusServerContext(slaves=store, single=True)

# 启动服务器
StartSerialServer(
    context,
    framer=ModbusRtuFramer,
    port="/dev/ttyUSB0",
    baudrate=9600,
    parity="N",
    stopbits=1,
    request_handler=CustomRtuRequestHandler
)

这种方式下，只有外部设备发送的写入请求会触发持久化逻辑，内部线程调用setValues不会经过请求处理器，自然实现了来源区分。

二、解决RTU帧处理器的不完整帧问题 #

你之前直接在processIncomingPacket解析原始字节的方式不可靠，因为该方法可能接收分片的不完整数据（RTU帧依赖超时或CRC校验拼接完整帧）。

最优方案是直接使用上述请求处理器——请求处理器拿到的request对象是Pymodbus已解析完成的完整请求，包含功能码、地址、值等所有字段，无需手动解析原始字节，从根源避免不完整帧问题。

如果一定要自定义帧处理器，必须先判断数据长度是否满足对应功能码的最小要求再解析：

class MyFramer(ModbusRtuFramer):
    def processIncomingPacket(self, data, callback, unit, **kwargs):
        res = super().processIncomingPacket(data, callback, unit, **kwargs)
        
        # RTU帧最小长度为6字节（地址1+功能码1+数据2+CRC2）
        if len(data) >= 6:
            function = int(data[1])
            # 根据功能码判断后续字段是否完整
            if function in [0x05, 0x06] and len(data) >= 4:
                address = (data[2] << 8) | data[3]
                # 执行持久化逻辑
            elif function in [0x0F, 0x10] and len(data) >= 6:
                address = (data[2] << 8) | data[3]
                # 执行持久化逻辑
        return res

但这种手动判断的方式仍存在遗漏风险，不如请求处理器方案稳定。

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