实时串口绘图APP读取超10000数据点后出现严重延迟求助

我正在开发一款从Arduino UNO板读取数据的实时串口数据读取与绘图APP，但遇到一个问题：读取超过10000个数据点后，绘图出现严重延迟。已确认电脑内存充足，排查了通信接口，排除硬件问题，附上代码寻求帮助：

import re
import sys
import time
import serial
import threading
import numpy as np
import pyqtgraph as pg
from queue import Queue
from pyqtgraph.Qt import QtCore, QtWidgets

class MyApp(QtWidgets.QWidget):
    def __init__(self):
        QtWidgets.QWidget.__init__(self)
        ## Creating the Widgets and Layouts
        self.plot_widget = pg.PlotWidget()
        self.layout = QtWidgets.QVBoxLayout()
        self.sbutton = QtWidgets.QPushButton("Start / Continue")
        self.ebutton = QtWidgets.QPushButton("Stop")
        self.timer = pg.QtCore.QTimer()
        self.scroll = QtWidgets.QScrollBar(QtCore.Qt.Horizontal)
        ## Creating the variables and constants
        self.stop_thread = False
        self.data_x = [0]  
        self.data_y = [0] 
        self.data_z = [0] 
        self.vsize = 500
        self.psize = 300000000
        self.mag_x = Queue(maxsize=0)
        self.mag_y = Queue(maxsize=0)
        self.mag_z = Queue(maxsize=0)
        self.Labelstyles = {"color": "black", "font-size": "30px", "font-weight": "bold"}
        self.plot_widget.addLegend(labelTextSize ='30px')
        self.plot_item_x = self.plot_widget.plot(self.data_x, pen=pg.mkPen('#89D99D', width=4), name='Bx')  
        self.plot_item_y = self.plot_widget.plot(self.data_y, pen=pg.mkPen('#FA7F08', width=4), name='By')
        self.plot_item_z = self.plot_widget.plot(self.data_z, pen=pg.mkPen('#F24405', width=4), name='Bz')
        ## Building the Widget
        self.setLayout(self.layout)
        self.layout.addWidget(self.sbutton)
        self.layout.addWidget(self.ebutton)
        self.layout.addWidget(self.plot_widget)
        self.layout.addWidget(self.scroll)
        ## Changing some properties of the widgets 
        self.plot_widget.setYRange(-3000, 3000)
        self.plot_widget.setFixedSize(1500, 800)
        self.plot_widget.setBackground('w')
        self.plot_widget.showGrid(x=True, y=True, alpha=0.5)  
        self.plot_widget.setLabel('left', 'Magnetic Intensity (mT)', bold=True, **self.Labelstyles)
        self.plot_widget.setLabel('bottom', 'Counts', bold=True, **self.Labelstyles)
        self.plot_widget.setMouseEnabled(x=False, y=False)
        self.ebutton.setEnabled(False)
        self.scroll.setEnabled(False)
        self.scroll.setMaximum(self.vsize)
        self.scroll.setMinimum(0)
        self.scroll.setValue(self.vsize)
        self.plot_widget.setXRange(0, self.vsize)
        ## Coneccting the signals
        self.mSerial = serial.Serial('COM6', 2000000)
        self.sbutton.clicked.connect(self.start)
        self.ebutton.clicked.connect(self.stop)
        self.timer.timeout.connect(self.update)
        self.scroll.valueChanged.connect(self.upd_scroll)
        self.th1 = threading.Thread(target=self.serial, daemon=True)

    def upd_scroll(self):                                      
        xmax = np.ceil(len(self.data_x) * (self.scroll.value() / self.scroll.maximum()))-2
        xmin = (xmax-self.vsize)
        self.plot_widget.setXRange(xmin, xmax)

    def update(self):
        self.data_x.append(self.mag_x.get())
        self.data_y.append(self.mag_y.get())
        self.data_z.append(self.mag_z.get())
        num = len(self.data_x)
        if num <= self.psize:
            self.plot_item_x.setData(self.data_x)
            self.plot_item_y.setData(self.data_y)
            self.plot_item_z.setData(self.data_z)
        else:
            self.plot_item_x.setData(self.data_x[-self.psize:])
            self.plot_item_y.setData(self.data_y[-self.psize:])
            self.plot_item_z.setData(self.data_z[-self.psize:]) 
        if num == self.vsize:
            self.scroll.setEnabled(True)
        if num > self.vsize:
            self.upd_scroll() 
     
    def start(self):
        self.sbutton.setEnabled(False)
        self.ebutton.setEnabled(True)
        self.timer.start()
        self.stop_thread = False
        window.mSerial.flushInput() 
        if not self.th1.is_alive():
            self.th1 = threading.Thread(target=self.serial, daemon=True)
        self.th1.start()

    def stop(self):
        self.sbutton.setEnabled(True)
        self.ebutton.setEnabled(False)
        self.timer.stop()
        self.stop_thread = True
        self.th1.join()
        self.upd_scroll()
  
    def closeEvent(self, event):
        self.timer.stop()
        event.accept()
    
    def serial(self):
        while not self.stop_thread:
            ret = b''
            n = self.mSerial.inWaiting()
            if(n):
                ret = self.mSerial.readline()
                if len(ret):
                    data_get = ret.decode('UTF-8')
                    pattern = re.compile(r"[+-]?\d+(?:\.\d+)?")
                    data_all = pattern.findall(data_get)
                    for j in range(len(data_all)):
                        if j==0:
                            self.mag_x.put(float(data_all[j]))
                        if j==1:
                            self.mag_y.put(float(data_all[j]))
                        if j==2:
                            self.mag_z.put(float(data_all[j]))
        
