上个月给一块 STM32F103 的小板子调 MPU6050,需要连着看三路采样值。串口助手能把 printf 出来的东西显示出来,但它有三个我忍不了的地方:数据是纯文本,想算个滑动平均得先复制粘贴进 Excel;加了时间戳和通道号之后每行变得很长,滚动起来眼睛跟不上;偶发丢帧的时候只能靠一行行肉眼比对,根本判断不了是数据错了还是显示错了。

于是一个晚上写了个不到 300 行的上位机。这篇文章把它拆开讲,重点只有三个地方:读串口为什么必须放子线程、帧怎么分、界面怎么刷新才不卡。

目录结构

serial-monitor/
├── main.py            # 入口:装配 worker 和窗口
├── serial_worker.py   # 读串口线程 + 分帧循环
├── protocol.py        # crc8 与结构体打包
└── requirements.txt

requirements.txt 只有一行 pyserial。tkinter 是标准库,不用装。

为什么不用 Qt 或网页

这类小工具最怕的是发给同学装不上。PyQt 功能确实强,但要把 Qt 运行库一起打包,分发体积明显更大,而且 PyQt 的 GPL 授权在实验室以外的地方需要留意;换成网页方案(本地 Flask 加浏览器)跨平台最省事,可又得多维护一个进程和端口,插拔串口时页面还得自己处理重连。我的使用场景就是插上开发板看几路数据,界面只有下拉框、按钮和文本框,tkinter 够用,而且它在 Python 标准库里,别人拿到源码双击就能跑。选型的标准不是”哪个更专业”,而是”谁维护成本最低”。

方案 依赖 分发成本 适合场景
tkinter Python 标准库 最小,PyInstaller 单文件即可 内部调试工具、界面简单
PyQt/PySide 第三方 Qt 运行库 体积更大,需注意许可证 复杂交互、长期维护的上位机
本地 Web 页面 浏览器 + Web 框架 多一个常驻服务 多人访问、远程查看

第一步:把读串口放进子线程

这是整个项目最容易写错的地方。tkinter 的 mainloop() 是一个单线程事件循环,所有控件操作都必须在这个线程里完成。如果直接在按钮回调里写 while True: ser.readline(),mainloop 就再也拿不到 CPU,窗口几秒内变成”未响应”。

反过来,在子线程里直接 text.insert() 同样不行。tkinter 内部没有对控件加锁,跨线程改控件在 Windows 上往往不报错,但会表现为偶发的界面不刷新、滚动条卡死,甚至进程直接退出。这类问题比崩溃更难查,因为它不复现。

所以我给自己定的规矩是:子线程只碰 queue,主线程只碰控件。

# serial_worker.py
import queue
import threading
import serial

from protocol import unpack_frame


class SerialWorker:
    """只负责 IO:解析好的帧丢进 rx,错误丢进 err"""

    def __init__(self, rx: queue.Queue, err: queue.Queue):
        self.rx = rx
        self.err = err
        self._ser = None
        self._stop = threading.Event()
        self._thread = None

    def open(self, port: str, baud: int = 115200):
        self._ser = serial.Serial(
            port=port, baudrate=baud,
            bytesize=8, parity='N', stopbits=1,
            timeout=0.05,          # 关键:不能是 None
        )
        self._stop.clear()
        self._thread = threading.Thread(target=self._loop, daemon=True)
        self._thread.start()

    def _loop(self):
        buf = bytearray()
        while not self._stop.is_set():
            try:
                chunk = self._ser.read(256)
            except serial.SerialException as e:
                self.err.put(f'串口断开:{e}')
                return
            if not chunk:
                continue
            buf.extend(chunk)
            while True:
                frame = unpack_frame(buf)   # 不够一帧返回 None
                if frame is None:
                    break
                self.rx.put(frame)

    def close(self):
        self._stop.set()
        if self._thread:
            self._thread.join(timeout=1.0)
        if self._ser and self._ser.is_open:
            self._ser.close()

timeout=0.05 是这段代码里唯一必须解释的参数。pyserial 的 read() 在 timeout 为 None 时会一直阻塞到读够字节数,这时 _stop 设了也没用,关窗口的瞬间线程还卡在 read 上,进程退不掉,任务管理器里留一个僵尸。给个短超时,read 每 50ms 就返回一次空字节串,循环重新检查 _stop,退出才是干净的。

第二步:帧怎么分

一开始我用 printf("%d,%d\n", ch, val) 加换行分隔,图省事。问题出在两处:某天我想在帧里塞一个二进制状态字,分帧就全乱了;另外一旦因为干扰丢了一个字节,后面所有行都错位,而且错得很像”对的”,肉眼几乎发现不了。

改成带包头、长度和校验的帧:

