手把手教你用PyQt5+QtSerialPort打造一个能实时绘图的数据采集器（附完整源码）-开发者社区

从零构建PyQt5串口数据采集与实时可视化系统

在物联网和嵌入式开发领域，串口通信是最基础也最常用的数据传输方式之一。无论是Arduino、STM32还是各种传感器模块，开发者经常需要通过串口获取实时数据并进行可视化分析。本文将带领读者使用PyQt5和QtSerialPort构建一个完整的串口数据采集系统，实现从硬件通信到图形化显示的完整链路。

1. 开发环境准备与项目架构

构建一个稳定的串口数据采集系统需要合理选择技术栈。PyQt5作为Python下最成熟的GUI框架之一，配合其内置的QtSerialPort模块，能够提供跨平台的串口通信能力。以下是基础环境配置步骤：

pip install pyqt5 pyqtchart

项目采用经典的三层架构设计：

通信层：处理串口连接、数据读取和协议解析
业务层：管理数据缓冲、线程安全和事件分发
展示层：实现动态图表更新和用户交互

核心模块依赖关系如下表所示：

模块	功能	关键类
QtSerialPort	串口通信	QSerialPort, QSerialPortInfo
QtChart	数据可视化	QChart, QChartView, QLineSeries
QtCore	基础功能	QThread, QTimer, pyqtSignal

提示：建议使用Python 3.8+环境，某些PyQt5版本在Python 3.10+上可能存在兼容性问题

2. 串口通信模块实现

QtSerialPort提供了跨平台的串口访问能力，相比Python标准库中的serial模块，它与PyQt5的信号槽机制天然集成，更适合GUI应用开发。

2.1 串口设备发现与配置

首先实现串口设备的自动发现功能：

from PyQt5.QtSerialPort import QSerialPort, QSerialPortInfo class SerialManager: def __init__(self): self.serial = QSerialPort() def scan_ports(self): """返回可用串口列表""" ports = [] for port in QSerialPortInfo.availablePorts(): ports.append({ 'name': port.portName(), 'description': port.description(), 'manufacturer': port.manufacturer() }) return ports

串口参数配置需要特别注意波特率匹配问题：

def configure_port(self, port_name, baud_rate): self.serial.setPortName(port_name) self.serial.setBaudRate(baud_rate) self.serial.setDataBits(QSerialPort.Data8) self.serial.setParity(QSerialPort.NoParity) self.serial.setStopBits(QSerialPort.OneStop) if not self.serial.open(QSerialPort.ReadWrite): raise Exception(f"无法打开串口 {port_name}")

2.2 数据接收与解析

串口数据接收需要处理字节流解析和粘包问题：

def start_receiving(self): self.serial.readyRead.connect(self._handle_data) def _handle_data(self): while self.serial.bytesAvailable(): raw_data = self.serial.readAll().data() try: # 假设数据格式为"value1,value2,value3\n" decoded = raw_data.decode('ascii').strip() if decoded: values = [float(x) for x in decoded.split(',')] self.data_received.emit(values) # 发射信号 except UnicodeDecodeError: print("非ASCII数据:", raw_data.hex())

常见的数据解析问题及解决方案：

数据不完整：实现缓冲区管理，等待完整帧
编码问题：明确硬件端使用的编码格式（通常ASCII或UTF-8）
校验失败：添加CRC校验或协议确认机制

3. 线程安全与数据缓冲

GUI应用中，串口数据的持续接收必须考虑线程安全问题。PyQt5提供了多种线程间通信机制。

3.1 工作线程实现

创建专用的串口工作线程：

from PyQt5.QtCore import QThread, pyqtSignal class SerialThread(QThread): data_ready = pyqtSignal(list) def __init__(self, serial_manager): super().__init__() self.serial = serial_manager def run(self): self.serial.start_receiving() self.exec_() # 进入事件循环

3.2 环形缓冲区设计

为应对数据突增情况，实现固定大小的环形缓冲区：

class CircularBuffer: def __init__(self, size=1000): self.size = size self.buffer = [0] * size self.index = 0 def add(self, value): self.buffer[self.index % self.size] = value self.index += 1 def get_last(self, count): start = max(0, self.index - count) return self.buffer[start % self.size : self.index % self.size]

4. 动态图表实现

QtChart模块提供了丰富的可视化组件，特别适合实时数据显示场景。

4.1 图表初始化

创建支持多曲线的动态图表：

from PyQt5.QtChart import QChart, QChartView, QLineSeries from PyQt5.QtGui import QPen class RealtimeChart(QChartView): def __init__(self): self.chart = QChart() super().__init__(self.chart) self.series = [] colors = [Qt.red, Qt.blue, Qt.green] for i, color in enumerate(colors): series = QLineSeries() series.setName(f"Series {i+1}") series.setPen(QPen(color, 2)) self.chart.addSeries(series) self.series.append(series) self.chart.createDefaultAxes() self.chart.legend().setVisible(True)

4.2 数据更新优化

实现高效的数据更新策略：

def update_chart(self, new_values): for i, series in enumerate(self.series): # 只保留最近100个数据点 if series.count() > 100: series.remove(0) # 添加新数据点 x = series.count() series.append(x, new_values[i]) # 自动调整坐标轴范围 self.chart.axisX().setRange(0, max(100, self.series[0].count())) # 触发重绘 self.chart.update()

性能优化技巧：

双缓冲技术：减少绘图操作频率
数据采样：大数据量时进行降采样显示
异步渲染：使用QTimer控制刷新率

5. 完整系统集成

将各模块组合成完整的应用程序，使用Qt Designer设计界面布局。

5.1 UI界面设计

推荐界面元素布局：

区域	组件	功能
顶部	串口选择下拉框	选择可用串口
波特率设置	设置通信速率
中部	图表显示区	实时数据曲线
底部	状态栏	显示连接状态
控制按钮	开始/停止采集

5.2 信号槽连接

核心业务逻辑的信号槽连接：

# 串口线程到主线程 self.serial_thread.data_ready.connect(self.on_new_data) # UI控制信号 self.ui.start_button.clicked.connect(self.start_acquisition) self.ui.stop_button.clicked.connect(self.stop_acquisition) # 定时刷新界面 self.refresh_timer = QTimer() self.refresh_timer.timeout.connect(self.update_ui) self.refresh_timer.start(50) # 20Hz刷新率

5.3 异常处理机制

健壮的错误处理流程：

def start_acquisition(self): try: port = self.ui.port_combo.currentText() baud = int(self.ui.baud_combo.currentText()) self.serial.configure_port(port, baud) self.serial_thread.start() except Exception as e: QMessageBox.critical(self, "错误", f"启动失败: {str(e)}") self.serial.close()