Modbus ASCII协议读取9926字节数据的分段读取方案咨询

需要读取9926字节数据，硬件采用UART串口通信，软件协议为Modbus ASCII。由于该协议无法一次性传输大量数据，咨询是否可以采用分包读取方式，或有无更优方案？

我正尝试从规格为80622+6的热电堆红外阵列读取数据，当前使用的Python代码如下：

import serial
import time

try:
    # 打开串口
    ser = serial.Serial(
        port='/dev/ttyUSB0',
        baudrate=115200,
        stopbits=1,
        parity='N',
        bytesize=8,
        timeout=1
    )

    while True:
        # 发送Modbus ASCII请求
        request_message = "51 75 65 72 79 54 0D 0A"
        request_bytes = bytes.fromhex(request_message.replace(" ", ""))
        print("Sent data:", request_bytes)
        ser.write(request_bytes)

        # 等待响应
        time.sleep(20)

        expected_length = 80 * 62 * 2 + 6
        response_bytes = b''

        while len(response_bytes) < expected_length:
            remaining_bytes = expected_length - len(response_bytes)
            print("Before reading response")
            try:
                chunk = ser.read(remaining_bytes)
                len_response_bytes = len(response_bytes)
                print("读到的数据:", chunk)
                print("读到的数据的长度:", len_response_bytes)

            except Exception as e:
                print(f"Error reading response:{e}")
            print("After reading response")
            response_bytes += chunk

        # 处理响应数据
        # 显示响应数据的十六进制表示
        print("Received data (hex):", response_bytes)
        response_hex = ' '.join(format(byte, '02X') for byte in response_bytes)
        print("Received data (hex):", response_hex)
        # 间隔5秒再次读取
        time.sleep(5)

except Exception as e:
    print(f"Error: {e}")
finally:
    # 关闭串口
    ser.close()

1. 分包读取是可行的，且为Modbus大场景下的标准方案 #

Modbus ASCII协议单帧长度有限（ASCII编码后通常不超256字符，对应实际数据约120字节），必须通过拆分多个Modbus请求获取全部数据：

• 9926字节对应4963个16位寄存器（每寄存器2字节），加上额外6字节数据，可按寄存器地址分段，每次请求一段连续寄存器。
• 使用Modbus功能码03（读保持寄存器），每次指定起始地址和读取数量，比如每次读200个寄存器（对应400字节），分25次即可获取全部数据。

2. 当前代码的核心问题 #

• 非标准Modbus帧：你发送的request_message不符合Modbus ASCII规范，标准帧格式为:[地址][功能码][起始地址][寄存器数量][LRC校验]\r\n，缺失标准格式会导致设备无法正确响应。
• 低效等待逻辑：固定time.sleep(20)完全冗余，可通过监听帧结束符\r\n判断单帧接收完成。
• 无校验机制：未验证Modbus帧的LRC校验码，易接收错误数据。

3. 优化后的实现思路 #

（1）构造标准Modbus ASCII请求帧

手动构造帧需计算LRC校验码，示例函数如下：

def calculate_lrc(data):
    lrc = 0
    for byte in data:
        lrc ^= byte
    return lrc

def build_modbus_ascii_request(slave_addr, func_code, start_addr, reg_count):
    # 转换为字节序列
    addr = slave_addr.to_bytes(1, byteorder='big')
    func = func_code.to_bytes(1, byteorder='big')
    start = start_addr.to_bytes(2, byteorder='big')
    count = reg_count.to_bytes(2, byteorder='big')
    # 拼接数据段
    data = addr + func + start + count
    # 计算LRC
    lrc = calculate_lrc(data).to_bytes(1, byteorder='big')
    # 转换为ASCII并添加帧头帧尾
    frame = b':' + data.hex().upper().encode() + lrc.hex().upper().encode() + b'\r\n'
    return frame

（2）分包读取逻辑

import serial
import time

def calculate_lrc(data):
    lrc = 0
    for byte in data:
        lrc ^= byte
    return lrc

