简介一份关于双路串口通信带帧头帧尾解析FF接收保存文件SRPHeadTail软件的设计说明文档适用于C语言开发、基于Windows平台的嵌入式串口通信工程师与学习串口协议解析的开发者。文档围绕软件的设计目的、基本功能、开发环境、使用说明、全局及运行流程以及串口通信设计展开重点介绍了接口层Port、协议层Protocol、数据帧结构帧头、数据域、帧尾等核心内容可帮助读者理清双路串口数据收发、解析与保存的完整实现思路。资源包内为1个docx文档大小约448KB内容结构清晰目录包含页面介绍、调试窗口使用、主函数流程、数据发送/接收流程、全局数据结构等模块。目前已有65人学习下载适合需要快速理解该软件设计方案或参考串口通信协议实现细节的开发者。1. 双路串口通信的帧头帧尾解析为什么非做不可双路串口通信在设备联调里很常见一个主机从两台从机同时收数据而串口协议本身是字节流没有消息边界。两台设备的报文经过底层缓冲后被切成不定长的片段接收端不定义帧头帧尾就说不清一帧从哪开始、到哪结束。FF 常被选作帧头但串口空闲态在 TTL 电平下正是 0xFF直接拿它做帧头容易被空闲干扰骗过所以还需要转义和校验。SRPHeadTail 这种工具要解决三件事双通道独立接收、按帧头帧尾解析流、把校验通过的载荷保存成文件。对搭产测软件和数据采集台架的工程师来说这套帧划分和落盘逻辑往往比业务本身更容易翻车。2. 帧头帧尾与 FF 转义双路串口协议设计2.1 帧格式长度、校验和帧头帧尾的定位先定义一帧里除了数据之外哪些是必须的。常见做法是把帧头、长度、数据、校验、帧尾五部分做成固定结构字段占用字节说明帧头1固定 0xFF表示一帧开始长度2大端存储表示 Data 字段字节数最大 65535DataN有效载荷可能包含任何 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 做转义符遇到需要转义的字节就改成两个字节。协议定义原始字节填充后含义0xFF0x7D 0x01帧头0xFE0x7D 0x02帧尾0x7D0x7D 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拼成 lengthdataxor然后遍历这个中间缓冲遇到 0xFF、0xFE、0x7D 时换成两字节转义序列最后在头和尾都加上原始帧头 0xFF 和帧尾 0xFE。这里帧头帧尾不再转义是为了接收端可以快速找到边界。参数说明MAP 是发送端的转义映射表对应关系与前文的表一致长度字段用to_bytes(2,big)如果你的从机用的小端要改成little。速度上这个 Python 逐字节循环在 115200 波特率下完全跑得动但到了 1M 以上的波特率建议换成translate表或 C 实现。2.4 接收端解析状态机从字节流里还原完整帧发送端是编码接收端要有状态机。常见做法是一个有限状态机状态迁移如下当前状态输入下一状态动作IDLE0xFFHEAD记录帧开始IDLE其他IDLE丢弃继续等头HEAD0x7DESC1等待转义数据HEAD其他LEN将输入作为长度第 1 字节LEN任意DATA填充长度第 2 字节切换进数据收集态DATA0x7DESC2转义等待DATA按长度接收完毕TAIL检查帧尾TAIL0xFEIDLE校验和通过就输出帧否则丢弃TAIL其他IDLE校验失败重置实际实现里LEN 和 DATA 会拆得更细因为长度是两字节数据长度 V 时需要读 V1校验和字节。不过状态表的核心是把“看门狗”放在边界上任何状态下收到一个既不是转义符也不是当前所需字节的数据都回到 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、COM7Linux 上是 /dev/ttyUSB0、/dev/ttyS1。下面的代码先定义两个配置对象然后启动两个线程import serial import threading import queue def make_serial(port, baud): ser serial.Serial( portport, baudratebaud, bytesizeserial.EIGHTBITS, parityserial.PARITY_NONE, stopbitsserial.STOPBITS_ONE, timeout0.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 不是必须但缓冲区大于串口驱动的默认值时可以避免单次读取把驱动缓冲塞满timeout0.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(maxsize2000) readers [] for ch in (0, 1): cfg {port: COM3 if ch 0 else COM7, baud: 115200} parser rx_frame() next(parser) # 启动生成器 t threading.Thread(targetreader, args(cfg, frame_q, parser), daemonTrue) t.start() readers.append(t)代码里maxsize2000给队列一个容量上限当写文件变慢时读线程会阻塞在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(timeout0.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, ...)就能切出时间和数据。队列空时用timeout0.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(fch{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,echo0,link/tmp/ttyS10 pty,raw,echo0,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 转接芯片驱动和系统中断绑定。本文还有配套的精品资源点击获取
双路串口通信帧头帧尾解析：从字节填充到Python实现
简介一份关于双路串口通信带帧头帧尾解析FF接收保存文件SRPHeadTail软件的设计说明文档适用于C语言开发、基于Windows平台的嵌入式串口通信工程师与学习串口协议解析的开发者。文档围绕软件的设计目的、基本功能、开发环境、使用说明、全局及运行流程以及串口通信设计展开重点介绍了接口层Port、协议层Protocol、数据帧结构帧头、数据域、帧尾等核心内容可帮助读者理清双路串口数据收发、解析与保存的完整实现思路。资源包内为1个docx文档大小约448KB内容结构清晰目录包含页面介绍、调试窗口使用、主函数流程、数据发送/接收流程、全局数据结构等模块。目前已有65人学习下载适合需要快速理解该软件设计方案或参考串口通信协议实现细节的开发者。1. 双路串口通信的帧头帧尾解析为什么非做不可双路串口通信在设备联调里很常见一个主机从两台从机同时收数据而串口协议本身是字节流没有消息边界。两台设备的报文经过底层缓冲后被切成不定长的片段接收端不定义帧头帧尾就说不清一帧从哪开始、到哪结束。FF 常被选作帧头但串口空闲态在 TTL 电平下正是 0xFF直接拿它做帧头容易被空闲干扰骗过所以还需要转义和校验。SRPHeadTail 这种工具要解决三件事双通道独立接收、按帧头帧尾解析流、把校验通过的载荷保存成文件。对搭产测软件和数据采集台架的工程师来说这套帧划分和落盘逻辑往往比业务本身更容易翻车。2. 帧头帧尾与 FF 转义双路串口协议设计2.1 帧格式长度、校验和帧头帧尾的定位先定义一帧里除了数据之外哪些是必须的。常见做法是把帧头、长度、数据、校验、帧尾五部分做成固定结构字段占用字节说明帧头1固定 0xFF表示一帧开始长度2大端存储表示 Data 字段字节数最大 65535DataN有效载荷可能包含任何 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 做转义符遇到需要转义的字节就改成两个字节。协议定义原始字节填充后含义0xFF0x7D 0x01帧头0xFE0x7D 0x02帧尾0x7D0x7D 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拼成 lengthdataxor然后遍历这个中间缓冲遇到 0xFF、0xFE、0x7D 时换成两字节转义序列最后在头和尾都加上原始帧头 0xFF 和帧尾 0xFE。这里帧头帧尾不再转义是为了接收端可以快速找到边界。参数说明MAP 是发送端的转义映射表对应关系与前文的表一致长度字段用to_bytes(2,big)如果你的从机用的小端要改成little。速度上这个 Python 逐字节循环在 115200 波特率下完全跑得动但到了 1M 以上的波特率建议换成translate表或 C 实现。2.4 接收端解析状态机从字节流里还原完整帧发送端是编码接收端要有状态机。常见做法是一个有限状态机状态迁移如下当前状态输入下一状态动作IDLE0xFFHEAD记录帧开始IDLE其他IDLE丢弃继续等头HEAD0x7DESC1等待转义数据HEAD其他LEN将输入作为长度第 1 字节LEN任意DATA填充长度第 2 字节切换进数据收集态DATA0x7DESC2转义等待DATA按长度接收完毕TAIL检查帧尾TAIL0xFEIDLE校验和通过就输出帧否则丢弃TAIL其他IDLE校验失败重置实际实现里LEN 和 DATA 会拆得更细因为长度是两字节数据长度 V 时需要读 V1校验和字节。不过状态表的核心是把“看门狗”放在边界上任何状态下收到一个既不是转义符也不是当前所需字节的数据都回到 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、COM7Linux 上是 /dev/ttyUSB0、/dev/ttyS1。下面的代码先定义两个配置对象然后启动两个线程import serial import threading import queue def make_serial(port, baud): ser serial.Serial( portport, baudratebaud, bytesizeserial.EIGHTBITS, parityserial.PARITY_NONE, stopbitsserial.STOPBITS_ONE, timeout0.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 不是必须但缓冲区大于串口驱动的默认值时可以避免单次读取把驱动缓冲塞满timeout0.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(maxsize2000) readers [] for ch in (0, 1): cfg {port: COM3 if ch 0 else COM7, baud: 115200} parser rx_frame() next(parser) # 启动生成器 t threading.Thread(targetreader, args(cfg, frame_q, parser), daemonTrue) t.start() readers.append(t)代码里maxsize2000给队列一个容量上限当写文件变慢时读线程会阻塞在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(timeout0.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, ...)就能切出时间和数据。队列空时用timeout0.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(fch{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,echo0,link/tmp/ttyS10 pty,raw,echo0,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 转接芯片驱动和系统中断绑定。本文还有配套的精品资源点击获取
