Python实现YModem协议串口通信：原理与pyserial代码示例

一、YModem 协议核心原理与帧格式定义

在嵌入式开发中，若直接通过原始串口字节流发送固件，极易因偶发噪波、波特率抖动或接收端 Flash 写入延时出现丢字节或乱序，且接收端无法感知传输是否完整。YModem 在经典 XModem 协议基础上演进而来，引入了文件名与大小元数据传递、1024 字节大包拓展以及多重校验机制，成为各类 MCU Bootloader（如 STM32 官方 IAP 示例）的标准升级协议。

1. 核心控制字符

YModem 协议在通信链路中通过单字节控制字符完成会话发起、确认、重发及中断协调。下表列出通信两端必须识别的标准控制字符定义：

2. 数据帧通用组成结构

无论是起始帧还是后续数据帧，YModem 的物理帧结构均由五个固定部分拼接而成：

• 帧起始标识（1 字节）：取值为 SOH (0x01) 或 STX (0x02)，同时决定了该帧的数据净载荷长度是 128 字节还是 1024 字节。
• 包序号（1 字节）：取值范围为 0x00 ~ 0xFF。起始帧固定为 0x00，真实数据块从 0x01 递增，超过 255 后自动回绕到 0 循环计数。
• 包序号反码（1 字节）：取值为 255 - 包序号。用于接收端校验序号本身的电气完整性。接收端计算 (seq + ~seq) & 0xFF == 0xFF，若不相等则直接判定帧头错位，丢弃并回复 NAK。
• 数据有效载荷（128 或 1024 字节）：固定长度。若实际数据不足包长，末尾使用 0x00（或 0x1A CPMEOF）补齐。
• CRC-16 校验码（2 字节）：采用大端模式（Big-Endian），先传高 8 位，后传低 8 位。校验计算范围仅包含数据载荷部分，不计算前置的 3 字节帧头。

3. 起始帧（Block 0）的特殊设计

与普通数据包不同，包序号为 0x00 的数据包被称为 Block 0。它不承载固件主体，而是承载传输任务的元数据：

• 文件名字段：以 ASCII 编码存储文件名（例如 app_v1.0.bin），后紧跟一个 0x00 作为结束分隔符。
• 文件大小字段：紧随其后，以十进制 ASCII 字符串记录文件的总字节数（例如 "32768"），同样以 0x00 分隔。接收端根据此大小即可在写入 Flash 时准确截断末尾填充字节，避免误将补齐填充写入镜像。
• 剩余空间填充：将该数据块的剩余部分全部填充为 0x00，直至凑满 128 字节（或 1024 字节）。
• 任务终结标识：当整个传输任务结束时，发送端会再次发送一个包序号为 0、载荷全为 0x00 的空 Block 0，明确通知接收端不会再有新文件进入批处理。

4. 关键差异对比：128 字节 SOH 帧与 1024 字节 STX 帧

在实际工程实现中，YModem 允许选用 128 字节（SOH）或 1024 字节（STX）作为数据切片单位。理解这两者的关键差异对于保证固件升级的效率与稳定性至关重要：

二、完整交互流程与状态机模型

YModem 是一种严格的“发一包、等一包、收一包确认”的停等式（Stop-and-Wait）协议。一个标准的文件会话严格划分为协商握手、数据流循环与终止确认三个阶段。

1. 三阶段时序拆解

1. 第一阶段：协商与元数据传输
   接收端启动后进入 Bootloader 监听状态，每隔 1~2 秒主动向串口发送字符 'C' (0x43)。上位机捕捉到 'C' 后，立即组装包序号为 0x00 的 Block 0（包含待发固件的文件名和大小），以 SOH 或 STX 发出。接收端收到后进行校验，若正确则回复 ACK (0x06)，紧接着主动发送第二个 'C'，指示发送端可以启动主体数据传输。
2. 第二阶段：数据流分包发送与逐包确认
   发送端从偏移量 0 开始读取文件二进制数据，每次截取 1024 字节（不足补齐），包序号从 0x01 递增，发送 [STX | seq | ~seq | data | CRC]。发送端阻塞等待接收端返回的响应字符： 若为 ACK，则推进包序号并截取下一块数据；若为 NAK 或超时，则保持包序号不变重新发包；若为 CAN 则立即熔断。
3. 第三阶段：文件结束与双重握手终止
   文件全部数据发完并得到最后一个 ACK 后，发送端向接收端发送首个 EOT (0x04)。 注意协议细节：接收端为了防止偶发杂波误触 EOT，按 YModem 规范首次会回复 NAK (0x15)。发送端收到该 NAK 后，必须发送第二个 EOT；接收端此时确认真为文件尾，回复 ACK 并紧接着发送一个 'C'。发送端最后发送一个全 0x00 的空 Block 0，接收端再次回复 ACK，整个串口会话正式彻底终结。

三、Python 与 pyserial 完整代码实现

