Python Qt上位机通过pyserial实现串口数据收发教程

开发工业控制或嵌入式调试工具时，上位机需要通过串口与微控制器（如STM32、ESP32等）进行数据交互。使用 Python 开发上位机兼具快速原型设计与跨平台特性，其中 Qt 库（PySide6 或 PyQt5）负责构建人机交互界面，pyserial 库负责底层的串行端口通信。本文将从运行机制、通信架构、后台多线程读写以及完整的代码工程出发，带你一步步搭建一个稳定不卡顿的串口上位机程序。

一、串口通信上位机的核心架构

在 Qt 应用程序中，主线程负责绘制界面并响应用户的鼠标、键盘等交互事件。串口数据接收通常是一个阻塞或轮询操作：若直接在 Qt 主线程中调用 serial.read()，当端口暂无数据到达时，主线程将被挂起等待，导致操作界面完全冻结并出现系统无响应假死。因此，标准的架构方案是将 pyserial 的数据监听放入独立的后台工作线程（QThread），并通过 Qt 的信号与槽（Signal & Slot）机制安全地将接收到的字节流跨线程投递给主界面。

图1：Python Qt上位机与 pyserial 线程交互系统架构图

二、环境准备与依赖安装

在开始编写代码前，需要安装 Python 的图形界面包和串口驱动包。推荐使用当前主流的 PySide6（Qt 官方 Python 绑定），同时本代码结构也天然兼容 PyQt5。使用终端运行以下命令：

pip install pyserial PySide6

三、开发步骤与实现流程

编写一个标准上位机程序通常分为四个阶段：枚举本机有效串口、配置并打开串口通信、多线程持续监听与信号转发、以及安全关闭串口释放句柄资源。

图2：上位机通信生命周期与关键操作步骤

1. 枚举系统可用串口

pyserial 提供了 serial.tools.list_ports.comports() 方法，可以自动发现当前系统连接的所有物理及虚拟串口设备。在界面加载或用户点击“刷新”按钮时，上位机应清空下拉框并重新填入检测到的串口名称（例如 Windows 下的 COM3，Linux/macOS 下的 /dev/ttyUSB0）。

2. 后台接收线程与信号传递

为了保证数据收发的实时性并且不锁死界面，我们继承 QThread 创建一个专用的串口读取线程 SerialReaderThread。线程内部保留一个布尔标志位 running。在 run() 方法中，通过死循环执行 read_all() 或指定长度的 read()，并在成功读取到字节数据时通过自定义的 Signal(bytes) 信号向外部投递数据。

图3：主线程与串口工作线程之间的跨线程事件时序图

四、完整可运行代码实现

以下为一个单文件、完整具备图形界面、串口扫描、连接断开、数据收发及十六进制显示的上位机工程代码。可直接保存为 serial_tool.py 并运行。

import sys
import serial
import serial.tools.list_ports
from PySide6.QtCore import QThread, Signal, Qt
from PySide6.QtWidgets import (
    QApplication, QMainWindow, QWidget, QVBoxLayout,
    QHBoxLayout, QLabel, QComboBox, QPushButton,
    QTextEdit, QLineEdit, QCheckBox, QMessageBox
)

class SerialReaderThread(QThread):
    # 定义自定义信号，用于将读取到的字节流传输到主线程
    data_received = Signal(bytes)
    error_occurred = Signal(str)

    def __init__(self, ser: serial.Serial):
        super().__init__()
        self.ser = ser
        self.is_running = True

    def run(self):
        while self.is_running and self.ser and self.ser.is_open:
            try:
                # 检查输入缓冲区中是否有数据待读
                waiting_bytes = self.ser.in_waiting
                if waiting_bytes > 0:
                    data = self.ser.read(waiting_bytes)
                    if data:
                        self.data_received.emit(data)
                else:
                    # 避免空转过多占用 CPU
                    self.msleep(10)
            except Exception as e:
                self.error_occurred.emit(str(e))
                break

    def stop(self):
        self.is_running = False
        self.wait()

class SerialMasterWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Python Qt 串口上位机")
        self.resize(720, 520)

        self.ser = None
        self.read_thread = None

        self._init_ui()
        self.refresh_ports()

    def _init_ui(self):
        main_widget = QWidget()
        self.setCentralWidget(main_widget)
        main_layout = QVBoxLayout(main_widget)

        # 1. 顶部控制栏：端口、波特率、刷新、连接/断开
        ctrl_layout = QHBoxLayout()

        ctrl_layout.addWidget(QLabel("串口号:"))
        self.port_combo = QComboBox()
        ctrl_layout.addWidget(self.port_combo)

        refresh_btn = QPushButton("刷新")
        refresh_btn.clicked.connect(self.refresh_ports)
        ctrl_layout.addWidget(refresh_btn)

        ctrl_layout.addWidget(QLabel("波特率:"))
        self.baud_combo = QComboBox()
        self.baud_combo.addItems(["9600", "19200", "38400", "57600", "115200", "921600"])
        self.baud_combo.setCurrentText("115200")
        ctrl_layout.addWidget(self.baud_combo)

        self.connect_btn = QPushButton("打开串口")
        self.connect_btn.clicked.connect(self.toggle_serial)
        ctrl_layout.addWidget(self.connect_btn)

        ctrl_layout.addStretch()
        main_layout.addLayout(ctrl_layout)

        # 2. 数据接收区域
        main_layout.addWidget(QLabel("接收区:"))
        self.recv_text = QTextEdit()
        self.recv_text.setReadOnly(True)
        main_layout.addWidget(self.recv_text)

        # 接收区控制：清空、HEX显示
        recv_ctrl_layout = QHBoxLayout()
        self.hex_recv_check = QCheckBox("十六进制显示 (HEX)")
        recv_ctrl_layout.addWidget(self.hex_recv_check)

        clear_recv_btn = QPushButton("清空接收")
        clear_recv_btn.clicked.connect(self.recv_text.clear)
        recv_ctrl_layout.addWidget(clear_recv_btn)
        recv_ctrl_layout.addStretch()
        main_layout.addLayout(recv_ctrl_layout)

        # 3. 数据发送区域
        main_layout.addWidget(QLabel("发送区:"))
        send_layout = QHBoxLayout()
        self.send_input = QLineEdit()
        send_layout.addWidget(self.send_input)

        self.send_btn = QPushButton("发送")
        self.send_btn.setEnabled(False)
        self.send_btn.clicked.connect(self.send_data)
        send_layout.addWidget(self.send_btn)

        main_layout.addLayout(send_layout)

    def refresh_ports(self):
        """扫描并刷新可用串口列表"""
        self.port_combo.clear()
        ports = serial.tools.list_ports.comports()
        for p in ports:
            self.port_combo.addItem(p.device)
        if not ports:
            self.port_combo.addItem("无可用串口")

    def toggle_serial(self):
        """打开或关闭串口"""
        if self.ser and self.ser.is_open:
            self.close_serial()
        else:
            self.open_serial()

    def open_serial(self):
        port = self.port_combo.currentText()
        if not port or port == "无可用串口":
            QMessageBox.warning(self, "提示", "请选择有效的串口！")
            return

        baudrate = int(self.baud_combo.currentText())

        try:
            # 初始化串口对象
            self.ser = serial.Serial(
                port=port,
                baudrate=baudrate,
                bytesize=serial.EIGHTBITS,
                parity=serial.PARITY_NONE,
                stopbits=serial.STOPBITS_ONE,
                timeout=0.1
            )

            # 启动监听线程
            self.read_thread = SerialReaderThread(self.ser)
            self.read_thread.data_received.connect(self.handle_received_data)
            self.read_thread.error_occurred.connect(self.handle_serial_error)
            self.read_thread.start()

            # 切换界面状态
            self.connect_btn.setText("关闭串口")
            self.port_combo.setEnabled(False)
            self.baud_combo.setEnabled(False)
            self.send_btn.setEnabled(True)

        except Exception as e:
            QMessageBox.critical(self, "错误", f"打开串口失败: {e}")
            self.ser = None

    def close_serial(self):
        """关闭串口并回收工作线程"""
        if self.read_thread:
            self.read_thread.stop()
            self.read_thread = None

        if self.ser and self.ser.is_open:
            try:
                self.ser.close()
            except Exception:
                pass
            self.ser = None

        self.connect_btn.setText("打开串口")
        self.port_combo.setEnabled(True)
        self.baud_combo.setEnabled(True)
        self.send_btn.setEnabled(False)

    def handle_received_data(self, raw_bytes: bytes):
        """处理线程回传的原始字节数据"""
        if self.hex_recv_check.isChecked():
            display_str = " ".join([f"{b:02X}" for b in raw_bytes]) + " "
        else:
            # 兼容非 UTF-8 字符，防止解码抛出异常崩溃
            display_str = raw_bytes.decode('utf-8', errors='replace')

        self.recv_text.insertPlainText(display_str)
        # 自动滚动到最新文本
        self.recv_text.moveCursor(self.recv_text.textCursor().End)

    def handle_serial_error(self, err_msg: str):
        """处理串口异常断开或拔出"""
        QMessageBox.warning(self, "串口断开", f"通信异常: {err_msg}")
        self.close_serial()

    def send_data(self):
        """向串口发送输入框中的内容"""
        text = self.send_input.text()
        if not text or not self.ser or not self.ser.is_open:
            return

        try:
            payload = text.encode('utf-8')
            self.ser.write(payload)
            self.send_input.clear()
        except Exception as e:
            QMessageBox.critical(self, "发送失败", f"数据发送异常: {e}")

    def closeEvent(self, event):
        """窗口关闭时确保退出子线程释放硬件句柄"""
        self.close_serial()
        event.accept()

if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = SerialMasterWindow()
    window.show()
    sys.exit(app.exec())

五、关键避坑要点与总结

在实际编写与调试上位机时，有几个非常普遍但容易导致程序异常的细节需要注意：

• 串口热插拔与设备断开：若在通信过程中 USB 转串口线被用户意外拔出，read() 或 write() 会抛出 SerialException。线程中必须捕获该异常并通过信号通知主窗口关闭句柄并复位按钮状态，否则上位机将陷入持续死锁或报错。
• 字符串解码策略：工业单片机回传的数据常常包含非标准编码或纯二进制校验码。如果直接使用 data.decode('utf-8')，遇到不合规字节就会直接让程序崩溃。务必添加 errors='replace' 或 errors='ignore'。
• 窗口退出事件清理：必须在 closeEvent 回调中显式调用线程的 stop() 和串口的 close()。若直接关闭 UI 窗口而不终止后台线程，Python 解释器进程可能会残留在后台继续占用串口，导致下一次启动程序提示端口已被占用。

图4：串口上位机核心要素与工程实践总结