Обновление диаграммы pqtgraph из RxPy наблюдаемого

Question

В следующем сценарии создается окно графика и values правильно передано plot.update. Однако сюжет не обновляется. Что я делаю не так?

import sys 
import time 

import numpy 
from numpy import pi 

import rx 
from rx.concurrency import QtScheduler 

import pyqtgraph as pg 
from pyqtgraph.Qt import QtCore, QtGui 


class DemoData: 

    def __init__(self, window, steps): 

     def signal(t): 
      omega = 5 
      return numpy.sin(2*pi*omega*t) + numpy.random.random() 

     self.stream = rx.Observable.range(
      0, steps 
     ).map(
      signal 
     ).buffer_with_count(
      window, 1 
     ) 


class UpdatingPlot: 

    def __init__(self): 
     self.plot = pg.plot() 
     self.curve = self.plot.plot() 

    def update(self, values): 
     self.curve.setData(values) 


def main(): 

    app = QtGui.QApplication([]) 
    scheduler = QtScheduler(QtCore) 

    window, steps = 50, 100 
    data = DemoData(window, steps) 
    plot = UpdatingPlot() 

    data.stream.subscribe_on(scheduler=scheduler).subscribe(plot.update) 

    sys.exit(app.exec_()) 


if __name__ == '__main__': 

    main() 

Answer 1

Это не исправит вашу проблему, но это может повлиять на вашу потребность. Вы можете использовать пакет joystick, установить его с помощью pip install joystick. Он предназначен для создания рамок реального времени и интерактивного построения графиков.

Вот пример использования. Он создает графический фрейм, который «прослушивает» и отображает вашу синусоидальную функцию как функцию времени; через каждые 0,2 секунды добавляется новая точка данных.

import joystick as jk 
import numpy as np 
import time 

class DemoData(jk.Joystick): 
    # initialize the infinite loop and callit decorators so they can auto- 
    # register methods they decorate 
    _infinite_loop = jk.deco_infinite_loop() 
    _callit = jk.deco_callit() 

    @_callit('before', 'init') 
    def _init_data(self, *args, **kwargs): 
     # Function automatically called at initialization, thanks to the 
     # decorator 
     self.tdata = np.array([]) # time x-axis 
     self.ydata = np.array([]) # data y-axis 
     self.omega = 5 

    @_callit('after', 'init') 
    def _build_frames(self, *args, **kwargs): 
     # Function automatically called at initialization, thanks to the 
     # decorator. 
     # Creates a graph frame 
     self.mygraph = self.add_frame(
        jk.Graph(name="DemoData", size=(500, 500), pos=(50, 50), 
          fmt="go-", xnpts=50, freq_up=7, bgcol="w", 
          xylim=(None,None,-1.1,2.1), xlabel='t', ylabel='sine')) 

    @_callit('before', 'start') 
    def _set_t0(self): 
     # initialize t0 at start-up 
     self._t0 = time.time() 

    @_infinite_loop(wait_time=0.2) 
    def _get_data(self): 
     # This method will automatically be called with simulation start 
     # (t.start()), and looped every 0.2 in a separate thread as long as 
     # the simulation runs (running == True) 
     # It generates new data and pushes it to the frame. 
     # concatenate data on the time x-axis 
     timestamp = time.time() - self._t0 
     self.tdata = jk.add_datapoint(self.tdata, 
             timestamp, 
             xnptsmax=self.mygraph.xnptsmax) 
     new_y_data = np.sin(2*np.pi*self.omega*timestamp) + np.random.random() 
     self.ydata = jk.add_datapoint(self.ydata, 
             new_y_data, 
             xnptsmax=self.mygraph.xnptsmax) 
     # push new data to the graph 
     self.mygraph.set_xydata(np.round(self.tdata, 1), self.ydata) 

d = DemoData() 
d.start() 
... 
d.stop()