编写可靠的 YModem 传输工具，关键在于严谨的 CRC-16 纯纯度计算、精准的超时重发控制以及完备的状态机收尾处理。以下基于标准库和 pyserial 进行完整实现。

1. 环境准备与依赖

# 安装 pyserial 串口通信核心库
pip install pyserial

2. 完整发送端类实现（可直接运行）

# -*- coding: utf-8 -*-
"""
Python YModem 发送端实现示例
基于 pyserial，支持 128B (SOH) 与 1024B (STX) 模式、完整握手、超时重传与中断响应
"""

import os
import sys
import time
import struct
import serial

class YModemError(Exception):
    """YModem 协议基础异常"""
    pass

class YModemTimeoutError(YModemError):
    """超时未收到应答异常"""
    pass

class YModemAbortedError(YModemError):
    """接收端或发送端强制中止异常"""
    pass


def calc_crc16(data: bytes) -> int:
    """
    计算 CRC-16-CCITT 校验码 (多项式 0x1021, 初始值 0x0000, 高位在先)
    :param data: 待计算的二进制字节串
    :return: 16位无符号整数校验和
    """
    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


class YModemSender:
    # 常用协议控制字符定义
    SOH = b'\x01'  # 128 字节帧头
    STX = b'\x02'  # 1024 字节帧头
    EOT = b'\x04'  # 结束传输
    ACK = b'\x06'  # 确认应答
    NAK = b'\x15'  # 否认重发
    CAN = b'\x18'  # 取消传输
    CRC_C = b'\x43'  # 字符 'C'

    def __init__(self, ser: serial.Serial, max_retries: int = 10, packet_timeout: float = 3.0):
        """
        初始化 YModem 发送引擎
        :param ser: 已成功初始化的 serial.Serial 实例
        :param max_retries: 单包最大允许重试次数
        :param packet_timeout: 单字符/应答等待超时时间（秒）
        """
        self.ser = ser
        self.max_retries = max_retries
        self.packet_timeout = packet_timeout

    def _read_byte(self, timeout: float = None) -> bytes:
        """从串口读取单个字节，带超时检查"""
        prev_timeout = self.ser.timeout
        if timeout is not None:
            self.ser.timeout = timeout
        try:
            char = self.ser.read(1)
            return char
        finally:
            self.ser.timeout = prev_timeout

    def _build_packet(self, header: bytes, seq: int, data: bytes) -> bytes:
        """
        组装符合规范的单个 YModem 帧
        格式: [Header(1B)] [Seq(1B)] [~Seq(1B)] [Data(NB)] [CRC_H(1B)] [CRC_L(1B)]
        """
        seq_byte = seq & 0xFF
        seq_inv_byte = (~seq) & 0xFF
        crc_val = calc_crc16(data)
        crc_bytes = struct.pack(">H", crc_val)
        return header + bytes([seq_byte, seq_inv_byte]) + data + crc_bytes

    def _send_and_wait_ack(self, packet: bytes) -> bool:
        """发送单包数据并等待 ACK，遇 NAK 或超时自动重传"""
        retries = 0
        while retries < self.max_retries:
            self.ser.write(packet)
            self.ser.flush()

            # 等待接收端响应字符
            resp = self._read_byte(self.packet_timeout)

            if resp == self.ACK:
                return True
            elif resp == self.NAK:
                # 接收端指示校验失败，重发当前包
                retries += 1
                continue
            elif resp == self.CAN:
                # 检查是否为双连续 CAN 中止
                next_char = self._read_byte(1.0)
                if next_char == self.CAN:
                    raise YModemAbortedError("接收端发送连续 CAN，传输已强制终止")
                retries += 1
            elif not resp:
                # 响应超时，清空输入缓冲重试
                self.ser.reset_input_buffer()
                retries += 1
            else:
                # 收到其他非预期字符（如残留杂波），重试
                retries += 1

        raise YModemTimeoutError(f"包重发超过最大上限 {self.max_retries} 次，通信中断")

    def send_file(self, file_path: str, use_1k: bool = True, progress_callback=None):
        """
        执行完整 YModem 协议发送流程
        :param file_path: 本地待发送固件文件绝对或相对路径
        :param use_1k: True 使用 1024 字节 STX 包，False 使用 128 字节 SOH 包
        :param progress_callback: 进度回调函数 callback(sent_bytes, total_bytes)
        """
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"目标文件不存在: {file_path}")

        file_size = os.path.getsize(file_path)
        file_name = os.path.basename(file_path)

        # 阶段一：等待接收端发送初始 'C' 字符发起握手
        print("[-] 等待接收端就绪信号 ('C')...")
        waiting_c_retries = 0
        while waiting_c_retries < 30:
            char = self._read_byte(timeout=2.0)
            if char == self.CRC_C:
                print("[+] 捕获 'C' 握手信号，准备发送 Block 0...")
                break
            elif char == self.CAN:
                raise YModemAbortedError("接收端在握手阶段发送取消字符")
            waiting_c_retries += 1
        else:
            raise YModemTimeoutError("等待接收端启动握手超时（未收到 'C'）")

        # 组装 Block 0（包含文件名与文件大小字符串）
        block0_payload = file_name.encode("utf-8") + b'\x00' + str(file_size).encode("ascii") + b'\x00'
        # 起始帧统一填充至 128 字节（部分单片机仅按 128B 解析 Block 0）
        block0_payload = block0_payload.ljust(128, b'\x00')
        block0_packet = self._build_packet(self.SOH, 0, block0_payload)

        # 发送 Block 0
        self._send_and_wait_ack(block0_packet)

        # 接收端确认 Block 0 后，必须再次发送 'C' 请求数据
        resp = self._read_byte(self.packet_timeout)
        if resp != self.CRC_C:
            raise YModemError(f"Block 0 确认后未收到数据引导字符 'C'，收到: {resp}")

        # 阶段二：读取文件并分包发送
        data_block_size = 1024 if use_1k else 128
        header_byte = self.STX if use_1k else self.SOH

        seq = 1
        sent_bytes = 0

        with open(file_path, "rb") as f:
            while True:
                chunk = f.read(data_block_size)
                if not chunk:
                    break

                # 末包不足补齐 0x00
                if len(chunk) < data_block_size:
                    padded_chunk = chunk.ljust(data_block_size, b'\x00')
                else:
                    padded_chunk = chunk

                packet = self._build_packet(header_byte, seq, padded_chunk)
                self._send_and_wait_ack(packet)

                sent_bytes += len(chunk)
                if progress_callback:
                    progress_callback(min(sent_bytes, file_size), file_size)

                seq = (seq + 1) % 256

        # 阶段三：结束文件传输（双 EOT 机制）
        print("\n[-] 数据发送完成，开始结束握手...")
        # 发送第 1 个 EOT
        self.ser.write(self.EOT)
        self.ser.flush()

        resp = self._read_byte(self.packet_timeout)
        if resp == self.NAK:
            # 协议要求：收到 NAK 后发第 2 个 EOT
            self.ser.write(self.EOT)
            self.ser.flush()
            resp = self._read_byte(self.packet_timeout)

        if resp != self.ACK:
            raise YModemError(f"EOT 确认失败，未收到 ACK，收到: {resp}")

        # 接收端回复 ACK 后会发送 'C'，请求下一文件或终结
        resp = self._read_byte(self.packet_timeout)
        if resp == self.CRC_C:
            # 发送全 0 空 Block 0 终结整个批处理会话
            empty_block0 = self._build_packet(self.SOH, 0, b'\x00' * 128)
            self.ser.write(empty_block0)
            self.ser.flush()
            final_ack = self._read_byte(self.packet_timeout)
            if final_ack != self.ACK:
                print(f"[!] 警告: 终结帧未收到 ACK，收到: {final_ack}")

        print("[+] YModem 文件传输流程全部顺利完成！")


