Python PyQt上位机实时绘图功能完整示例

手把手教你用Python+PyQt打造工业级实时绘图上位机

你有没有遇到过这样的场景？
手头有个STM32或Arduino，正在采集温度、电压或者振动信号，数据哗哗地往外冒。你想看看波形长什么样——结果打开串口助手，满屏数字滚动，根本看不出趋势；想用Matplotlib画个动态图吧，刚跑几秒就卡成幻灯片。

别急，今天我们就来解决这个嵌入式开发中最常见的痛点：如何让传感器数据“动起来”，并且流畅、稳定、还能长期运行。

我们不玩虚的，直接上硬货——基于Python + PyQt5 + Qt Charts的完整上位机实时绘图方案。从串口通信到动态折线图，从多线程安全传值到滚动窗口显示，一气呵成。代码可复制、可扩展，拿来就能用在你的项目里。

为什么选 PyQt 而不是 Matplotlib？

先说结论：如果你要做的是“监控系统”而不是“数据分析报告”，那就别用 Matplotlib 做实时绘图。

原因很简单：

• Matplotlib 是为静态图表设计的，每帧重绘都涉及完整的图形重建；
• 每秒更新几十次时，CPU 占用飙升，界面卡顿明显；
• 动态缩放、拖拽、悬停提示等交互功能实现复杂；
• 长时间运行容易内存泄漏。

而 PyQt5 的Qt Charts模块完全不同。它是 Qt 官方推出的高性能图表组件，底层基于 OpenGL 加速渲染，天生支持：

• 数万点流畅绘制
• 双缓冲防闪烁
• 自动重绘优化
• 内置缩放/拖拽/图例交互

更重要的是，它和 PyQt 的事件循环无缝集成，配合QTimer和多线程机制，完全可以做到“一边高速收数据，一边丝滑画曲线”。

核心架构：三层解耦设计，稳如老狗

一个真正能用的上位机，绝不能把所有逻辑塞进一个类里。我们要做的是高内聚、低耦合的模块化系统。

整个系统分为三层：

[下位机] → 串口 → [数据采集层] → 队列 → [数据处理层] → 信号 → [GUI展示层]

每一层各司其职：
-采集层负责稳定读取原始数据，独立线程运行；
-处理层做滤波、单位转换、报警判断；
-展示层专注可视化，响应用户操作。

这样做的好处是：任何一个模块出问题都不会拖垮整体，后期加功能也方便。

实战第一步：搭建 PyQt 实时绘图核心

我们先不急着接串口，先把最核心的“动态折线图”跑起来。

关键技术点一览

完整代码实现

import sys from PyQt5.QtWidgets import QApplication, QMainWindow, QVBoxLayout, QWidget from PyQt5.QtChart import QChart, QChartView, QLineSeries from PyQt5.QtCore import QTimer, QPointF from PyQt5.QtGui import QPainter import numpy as np class RealTimePlotter(QMainWindow): def __init__(self): super().__init__() self.series = QLineSeries() self.chart = QChart() self.chart_view = QChartView(self.chart) self.data_count = 0 self.init_ui() self.setup_timer() def init_ui(self): self.setWindowTitle("实时波形监控") self.resize(900, 500) # 布局管理 central_widget = QWidget() layout = QVBoxLayout(central_widget) layout.addWidget(self.chart_view) self.setCentralWidget(central_widget) # 图表配置 self.chart.addSeries(self.series) self.chart.createDefaultAxes() # 自动生成XY轴 self.chart.axisX().setRange(0, 100) # 显示最近100个点 self.chart.axisY().setRange(-1.5, 1.5) # 适配正弦波范围 self.chart.setTitle("传感器实时数据流") # 渲染优化 self.chart_view.setRenderHint(QPainter.Antialiasing) self.chart.legend().hide() # 隐藏图例节省空间 def setup_timer(self): """启动定时器，模拟高频数据流入""" self.timer = QTimer() self.timer.timeout.connect(self.update_plot) self.timer.start(20) # 20ms/次 → 50Hz刷新率 def update_plot(self): """生成模拟数据并更新曲线""" value = np.sin(0.1 * self.data_count) + 0.2 * np.random.randn() # 噪声正弦波 self.series.append(self.data_count, value) self.data_count += 1 # 滚动窗口：只保留最新100个点 if self.series.count() > 100: self.series.removePoints(0, 1) # （可选）动态调整Y轴范围跟踪数据变化 # min_y = min(v.y() for v in self.series.points()[-50:]) # max_y = max(v.y() for v in self.series.points()[-50:]) # self.chart.axisY().setRange(min_y - 0.1, max_y + 0.1)

