简介:一份关于双路串口通信带帧头帧尾解析FF接收保存文件SRPHeadTail软件的设计说明文档,适用于C语言开发、基于Windows平台的嵌入式串口通信工程师与学习串口协议解析的开发者。文档围绕软件的设计目的、基本功能、开发环境、使用说明、全局及运行流程以及串口通信设计展开,重点介绍了接口层Port、协议层Protocol、数据帧结构(帧头、数据域、帧尾)等核心内容,可帮助读者理清双路串口数据收发、解析与保存的完整实现思路。资源包内为1个docx文档,大小约448KB,内容结构清晰,目录包含页面介绍、调试窗口使用、主函数流程、数据发送/接收流程、全局数据结构等模块。目前已有65人学习下载,适合需要快速理解该软件设计方案或参考串口通信协议实现细节的开发者。
1. 双路串口通信的帧头帧尾解析,为什么非做不可
双路串口通信在设备联调里很常见:一个主机从两台从机同时收数据,而串口协议本身是字节流,没有消息边界。两台设备的报文经过底层缓冲后,被切成不定长的片段,接收端不定义帧头帧尾,就说不清一帧从哪开始、到哪结束。FF 常被选作帧头,但串口空闲态在 TTL 电平下正是 0xFF,直接拿它做帧头容易被空闲干扰骗过,所以还需要转义和校验。SRPHeadTail 这种工具要解决三件事:双通道独立接收、按帧头帧尾解析流、把校验通过的载荷保存成文件。对搭产测软件和数据采集台架的工程师来说,这套帧划分和落盘逻辑,往往比业务本身更容易翻车。
2. 帧头帧尾与 FF 转义:双路串口协议设计
2.1 帧格式:长度、校验和帧头帧尾的定位
先定义一帧里除了数据之外,哪些是必须的。常见做法是把帧头、长度、数据、校验、帧尾五部分做成固定结构:
| 字段 | 占用(字节) | 说明 |
|---|---|---|
| 帧头 | 1 | 固定 0xFF,表示一帧开始 |
| 长度 | 2 | 大端存储,表示 Data 字段字节数,最大 65535 |
| Data | N | 有效载荷,可能包含任何 0x00~0xFF |
| 校验和 | 1 | 对 Data 逐字节异或(XOR),用于检测传输错误 |
| 帧尾 | 1 | 固定 0xFE,表示一帧结束 |
为什么帧尾在长度已知时还得要?因为长度字段只能用来切分,不能防错。如果长度在噪声里被改坏,接收端会按错误长度一直等下去;帧尾则提供第二道边界,一旦在预期位置没收到帧尾,可以直接丢弃本轮,回到同步状态。
这里把长度限制在 2 字节,对绝大多数传感器、PLC 和变位机报文都够用。如果你面对的协议帧更大,可以把长度扩成 3 字节,但代价是解析循环里多两次移位和判断,性能敏感时不一定划算。校验和选异或而不是 CRC 是出于简单:一轮循环就能算完,出错概率对低速链路足够低,而且测试时用手算比对也方便。真要传金融文件或医疗数据,再考虑 CRC32。
2.2 为什么数据中的 0xFF 必须转义
帧头是 0xFF,可数据负载里也可能出现 0xFF。如果不处理,接收端在 Data 途中碰到 0xFF,会误认为是下一帧的头,直接丢掉当前帧的后半段,导致每帧都被切碎。简单的“用长度跳过数据”也有隐患:长度本身受干扰后,跳转位置错误,后面就全错。
解决这个问题的基础手段是字节填充(Byte Stuffing),也叫转义。RFC 1662 的 PPP 协议就是这么做的:用 0x7D 做转义符,遇到需要转义的字节就改成两个字节。协议定义:
| 原始字节 | 填充后 | 含义 |
|---|---|---|
| 0xFF | 0x7D 0x01 | 帧头 |
| 0xFE | 0x7D 0x02 | 帧尾 |
| 0x7D | 0x7D 0x00 | 转义符本身 |
选择 0x7D 是为了避免和 0xFF/0xFE 撞车,转义后的第二个字节是“转义映射表索引”,接收端查表还原。映射关系也可以换成异或一套算法,比如第二个字节为原字节异或 0x20,恢复时同样异或回来。查表更直观,代码里用一个数组就搞定。
2.3 发送端编码实现:对一帧数据做转义
下面这段 Python 代码把“原始数据 + 校验和 + 帧头帧尾”打包成可以发到串口的字节流:
MAP = {0xFF: 0x01, 0xFE: 0x02, 0x7D: 0x00} def tx_frame(data: bytes) -> bytes: if len(data) > 0xFFFF: raise ValueError("data too long") xor = 0 for b in data: xor ^= b # 计算 XOR 校验和 raw = b'' raw += len(data).to_bytes(2, 'big') # 长度字段,大端 raw += data # 有效载荷 raw += bytes([xor]) # 校验和 escaped = b'' for b in raw: if b in (0xFF, 0xFE, 0x7D): escaped += bytes([0x7D, MAP[b]]) # 遇到保留字节就转义 else: escaped += bytes([b]) return b'\xFF' + escaped + b'\xFE' # 帧头 + 转义体 + 帧尾逻辑说明:先计算原始数据的 XOR,拼成 length+data+xor;然后遍历这个中间缓冲,遇到 0xFF、0xFE、0x7D 时换成两字节转义序列;最后在头和尾都加上原始帧头 0xFF 和帧尾 0xFE。这里帧头帧尾不再转义,是为了接收端可以快速找到边界。
参数说明:MAP 是发送端的转义映射表,对应关系与前文的表一致;长度字段用to_bytes(2,'big'),如果你的从机用的小端,要改成'little'。速度上,这个 Python 逐字节循环在 115200 波特率下完全跑得动,但到了 1M 以上的波特率,建议换成translate表或 C 实现。
2.4 接收端解析状态机:从字节流里还原完整帧
发送端是编码,接收端要有状态机。常见做法是一个有限状态机,状态迁移如下:
| 当前状态 | 输入 | 下一状态 | 动作 |
|---|---|---|---|
| IDLE | 0xFF | HEAD | 记录帧开始 |
| IDLE | 其他 | IDLE | 丢弃,继续等头 |
| HEAD | 0x7D | ESC1 | 等待转义数据 |
| HEAD | 其他 | LEN | 将输入作为长度第 1 字节 |
| LEN | 任意 | DATA | 填充长度第 2 字节,切换进数据收集态 |
| DATA | 0x7D | ESC2 | 转义等待 |
| DATA | 按长度接收完毕 | TAIL | 检查帧尾 |
| TAIL | 0xFE | IDLE | 校验和通过就输出帧,否则丢弃 |
| TAIL | 其他 | IDLE | 校验失败,重置 |
实际实现里,LEN 和 DATA 会拆得更细,因为长度是两字节,数据长度 V 时需要读 V+1(校验和)字节。不过状态表的核心是把“看门狗”放在边界上:任何状态下收到一个既不是转义符也不是当前所需字节的数据,都回到 IDLE 重新找帧头。这里不推荐用“找到帧头就一路按长度走”的简单逻辑,因为噪声会让长度字段失真,必须用帧尾和校验双重确认。
接收端核心代码用一个生成器函数实现,每调用一次可以喂入任意字节,返回解析出的完整帧:
MAP_RX = {0x01: 0xFF, 0x02: 0xFE, 0x00: 0x7D} def rx_frame(): buf = bytearray() state = 'IDLE' length = 0 cnt = 0 xor = 0 data = bytearray() while True: b = yield if state == 'IDLE': if b == 0xFF: state = 'HEAD' elif state == 'HEAD': if b == 0x7D: state = 'ESC1' else: length = b << 8 state = 'LEN' elif state == 'LEN': length |= b state = 'DATA' cnt = 0 data = bytearray() xor = 0 elif state == 'DATA': if b == 0x7D: state = 'ESC2' else: data.append(b) xor ^= b cnt += 1 if cnt == length + 1: # 数据+校验和收完 state = 'TAIL' elif state == 'ESC1': length = (MAP_RX[b]) << 8 # 恢复长度高字节 state = 'LEN' elif state == 'ESC2': data.append(MAP_RX[b]) xor ^= MAP_RX[b] cnt += 1 if cnt == length + 1: state = 'TAIL' elif state == 'TAIL': if b == 0xFE and xor == 0: yield bytes(data[:-1]) # 去掉校验和输出帧 state = 'IDLE'逻辑说明:生成器用yield接收外部字节,内部把一帧拆成若干状态。长度字段先在 HEAD 拿高字节,在 LEN 拿低字节;DATA 阶段每收到一个数据字节就追加并累加异或,当数量达到 length 时再额外收一个校验字节;打到 TAIL 后,只有帧尾是 0xFE 且异或归零才认为帧有效,否则整个状态机重置。注意yield在最后一行会再次交出一个帧,因此调用方要连续next或send。
参数说明:MAP_RX是反向映射表,对应发送端 MAP。生成器状态保存在局部变量里,天然支持多路并发:每路串口实例化一个独立生成器,互不干扰,这正是双路接收需要的。如果你用 C 实现,直接把这套状态迁移写成一个handler(int byte)函数即可。
3. 双路串口接收:并发读串口与帧边界处理
3.1 串口参数怎么设,双路才不会相互干扰
双路串口通信里第一件事是把两路看成完全独立的物理通道,但共享一个进程和 CPU。常见参数如下:
| 参数 | 常用值 | 说明 |
|---|---|---|
| 波特率 | 115200 / 460800 | 从机和线缆长度决定,两路可以不同 |
| 数据位 | 8 | 大多数 Modbus/自定义协议固定 8 位 |
| 停止位 | 1 | 少数 1.5/2,由从机要求决定 |
| 校验位 | N | 帧内已经有 XOR,链路校验关掉 |
| 流控 | None | 接收数据量大时建议 RTS/CTS 硬件流控 |
| 超时 | 0.05~0.1s | 读不到数据时给线程让出 CPU |
两路串口的波特率不同不影响接收,因为每路有各自的驱动缓冲。真正会打架的是读串口的代码:不要让两路共用同一个 DataFrame,或者在某一路的读线程里做文件写入。IO 一旦被阻塞,另一路串口驱动缓冲区就溢出丢数据。所以双路接收最稳的结构是每路一个读线程,把解析出的帧放进独立队列,再由一个写线程统一落盘。
3.2 用 pyserial 打开两个串口,各起一个读线程
Windows 上串口是 COM3、COM7,Linux 上是 /dev/ttyUSB0、/dev/ttyS1。下面的代码先定义两个配置对象,然后启动两个线程:
import serial import threading import queue def make_serial(port, baud): ser = serial.Serial( port=port, baudrate=baud, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=0.05 ) ser.reset_input_buffer() return ser def reader(config, frame_q, parser): ser = make_serial(config['port'], config['baud']) while not stop_event.is_set(): chunk = ser.read(4096) # 一次最多读 4K if not chunk: continue for b in chunk: frame = parser.send(b) # 逐字节喂给状态机 if frame is not None: frame_q.put(frame)逻辑说明:reader是每路串口的读线程,parser是上一章那个生成器实例,调用send(b)逐字节喂数据。ser.read(4096)在串口驱动里会用 timeout 等待,最大一次取 4K 字节,减少系统调用次数。解析出完整帧后放进队列frame_q,写线程再从队列取,实现读写分离。
参数说明:ser.read(4096)的 4096 不是必须,但缓冲区大于串口驱动的默认值时,可以避免单次读取把驱动缓冲塞满;timeout=0.05让读线程每 50ms 醒来一次,如果没数据就继续循环,这样stop_event可以在 50ms 内被响应,方便程序退出。
3.3 粘包和半包:为什么状态机必须连续处理
串口通信最典型的坑是半个包。底层驱动可能在一个 read 调用里返回两个帧的一部分,比如第一帧的后 5 个字节加第二帧的前 20 个字节。如果程序按“一次 read 一帧”来写,就会在第一帧还没收完时误判超时。
上面的for b in chunk逐字节喂给状态机,天然解决半包和粘包:状态机内部保存了state、length、cnt,即使 chunk 只包含半帧,下一次 read 回来接着喂即可。这里要特别注意:生成器的yield返回值不能靠send的返回值拿,因为send的返回值是生成器跑到的下一个yield表达式,而这个设计里最后一行yield bytes(...)才是对外输出。如果你发现frame is None一直为真,检查是不是少了一次next(parser)初始化。
还有一种常见误用是“清空串口缓冲”。有人担心接收慢导致缓冲里积累旧数据,于是每次读之前调用reset_input_buffer()。这在双路通信里是自杀行为:它会丢掉一帧的后半段。正确做法是提高轮询频率,或者启用硬件流控,而不是清缓冲。
3.4 双路帧数据合并保存的队列设计
两路解析出来的帧如果都放进同一个队列,会丢失通道信息。常见做法是frame_q里放(channel, frame_bytes, timestamp)这样的三元组。写线程只需要按 channel 分发到不同文件,不必关心数据来自哪路。
frame_q = queue.Queue(maxsize=2000) readers = [] for ch in (0, 1): cfg = {'port': 'COM3' if ch == 0 else 'COM7', 'baud': 115200} parser = rx_frame() next(parser) # 启动生成器 t = threading.Thread(target=reader, args=(cfg, frame_q, parser), daemon=True) t.start() readers.append(t)代码里maxsize=2000给队列一个容量上限:当写文件变慢时,读线程会阻塞在frame_q.put,避免无限制堆积内存。这里不选择丢弃帧,是因为产测数据丢了没法补,宁可让读线程短暂等待,也不能静默丢帧。如果业务能容忍丢帧,可以把put换成put_nowait并捕获Full异常。
4. 接收保存文件:帧写入结构与落盘策略
4.1 按通道分文件,文件名带时间序号
双路串口通信保存文件时,建议一路一个文件,不要混存。文件命名里带上起始时间,方便后续对齐回放。命名规则:
| 内容 | 示例 | 说明 |
|---|---|---|
| 通道号 | ch0 | 两路用 0/1 区分 |
| 起始时间 | 20250701_153000 | 本地时间,精确到秒 |
| 序号 | 00001 | 程序重启避免覆盖 |
| 扩展名 | .bin | 二进制原始帧 |
示例:ch0_20250701_153000_00001.bin。如果两路数据需要后期合并,可以在文件里额外写入每帧的绝对时间戳,例如每帧前固定加 4 字节 Unix 秒和 2 字节毫秒,这样比靠帧顺序对齐更可靠。
4.2 落盘线程:从队列批量写二进制文件
文件写得太频繁会拖慢整个程序,写得太少又怕进程崩溃丢数据。常见做法是让写线程用accumulate + flush策略:每积累 64KB 或每 500ms,把缓冲区刷到磁盘。
import time import queue def file_saver(frame_q, out_dir): files = {0: None, 1: None} counts = {0: 0, 1: 0} while True: try: ch, frame, ts = frame_q.get(timeout=0.2) except queue.Empty: continue if files[ch] is None: fname = f"{out_dir}/ch{ch}_{time.strftime('%Y%m%d_%H%M%S')}_{counts[ch]:05d}.bin" files[ch] = open(fname, 'wb') files[ch].write(ts) # 8 字节 double 时间戳 files[ch].write(frame) counts[ch] += 1逻辑说明:file_saver是单线程循环,从队列取出三元组,先判断当前通道的文件是否已打开,没有则创建新文件。写入顺序是时间戳在前、帧载荷在后,这样一个文件就是完整的时间序列,回放时用struct.unpack('d', ...)就能切出时间和数据。队列空时用timeout=0.2让线程休眠 200ms,避免忙等。
参数说明:每帧写一个时间戳的开销很小,但如果你每秒有上万帧,逐个write会成为瓶颈。这时可以把若干帧拼接成一个大bytes再write,或者改用预分配缓冲。counts只是文件序号,如果程序运行过程中文件超过 100MB,建议切分新文件,避免 FAT32 的 4GB 限制。
4.3 保存文件的校验:写完一帧再验一遍
文件保存不等于数据正确。为确认没有错位,可以在写线程里对解析出的帧再做一次 CRC 校验。虽然帧内已经做了 XOR,但 XOR 检错能力有限,多字节连续翻转可能漏检。生产环境我一般加一层 CRC32,在保存文件之前把原始数据和 CRC 结果写入日志,或者干脆把帧校验和从 XOR 改成 CRC32。
import zlib crc = zlib.crc32(frame) & 0xffffffff if crc != frame_crc_from_protocol: log_warning(f"ch{ch} frame crc mismatch, skip") continue逻辑说明:zlib.crc32返回 32 位整数,按位与0xffffffff是为了统一 Python 不同版本的无符号返回值。这里假设协议里帧载荷最后已经带了 CRC32,如果只有 XOR,就需要在协议端把校验字段升级为 4 字节。对 115200 波特率来说,CRC32 的计算开销完全可以忽略;对大数据量双路,建议用binascii.crc32或硬件 CRC 指令。
参数说明:校验失败要跳过该帧还是保留到单独错误文件?我一般把错误帧写入chX_err.bin,因为往往错误帧里保留着故障现场的上下文,直接丢弃会让问题排查无从下手。
4.4 文件写入性能:打开文件、flush 与重试
Windows 上写串口数据文件会碰到两个常见问题:写入时被杀毒软件扫描、磁盘休眠导致第一次写很慢。常见做法是打开文件时用标准缓冲,每秒flush一次,并在程序退出时统一close。如果担心断电丢数据,可以加一行fsync,但要清楚fsync会把整个文件系统事务刷下去,频繁调用会明显掉速。一般我会把flush间隔做成可配置参数,默认 1 秒,在需要极低延迟的场景调到 100ms,但会牺牲写入吞吐。
5. 帧头帧尾解析排错与验证的三个技巧
5.1 用虚拟串口做回环测试
没有真实设备时,用 socat 创建一对虚拟串口,把两个串口连接起来:一个串口写数据,另一个串口跑接收程序。常见做法是先用 Python 脚本按协议发送构造帧,再检查接收保存的文件是否符合预期。比如:
socat -d -d pty,raw,echo=0,link=/tmp/ttyS10 pty,raw,echo=0,link=/tmp/ttyS11 &然后程序里一个线程打开/tmp/ttyS10,另一个打开/tmp/ttyS11。回环测出的问题多半不在协议,而在串口参数:数据位 7/8、校验位和停止位不一致时,帧头 0xFF 都会被改成别的值,解析器永远等不到头。
5.2 怎么判断是丢帧还是解析错位
保存文件的大小和帧数不一致时,先看文件里是否存在连续的 0xFF 0xFE 对。错误定位到协议层后,把收到的原始字节流单独记录一份:在解析器入口加一行raw_log.write(bytes([b])),用原始流和解析结果比对。如果原始流里能找到完整帧但解析器没输出,说明状态机在某个状态卡住;如果原始流本身就缺字节,那就是串口驱动缓冲区溢出,需要调大相关参数或启用流控。
5.3 给解析器加时间戳的最后一招
我要重点讲这个技巧:让解析器在输出帧时附带“帧头到达时刻”,而不是在写文件时才取当前时间。这样即使数据在队列里排队几分钟,文件里的时间戳依然能反映物理接收时刻,用于回放和对时都非常有用。实现很简单,在reader里记录遇到裸 0xFF 的瞬间:
head_ts = time.time() for b in chunk: if b == 0xFF: # 数据中的 0xFF 已被转义,裸 0xFF 即帧头 head_ts = time.time() frame = parser.send(b) if frame is not None: ts_packed = struct.pack('d', head_ts) frame_q.put((ch, frame, ts_packed))时间戳用double结构体保存,精度可以到微秒级,存文件后在 Python 里用struct.unpack('d', ...)读出来,转成 UTC 字符串。这个做法不依赖队列长度,即使系统负载高也不会失真。把head_ts和帧数据同时落盘后,用脚本按时间戳排序,如果发现接收顺序和时间戳倒挂,说明双路驱动中断调度有问题,这时才需要去排查 USB-232 转接芯片驱动和系统中断绑定。
本文还有配套的精品资源,点击获取