如何使用matplotlib来动画化几个图形,每个图形都有一个实时数据的前进窗口,并利用位块传输?

tquggr8v  于 2023-01-26  发布在  其他
关注(0)|答案(1)|浏览(103)

我目前有一个正在运行的python程序,它可以同时动画一个或多个图形,每个图形都有一个实时数据的前进窗口。该程序利用FuncAnimation,并使用轴绘图例程重新绘制每个图形。希望在动画中显示每秒的更新,并且该程序在动画几个图形时能够按预期执行。然而,当尝试为多个(〉5个)图形制作动画时,matplotlib无法在1秒时间范围内完成更新。
了解到完整地更新图需要时间,我尝试在动画过程中使用位块传输。
我试图简化和注解代码以便于理解。我使用的数据是来自文件的二进制流。在运行下面的代码之前,流中的 Dataframe 被标识和标记。在每个 Dataframe 中驻留要绘制的电子信号值。每个电子信号在单个二进制 Dataframe 中具有一个或多个数据点。同时查看多达十几个标绘信号的能力是理想的。代码如下并进行了注解。
我使用一个python deque来模拟一个10秒的数据窗口。对于每次调用FuncAnimation例程,将1秒的数据放入双端队列,然后处理双端队列以生成数据点的xValues和yValues数组。代码底部是FuncAnimation例程,每1秒调用一次在该例程中有2个print语句,我使用它们来确定来自双端队列的xValues和yValues数组中的数据是否正确,以及绘图集_每个图形的Xlim以推进动画数据窗口的方式被正确地改变。
绘图工作正常。但是,在使用set_xlim调用正确应用了一组初始刻度值之后,x轴刻度值并没有更新。而且,我希望y轴ylim自动缩放到数据。但它没有。如何使x轴刻度值随着数据窗口的前进而前进?如何使y轴刻度值正确显示?最后,你会注意到代码隐藏了所有图形的x轴,除了最后一个。2我设计这个的时候考虑到了,虽然每次通过FuncAnimation都会调用每个图形的set_xlim,但是除了一个x轴之外,没有时间花在重绘上。3我希望这会提高性能。4希望你的见解能有所帮助。

from matplotlib import animation
from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas
from matplotlib.figure import Figure
from collections import deque
from PyQt5 import QtCore, QtGui, QtWidgets

#PlotsUI is code created via Qt Designer
class PlotsUI(object):
    def setupUi(self, PlotsUI):
        PlotsUI.setObjectName("PlotsUI")
        PlotsUI.setWindowModality(QtCore.Qt.NonModal)
        PlotsUI.resize(1041, 799)
        sizePolicy = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Expanding, 
                     QtWidgets.QSizePolicy.Expanding)
        sizePolicy.setHorizontalStretch(0)
        sizePolicy.setVerticalStretch(0)
        sizePolicy.setHeightForWidth(PlotsUI.sizePolicy().hasHeightForWidth())
        PlotsUI.setSizePolicy(sizePolicy)
        self.gridLayout_2 = QtWidgets.QGridLayout(PlotsUI)
        self.gridLayout_2.setObjectName("gridLayout_2")
        self.plotLayout = QtWidgets.QVBoxLayout()
        self.plotLayout.setObjectName("plotLayout")
        self.gridLayout_2.addLayout(self.plotLayout, 0, 0, 1, 1)

        self.retranslateUi(PlotsUI)
        QtCore.QMetaObject.connectSlotsByName(PlotsUI)

    def retranslateUi(self, PlotsUI):
        _translate = QtCore.QCoreApplication.translate
        PlotsUI.setWindowTitle(_translate("PlotsUI", "Plots"))