0xAA 0x55 | len(1) | payload(len) | crc8(1)

payload 固定 3 字节:channel(int8) + value(int16 小端,单位 0.01)。

# protocol.py
import struct

HEAD = b'\xaa\x55'


def crc8(data: bytes) -> int:
    """CRC-8/ATM:poly=0x07, init=0x00"""
    crc = 0x00
    for b in data:
        crc ^= b
        for _ in range(8):
            crc = ((crc << 1) ^ 0x07) & 0xFF if crc & 0x80 else (crc << 1) & 0xFF
    return crc


def pack_frame(ch: int, value: float) -> bytes:
    payload = struct.pack('<Bh', ch, int(round(value * 100)))
    return HEAD + bytes([len(payload)]) + payload + bytes([crc8(payload)])


def unpack_frame(buf: bytearray):
    """从缓冲区头部切一帧出来;不够一帧返回 None,脏字节直接丢"""
    while True:
        start = buf.find(HEAD)
        if start < 0:
            buf.clear()          # 全是垃圾,清空重来
            return None
        if start:
            del buf[:start]      # 丢掉包头之前的噪声
        if len(buf) < 4:
            return None          # 连头部都还没收全
        length = buf[2]
        total = 3 + length + 1
        if len(buf) < total:
            return None          # 半包,等下一次 read 补齐
        payload = bytes(buf[3:3 + length])
        crc = buf[3 + length]
        del buf[:total]
        if crc8(payload) != crc:
            continue             # 校验错,继续往后找包头
        return struct.unpack('<Bh', payload) if length == 3 else None

这段里真正重要的是两个 return None:一个在”长度不够”时返回,一个在”包头还没来”时返回。而且 buf 必须留在函数外面(_loop 里的那个 bytearray),因为一次 read(256) 拿到的可能是 1.7 帧,也可能只有半帧。我最早的版本把 buf 写在函数内部,半包永远拼不起来——表现就是连上之后三分钟内必然丢一段数据。

STM32 那边对应着发就行,HAL 库版本:

uint8_t crc8(const uint8_t *data, uint8_t len)
{
    uint8_t crc = 0x00;
    while (len--) {
        crc ^= *data++;
        for (uint8_t i = 0; i < 8; i++)
            crc = (crc & 0x80) ? (uint8_t)((crc << 1) ^ 0x07) : (uint8_t)(crc << 1);
    }
    return crc;
}

void report_sample(int8_t ch, int16_t raw)
{
    uint8_t payload[3] = { (uint8_t)ch, (uint8_t)(raw & 0xFF), (uint8_t)((raw >> 8) & 0xFF) };
    uint8_t frame[7] = { 0xAA, 0x55, sizeof(payload) };
    memcpy(&frame[3], payload, sizeof(payload));
    frame[6] = crc8(payload, sizeof(payload));
    HAL_UART_Transmit(&huart1, frame, sizeof(frame), 10);
}

这里唯一容易错的是 frame[7] 的下标:包头 2 + 长度 1 + payload 3 + 校验 1。我第一版写成 uint8_t frame[5],编译不报错,运行时把栈上的相邻变量踩掉,查了两个小时。

第三步:界面怎么刷新才不卡

布局本身没什么好讲的:一个 Combobox 选串口、一个按钮开关、一个 Text 显示。有讲究的是刷新策略。

如果每收到一帧就 insert 一次,115200 下数据一密,mainloop 的时间几乎全花在 Text 重排上,窗口拖都拖不动。我的做法是每 30ms 从队列里批量取一次,一次最多 200 条:

def _poll(self):
    lines = []
    for _ in range(200):
        try:
            ch, value = self.rx.get_nowait()
        except queue.Empty:
            break
        lines.append(f'CH{ch}  {value:8.2f}')
    if lines:
        self.text.configure(state='normal')
        self.text.insert('end', '\n'.join(lines) + '\n')
        # 只保留最近 2000 行,否则重排会越来越慢
        if int(self.text.index('end-1c').split('.')[0]) > 2000:
            self.text.delete('1.0', '500.0')
        self.text.see('end')
        self.text.configure(state='disabled')
    self.after(30, self._poll)

after(30) 是在 mainloop 里排一个定时回调,语义就是”30 毫秒后让主线程再跑一次这个函数”,它不占线程也不阻塞事件处理。这比 threading.Timer 好:后者又开一个线程,等于把最开始的问题绕回来了。

另外 Text 设成 state='disabled' 后用户不能编辑,但 insert 之前必须临时切回 normal,忘了就是”插不进去还不报错”。

采样数据怎么落盘

