基于PyQtGraph的串口实时绘图程序数据丢失问题求助

算力阿峰

2026-4-14

我写了一个从串口接收数据并实时绘图的PyQtGraph程序，但发现locals()[self.axes_mapping[axis]]的值并没有包含receive_data函数收到的所有数据，似乎有部分数据丢失了。如果我打印x的值，它是和收到的数据对应的，但绘图时就丢了数据。请问如何确保所有接收的数据都能被绘制出来？

我的代码如下：

import sys
from PyQt6.QtWidgets import QApplication, QMainWindow
import pyqtgraph as pg
from pyqtgraph.Qt import QtCore
import numpy as np
from collections import deque
import threading
import queue
import serial

BUFFER_SIZE = 100

class RealTimePlot(QMainWindow):
    def __init__(self):
        super().__init__()

        self.setup_ui()
    
        self.serial_port = serial.Serial('COM6', 115200)
        self.N = 100
        self.fs = 1000  # Sampling frequency in Hz (adjust according to your setup)
        self.T = 1/self.fs
        self.x_values = np.arange(0, self.N*self.T, self.T)

        # Circular buffers for time domain plots
        self.axes = ['X', 'RMS']

        self.z_values = {axis: deque([0] * self.N, maxlen=self.N) for axis in self.axes}
        self.z_index = {axis: 0 for axis in self.axes}

        # Axes variable mapping
        self.axes_mapping = {'X': 'x', 'RMS': 'rms'}

        # Plotting setup
        self.setup_plots()

        self.data_queue = queue.Queue()
        # Lock for synchronizing access to the data queue
        self.data_queue_lock = threading.Lock()

        # Create and start the receiving thread
        self.receive_thread = threading.Thread(target=self.receive_data)
        self.receive_thread.daemon = True
        self.receive_thread.start()

        # Start the animation
        self.timer = QtCore.QTimer(self)
        self.timer.timeout.connect(self.update_plot)
        self.timer.start(10)

    def setup_ui(self):
        self.central_widget = pg.GraphicsLayoutWidget()
        self.setCentralWidget(self.central_widget)

    def setup_plots(self):
        self.plots = {axis: self.central_widget.addPlot(row=i, col=0, title=f"<span style='color: #ffffff; font-weight: bold; font-size: 15px'>Time Domain - {axis} Axis</span>")
                      for i, axis in enumerate(self.axes)}

        for plot in self.plots.values():
            plot.setLabel('bottom', 'Time', 's')
            # plot.setLabel('left', 'Amplitude', 'g')
            plot.setYRange(-2, 5000)


        linha1 = pg.mkPen((52, 255, 52), width=2) # R G B & width
        # linha4 = pg.mkPen((255, 255, 255), width=2) 
        self.lines = {axis: plot.plot(pen=linha1) for axis, plot in self.plots.items()}
        # self.lines_fft = {axis: plot.plot(pen=linha4) for axis, plot in self.plots_fft.items()}
        self.plots['RMS'].setYRange(0, 5000)

    def receive_data(self):
        data_buffer = np.zeros((BUFFER_SIZE), dtype=int)
        data_cnt = 0
        rx_flag = True
        while True:
            if self.serial_port.in_waiting > 0.0:  # Check if there is data waiting
                    data_str =  int.from_bytes(self.serial_port.read(2), byteorder='little', signed = False)
                    rx_flag = True

            with self.data_queue_lock:
                if rx_flag == True:
                    data_buffer[data_cnt] = data_str
                    data_cnt = data_cnt+1
                    rx_flag = False

                # Check if the buffer size is reached, then update the plot
                if data_cnt >= BUFFER_SIZE:
                    self.data_queue.put(data_buffer)
                    data_buffer = np.zeros((BUFFER_SIZE), dtype=int)
                    data_cnt = 0
                    

    def calculate_rms(self, x):
        return np.sqrt(np.mean(np.square([x])))

    def update_plot(self):
        with self.data_queue_lock:
            while not self.data_queue.empty():
                data_buffer = self.data_queue.get()

                for data_str in data_buffer:
                    x = data_str
                    rms = self.calculate_rms(data_buffer)

                for axis in self.axes:
                    self.z_values[axis].append(locals()[self.axes_mapping[axis]])
                    self.lines[axis].setData(self.x_values, self.z_values[axis])
                    print(locals()[self.axes_mapping[axis]])
        return self.lines.values()
    