if __name__ == "__main__":
    app = QtWidgets.QApplication(sys.argv)
    window = MyApp()
    window.show()
    #access serial port
    if (window.mSerial.isOpen()):
        print("open success")
        window.mSerial.flushInput()  #clear the input buffer
    else:
        print("open failed")
        serial.close()

    sys.exit(app.exec_()) 

核心问题分析 #

1. 全量数据重绘：每次update都传递完整数据列表给setData，数据量越大，pyqtgraph渲染开销呈线性增长，这是延迟的主因。
2. 不合理的数据存储上限：psize=300000000完全超出实时绘图需求，即使内存足够，也会导致数据处理和渲染的巨大开销。
3. 单数据点更新效率低：update每次仅处理一个数据点，UI线程频繁触发，上下文切换成本高。
4. 滚动条计算冗余：依赖np.ceil做浮点运算，数据量越大计算成本越高，且逻辑可简化。

优化方案与修改后代码 #

import re
import sys
import serial
import threading
import numpy as np
import pyqtgraph as pg
from queue import Queue
from pyqtgraph.Qt import QtCore, QtWidgets

class MyApp(QtWidgets.QWidget):
    def __init__(self):
        QtWidgets.QWidget.__init__(self)
        # 组件与布局创建
        self.plot_widget = pg.PlotWidget()
        self.layout = QtWidgets.QVBoxLayout()
        self.sbutton = QtWidgets.QPushButton("Start / Continue")
        self.ebutton = QtWidgets.QPushButton("Stop")
        self.timer = QtCore.QTimer()
        self.scroll = QtWidgets.QScrollBar(QtCore.Qt.Horizontal)
        
        # 变量与常量定义
        self.stop_thread = False
        self.max_data_len = 100000  # 限制最大存储数据量，避免内存过载
        self.view_size = 500        # 可视窗口大小
        self.data_x = [0]
        self.data_y = [0]
        self.data_z = [0]
        self.mag_x = Queue(maxsize=0)
        self.mag_y = Queue(maxsize=0)
        self.mag_z = Queue(maxsize=0)
        
        # 绘图样式配置
        self.label_styles = {"color": "black", "font-size": "30px", "font-weight": "bold"}
        self.plot_widget.addLegend(labelTextSize='30px')
        self.plot_item_x = self.plot_widget.plot(self.data_x, pen=pg.mkPen('#89D99D', width=4), name='Bx')
        self.plot_item_y = self.plot_widget.plot(self.data_y, pen=pg.mkPen('#FA7F08', width=4), name='By')
        self.plot_item_z = self.plot_widget.plot(self.data_z, pen=pg.mkPen('#F24405', width=4), name='Bz')
        
        # 界面布局构建
        self.setLayout(self.layout)
        self.layout.addWidget(self.sbutton)
        self.layout.addWidget(self.ebutton)
        self.layout.addWidget(self.plot_widget)
        self.layout.addWidget(self.scroll)
        
        # 组件属性设置
        self.plot_widget.setYRange(-3000, 3000)
        self.plot_widget.setFixedSize(1500, 800)
        self.plot_widget.setBackground('w')
        self.plot_widget.showGrid(x=True, y=True, alpha=0.5)
        self.plot_widget.setLabel('left', 'Magnetic Intensity (mT)', bold=True, **self.label_styles)
        self.plot_widget.setLabel('bottom', 'Counts', bold=True, **self.label_styles)
        self.plot_widget.setMouseEnabled(x=False, y=False)
        self.ebutton.setEnabled(False)
        self.scroll.setEnabled(False)
        self.scroll.setMaximum(100)  # 用百分比计算滚动范围，适配数据量变化
        self.scroll.setMinimum(0)
        self.scroll.setValue(100)
        self.plot_widget.setXRange(0, self.view_size)
        
        # 信号连接
        self.mSerial = serial.Serial('COM6', 2000000)
        self.sbutton.clicked.connect(self.start)
        self.ebutton.clicked.connect(self.stop)
        self.timer.timeout.connect(self.update_plot)
        self.scroll.valueChanged.connect(self.update_view_range)
        self.data_pattern = re.compile(r"[+-]?\d+(?:\.\d+)?")  # 预编译正则表达式
        self.serial_thread = threading.Thread(target=self.read_serial, daemon=True)

    def update_view_range(self):
        data_len = len(self.data_x)
        if data_len <= self.view_size:
            self.plot_widget.setXRange(0, self.view_size)
            return
        
        scroll_percent = self.scroll.value() / 100.0
        xmax = int(data_len * scroll_percent)
        xmin = max(xmax - self.view_size, 0)
        self.plot_widget.setXRange(xmin, xmax)

    def update_plot(self):
        # 批量处理队列中所有待处理数据
        new_count = self.mag_x.qsize()
        if new_count == 0:
            return
        
        new_x = [self.mag_x.get() for _ in range(new_count)]
        new_y = [self.mag_y.get() for _ in range(new_count)]
        new_z = [self.mag_z.get() for _ in range(new_count)]
        
        # 更新数据列表并截断超出上限的旧数据
        self.data_x.extend(new_x)
        self.data_y.extend(new_y)
        self.data_z.extend(new_z)
        if len(self.data_x) > self.max_data_len:
            self.data_x = self.data_x[-self.max_data_len:]
            self.data_y = self.data_y[-self.max_data_len:]
            self.data_z = self.data_z[-self.max_data_len:]
        
        data_len = len(self.data_x)
        # 仅渲染当前可视范围附近的数据，减少渲染压力
        if data_len <= self.view_size:
            self.plot_item_x.setData(self.data_x)
            self.plot_item_y.setData(self.data_y)
            self.plot_item_z.setData(self.data_z)
        else:
            view_range = self.plot_widget.viewRange()[0]
            start_idx = max(0, int(view_range[0]) - 100)
            end_idx = min(data_len, int(view_range[1]) + 100)
            self.plot_item_x.setData(self.data_x[start_idx:end_idx])
            self.plot_item_y.setData(self.data_y[start_idx:end_idx])
            self.plot_item_z.setData(self.data_z[start_idx:end_idx])
        
        # 数据量达标后启用滚动条，滚动条在底部时自动跟随最新数据
        if data_len == self.view_size:
            self.scroll.setEnabled(True)
        elif data_len > self.view_size and self.scroll.value() == 100:
            self.plot_widget.setXRange(data_len - self.view_size, data_len)

    def start(self):
        self.sbutton.setEnabled(False)
        self.ebutton.setEnabled(True)
        self.timer.start(30)  # 设置定时器间隔为30ms，平衡实时性与性能
        self.stop_thread = False
        self.mSerial.flushInput()
        if not self.serial_thread.is_alive():
            self.serial_thread = threading.Thread(target=self.read_serial, daemon=True)
            self.serial_thread.start()

    def stop(self):
        self.sbutton.setEnabled(True)
        self.ebutton.setEnabled(False)
        self.timer.stop()
        self.stop_thread = True
        self.serial_thread.join()
        self.update_view_range()

    def closeEvent(self, event):
        self.timer.stop()
        self.mSerial.close()
        event.accept()

    def read_serial(self):
        while not self.stop_thread:
            n = self.mSerial.inWaiting()
            if n:
                ret = self.mSerial.readline()
                if len(ret) == 0:
                    continue
                try:
                    data_get = ret.decode('UTF-8')
                    data_all = self.data_pattern.findall(data_get)
                    if len(data_all) >=3:
                        self.mag_x.put(float(data_all[0]))
                        self.mag_y.put(float(data_all[1]))
                        self.mag_z.put(float(data_all[2]))
                except (UnicodeDecodeError, ValueError):
                    # 处理解码或数据格式错误，避免线程崩溃
                    continue

if __name__ == "__main__":
    app = QtWidgets.QApplication(sys.argv)
    window = MyApp()
    window.show()
    if window.mSerial.isOpen():
        print("串口打开成功")
        window.mSerial.flushInput()
    else:
        print("串口打开失败")
        sys.exit(1)
    sys.exit(app.exec_())

优化说明 #

1. 批量数据处理：一次性处理队列中所有新数据，减少UI线程触发次数，降低上下文切换开销。
2. 数据量限制：设置合理的最大存储长度，避免内存无限增长。
3. 局部渲染优化：仅渲染当前可视范围附近的数据，大幅降低pyqtgraph的渲染负载。
4. 滚动条逻辑简化：改用百分比计算可视范围，适配数据量变化，同时增加自动跟随最新数据的功能。
5. 串口线程优化：预编译正则表达式，添加异常处理避免线程崩溃，简化数据判断逻辑。

内容的提问来源于stack exchange，提问作者郑英杰