Implementing Threads on PyQtGraph

[Şener Yılmaz]

In the code below, I'm trying to plot data coming to UDP port of my pc from a RPi 3 model B. At first I used deque object to store data coming from UDP port since you can limit their max length to use them as a circular buffer. However its performance for fetching data in real time was not that impressive. Then I decided to use a CircularBuffer class someone on the internet wrote to replace deque objects. It significantly improved the performance. Next thing I wanna do is implementing Threading to seperate both data acquisition and plotting processes. As you can see below in my script, init calls updateplot function then updateplot function calls getdata. Since they're chained like this together, how can I implement Threading on this? Any other suggestions to improve performance is appreciated.

#!/usr/bin/env python
from __future__ import division
from pyqtgraph.Qt import QtGui, QtCore
import pyqtgraph as pg
import numpy as np
import socket

UDP_IP = "192.168.180.64"
UDP_PORT=[5013,5012]
data=["",""]

sock1= socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock2= socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock1.bind((UDP_IP, UDP_PORT[0]))
sock2.bind((UDP_IP, UDP_PORT[1]))
t=0
time_step=1/1024


class RingBuffer():
    def __init__(self, length):
        self.data = np.zeros(length, dtype='f')
        self.index = 0

    def extend(self, x):
        x_index = (self.index + np.arange(x.size)) % self.data.size
        self.data[x_index] = x
        self.index = x_index[-1] + 1

    def get(self):
        idx = (self.index + np.arange(self.data.size)) % self.data.size
        return self.data[idx]

class DynamicPlotter():

    def __init__(self, sampleinterval, timewindow, size=(1024,350)):
        # Data
        self._interval = int(sampleinterval*1000)
        self._bufsize = int(timewindow/sampleinterval)
        self.databuffer1 = RingBuffer(self._bufsize)
        self.databuffer2 = RingBuffer(self._bufsize)
        self.databuffer3 = RingBuffer(1024)
        for i in range(0,1024):
            self.databuffer3.extend(self.getdata(0))
        self.fourier = np.fft.fft(self.databuffer3.get())
        self.freqs = np.fft.fftfreq(len(self.databuffer3.get()), float(time_step))
        self.idx = np.argsort(self.freqs)
        self.x = np.linspace(-timewindow, 0.0, self._bufsize)
        self.y1 = np.zeros(self._bufsize, dtype=np.float)
        self.y2 = np.zeros(self._bufsize, dtype=np.float)
        self.x3 = self.freqs[self.idx]
        self.y3 = np.abs(self.fourier)[self.idx]
        # PyQtGraph Init
        self.app = QtGui.QApplication([])
        pg.setConfigOptions(antialias=True)
        self.plt1 = pg.plot(title='ADC Kanal 0')
        self.plt1.resize(*size)
        self.plt1.showGrid(x=True, y=True)
        self.plt1.setLabel('left', 'Genlik', 'V')
        self.plt1.setLabel('bottom', 'Zaman')
        self.curve1 = self.plt1.plot(self.x, self.y1, pen=(0,255,0))
        self.plt2 = pg.plot(title='ADC Kanal 1')
        self.plt2.resize(*size)
        self.plt2.showGrid(x=True, y=True)
        self.plt2.setLabel('left', 'Genlik', 'V')
        self.plt2.setLabel('bottom', 'Zaman')
        self.curve2 = self.plt2.plot(self.x, self.y2, pen=(0,0,255))
        self.plt3 = pg.plot(title='Genlik Spektrumu')
        self.plt3.showGrid(x=True, y=True)
        self.plt3.setLabel('left', 'Genlik')
        self.plt3.setLabel('bottom', 'Frekans','Hz')
        self.plt3.resize(*size)
        self.curve3 = self.plt3.plot(self.x3, self.y3, pen=(0,255,255))

        # QTimer
        self.timer = QtCore.QTimer()
        self.timer.timeout.connect(self.updateplot)
        self.timer.start(self._interval)

    def getdata(self,i):
        data[0], addr = sock1.recvfrom(1024)
        data[1], addr = sock2.recvfrom(1024)
        return np.array([data[i]],dtype=np.float)

    def updateplot(self):
        global t
        if t==102:
            self.fourier=np.fft.fft(self.databuffer3.get())
            self.freqs = np.fft.fftfreq(len(self.databuffer3.get()), float(time_step))
            self.idx = np.argsort(self.freqs)
            t=0
        self.databuffer1.extend( self.getdata(0) )
        self.databuffer2.extend( self.getdata(1) )
        self.databuffer3.extend( self.getdata(0) )
        self.y1[:] = self.databuffer1.get()
        self.y2[:] = self.databuffer2.get()
        self.x3[:] = self.freqs[self.idx]
        self.y3[:] = np.abs(self.fourier)[self.idx]
        self.y3[512] = 0

        self.curve1.setData(self.x, self.y1)
        self.curve2.setData(self.x, self.y2)
        self.curve3.setData(self.x3[512:1024], self.y3[512:1024])
        self.app.processEvents()
        t+=1

    def run(self):
        self.app.exec_()

if __name__ == '__main__':

    m = DynamicPlotter(sampleinterval=0.01, timewindow=5)
    m.run()

[vas...@gmail.com]

I can't run your code on my side, but I think you can find some ideas from the next code. Update buffers thread is just for calculation, graphic sequence must remain in the main app (thread).

#!/usr/bin/env python
from __future__ import division
from pyqtgraph.Qt import QtGui, QtCore
import pyqtgraph as pg
import numpy as np
import socket

import threading

UDP_IP = "192.168.180.64"
UDP_PORT = [5013, 5012]
data = ["", ""]

sock1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock2 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

        self.app.processEvents()
        self.updatebuffers_thread = threading.Thread(self.updatebuffers)
        self.updatebuffers_thread.start()

        # QTimer
        self.timer = QtCore.QTimer()
        self.timer.timeout.connect(self.updateplot)
        self.timer.start(self._interval)

    def getdata(self, i):
        data[0], addr = sock1.recvfrom(1024)
        data[1], addr = sock2.recvfrom(1024)
        return np.array([data[i]], dtype=np.float)

    def updatebuffers(self):
        global t
        while True:  # or while self.plt1.isVisible():
            if t == 102:

                self.fourier = np.fft.fft(self.databuffer3.get())
                self.freqs = np.fft.fftfreq(len(self.databuffer3.get()), float(time_step))
                self.idx = np.argsort(self.freqs)

                t = 0
            self.databuffer1.extend(self.getdata(0))
            self.databuffer2.extend(self.getdata(1))
            self.databuffer3.extend(self.getdata(0))

            self.y1[:] = self.databuffer1.get()
            self.y2[:] = self.databuffer2.get()
            self.x3[:] = self.freqs[self.idx]
            self.y3[:] = np.abs(self.fourier)[self.idx]
            self.y3[512] = 0

            t += 1

    def updateplot(self):

        self.curve1.setData(self.x, self.y1)
        self.curve2.setData(self.x, self.y2)
        self.curve3.setData(self.x3[512:1024], self.y3[512:1024])
        self.app.processEvents()

[Şener Yılmaz]

The only thing I changed on your suggestion is 'self.updatebuffers_thread = threading.Thread(self.updatebuffers)' to self.updatebuffers_thread = threading.Thread(target=self.updatebuffers) and it worked like a charm. I appreciate your help.