#DataSeriesMgr is given a collection of values for a user selected electronic signal 
#found in the stream of binary data frames. One instance of this class is dedicated to
#manage the values of one electronic signal.
class DataSeriesMgr:
    def __init__(self, frameMultiple, timeRange, dataSeries):
        self._dataSeries = dataSeries 
        #frame multiple will typically be number of binary data frames required
        #for 1 second of data (default 100 frames)
        self._frameMultiple = frameMultiple

        #create a data deque to support the windowing of animated data
        #timeRange is the number of framesMultiples(seconds) of data stored in deque
        self._dataDeque = deque(maxlen=timeRange)
        self._timeRange = timeRange

        #index into dataSeries
        #keep track of what data has been processed
        self._xValueIndex = 0 #byte number in buffer from binary file
        self._dataSeriesSz = len(dataSeries)

        #get the first available xvalue and yvalue arrays to help facilitate
        #the calculation of x axis limits (by default 100 frames of data at a time)
        self._nextXValues, self._nextYValues = self.XYDataSetsForAnimation()
        if self._nextXValues is not None:
            self._nextXLimits = (self._nextXValues[0], self._nextXValues[0] + 
                                 self._timeRange)
        else:
            self._nextXLimits = (None, None)

    @property
    def DataDeque(self):
        return self._dataDeque

    @property
    def TimeRange(self):
        return self._timeRange

    @property
    def NextXValues(self):
        return self._nextXValues

    def GetXYValueArrays(self):
        allXValues = []
        allYValues = []
        #xyDataDeque is a collection of x values, y values tuples each 1 sec in duration
        #convert what's in the deque to arrays of x and y values
        xyDataArray = list(self._dataDeque)
        for dataSet in xyDataArray:
            for xval in dataSet[0]:
                allXValues.append(xval)
            for yval in dataSet[1]:
                allYValues.append(yval)
        #and set the data for the plot line
        #print(f'{key}-NumOfX:  {len(allXValues)}\n\r')
        return allXValues,allYValues

    def GatherFrameData(self, dataSubSet):
        consolidatedXData = []
        consolidatedYData = []
        for frameData in dataSubSet:  # each frame of data subset will have one or more data points
            for dataPointTuple in frameData:  # (unimportantValue, x, y) values
                if dataPointTuple[0] is None: #no data in this frame
                    continue
                consolidatedXData.append(dataPointTuple[1])
                consolidatedYData.append(dataPointTuple[2])
        return consolidatedXData,consolidatedYData

    def XYDataSetsForAnimation(self):
        index = self._xValueIndex #the current location in the data array for animation
        nextIndex = index + self._frameMultiple
        if nextIndex > self._dataSeriesSz: #we are beyond the number of frames
            #there are no more data points to plot for this specific signal
            return None, None
        dataSubset = self._dataSeries[index:nextIndex]
        self._xValueIndex = nextIndex #prepare index for next subset of data to be animated
        #gather data points from data subset
        xyDataSet = self.GatherFrameData(dataSubset)
        #add it to the deque
        # the deque holds a window of a number of seconds of data
        self._dataDeque.append(xyDataSet)
        #convert the deque to arrays of x and y values
        xValues, yValues = self.GetXYValueArrays()
        return xValues, yValues

    def NextXYDataSets(self):
        xValues = self._nextXValues
        yValues = self._nextYValues
        xlimits = self._nextXLimits
        self._nextXValues, self._nextYValues = self.XYDataSetsForAnimation()
        if self._nextXValues is not None:
            self._nextXLimits = (self._nextXValues[0], self._nextXValues[0] + 
                                 self._timeRange)
        else:
            self._nextXLimits = (None, None)
        return xValues, yValues, xlimits

