Reads data from a thread, plots it in another Process, with main process being free all the time ! Using PyQtGraph, python standard multiprocessing and multiprocessing.Queue

Plot.py

# -*- coding: utf-8 -*-
from pyqtgraph.Qt import QtGui, QtCore
import numpy as np
import pyqtgraph as pg
from multiprocessing import Process, Manager, Queue
import sched, time, threading

# This function is responsible for displaying the data
# it is run in its own process to liberate main process
def display(name,q):
    app2 = QtGui.QApplication([])

    win2 = pg.GraphicsWindow(title="Basic plotting examples")
    win2.resize(1000,600)
    win2.setWindowTitle('pyqtgraph example: Plotting')
    p2 = win2.addPlot(title="Updating plot")
    curve = p2.plot(pen='y')

    x_np = []
    y_np = []

    def updateInProc(curve,q,x,y):
        item = q.get()
        x.append(item[0])
        y.append(item[1])
        curve.setData(x,y)

    timer = QtCore.QTimer()
    timer.timeout.connect(lambda: updateInProc(curve,q,x_np,y_np))
    timer.start(50)

    QtGui.QApplication.instance().exec_()

# This is function is responsible for reading some data (IO, serial port, etc)
# and forwarding it to the display
# it is run in a thread
def io(running,q):
    t = 0
    while running.is_set():
        s = np.sin(2 * np.pi * t)
        t += 0.01
        q.put([t,s])
        time.sleep(0.01)
    print("Done")

if __name__ == '__main__':
    q = Queue()
    # Event for stopping the IO thread
    run = threading.Event()
    run.set()

    # Run io function in a thread
    t = threading.Thread(target=io, args=(run,q))
    t.start()

    # Start display process
    p = Process(target=display, args=('bob',q))
    p.start()
    input("See ? Main process immediately free ! Type any key to quit.")
    run.clear()
    print("Waiting for scheduler thread to join...")
    t.join()
    print("Waiting for graph window process to join...")
    p.join()
    print("Process joined successfully. C YA !")

Hello @Overdrivr , I ran your code on Windows10 Pycharm and got the following error:

Traceback (most recent call last):
  File "F:\Anaconda\envs\empty\lib\site-packages\IPython\core\interactiveshell.py", line 3441, in run_code
    exec(code_obj, self.user_global_ns, self.user_ns)
  File "<ipython-input-2-333432fa0c82>", line 1, in <module>
    runfile('G:/0_mycode/empty_pc/git/Empty/core/test.py', wdir='G:/0_mycode/empty_pc/git/Empty/core')
  File "F:\PyCharm 2021.2\plugins\python\helpers\pydev\_pydev_bundle\pydev_umd.py", line 198, in runfile
    pydev_imports.execfile(filename, global_vars, local_vars)  # execute the script
  File "F:\PyCharm 2021.2\plugins\python\helpers\pydev\_pydev_imps\_pydev_execfile.py", line 18, in execfile
    exec(compile(contents+"\n", file, 'exec'), glob, loc)
  File "G:/0_mycode/empty_pc/git/Empty/core/test.py", line 57, in <module>
    p.start()
  File "F:\Anaconda\envs\empty\lib\multiprocessing\process.py", line 121, in start
    self._popen = self._Popen(self)
  File "F:\Anaconda\envs\empty\lib\multiprocessing\context.py", line 224, in _Popen
    return _default_context.get_context().Process._Popen(process_obj)
  File "F:\Anaconda\envs\empty\lib\multiprocessing\context.py", line 327, in _Popen
    return Popen(process_obj)
  File "F:\Anaconda\envs\empty\lib\multiprocessing\popen_spawn_win32.py", line 93, in __init__
    reduction.dump(process_obj, to_child)
  File "F:\Anaconda\envs\empty\lib\multiprocessing\reduction.py", line 60, in dump
    ForkingPickler(file, protocol).dump(obj)
_pickle.PicklingError: Can't pickle <function display at 0x0000027AB237DF70>: attribute lookup display on __main__ failed
Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "F:\Anaconda\envs\empty\lib\multiprocessing\spawn.py", line 116, in spawn_main
    exitcode = _main(fd, parent_sentinel)
  File "F:\Anaconda\envs\empty\lib\multiprocessing\spawn.py", line 126, in _main
    self = reduction.pickle.load(from_parent)
