使用外部数据绘制 pyqtgraph

Posted 2023-03-25

【中文标题】使用外部数据绘制 pyqtgraph

【英文标题】：Plotting with pyqtgraph using external data

【发布时间】：2018-10-23 04:55:20

【问题描述】：

我正在尝试实时绘制不同传感器的数据，因此我决定在 PyQt 中使用 PyQtGraph 绘制数据，以使其与来自不同来源的多个传感器数据一起工作。 在互联网上搜索示例，我找到了一个并尝试对其进行调整， 因为 QtGui.QApplication.instance().exec_()，它带来了不方便的副作用，即阻止执行它之后的其余代码。我尝试使用 Multiproccessing 来管理使用过的线程。我可以使其余代码工作，但是我如何使用外部数据（Plo2D.update2）更新绘图，我尝试使用 multiprocessing.Queue ，但我没有工作，而是出现窗口的按摩必须关闭.

from pyqtgraph.Qt import QtGui, QtCore
import numpy as np
from numpy import arange
import pyqtgraph as pg
import sys
import multiprocessing

class Plot2D():
    def __init__(self,):
        self.traces = dict()
        self.app = QtGui.QApplication([])
        self.win = pg.GraphicsWindow(title="Dibujar")
        self.win.resize(1000, 600)
        self.win.setWindowTitle('Ejemplo')
        pg.setConfigOptions(antialias=True)
        #self.canvas = self.win.addPlot(title="Pytelemetry")
        self.waveform1 = self.win.addPlot(title='WAVEFORM1', row=1, col=1)
        self.waveform2 = self.win.addPlot(title='WAVEFORM2', row=2, col=1)

    def start(self):
        if (sys.flags.interactive != 1) or not hasattr(QtCore, 'PYQT_VERSION'):
            QtGui.QApplication.instance().exec_()

    def set_plotdata(self, name, datax, datay):
        if name in self.traces:
            self.traces[name].setData(datax, datay)
        else:
            if name == '910D':
                self.traces[name] = self.waveform1.plot(pen='c', width=3)
            if name == 'MPU':
                self.traces[name] = self.waveform2.plot(pen='c', width=3)

    def update2(self):
        # Trying to get external data
        ptm1 = globals()['DatExt1']
        ptm2 = globals()['DatExt2']
        while ptm1.empty() is False:
            self.data1 = ptm1.get()
            self.set_plotdata('MPU', self.data1[0], self.data1[1])
            # csvWriterG910D.writerows(Informa)
            # file1.flush()
        while ptm2.empty() is False:
            self.data2 = ptm2.get()
            self.set_plotdata('910D', self.data1[0], self.data1[1])

    def animation(self):
        timer = QtCore.QTimer()
        timer.timeout.connect(self.update2)
        timer.start(60)
        self.start()

# It is thread started from main.py
def ShowData(Data1, Data2): # Data1,Data2 : multiprocessing.Queue
    DatExt1 = Data1
    DatExt2 = Data2
    p = Plot2D()
    p.animation()

main.py：

    if __name__ == '__main__':

    Data1 = multiprocessing.Queue()
    Data2 = multiprocessing.Queue()

    Plottingdata = Process(target=PlotData.ShowData, args=(Data1, Data2, ))
    Plottingdata.start()

    t = np.arange(-3.0, 2.0, 0.01)
    i = 0.0
    while True:
        s = np.sin(2 * 2 * 3.1416 * t) / (2 * 3.1416 * t + i)
        time.sleep(1)
        Data1.put([t, s])
        i = i + 0.1

感谢ind的帮助

【参考方案1】：

您应该使用多线程，而不是使用多处理，也就是说，创建负责收集数据的线程（在您的示例中模拟数据），然后通过信号将数据发送到 GUI。

PlotData.py

from pyqtgraph.Qt import QtGui, QtCore
import numpy as np
import pyqtgraph as pg
import sys

class Plot2D(pg.GraphicsWindow):
    def __init__(self):
        pg.GraphicsWindow.__init__(self, title="Dibujar")
        self.traces = dict()
        self.resize(1000, 600)
        pg.setConfigOptions(antialias=True)
        #self.canvas = self.win.addPlot(title="Pytelemetry")
        self.waveform1 = self.addPlot(title='WAVEFORM1', row=1, col=1)
        self.waveform2 = self.addPlot(title='WAVEFORM2', row=2, col=1)

    def set_plotdata(self, name, x, y):
        if name in self.traces:
            self.traces[name].setData(x, y)
        else:
            if name == "910D":
                self.traces[name] = self.waveform1.plot(x, y, pen='y', width=3)
            elif name == "MPU":
                self.traces[name] = self.waveform2.plot(x, y, pen='y', width=3)

    @QtCore.pyqtSlot(str, tuple)
    def updateData(self, name, ptm):
        x, y = ptm
        self.set_plotdata(name, x, y)

if __name__ == '__main__':
    app = QtGui.QApplication(sys.argv)
    plot = Plot2D()
    if (sys.flags.interactive != 1) or not hasattr(QtCore, 'PYQT_VERSION'):
        QtGui.QApplication.instance().exec_()

ma​​in.py

import sys

from pyqtgraph.Qt import QtCore, QtGui
import threading
import numpy as np
import time

from PlotData import Plot2D

class Helper(QtCore.QObject):
    changedSignal = QtCore.pyqtSignal(str, tuple)

def create_data1(helper, name):
    t = np.arange(-3.0, 2.0, 0.01)
    i = 0.0
    while True:
        s = np.sin(2 * 2 * 3.1416 * t) / (2 * 3.1416 * t + i)
        time.sleep(.1)
        helper.changedSignal.emit(name, (t, s))
        i = i + 0.1

def create_data2(helper, name):
    t = np.arange(-3.0, 2.0, 0.01)
    i = 0.0
    while True:
        s = np.cos(2 * 2 * 3.1416 * t) / (2 * 3.1416 * t - i)
        time.sleep(.1)
        helper.changedSignal.emit(name, (t, s))
        i = i + 0.1

if __name__ == '__main__':
    app = QtGui.QApplication(sys.argv)
    helper = Helper()
    plot = Plot2D()
    helper.changedSignal.connect(plot.updateData, QtCore.Qt.QueuedConnection)
    threading.Thread(target=create_data1, args=(helper, "910D"), daemon=True).start()
    threading.Thread(target=create_data2, args=(helper, "MPU"), daemon=True).start()
    if (sys.flags.interactive != 1) or not hasattr(QtCore, 'PYQT_VERSION'):
        QtGui.QApplication.instance().exec_()

【讨论】：

非常感谢，我尝试修改代码以在其他函数中执行 QtGui.QApplication，但我注意到 QtGui.QApplication 必须在 main() 中执行。