Python多串口并发批量升级固件实战

在产线待过的朋友应该都对批量烧录这件事有心理阴影。几十台设备摆在那里一台一台抠串口、点烧录、等结果运气好一下午运气不好就是加班到深夜。后来我把升级流程改成了 python 支持多个串口同时升级情况才彻底改观一台电脑接上多个设备的串口运行一个脚本所有设备并行烧录进度和结果自动汇总。这篇文章就把这套方案的思路、代码和踩坑过程完整写出来给同样被批量升级折磨的朋友一个参考。适合三类人看被产线批量烧录逼疯的硬件工程师、需要远程维护大量设备的运维朋友以及刚接触串口通信、想搞明白“并发究竟怎么写”的Python开发者。1. 先想清楚多串口并发升级的整体设计思路1.1 为什么用 Python 做批量升级聊实现之前先回答一个很多人会问的问题批量升级这种事用现成的烧录工具不好吗为什么非要自己写Python我的回答是现成工具大多只管“单台烧录”批量并发要么贵得要命要么是定制方案改一个协议就要等厂商排期。而Python这边pyserial是串口通信的事实标准安装完就能用跨Windows、Linux都能跑脚本写起来快改起来更快今天设备换了个升级协议改个函数重跑一遍就行。对于产线、实验室这类“协议经常变、设备型号经常换”的场景这种灵活性比什么都重要。还有一个现实问题现有工具基本不做结果汇总。几十台设备烧完哪些成功哪些失败全靠人肉记录非常容易漏。用Python就不一样结果直接落成Excel、CSV或者JSON谁烧成功了、谁失败了、失败在哪个包一目了然。这是很多商业工具给不了的。当然Python也有短板比如CPU密集任务扛不住但对串口升级这种I/O密集场景来说完全够用。瓶颈通常根本不在Python本身而在USB带宽、供电和硬件稳定性。1.2 并发模型选型多线程、多进程还是 asyncio多串口并发第一个绕不开的问题就是用多线程、多进程还是asyncio我把三种方案实测过一遍结论很明确串口数量在几十台以内直接上多线程简单、可靠、好排查。先看多进程。每个进程独立一个Python解释器确实能躲开GIL但代价是要把固件数据拷贝到每个子进程里烧录结果还要通过IPC回传代码复杂不少。我是这么理解这个问题的串口升级的大部分时间都花在“等待设备应答”上CPU基本闲着就算单线程、单进程它也只在等I/OGIL根本没有成为瓶颈。既然瓶颈不在CPU多进程的收益就非常有限纯粹是给自己找麻烦。再看asyncio。这个方案表面上很美——单线程异步I/O并发量可以很高。但坑在于asyncio和pyserial的配合并不直观要么用pyserial-asyncio这种第三方封装要么自己把串口读操作包成异步任务。设备端一旦出现异常异步回调式的错误排查会让一个简单的“超时重传”变得很难跟踪。对于产线这种需要马上定位问题的环境我不太推荐。多线程就舒服多了。串口操作天然是阻塞式I/O线程在read等待时会主动让出CPU互相之间干扰很小。用concurrent.futures.ThreadPoolExecutor每个串口提交一个升级任务代码就是普通同步写法问题也好复现。我的实测感受是带动几十个串口轻轻松松线程调度开销根本感觉不到。1.3 升级流程的主线设计不管用什么并发模型升级主流程其实是固定的我用一句话概括扫描设备、握手识别、读取固件、并发烧录、汇总结果。扫描设备是第一步。脚本启动后先遍历当前系统所有串口拿到端口列表。这里要注意插上设备的端口不一定都能烧录可能是调试口、AT指令口之类的所以必须做握手识别——往每个串口发一条查询命令只有返回了正确设备型号和固件版本的端口才加入待升级列表。这一步能过滤掉大量无关串口避免误烧。握手完成之后读取固件文件到内存按设备数量分发。真正并发执行的是“单设备升级”这个函数每个函数负责一个串口的完整烧录流程。全部跑完后主线程收集每个任务的结果生成报表。整个设计把“单台升级”和“批量调度”解耦开单台逻辑出问题只影响单台和“多个用户同时下单但每个订单独立处理”是同一个思路边界清晰改起来也方便。2. 串口通信基础与工具选型2.1 pyserial 的核心用法复习pyserial的核心其实就一个类serial.Serial。初始化参数要重点关注几个因为80%的通信问题都出在这里。端口号是第一步Windows下一般是COM3、COM4这种Linux下是/dev/ttyUSB0、/dev/ttyACM0。波特率是双方约定好的升级场景常用115200或921600必须和设备端Bootloader保持一致。数据位、校验位、停止位一般设备默认都是8N1也就是8数据位、无校验、1停止位绝大多数场景不用改。超时参数一定要设置这是多串口并发时最关键的一项没有超时线程可能会永远卡在读操作上整个并发任务就挂死了。我习惯把读超时和写超时都设成2秒。用法上推荐用with语句打开串口退出时自动关闭避免程序异常后串口被占死。读写前最好执行reset_input_buffer()和reset_output_buffer()把启动瞬间的脏数据清空否则后面解析协议时会被垃圾字节干扰。这是新手最容易漏、也最容易踩坑的一步。import serial def open_serial(port: str) - serial.Serial: ser serial.Serial( portport, baudrate115200, bytesizeserial.EIGHTBITS, parityserial.PARITY_NONE, stopbitsserial.STOPBITS_ONE, timeout2, write_timeout2 ) ser.reset_input_buffer() ser.reset_output_buffer() return ser2.2 设备发现与身份识别设备发现用serial.tools.list_ports.comports()遍历系统串口这个函数会返回端口信息列表每个元素包含设备路径、描述、硬件ID等。我通常会按照设备描述或VID/PID过滤只保留目标设备对应的串口。比如某批次模组用的USB转串口芯片是常见的CH340、CP2102这类那它的描述和VID/PID就相对固定过滤条件写死即可。不过不同厂家的芯片驱动版本不同描述字段可能带厂商名所以我更推荐在拿到端口列表之后再做一次应用层握手确认。握手命令不能随便设计。要有一个包含固定帧头、命令字、设备地址的查询帧设备收到后回复包含设备型号和固件版本的数据包。脚本只有收到合法应答才能把这个端口判定为“待升级设备”。这样设计的好处是即使同一台电脑上接了调试串口、传感器串口也能准确区分出哪些是可以烧录的目标不会误操作。实际使用中我还会加一个参数允许手动指定端口列表。产线上经常出现USB枚举顺序变化导致端口号漂移的情况自动扫描能解决大部分问题但偶尔还得人工干预一下。自动扫描加手动指定双保险比单纯依赖扫描更稳妥。2.3 固件升级协议的设计要点升级协议是整套方案里最依赖具体设备的部分不同设备的Bootloader协议差异很大但几个核心要素是共通的。帧格式要统一。我常用的帧结构是帧头2字节 数据长度1字节 命令字1字节 数据区可变 CRC校验2字节。帧头固定用0xAA 0x55双方约定好设备端收到后先校验帧头再校验CRC通过才执行命令。长度字段用来界定一帧数据的边界防止粘包时解析错位。命令字按流程划分至少要覆盖这几类握手查询、擦除Flash准备、写入数据块、结束校验、复位重启。每个命令对应一个操作阶段设备端收到后执行并回复ACK。这里有个容易忽略的细节ACK一定不能只在命令正确时回复执行失败时也要回复NAK并带上错误码。否则主机侧只能靠超时判断失败效率极低还会把“设备忙”和“通信断了”混为一谈。应答机制和重传策略也是协议的一部分。我惯用的方式是主机发出命令后必须收到ACK才算完成超时未收到就重发最多重试3次。重试做好退避第一次等2秒第二次等3秒第三次等5秒避免设备刚复位还没准备好就被连续轰炸。3. 多串口并发升级核心实现3.1 项目代码结构一览我习惯把代码拆成四个模块串口层、协议层、升级逻辑、调度入口。串口层只管打开、关闭、读写协议层负责组帧、解析、CRC校验升级逻辑把协议层串起来实现完整的单设备烧录流程调度入口负责扫描端口、创建线程池、汇总结果。分离是为了后期维护方便设备协议一旦变化只需要改协议层其他模块完全不用动。实际文件结构大概是这样的upgrade_tool/ ├── serial_layer.py # 串口封装 ├── protocol.py # 帧结构、CRC、命令封装 ├── device_upgrader.py # 单设备升级逻辑 ├── batch_upgrader.py # 线程池调度、结果汇总 └── main.py # 命令行入口不要把所有代码堆在一个文件里除非你的协议永远不变、设备永远不换。我在早期版本就是单文件快速实现改到第三版就后悔了每次协议调整都要在一千多行代码里找逻辑非常痛苦。3.2 单设备升级函数详解单设备升级是整个系统的心脏。这个函数做这几件事打开串口、握手确认、按块发送固件、等待每块ACK、发送结束校验、复位设备。发数据块时有一个关键的偏移量处理。固件文件可能是几MB的二进制设备Flash地址不一定从0开始。所以我发送每一块时除了传给设备“当前数据的长度”还要带上“数据在固件文件里的偏移量”。设备端根据偏移量决定写入Flash的地址。这个设计在固件包含Bootloader分区、App分区时尤其重要不同分区对应不同起始地址。以下是我实际在用的简化版单设备升级核心逻辑FRAME_HEADER b\xAA\x55 def crc16_modbus(data: bytes) - int: crc 0xFFFF for byte in data: crc ^ byte for _ in range(8): crc (crc 1) ^ 0xA001 if crc 1 else crc 1 return crc def build_frame(cmd: int, payload: bytes) - bytes: body bytes([cmd]) payload frame FRAME_HEADER bytes([len(body)]) body crc crc16_modbus(frame[2:]) return frame crc.to_bytes(2, little) def upgrade_one(port: str, firmware: bytes, chunk_size: int 4096, max_retry: int 3): result {port: port, status: fail, message: } try: with serial.Serial(port, baudrate115200, timeout2, write_timeout2) as ser: ser.reset_input_buffer() ser.reset_output_buffer() # 握手确认设备就绪 ser.write(build_frame(0x01, b)) resp ser.read(64) if not resp or resp[4] ! 0x06: # 0x06 表示 ACK raise TimeoutError(握手无应答) total len(firmware) for offset in range(0, total, chunk_size): chunk firmware[offset:offset chunk_size] payload offset.to_bytes(4, little) chunk for retry in range(max_retry): ser.write(build_frame(0x02, payload)) ack ser.read(8) if ack and ack[4] 0x06: break else: raise TimeoutError(f数据块写入失败 offset{offset}) printed f {port}: {offset / total * 100:.1f}% print(printed, end\r, flushTrue) # 结束校验 firmware_crc crc16_modbus(firmware).to_bytes(2, little) ser.write(build_frame(0x03, firmware_crc)) ack ser.read(8) if not ack or ack[4] ! 0x06: raise TimeoutError(整体校验失败) # 复位重启 ser.write(build_frame(0x04, b)) result[status] success result[message] 升级成功 except Exception as e: result[message] str(e) return result这里有个小细节打印进度时光标用\r回到行首几十台设备同时刷屏会非常乱。所以我后来不给每台设备单独打印进度而是只把汇总状态打出来具体每台进度记录到日志文件需要看细节时再查。3.3 线程池并发与结果汇总并发执行用ThreadPoolExecutor它是Python标准库自带的生产者-消费者封装。提交任务时把每个串口对应一个任务收集返回结果后汇总成字典键是端口名值是状态和消息。具体代码非常短from concurrent.futures import ThreadPoolExecutor, as_completed def upgrade_multi(ports, firmware: bytes, max_workers: int 8): results {} with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_port { executor.submit(upgrade_one, port, firmware): port for port in ports } for future in as_completed(future_to_port): port future_to_port[future] try: results[port] future.result() except Exception as e: results[port] {port: port, status: fail, message: str(e)} return results线程数怎么定我个人的经验是不要一上来就把所有串口全开并发。max_workers可以设为端口数量的一半甚至三分之一尤其在USB转串口HUB上多个设备共享一条USB链路同时传输的线程太多会互相抢带宽导致单个设备超时概率明显上升。如果每个串口是独立的PCIe转串口卡或者原生串口那可以多开一些。这个需要根据实际硬件调整我一般先用端口数的一半试一轮再逐步加用成功率来找最优值。结果汇总之后建议立刻落盘。CSV是最通用的格式把端口号、状态、消息、耗时都写进去。产线复盘时直接看CSV就知道是哪一台、哪一步出了问题比盯着屏幕抄结果强一百倍。3.4 并发日志与状态管理日志是多串口并发最容易翻车的地方。直接用print()打印没问题但如果同时处理几十台设备日志会交错成一团乱麻。我的做法是两层日志一是控制台只显示“总进度”和“失败汇总”二是文件日志按设备单独记录。文件日志用Python标准库的logging就能实现。我加了两个handler一个FileHandler写入统一的upgrade.log每一条日志都带上端口名和时间戳另一个StreamHandler只输出汇总信息。真到排查的时候grep 端口名 upgrade.log就能把单台设备的完整升级过程拉出来。这个习惯救过我很多次尤其在设备偶发失败、又无法稳定复现的时候。还有一个细节每个线程里如果抛了异常不能让线程静默吞掉。我在upgrade_one里把异常捕获并转成result返回就是为了让主线程能准确知道哪台失败、为什么失败。这比让线程打印完异常就结束时好得多至少结果汇总表里能留下记录。4. 校验、容错与异常处理4.1 分块传输的块大小怎么定固件必须分块发送原因主要有两个一是设备端的RAM有限不可能一次性接收几百KB的数据然后写Flash二是串口传输容易受干扰一旦出错重传整个固件文件代价太大分块之后只需要重传出错的那一小块。块大小怎么选要看设备端的接收缓冲区和Flash写入单位。通常设备Bootloader会声明每包最大长度比如512字节、1024字节、4096字节。如果设备文档没写我一般先用4096字节做实验看设备应答是否稳定。包太大设备缓冲区溢出直接表现为频繁丢包包太小传输次数变多握手和ACK的开销占比变大总耗时变长。以256KB固件为例4096字节分块就是64包1024字节分块就是256包后者光是ACK等待时间就多了好几倍。实测下来115200波特率下4096字节是个比较舒服的折中。另外要注意最后一块往往不足一个整块大小代码里要用切片截取实际长度不要假设每块都等长。我就见过同事用固定缓冲区直接读最后一块读到了上一包的残留数据导致整个升级校验失败。4.2 CRC 校验与应答机制CRC是整个协议里最后一道大门。每一帧的CRC负责确认“这一帧没有传错”帧级校验通过后才执行对应操作。升级结束后还要对整个固件文件计算一次CRC发送给设备端设备端将写入Flash的数据整体计算CRC并对比两者一致才认为烧录成功。帧级校验防止单包出错整体校验防止“某个包虽然没错但中间漏掉了一个包”这种隐蔽问题。两个都不能省。CRC算法要和设备端保持一致。常用的有CRC16/MODBUS、CRC32等具体用哪个以设备协议文档为准。我上面代码里用的是CRC16/MODBUS实现起来简单在8位单片机上也能撑得住。CRC16的冲突概率对固件校验来说已经足够低不需要为了“更安全”强上CRC32增加设备端的计算负担。应答机制要注意区分ACK和NAK。ACK表示设备正确处理了命令NAK表示命令格式错误、地址越界、Flash写入失败等。我在协议里把NAK后面跟了一个字节的错误码这样主机能直接把失败原因反馈给用户而不是只看到一个笼统的“校验失败”。4.3 超时、重传与失败隔离策略超时和重传是多串口并发里决定成败的关键。没有超时机制一个信号线松了的设备就可能让整批升级卡住几分钟。我的策略是两层单帧等待超时和总重试次数限制。单帧等待超时设2秒设备处理一个4KB数据块的时间通常远少于1秒2秒已经足够宽裕。超时后重发当前帧最多重试3次3次都不应答就判定该设备失败不再往下发。这样即使设备彻底掉线这个线程也会在十几秒内结束并返回失败原因不会拖住整个线程池。失败的设备要隔离处理不能让一个失败设备影响其他设备。每个设备的数据包发送都在各自的串口线程里完成天然隔离线程池层面更要保证——某个任务抛异常as_completed循环里接住并记录然后继续处理下一个任务。我在实际代码里专门加了except Exception分支确保任何异常都不会中断整个批量任务。还有一个听上去有点矛盾但很实用的建议不要过度重传。有些设备在写Flash出错后会直接进入错误状态重传再多次也没用。我见过有人把重试次数改成99次结果99次全部失败白白等了十几分钟。重传的意义在于应对瞬时干扰如果3次都不行大概率是硬件问题不值得再耗时间。5. 完整实操24 台设备同时烧录5.1 环境准备与硬件接法我拿自己实测过的一套场景来举例某物联网模组X200虚构型号固件文件约256KBBootloader升级协议支持分块写入块大小4KB波特率115200。环境是一台普通Windows工控机接了一个16口的USB转串口HUB和一个8口HUB一共接了24台模组。这个硬件接法很常见但有一个大坑USB HUB必须自供电如果靠USB总线的5V供电24台设备同时写入Flash时电流峰值会直接把电压拉垮设备表现为随机掉线、枚举失败、升级失败率飙升。电源这块我专门吃过亏。第一版测试用的是一个不带电源的USB HUB8台设备同时升级成功率只有六成。换成自供电HUB之后成功率直接回到了99%以上。所以如果要做多串口并发先检查供电是否扎实再谈代码优化。软件环境就是Python 3.10 pyserialpip install pyserial一条命令搞定没有其他依赖。Linux下还要注意用户权限普通用户访问串口需要加入dialout用户组命令行执行sudo usermod -aG dialout $USER重新登录后生效。不做这一步打开串口会直接报权限错误。5.2 运行参数与命令行设计命令行入口我用argparse写了几个参数固件路径、端口列表、线程数、是否自动扫描。自动扫描模式下会先列出所有串口并发送握手命令筛选目标设备筛选结果打印出来让用户确认再开始升级。python main.py --firmware app_v2.3.1.bin --auto-scan --workers 8实际跑起来脚本会先打印“发现24台设备正在准备升级”然后每台设备升级结束打印一行汇总。整个过程中日志文件upgrade.log不断追加模块级日志方便事后复盘。这里有一个经验首次使用先加一个--dry-run参数只做扫描和握手不真正写Flash用来验证设备连接和协议握手。产线首次环境搭建时这个参数能帮你快速排除“设备没连好”还是“协议没对上”的疑问。5.3 实测结果与瓶颈分析实测数据大概是这个量级单台升级耗时约30秒其中磨蹭时间主要在握手、数据帧ACK等待和设备擦写Flash上。24台设备按8线程并发总耗时约70秒四批设备每批6台轮换下来的整体吞吐相当可观。整个过程中CPU占用率不到10%内存占用主要来自固件文件缓存和日志缓冲完全可以忽略。瓶颈在USB链路。几台设备共享一个HUB时同时传输的线程多了单台设备的实际传输带宽会被稀释等待ACK的间隔变长个别设备会出现偶发超时。解决办法我刚才说过控制max_workers或者按HUB物理分组每个组内并发不超过4到6台。另一个有效手段是把波特率从115200提高到460800前提是设备端Bootloader支持而且线材质量足够好。提高波特率后同等工作量下传输时间缩短到原来的四分之一但线材差的话反而会频繁出错。从结果汇总的角度看24台设备里稳定环境下一次成功率通常在95%以上。失败的多半是固定某一台几乎可以肯定是硬件问题比如排线松动、母座氧化。脚本给出的失败信息如果能精确到“offset某个值超时”你基本可以按图索骥去检查那一台的接线不用再做全量排查。6. 常见问题排查与避坑实录6.1 典型问题速查表我把实际运行中遇到的高频问题整理成了表每条都是真金白银换来的现象可能原因解决办法扫描不到任何串口USB转串口驱动未装HUB未插电源导致枚举失败安装对应驱动换自供电HUB检查设备管理器是否出现未知设备打开串口报权限/占用错误其他软件占用了端口Linux用户不在dialout组关闭串口监视器sudo usermod -aG dialout $USER后重登握手命令无应答波特率不匹配接线接错成RS232而不是TTL检查设备文档确认波特率和电平标准先接单台设备验证升级中频繁超时线材过长/质量差USB HUB供电不足换短而粗的USB线改用自供电HUB适当降低并发数并发后大批量失败同时写入挤爆USB带宽供电瞬时跌落控制max_workers按HUB分组提高波特率设备升级后无法启动固件文件损坏升级中断导致Flash分区坏确认固件MD5重新升级设备端考虑双区备份策略Windows下COM10以上打不开旧版pyserial或路径写法问题端口路径写成\\.\COM10这种显式格式6.2 几条压箱底的实操经验第一条永远先把环境跑稳再谈并发。我第一次做24台批量升级上来就全线程并发结果一大半设备握手失败。后来老老实实单台测通再按2台、8台、16台递增才找到供电和带宽的临界点。这类问题用再好的代码也掩盖不了只能靠环境排查。第二条结果一定要落盘而且日志里要带足够上下文。早期我只在控制台看结果设备偶发失败时根本没法追踪。后来每台设备的每一步操作都写入upgrade.log失败排查快多了。日志记得带上线程名和端口名这是多线程日志最基本的要求不然你根本不知道这几行日志是哪台设备打的。第三条升级方案里要预留“后门”。比如在设备端做双区备份升级失败还能从备份区启动或者在升级流程最前面加一个“是否已是最新版本”的判断避免重复烧录。产线上经常会出现同型号设备固件版本不一致的情况让脚本先读取设备版本号和要烧的固件版本比对相同就跳过能省下大量时间。最后一条批处理脚本要能中断后安全退出。按下CtrlC时ThreadPoolExecutor的退出逻辑要处理好不能让已经打开的串口还占着端口。我在脚本里加了信号处理和上下文管理器确保退出时所有串口被关闭否则下一次运行会报“端口被占用”又得多一个排查步骤。多串口并发升级这套方案到这一步基本就完整了。从整体设计、并发选型、协议设计到实测排坑每一步都有它的取舍逻辑。说实话刚开始我也犹豫过要不要用现成商业工具但用过一轮之后发现自己写的这套工具才是真正贴合产线需求的协议改了改协议层设备换了换握手逻辑结果报表随便定制。如果你也正被批量升级折磨不妨照这个思路先搭一个最小版本试起来遇到具体的坑再回来查这篇文章的速查表大概率能少走不少弯路。