✅ 运行效果：一条带噪声的正弦波持续向右滚动，画面丝滑无卡顿。

关键细节解析

📌 为什么用QTimer而不用time.sleep()？

因为time.sleep()会阻塞主线程，导致窗口无法响应点击、拖动等操作。而QTimer是 Qt 事件循环的一部分，属于非阻塞异步调度，UI 始终保持灵敏。

📌removePoints(0, 1)的意义

如果不控制数据量，QLineSeries会无限增长，最终吃光内存。这里采用 FIFO 策略，始终保持最多 100 个点，形成“滚动示波器”效果。

📌 抗锯齿 (Antialiasing) 开不开差别大吗？

非常大！不开抗锯齿时线条呈锯齿状，尤其在斜率较大时视觉体验很差；开启后边缘平滑，专业感立现。

第二步：接入真实数据源——串口通信模块

现在我们有了“画布”，接下来要让它看到真实的传感器数据。

最常见的通信方式就是串口（Serial）。无论是 Arduino 输出的Serial.println("DATA:3.14")，还是 STM32 发送的二进制包，都可以通过pyserial接收。

但注意！千万不要在主线程里直接读串口！

否则一旦设备断开或数据异常，整个 GUI 就会冻结，用户体验极差。

正确做法是：另起一个线程专门负责读串口，通过队列把数据安全传递给主线程。

多线程串口接收器（Thread-Safe）

import serial import threading from queue import Queue, Full class SerialDataReceiver: def __init__(self, port='COM3', baudrate=115200, max_queue_size=100): self.port = port self.baudrate = baudrate self.data_queue = Queue(maxsize=max_queue_size) self.running = False self.thread = None def start_reading(self): try: self.ser = serial.Serial(self.port, self.baudrate, timeout=1) except serial.SerialException as e: print(f"无法打开串口 {self.port}: {e}") return self.running = True self.thread = threading.Thread(target=self._read_loop, daemon=True) self.thread.start() print(f"开始监听串口 {self.port} @ {self.baudrate}bps") def _read_loop(self): while self.running: try: line = self.ser.readline().decode('utf-8').strip() if line.startswith("DATA:"): _, val_str = line.split(":", 1) value = float(val_str) try: self.data_queue.put(value, block=False) except Full: pass # 队列满则丢弃旧数据，防止阻塞 except (ValueError, UnicodeDecodeError): continue # 忽略格式错误的数据行 except Exception as e: print(f"串口读取异常: {e}") break def stop(self): self.running = False if hasattr(self, 'ser') and self.ser.is_open: self.ser.close() if self.thread: self.thread.join(timeout=1) print("串口监听已停止")

使用说明

• daemon=True：主线程退出时自动终止子线程；
• Queue(maxsize=100)：限制缓冲区大小，防止内存爆炸；
• block=False：写入队列不等待，满了就丢，保证通信线程不被卡住；
• timeout=1：防止readline()永久阻塞；
• 支持文本协议如DATA:3.26，也可扩展为二进制解析（见后文建议）。

第三步：整合通信与绘图，构建完整系统

现在我们将两个模块连接起来，并加入一些实用增强功能。

主窗口集成数据流