界面解决的是”我看见”,分析解决的是”我回看”。只把数据留在 Text 控件里,关掉程序就没了,所以我在同一个工具里加了 CSV 记录:勾选”记录”后,每次从队列取出的帧顺手写一行,文件名按启动时间生成。

import csv
import time


class CsvRecorder:
    """只在主线程调用:一次写一批,定时 flush"""

    def __init__(self, path: str):
        self._f = open(path, 'w', newline='', encoding='utf-8-sig')
        self._w = csv.writer(self._f)
        self._w.writerow(['t_s', 'channel', 'value'])
        self._t0 = time.time()
        self._last_flush = time.time()

    def write(self, rows):
        for ch, value in rows:
            self._w.writerow([f'{time.time() - self._t0:.3f}', ch, f'{value:.2f}'])
        # 每秒落一次盘,避免每行 flush 卡住界面
        if time.time() - self._last_flush >= 1.0:
            self._f.flush()
            self._last_flush = time.time()

    def close(self):
        self._f.flush()
        self._f.close()

三个细节值得说清楚:

  1. encoding='utf-8-sig' 不是为了好看。Windows 版 Excel 默认不认纯 UTF-8 的 CSV,表头里的中文会变成乱码;加 BOM 之后双击就能正确打开,这个和以前折腾 GBK 是同一类问题。
  2. 不在串口线程里写文件。磁盘偶尔卡一下,如果发生在读线程里,缓冲区就会堆起来,看起来像串口丢数据。写文件统一放到主线程的批量回调里,读线程只负责收字节。
  3. 用 flush() 控制频率而不是每行都刷。每行 flush 等于每 30 毫秒强制一次磁盘写入,界面在机械硬盘上会明显发顿;每秒刷一次,最坏丢一秒数据,对调 PID 这类场景完全可以接受。

关窗口时也要把记录器关掉,否则 Python 退出时可能还没来得及刷缓冲区:

def on_close(self):
    self.worker.close()
    if self.recorder:
        self.recorder.close()
    self.root.destroy()

self.root.protocol('WM_DELETE_WINDOW', self.on_close)

WM_DELETE_WINDOW 是 tkinter 的关闭钩子,不注册它,点右上角的叉只会销毁窗口,后台线程和文件对象都还活着。

打包成 exe

pyinstaller -F -w -n SerialMonitor --icon app.ico \
  --hidden-import serial.tools.list_ports main.py

-F 打单文件,-w 去掉黑色控制台窗口。加 --hidden-import 是因为我用 serial.tools.list_ports.comports() 枚举串口,PyInstaller 的静态分析抓不到这种间接 import,不打这个参数,生成的 exe 一运行就 ModuleNotFoundError。

单文件 exe 每次启动会先把自己解压到临时目录,所以启动比源码慢一点。这是个可以接受的交换:发给实验室同学一个文件就能用,不用教他们装 Python 和驱动。

三种分帧方式的取舍

分帧方式 优点 缺点 适合
换行分隔文本 一行 printf 就能用,最省事 二进制里出现 0x0A 就误判;解析要 atof 只打日志、不带结构化数据
定长帧 解析最快,不用搜包头 改一个字段两端都得改;丢一字节就永久错位 固定结构的高频采样
包头 + 长度 + CRC 能自恢复,能识别坏帧 多二十行代码,要处理半包 长时间运行、有干扰的现场

我最后选第三种,理由不是”更专业”,而是板子和电脑之间那根杜邦线经常被碰。第二种方案里,碰一下之后必须重启程序才能恢复,这个代价比多写二十行代码高得多。

调试时最常遇到的几种情况

现象 大概率原因 处理
打开串口报 PermissionError 串口助手或调试器还占着这个口 关掉占用它的程序
收到一堆乱码 两端波特率不一致,或 STM32 时钟树配错 都设 115200,核对 HSE 与 PLL
前几帧正常,后面全错位 半包没处理,把半帧当整帧切了 按长度字段判断是否收全
拔掉 USB 后程序卡死 子线程在 read 上抛异常没捕获 捕获 SerialException,向队列发错误

最后一条我真踩过:线被碰掉时 _ser.read() 抛 SerialException,如果 _loop 里没有 try,线程会静默死掉,界面还显示”已连接”,然后一直等一个再也不会来的数据。

小结

整个工具不到 300 行,但把几件事说清楚了:

  • 串口 IO 一定放子线程,线程和数据的边界画清:子线程只往 queue 里放,主线程只从 queue 里取;
  • 分帧要有长度和校验,缓冲区放在循环外,半包要能等下一次 read;
  • 界面刷新要合并,after() 批处理比”来一帧刷一次”稳。

再往下能加的东西不少:把数据写成 CSV、用 matplotlib 画实时曲线、命令下发也走同一套帧格式。但先把这三件事做对,这个工具已经能天天用了。