[Python] How to best plot real-time data with PyQtGraph

What is the best way to update a PyQtGraph plot when the upstream data generation produces one sample at a time that can come in at various rates, from 1 Hz to as fast as 1000 Hz (i.e., 1000 samples per second). The samples will only come in 1 sample at a time. Based on some experimentation, plotting each sample as it comes in is very slow. A workaround is to buffer the data prior to plotting.

Below is some sample code. The BUF_SZ parameter can be changed from between {1, 5, 10, 20} to show how buffering ( BUF_SZ > 1) compares to not buffering (BUF_SZ = 1).

The question is - what is the best practice for plotting data when one sample comes in from an upstream process?

import pyqtgraph as pg
from pyqtgraph.Qt import QtCore, QtWidgets
from collections import deque
import random
import signal
import numpy as np


# Configure global appearance: black background, white text
pg.setConfigOption('background', 'k')
pg.setConfigOption('foreground', 'w')

# Handle keyboard interrupt
signal.signal(signal.SIGINT, signal.SIG_DFL)


# Global variables for plot elements and data buffers
window_size = 500 # Plot history
data = None
win = None
data_plot = None
data_curves = None
buf_cnt = 0
app = None


def init_plot(update_interval=None):
"""Initialize the pyqtgraph window and plots"""
global data
global win, data_plot
global data_curves, app
global BUF_SZ

if update_interval:
BUF_SZ = update_interval

# Initialize data buffers
data = {
"x": deque([0]*window_size, maxlen=window_size),
"y": deque([0]*window_size, maxlen=window_size),
"z": deque([0]*window_size, maxlen=window_size)
}

# Create the Qt application if not already created
app = QtWidgets.QApplication.instance()
if app is None:
app = QtWidgets.QApplication([])

# Create a window using GraphicsLayoutWidget
win = pg.GraphicsLayoutWidget(title="Real-Time Data")
win.resize(900, 600)

# ------------------ Accelerometer Plot ------------------
data_plot = win.addPlot(title="Data")
data_plot.showGrid(x=True, y=True, alpha=0.3)
data_curves = {
"x": data_plot.plot(pen=pg.mkPen(color=(31, 119, 180), width=2), name=' X'), # Blue
"y": data_plot.plot(pen=pg.mkPen(color=(255, 127, 14), width=2), name=' Y'), # Orange
"z": data_plot.plot(pen=pg.mkPen(color=(44, 160, 44), width=2), name=' Z') # Green
}

# Show the window
win.show()
return app


def update_plot(in_data):
global buf_cnt

# Append new data to buffers
data["x"].append(in_data["x"])
data["y"].append(in_data["y"])
data["z"].append(in_data["z"])


# Buffer data to reduce plotting overhead
buf_cnt += 1
if buf_cnt % BUF_SZ == 0:

# Update accelerometer and gyroscope curves
data_curves["x"].setData(list(data["x"]))
data_curves["y"].setData(list(data["y"]))
data_curves["z"].setData(list(data["z"]))

# Process pending Qt events so the window updates immediately
QtWidgets.QApplication.processEvents()

def sim_data():
x = random.uniform(-0.02, 0.02)
y = random.uniform(0.2, 0.3)
z = random.uniform(0.98, 1.02)


return {
"x": x,
"y": y,
"z": z,
}

if __name__ == "__main__":
import time
# !
# ! Setting BUF_SZ to 1 (update 1 sample at a time) slows down plot
# !
BUF_SZ = 5 # Buffer N samples before plotting
app = init_plot(update_interval=BUF_SZ)
rate = 10000.0 # Update rate in Hz
num_samples = 500 # Set a finite number of samples
sample_count = 0 # Track number of generated samples

timer = QtCore.QTimer()
t0 = time.time()

def generate_data():
"""Generate data """
global sample_count
if sample_count >= num_samples:
timer.stop() # Stop after reaching limit
print(f"Finished generating {num_samples} samples.")
elapsed = time.time() - t0
print(f"Elapsed: {elapsed:0.2f} sec")
exit(1)
update_plot(sim_data())
sample_count += 1

timer.timeout.connect(generate_data)
timer.start(int(1000 / rate)) # Interval in ms
app.exec_() # Start app

Continue reading...