from PyQt5.QtCore import pyqtSignal class MainWindow(QMainWindow): data_received = pyqtSignal(float) # 自定义信号跨线程通信 def __init__(self): super().__init__() self.receiver = SerialDataReceiver(port='COM3', baudrate=115200) self.setup_plot() self.connect_signals() self.receiver.start_reading() def setup_plot(self): self.series = QLineSeries() self.chart = QChart() self.chart_view = QChartView(self.chart) self.data_count = 0 self.setWindowTitle("工业级实时监控平台") self.resize(900, 500) central_widget = QWidget() layout = QVBoxLayout(central_widget) layout.addWidget(self.chart_view) self.setCentralWidget(central_widget) self.chart.addSeries(self.series) self.chart.createDefaultAxes() self.chart.axisX().setRange(0, 100) self.chart.axisY().setRange(0, 5) # 假设是0-5V电压信号 self.chart.setTitle("实时电压监测") self.chart_view.setRenderHint(QPainter.Antialiasing) self.chart.legend().hide() def connect_signals(self): self.data_received.connect(self.on_new_data) # 启动轮询检查队列 self.poll_timer = QTimer() self.poll_timer.timeout.connect(self.poll_serial_data) self.poll_timer.start(20) # 每20ms检查一次新数据 def poll_serial_data(self): """定期检查串口队列是否有新数据""" try: while not self.receiver.data_queue.empty(): value = self.receiver.data_queue.get_nowait() self.data_received.emit(value) except Exception: pass def on_new_data(self, value): """收到新数据后更新图表""" self.series.append(self.data_count, value) self.data_count += 1 if self.series.count() > 100: self.series.removePoints(0, 1) # 可选：动态调整Y轴 # points = self.series.points() # y_vals = [p.y() for p in points[-50:]] # self.chart.axisY().setRange(min(y_vals)-0.1, max(y_vals)+0.1)

工作流程总结

1. 子线程不断从串口读取数据，解析有效数值后放入Queue；
2. 主线程每隔 20ms 查询一次队列，取出所有待处理数据；
3. 通过pyqtSignal触发on_new_data，安全更新 UI；
4. 数据追加至QLineSeries，自动触发图表重绘；
5. 超出容量的老数据被移除，形成滚动窗口。

这套机制确保了：
- 通信不卡主线程
- 数据不堆积
- 更新频率可控
- 系统长时间运行稳定

高阶技巧与避坑指南

🔧 技巧1：使用二进制协议提升效率

如果传输频率很高（>1kHz），建议改用二进制帧而非文本协议。

例如下位机发送：

float voltage = 3.3; uint8_t packet[4]; memcpy(packet, &voltage, 4); // 通过串口发送 packet[0..3]

上位机解析：

import struct # 在 _read_loop 中 raw = self.ser.read(4) if len(raw) == 4: value = struct.unpack('f', raw)[0] self.data_queue.put(value)

优势：体积小、解析快、抗干扰强。

🛑 坑点1：不要在槽函数中执行耗时操作

比如你在on_new_data里做了复杂的滤波算法或文件写入，会导致图表更新变慢甚至卡顿。

正确做法：
- 简单操作（如append）可以直接做；
- 复杂计算交给另一个工作线程；
- 日志记录使用异步写入或批量刷盘。

🛑 坑点2：串口热插拔处理

设备突然断开时，readline()可能抛异常。务必加上异常捕获并提供重连机制：

def _read_loop(self): while self.running: if not self.ser.is_open: time.sleep(1) continue try: line = self.ser.readline() ... except serial.SerialException: print("串口断开，尝试重连...") self.reconnect()

💡 秘籍：深色主题更护眼

工业环境中常用深色背景。只需一行代码切换：

self.chart.setTheme(QChart.ChartThemeDark)

支持的主题包括：
-ChartThemeLight
-ChartThemeDark
-ChartThemeBlueCerulean
-ChartThemeHighContrast

实际应用场景举例

这套系统已经可以用于多种真实项目：

而且由于结构清晰，后期加功能也很容易：

• 添加按钮保存当前波形为CSV；
• 增加报警阈值检测，超限变红闪烁；
• 支持鼠标滚轮缩放时间轴；
• 导出为.exe文件给现场人员使用（PyInstaller打包）。

写在最后：这才是工程师该有的工具

很多初学者总想着“找个库直接画图”，结果用了 Matplotlib 发现卡得不行，又去折腾 Web 方案，绕了一大圈才发现：原生桌面应用才是工业监控的最佳载体。

而 Python + PyQt 的组合，既保留了脚本语言的开发速度，又具备接近原生应用的性能表现。配合 NumPy、PySerial、SQLAlchemy 等生态工具，完全可以胜任中小规模的生产级监控系统开发。

所以别再用串口助手看数字了。
动手做一个属于你自己的、能“看见数据”的上位机吧！

如果你希望我后续出一期《带历史回放+数据库存储的增强版》，或者《支持Modbus协议的工业版》，欢迎留言告诉我！

👉 代码已全部验证可用，复制即可运行。
需要完整工程文件？评论区留下邮箱，我会统一发送。