解决Python串口YModem传输中的丢包、超时与校验错误

在嵌入式设备固件更新（IAP/Bootloader）和仪器数据交互中，利用 Python 编写上位机脚本配合 pyserial 执行 YModem 协议传输是常见方案。然而，在实际运行过程中，开发者极易遇到数据丢包、响应超时以及 CRC 校验错误三大痛点。解决这些问题的核心途径在于：建立严格的包间自适应延时流控防止硬件 FIFO 溢出、基于下位机 Flash 擦写耗时建立分阶段动态超时模型、以及统一严格遵循大端序的 CRC16-CCITT 算法与数据补齐约定。通过理顺状态机转换与异常重试逻辑，可以使 Python 串口 YModem 传输达到工业级的高可靠度。

一、YModem 数据帧结构与协议交互机制

YModem 协议是对传统 XModem 协议的扩展，它支持多包批量传输、文件名与文件大小元数据传递，并提供 128 字节（标准帧 SOH）与 1024 字节（长帧 STX）两种载荷格式。一次完整的 YModem 传输由起始帧（Block 0）、有效数据帧（Block 1..N）以及结束空帧组成。

每个数据帧的物理构成必须严格符合字节对齐规范：包含 1 字节帧起始符、1 字节正向包序号、1 字节序号补码、固定长度数据载荷（不足部分以 0x1A 或 0x00 填充），以及 2 字节 CRC16-CCITT 校验和。序号补码必须满足 Seq + (~Seq) == 0xFF，若序号与补码不匹配，接收端将直接判定为链路帧头受损并丢弃该包。

在校验机制上，YModem 规定使用 CRC16-CCITT（多项式为 0x1021，初始值为 0x0000），且计算区间仅覆盖数据有效载荷（Data Payload）本身，不计入帧起始符、序号及补码。计算得出的 16 位整型校验和必须以大端字节序（高位字节在前，低位字节在后）追加在帧尾。

二、三大核心故障根因分析与针对性治理

1. 串口丢包：硬件 FIFO 溢出与 Flash 写入关中断

故障表现：Python 上位机在无报错状态下持续发送数据，但下位机中途失联，或者下位机接收到的包序号出现跳跃，导致传输立即中断。

底层根因：

• 硬件 FIFO 深度有限：通用 MCU（如 STM32、GD32 或 ESP32）的串口外设硬件 FIFO 通常只有 8 到 16 字节甚至仅有单字节缓冲寄存器。当 Python 借助主机操作系统的输出缓冲全速突发写入时，下位机中断服务函数来不及搬移数据，引发串口硬件溢出错误（Overrun Error, ORE）。
• Flash 编程期间屏蔽中断：下位机在收到一个完整数据包后，需将数据解包并烧录至内部 Flash。许多微控制器的片内 Flash 擦写控制器在编程期间会挂起总线或要求关闭全局中断，这会导致正在串口线上送达的新数据位直接丢失。

2. 通信超时：擦除耗时失配与固定超时陷阱

故障表现：上位机成功捕获握手字符 'C' 并发送 Block 0 后，在等待从机的 ACK 时频繁抛出读取超时异常；或者传输在进行到特定扇区边界时突然断流。

底层根因：

• Flash 全片/扇区擦除耗时较长：下位机 Bootloader 在解析 Block 0 中的文件大小元数据后，会根据文件长度计算需要擦除的 Flash 扇区数量并执行擦除。Flash 扇区擦除是毫秒甚至秒级的慢速物理操作（多扇区擦除耗时通常在 1.5 秒至 4 秒之间）。如果 Python 代码将 serial.Serial(timeout=1.0) 全局固定为 1 秒，上位机将在单片机擦除完毕前提前报错退出。
• 串口开启时的 DTR/RTS 抖动引发复位延迟：pyserial 打开某些 USB 转串口芯片（如 CH340、CP2102）时，默认会触发 DTR/RTS 信号跳变，导致单片机硬件复位。若上位机未等待单片机 Bootloader 完成初始化就直接开始超时计时，极易漏过前几秒发出的 'C' 字符。