EOFError: Ran out of input

Could you help me please?

This works.!!! But I want to embed pyqtgraph in application and turn on and off the plotting. I wonder how will I be able to do this. Do you have any code to resemble the idea?

I was also having some issues getting this example to work, but managed to in the end by changing GraphicsWindow to GraphicsLayoutWidget, and app2 = QtGui.QApplication([]) to app2 = pg.mkQApp("Multiprocess plotter").

Nevertheless, when I tried adding a second process that was making data for a second curve, using the Queue got quite laggy.
Using an example I found that made a numpy array with shared memory, I was able to make a plotter which plots two curves, but the data it plots gets updated by two external processes.

Hope someone finds this useful.

import time
from multiprocessing import Process
from multiprocessing.managers import SharedMemoryManager
from multiprocessing.shared_memory import SharedMemory
from typing import Tuple

import numpy as np
from pyqtgraph.Qt import QtCore
import pyqtgraph as pg


def create_np_array_from_shared_mem(
    shared_mem: SharedMemory,
    shared_data_dtype: np.dtype,
    shared_data_shape: Tuple[int, ...],
) -> np.ndarray:
    arr = np.frombuffer(shared_mem.buf, dtype=shared_data_dtype)
    arr = arr.reshape(shared_data_shape)
    return arr


# This function is responsible for displaying the data
# it is run in its own process to liberate main process
def display(
    shared_mem: SharedMemory,
    shared_data_dtype: np.dtype,
    shared_data_shape: Tuple[int, ...],
):
    app = pg.mkQApp("Multiprocess plotter")

    arr = create_np_array_from_shared_mem(
        shared_mem, shared_data_dtype, shared_data_shape
    )

    win2 = pg.GraphicsLayoutWidget(title="Basic plotting examples")
    win2.resize(1000, 600)
    win2.setWindowTitle('pyqtgraph example: Plotting')
    p2 = win2.addPlot(title="Updating plot")
    curve1 = p2.plot(pen='y')
    curve2 = p2.plot(pen='b')

    def updateInProc():
        curve1.setData(arr[:, 0, 0], arr[:, 1, 0])
        curve2.setData(arr[:, 0, 1], arr[:, 1, 1])

    win2.show()
    timer = QtCore.QTimer()
    timer.timeout.connect(updateInProc)
    timer.start(50)
    pg.exec()


def make_data1(
    shared_mem: SharedMemory,
    shared_data_dtype: np.dtype,
    shared_data_shape: Tuple[int, ...],
):
    arr = create_np_array_from_shared_mem(
        shared_mem, shared_data_dtype, shared_data_shape
    )
    for i in range(1000):
        t = i / 100
        s = np.sin(2 * np.pi * t)
        arr[i, 0, 0] = t
        arr[i, 1, 0] = s
        time.sleep(0.01)
    print("Done")


def make_data2(
    shared_mem: SharedMemory,
    shared_data_dtype: np.dtype,
    shared_data_shape: Tuple[int, ...],
):
    arr = create_np_array_from_shared_mem(
        shared_mem, shared_data_dtype, shared_data_shape
    )
    for i in range(1000):
        t = i / 100
        s = np.sin(2 * np.pi * t + np.pi)
        arr[i, 0, 1] = t
        arr[i, 1, 1] = s
        time.sleep(0.01)
    print("Done")


if __name__ == '__main__':
    data_to_share = np.zeros((1000, 2, 2))

    SHARED_DATA_DTYPE = data_to_share.dtype
    SHARED_DATA_SHAPE = data_to_share.shape
    SHARED_DATA_NBYTES = data_to_share.nbytes

    with SharedMemoryManager() as smm:
        shared_mem = smm.SharedMemory(size=SHARED_DATA_NBYTES)

        writer1 = Process(
            target=make_data1, args=(shared_mem, SHARED_DATA_DTYPE, SHARED_DATA_SHAPE)
        )
        writer2 = Process(
            target=make_data2, args=(shared_mem, SHARED_DATA_DTYPE, SHARED_DATA_SHAPE)
        )
        reader = Process(
            target=display, args=(shared_mem, SHARED_DATA_DTYPE, SHARED_DATA_SHAPE)
        )
        writer1.start()
        writer2.start()
        reader.start()
        writer1.join()
        writer2.join()
        reader.join()