def build_modbus_ascii_request(slave_addr, func_code, start_addr, reg_count):
    addr = slave_addr.to_bytes(1, byteorder='big')
    func = func_code.to_bytes(1, byteorder='big')
    start = start_addr.to_bytes(2, byteorder='big')
    count = reg_count.to_bytes(2, byteorder='big')
    data = addr + func + start + count
    lrc = calculate_lrc(data).to_bytes(1, byteorder='big')
    frame = b':' + data.hex().upper().encode() + lrc.hex().upper().encode() + b'\r\n'
    return frame

def parse_modbus_ascii_response(response):
    # 去掉帧头帧尾
    payload = response[1:-2]
    # 转换为字节
    data_bytes = bytes.fromhex(payload.decode())
    # 分离地址、功能码、数据、LRC
    slave_addr = data_bytes[0]
    func_code = data_bytes[1]
    data_len = data_bytes[2]
    data = data_bytes[3:-1]
    received_lrc = data_bytes[-1]
    # 校验LRC
    calculated_lrc = calculate_lrc(data_bytes[:-1])
    if received_lrc != calculated_lrc:
        raise ValueError("LRC校验失败")
    return data

try:
    ser = serial.Serial(
        port='/dev/ttyUSB0',
        baudrate=115200,
        stopbits=1,
        parity='N',
        bytesize=8,
        timeout=2
    )

    total_registers = 4963  # 9926字节 / 2字节每寄存器
    read_per_request = 200  # 每次读取200个寄存器
    slave_address = 1  # 设备地址，根据实际修改
    all_data = b''

    while True:
        all_data = b''
        # 分包发送请求
        for i in range(0, total_registers, read_per_request):
            current_count = min(read_per_request, total_registers - i)
            request = build_modbus_ascii_request(slave_address, 3, i, current_count)
            print(f"发送请求: {request}")
            ser.write(request)
            
            # 读取响应直到收到换行符
            response = b''
            while b'\r\n' not in response:
                chunk = ser.read(100)
                if not chunk:
                    raise TimeoutError("读取响应超时")
                response += chunk
            
            # 解析响应数据
            data = parse_modbus_ascii_response(response)
            all_data += data
            time.sleep(0.1)  # 短间隔避免设备过载
        
        # 处理额外的6字节数据（根据设备协议补充读取逻辑）
        extra_request = build_modbus_ascii_request(slave_address, 3, total_registers, 3)  # 3个寄存器=6字节
        ser.write(extra_request)
        extra_response = ser.read_until(b'\r\n')
        extra_data = parse_modbus_ascii_response(extra_response)
        all_data += extra_data

        print(f"总接收数据长度: {len(all_data)}")
        # 后续处理数据...
        time.sleep(5)

except Exception as e:
    print(f"错误: {e}")
finally:
    ser.close()

4. 更优方案：使用成熟的Modbus库 #

直接使用pymodbus库可省去手动构造帧和校验的麻烦，简化代码：

from pymodbus.client import ModbusSerialClient
from pymodbus.payload import BinaryPayloadDecoder
from pymodbus.constants import Endian

client = ModbusSerialClient(
    method='ascii',
    port='/dev/ttyUSB0',
    baudrate=115200,
    parity='N',
    stopbits=1,
    bytesize=8,
    timeout=2
)

if client.connect():
    total_registers = 4963
    read_per_request = 200
    all_data = b''
    slave_addr = 1

    for i in range(0, total_registers, read_per_request):
        count = min(read_per_request, total_registers - i)
        response = client.read_holding_registers(address=i, count=count, slave=slave_addr)
        if not response.isError():
            decoder = BinaryPayloadDecoder.fromRegisters(response.registers, byteorder=Endian.Big)
            all_data += decoder.decode_bytes(count*2)
        else:
            print(f"请求失败: {response}")
    
    # 读取额外6字节数据
    extra_response = client.read_holding_registers(address=total_registers, count=3, slave=slave_addr)
    if not extra_response.isError():
        decoder = BinaryPayloadDecoder.fromRegisters(extra_response.registers, byteorder=Endian.Big)
        all_data += decoder.decode_bytes(6)
    
    print(f"总数据长度: {len(all_data)}")
    client.close()

内容的提问来源于stack exchange，提问作者李思怡