3. CRC 校验错误：算法多项式混淆与大小端字节序颠倒

故障表现：上位机能够正常发送数据帧，但下位机对每个数据包都返回 NAK (0x15)，连续重试达到阈值后传输被中止。

底层根因：

• CRC 变种算法不一致：工业界存在多种 16 位 CRC 标准。YModem 规范严格要求使用 CRC16-CCITT（Polynomial 0x1021，初始值 0x0000）。很多初学者误用了 MODBUS CRC（Poly 0x8005）或初始值为 0xFFFF 的 X.25 算法，导致算出的校验值截然不同。
• 高低字节序排列颠倒：YModem 规定校验和必须以大端序（Big-Endian / Network Byte Order）传输。如果上位机使用 struct.pack('<H', crc)（小端序）打包，单片机解析校验和时高低字节倒置，校验必定失败。
• 末包填充补齐不合规：传输文件的最后一个分片通常不足 128 或 1024 字节。YModem 要求必须将其垫补至完整帧长（传统使用 0x1A 填充，现代 Bootloader 亦接受 0x00）。若上位机未做定长补齐而发送短包，下位机按固定长度解帧会导致后续字节错位，进而引发校验崩溃。

三、128字节标准包 vs 1K字节包：差异对比与工况选型

在设计 Python YModem 发送程序时，开发者常面临选择 128 字节标准包（SOH）还是 1024 字节长包（STX）的权衡。在波特率 115200bps 的基准条件下，二者在开销、容错代价与硬件要求上存在明显分化。

关键差异解析：

• 协议有效载荷占比：128 字节包每帧需承载 5 字节固定控制头尾（1 字节 SOH + 2 字节序号 + 2 字节 CRC），控制开销约为 3.90%；而 1K 字节长包的控制开销仅为 0.49%。在洁净、短距离板级通信中，1K 帧能显著提升净吞吐率。
• 误码重传惩罚：当现场存在电磁脉冲或线路抖动时，若数据出现 1 字节突变导致 CRC 校验失败，128 字节包只需重传 133 字节（耗时约 11.5ms）；而 1K 帧一旦报错则必须重传整整 1029 字节（耗时约 89.3ms）。在高误码率链路上，1K 帧的频繁整包重传会导致有效传输速率呈断崖式下跌。
• 从机内存与写入颗粒度约束：低功耗 MCU（如 Cortex-M0 内核）通常只有 4KB 到 8KB 片上 RAM。若分配 1024 字节双接收缓冲，将占据大半内存空间；同时某些 Flash 每次仅支持写入 128 或 256 字节，1K 帧需要在内存中做切片写入。

选型结论：对于板载短距走线、USB 转串口直连且固件体积超过 100KB 的场景，推荐采用 1K 字节帧以缩短升级时间；对于长线缆、RS-485 工业多机总线或低 RAM 规格 MCU，应优先采用 128 字节标准帧以确保高抗噪稳定性。

四、状态机驱动的完整传输与重试操作时序

为了兼顾健壮性，Python 端必须实现具备完整状态回滚与最大重试次数保护的状态机。一次严谨的传输操作包含五个执行步骤：

1. 步骤一：监听握手信标（Handshake）
   打开串口并清空缓冲队列，以非阻塞或短轮询方式持续监听下位机发出的 'C' (0x43) 字符。捕获到 'C' 代表从机已就绪并要求以 CRC 校验模式启动传输。
2. 步骤二：发送文件名元数据帧（Block 0）
   构建序号为 0x00 的数据包。有效载荷前部为文件名字符串以 0x00 截断，后接文件十进制大小以 0x20 或 0x00 截断，其余空间以 0x00 补齐。发送后等待从机响应 ACK (0x06) 以及紧随其后的新 'C' 字符（提示从机已完成 Flash 扇区擦除并准备接收内容）。
3. 步骤三：循环发送数据载荷（Block 1..N）
   从偏移量 0 开始切片读取文件，包序号从 0x01 递增（超过 0xFF 自动回绕至 0x00）。每发一包后启动超时倒计时监听响应：收到 ACK 则推进包序号；收到 NAK 或触发超时则保留当前数据包就地重传，单包重试计数器自增 1。若重发达到预设上限（推荐 5 次），发送连续 CAN (0x18) 取消传输并退出。