class Graph:
    def __init__(self, title, dataSeriesMgr):
        self._title = title
        self._ax = None
        self._line2d = None
        self._xlimits = None
        self._dataSeriesMgr = dataSeriesMgr

    @property 
    def DataSeriesMgr(self):
        return self._dataSeriesMgr

    @DataSeriesMgr.setter
    def DataSeriesMgr(self, val):
        self._dataSeriesMgr = val

    @property
    def AX(self):
        return self._ax
    
    @AX.setter
    def AX(self, ax):
        self._ax = ax
        line2d, = self._ax.plot([], [], animated=True)
        self._line2d = line2d
        self._ax.set_title(self._title, fontweight='bold', size=10)

    @property
    def Line2D(self):
        return self._line2d

    @Line2D.setter
    def Line2D(self,val):
        self._line2d = val

    @property
    def Title(self):
        return self._title

    @property
    def ShowXAxis(self):
        return self._showXAxis

    @ShowXAxis.setter
    def ShowXAxis(self, val):
        self._showXAxis = val
        self._ax.xaxis.set_visible(val)

    @property
    def XLimits(self):
        return self._xlimits

    @XLimits.setter
    def XLimits(self, tup):
        self._xlimits = tup
        self._ax.set_xlim(tup[0], tup[1])

class Plotter(QtWidgets.QDialog):
    def __init__(self, parentWindow):
        super(Plotter, self).__init__()
        self._parentWindow = parentWindow

        #Matplotlib Figure
        self._figure = Figure()

        self._frameMultiple = 100 #there are 100 frames of data per second
        self._xaxisRange = 10 #make the graphs have a 10 second xaxis range
        self._animationInterval = 1000 #one second

        #PyQt5 UI
        #add the canvas to the UI
        self.ui = PlotsUI()
        self.ui.setupUi(self)
        self._canvas = FigureCanvas(self._figure)
        self.ui.plotLayout.addWidget(self._canvas)

        self.show()

    def PlaceGraph(self,aGraph,rows,cols,pos):
        ax = self._figure.add_subplot(rows,cols,pos)
        aGraph.AX = ax

    def Plot(self, dataSeriesDict):
        self._dataSeriesDict = {}
        self._graphs = {}
        #for this example, simplify the structure of the data to be plotted
        for binaryFileAlias, dataType, dataCode, dataSpec, dataTupleArray in dataSeriesDict.YieldAliasTypeCodeAndData():
            self._dataSeriesDict[dataCode] = DataSeriesMgr(self._frameMultiple, self._xaxisRange, dataTupleArray)
        self._numberOfGraphs = len(self._dataSeriesDict.keys())

        #prepare for blitting
        pos = 1
        self._lines = []
        lastKey = None
        for k,v in self._dataSeriesDict.items():
            #create a graph for each series of data
            aGraph = Graph(k,v)
            self._graphs[k] = aGraph
            #the last graph will show animated x axis
            lastKey = k
            #and place it in the layout
            self.PlaceGraph(aGraph, self._numberOfGraphs, 1, pos)
            aGraph.ShowXAxis = False
            #collect lines from graphs
            self._lines.append(aGraph.Line2D)
            pos += 1

        #show the x axis of the last graph
        lastGraph = self._graphs[lastKey]
        lastGraph.ShowXAxis = True

        #Animate
        self._animation = animation.FuncAnimation(self._figure, self.DisplayAnimatedData, 
                            None, interval=self._animationInterval, blit=True)

    def DisplayAnimatedData(self,i):
        indx = 0
        haveData = False
        for key, graph in self._graphs.items():
            allXValues, allYValues, xlimits = graph.DataSeriesMgr.NextXYDataSets() 
            if allXValues is None: #no more data
                continue
            # print(f'{key}-NumOfX:{len(allXValues)}')
            # print(f'{key}-XLimits: {xlimits[0]}, {xlimits[1]}')
            self._lines[indx].set_data(allXValues, allYValues)
            #call set_xlim on the graph.
            graph.XLimits = xlimits
            haveData = True
            indx += 1

        if not haveData: #no data ??
            self._animation.event_source.stop()
        return self._lines
hpcdzsge

hpcdzsge1#

通过在更新FuncAnimation方法DisplayAnimatedData中的行之前更新x轴限制,数据窗口的前进似乎按预期工作。但是,当尝试以一秒为增量动画显示10个单独的图形时,x轴的前进和图的更新花费几乎两倍的时间(大约2秒),即使在实现只重绘一个x轴而只绘制一次y轴的位块传送时也是如此。

相关问题