# 命令行测试驱动示例
if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Python YModem 固件串口发送工具")
    parser.add_argument("-p", "--port", required=True, help="串口号 (如 COM3 或 /dev/ttyUSB0)")
    parser.add_argument("-b", "--baud", type=int, default=115200, help="波特率 (默认 115200)")
    parser.add_argument("-f", "--file", required=True, help="待发送固件文件路径")
    parser.add_argument("--use-128b", action="store_true", help="强制使用 128 字节 SOH 帧（默认使用 1024 字节 STX）")

    args = parser.parse_args()

    # 打开串口连接
    try:
        serial_conn = serial.Serial(
            port=args.port,
            baudrate=args.baud,
            bytesize=serial.EIGHTBITS,
            parity=serial.PARITY_NONE,
            stopbits=serial.STOPBITS_ONE,
            timeout=2.0
        )
    except serial.SerialException as err:
        print(f"[x] 打开串口失败: {err}")
        sys.exit(1)

    def print_progress(current, total):
        pct = (current / total) * 100
        bar_len = 30
        filled = int(bar_len * current // total)
        bar = "=" * filled + ">" + "-" * (bar_len - filled - 1) if filled < bar_len else "=" * bar_len
        sys.stdout.write(f"\r[传输进度] [{bar}] {pct:.1f}% ({current}/{total} Bytes)")
        sys.stdout.flush()

    try:
        sender = YModemSender(serial_conn)
        use_stx = not args.use_128b
        print(f"[+] 串口 {args.port} 已就绪，波特率: {args.baud}，帧规格: {'1024B (STX)' if use_stx else '128B (SOH)'}")
        sender.send_file(args.file, use_1k=use_stx, progress_callback=print_progress)
    except Exception as e:
        print(f"\n[x] 传输异常中断: {e}")
    finally:
        serial_conn.close()
        print("[+] 串口资源已释放")

四、联调实战与嵌入式常见排查点

在实际连接 STM32 或其它 MCU 进行串口 IAP 联调时，常因物理线路或时序配合出现异常。下表总结了高频故障现象及精准定位方案：