4. 步骤四：文件结束协商（EOT Handshake）
   文件内容全部发送完成后，上位机发送 EOT (0x04)。标准 YModem 规定从机对第一个 EOT 会回应 NAK，上位机必须发送第二个 EOT，从机确认后回复 ACK 并紧接着发送一个 'C' 字符以询问后续文件。
5. 步骤五：终结空帧并退出（End of Session）
   上位机发送一个序号为 0x00 且载荷全为 0x00 的空数据包。从机校验通过后回复 ACK，一次稳健的 YModem 升级流程宣告圆满结束。

五、生产级高可靠 Python 实现代码

下面给出基于 pyserial 封装的高可用 YModem 发送器类。代码内嵌 CRC16-CCITT 快速计算引擎、分阶段动态超时控制、单包补齐与闭环重传容错逻辑：

import os
import time
import struct
import serial

class RobustYModemSender:
    SOH = b'\x01'   # 128 字节数据帧起始
    STX = b'\x02'   # 1024 字节数据帧起始
    EOT = b'\x04'   # 传输结束标志
    ACK = b'\x06'   # 肯定应答
    NAK = b'\x15'   # 否定应答（校验失败）
    CAN = b'\x18'   # 强制取消传输
    CRC_C = b'C'    # 请求 CRC 模式握手标信

    def __init__(self, port, baudrate=115200, use_1k=False):
        self.port = port
        self.baudrate = baudrate
        self.use_1k = use_1k
        self.ser = None

    @staticmethod
    def calc_crc16_ccitt(data: bytes) -> int:
        """标准 CRC16-CCITT (XModem 变种): Poly=0x1021, Init=0x0000"""
        crc = 0x0000
        for byte in data:
            crc ^= (byte << 8)
            for _ in range(8):
                if crc & 0x8000:
                    crc = ((crc << 1) ^ 0x1021) & 0xFFFF
                else:
                    crc = (crc << 1) & 0xFFFF
        return crc

    def connect(self):
        """开启串口并确保输入输出缓冲队列复位"""
        self.ser = serial.Serial(
            port=self.port,
            baudrate=self.baudrate,
            bytesize=serial.EIGHTBITS,
            parity=serial.PARITY_NONE,
            stopbits=serial.STOPBITS_ONE,
            timeout=1.0,
            rtscts=False,
            dsrdtr=False
        )
        time.sleep(0.15)  # 等待 USB 串口芯片电平稳定
        self.ser.reset_input_buffer()
        self.ser.reset_output_buffer()

    def wait_for_char(self, target: bytes, timeout: float = 3.0) -> bool:
        """带自定义超时与轮询的字符等待方法"""
        deadline = time.time() + timeout
        while time.time() < deadline:
            if self.ser.in_waiting > 0:
                char = self.ser.read(1)
                if char == target:
                    return True
            time.sleep(0.005)
        return False

    def build_packet(self, seq: int, payload: bytes) -> bytes:
        """构建单个符合 YModem 规范的数据帧"""
        header = self.STX if self.use_1k else self.SOH
        packet_size = 1024 if self.use_1k else 128
        
        # 补齐不足长度的负载数据（使用 0x1A 填充）
        if len(payload) < packet_size:
            payload = payload.ljust(packet_size, b'\x1A')
            
        seq_num = seq & 0xFF
        seq_inv = (0xFF - seq_num) & 0xFF
        crc = self.calc_crc16_ccitt(payload)
        
        # 组装：起始符(1B) + 序号(1B) + 反码(1B) + 载荷(NB) + 大端CRC(2B)
        return header + bytes([seq_num, seq_inv]) + payload + struct.pack('>H', crc)

    def send_file(self, filepath: str, max_retries: int = 5) -> bool:
        """执行完整稳健的 YModem 文件传输流程"""
        if not os.path.isfile(filepath):
            raise FileNotFoundError(f"未找到目标文件: {filepath}")

        filename = os.path.basename(filepath)
        filesize = os.path.getsize(filepath)

        try:
            print("[*] 等待从机握手信标 'C' (超时 15 秒)...")
            if not self.wait_for_char(self.CRC_C, timeout=15.0):
                print("[-] 握手失败: 未捕获到下位机发出的 'C'")
                return False

            # --- 步骤 2: 发送 Block 0 (包含文件名与大小) ---
            print("[*] 握手成功，构建并发送 Block 0 (元数据帧)...")
            header_payload = filename.encode('utf-8') + b'\x00' + str(filesize).encode('utf-8') + b'\x20'
            block0 = self.build_packet(0, header_payload)
            
            self.ser.write(block0)
            self.ser.flush()

            # 等待从机确认 ACK
            if not self.wait_for_char(self.ACK, timeout=2.0):
                print("[-] Block 0 未收到从机 ACK 确认")
                return False

            # 等待从机完成 Flash 擦除并发出的新 'C' (动态分配 5 秒安全超时)
            print("[*] 正在等待下位机完成 Flash 扇区擦除...")
            if not self.wait_for_char(self.CRC_C, timeout=5.0):
                print("[-] 等待 Flash 擦除完成信标 'C' 超时")
                return False

            # --- 步骤 3: 循环发送数据块 ---
            packet_size = 1024 if self.use_1k else 128
            seq = 1
            with open(filepath, 'rb') as f:
                while True:
                    chunk = f.read(packet_size)
                    if not chunk:
                        break  # 文件已全部读取

                    packet = self.build_packet(seq, chunk)
                    retry = 0
                    success = False

                    while retry < max_retries:
                        self.ser.write(packet)
                        self.ser.flush()

                        # 监听应答（数据块阶段配置 1.5 秒动态超时）
                        deadline = time.time() + 1.5
                        resp = b''
                        while time.time() < deadline:
                            if self.ser.in_waiting > 0:
                                resp = self.ser.read(1)
                                break
                            time.sleep(0.002)

                        if resp == self.ACK:
                            success = True
                            # 插入微小自适应包间延时，给从机写入留出安全窗口
                            time.sleep(0.008)
                            break
                        elif resp == self.NAK:
                            print(f"[!] 包 {seq} 收到 NAK (校验错误)，准备重发 ({retry + 1}/{max_retries})...")
                            retry += 1
                        elif resp == self.CAN:
                            print("[-] 传输被下位机主动取消 (收到 CAN)")
                            return False
                        else:
                            print(f"[!] 包 {seq} 等待响应超时 (丢包/无应答)，重发 ({retry + 1}/{max_retries})...")
                            retry += 1

                    if not success:
                        print(f"[-] 包 {seq} 连续重试超限，强制中止传输")
                        self.ser.write(self.CAN * 2)
                        return False

                    seq = (seq + 1) & 0xFF

            # --- 步骤 4: 结束传输 EOT 握手 ---
            print("[*] 文件内容发送完毕，开始 EOT 结束协商...")
            self.ser.write(self.EOT)
            self.ser.flush()
            # 首次 EOT 通常收到从机 NAK
            self.wait_for_char(self.NAK, timeout=1.5)

            # 再次发送 EOT
            self.ser.write(self.EOT)
            self.ser.flush()
            if not self.wait_for_char(self.ACK, timeout=2.0):
                print("[-] 第二次 EOT 未收到 ACK")
                return False

            # 接收从机发来的 'C'
            self.wait_for_char(self.CRC_C, timeout=2.0)

            # --- 步骤 5: 发送空 Block 0 终止会话 ---
            print("[*] 发送会话终结空数据帧...")
            empty_block = self.build_packet(0, b'')
            self.ser.write(empty_block)
            self.ser.flush()
            self.wait_for_char(self.ACK, timeout=2.0)

            print("[+] YModem 传输全流程圆满完成！")
            return True

        finally:
            if self.ser and self.ser.is_open:
                self.ser.close()

六、常见故障排查矩阵与加固速查

为便于在工程现场快速定位问题，下表将 Python 串口 YModem 传输的典型异常、排查手段与修复策略进行了系统梳理：