def closeEvent(self, event):
    self.csv_file.close()
    event.accept()

def main():
    app = QApplication(sys.argv)
    window = RealTimePlot()
    window.show()
    sys.exit(app.exec())

if __name__ == '__main__':
    main()

问题排查与分析

我帮你梳理了代码里的几个核心问题，正是这些问题导致了数据丢失：

绘图更新逻辑的致命错误
在update_plot里，你遍历data_buffer的时候，只是把每个data_str赋值给x，但没有即时把x添加到绘图缓冲区里——最后x只会保留整个data_buffer的最后一个值！同样，rms只计算了一次整个缓冲区的均值，但也没有对应每个数据点处理。这就导致每次处理100个数据的缓冲区时，只把最后1个x和1个rms加到绘图队列里，前面99个数据直接丢了，这肯定会出现数据缺失的情况。
RMS计算逻辑不符合预期
你的calculate_rms函数现在传入单个x，然后用[x]计算均方根，这其实和取x的绝对值没区别，不是真正意义上的均方根（通常RMS是一段数据的统计值）。如果你的需求是计算每段缓冲区的RMS，那得调整函数参数；如果就是要单个点的“瞬时RMS”，那这个函数没问题，但命名容易让人误解。
串口接收的小细节问题
if self.serial_port.in_waiting > 0.0这里应该写成> 0，因为in_waiting返回的是整数类型的待读字节数，用浮点数判断虽然不会报错，但不符合逻辑规范。

修复后的代码

下面是修改后的完整代码，我标注了关键修改点，你可以对比看看：

import sys
from PyQt6.QtWidgets import QApplication, QMainWindow
import pyqtgraph as pg
from pyqtgraph.Qt import QtCore
import numpy as np
from collections import deque
import threading
import queue
import serial

BUFFER_SIZE = 100

class RealTimePlot(QMainWindow):
    def __init__(self):
        super().__init__()

        self.setup_ui()
    
        self.serial_port = serial.Serial('COM6', 115200)
        self.N = 100
        self.fs = 1000  # Sampling frequency in Hz (adjust according to your setup)
        self.T = 1/self.fs
        self.x_values = np.arange(0, self.N*self.T, self.T)

        # Circular buffers for time domain plots
        self.axes = ['X', 'RMS']

        self.z_values = {axis: deque([0] * self.N, maxlen=self.N) for axis in self.axes}
        self.z_index = {axis: 0 for axis in self.axes}

        # Axes variable mapping
        self.axes_mapping = {'X': 'x', 'RMS': 'rms'}

        # Plotting setup
        self.setup_plots()

        self.data_queue = queue.Queue()
        # Lock for synchronizing access to the data queue
        self.data_queue_lock = threading.Lock()

        # Create and start the receiving thread
        self.receive_thread = threading.Thread(target=self.receive_data)
        self.receive_thread.daemon = True
        self.receive_thread.start()

        # Start the animation
        self.timer = QtCore.QTimer(self)
        self.timer.timeout.connect(self.update_plot)
        self.timer.start(10)

    def setup_ui(self):
        self.central_widget = pg.GraphicsLayoutWidget()
        self.setCentralWidget(self.central_widget)

    def setup_plots(self):
        self.plots = {axis: self.central_widget.addPlot(row=i, col=0, title=f"<span style='color: #ffffff; font-weight: bold; font-size: 15px'>Time Domain - {axis} Axis</span>")
                      for i, axis in enumerate(self.axes)}

        for plot in self.plots.values():
            plot.setLabel('bottom', 'Time', 's')
            # plot.setLabel('left', 'Amplitude', 'g')
            plot.setYRange(-2, 5000)


        linha1 = pg.mkPen((52, 255, 52), width=2) # R G B & width
        # linha4 = pg.mkPen((255, 255, 255), width=2) 
        self.lines = {axis: plot.plot(pen=linha1) for axis, plot in self.plots.items()}
        # self.lines_fft = {axis: plot.plot(pen=linha4) for axis, plot in self.plots_fft.items()}
        self.plots['RMS'].setYRange(0, 5000)

    def receive_data(self):
        data_buffer = np.zeros((BUFFER_SIZE), dtype=int)
        data_cnt = 0
        rx_flag = True
        while True:
            # 修改点1：将>0.0改为>0，符合整数类型的判断逻辑
            if self.serial_port.in_waiting > 0:  
                    data_str =  int.from_bytes(self.serial_port.read(2), byteorder='little', signed = False)
                    rx_flag = True

            with self.data_queue_lock:
                if rx_flag == True:
                    data_buffer[data_cnt] = data_str
                    data_cnt = data_cnt+1
                    rx_flag = False

                # Check if the buffer size is reached, then update the plot
                if data_cnt >= BUFFER_SIZE:
                    self.data_queue.put(data_buffer)
                    data_buffer = np.zeros((BUFFER_SIZE), dtype=int)
                    data_cnt = 0
                    
    # 修改点2：调整RMS计算逻辑，现在传入整个数据数组，计算这段数据的均方根
    def calculate_rms(self, data_array):
        return np.sqrt(np.mean(np.square(data_array)))

    def update_plot(self):
        with self.data_queue_lock:
            while not self.data_queue.empty():
                data_buffer = self.data_queue.get()

                # 修改点3：遍历每个数据点，逐个添加到绘图缓冲区
                # 这里假设你需要：X轴显示每个原始数据点，RMS显示当前整段缓冲区的统计值
                # 如果需要每个数据点对应滑动窗口的RMS，可以再调整逻辑
                rms_value = self.calculate_rms(data_buffer)
                for data_str in data_buffer:
                    x = data_str
                    # 把每个x和对应的RMS值添加到各自的循环缓冲区
                    self.z_values['X'].append(x)
                    self.z_values['RMS'].append(rms_value)
                    # 调试打印，确认每个数据都被处理
                    print(f"已处理数据 - X: {x}, RMS: {rms_value}")

                # 修改点4：更新绘图，现在缓冲区里已经有所有数据了
                for axis in self.axes:
                    self.lines[axis].setData(self.x_values, self.z_values[axis])
        return self.lines.values()
    

def closeEvent(self, event):
    self.csv_file.close()
    event.accept()

def main():
    app = QApplication(sys.argv)
    window = RealTimePlot()
    window.show()
    sys.exit(app.exec())

if __name__ == '__main__':
    main()

额外说明

如果你的需求是每个原始数据点对应一个滑动窗口的RMS（比如用最近10个点计算RMS），而不是整段缓冲区的RMS，那需要调整update_plot里的逻辑，比如给RMS单独维护一个滑动窗口缓冲区，每次添加新数据后计算窗口内的RMS。
串口接收线程的逻辑目前是攒够100个数据才放到队列，如果你的串口数据速率较低，可能会觉得绘图更新不及时，可以考虑减小BUFFER_SIZE，或者设置串口超时时间，避免等待太久。
你已经用了data_queue_lock来保护队列的线程安全，这部分做得很好，继续保持就好。

备注：内容来源于stack exchange，提问作者JOSEPH129009

基于PyQtGraph的串口实时绘图程序数据丢失问题求助

问题排查与分析 #

修复后的代码 #

额外说明 #

问题排查与分析

修复后的代码

额外说明