How do I update a matplotlib plot in realtime and also have a REPL?

I am in a situation where I have a primitive REPL with some simple commands. These commands generate data and put the data into a threadsafe queue. I then have a function that takes the data from the queue and plots it on a matplotlib plot.

What I would ultimately like to do is to be able to have a REPL command that creates a thread that generates data upon starting and puts the data into the queue. The data would then be plotted as it arrives from the queue into a matplotlib plot. I am having trouble understanding how to implement this.

My first idea was to just implement the REPL as while loop in the main thread and then have the function with matplotlib run in a separate thread. Unfortunately, matplotlib seems to require running in the main thread of the application. So I could not pursue that.

The next thing I tried was to put the REPL into its own thread and keep a matplotlib function in the main thread. Here's my main.py:

import threading
from queue import Queue
from typing import List

from repl import REPL
from stripchart import live_stripchart


def main():

    queue :Queue[List[float]] = Queue(maxsize = 0)
    quit: threading.Event = threading.Event()

    # A primitive REPL to control application. It runs
    # in a separate thread. And puts data into the queue.
    repl = REPL(queue, quit)
    repl.start()

    # live_stripchart is a function that plots data from the queue.
    # It runs in the main thread (because matplotlib seems to require this?)
    live_stripchart(queue, quit) # 
    repl.join()
    

if __name__ == "__main__":
    main()

The REPL consists of a simple while loop that runs in a separate thread. I feel like this is a problem. Is it OK to use input statements that take data from keyboard input in a separate thread? ¯\_(ツ)_/¯, but it seems like it works.

import re
import threading
from queue import Queue
from typing import List
import random

fake_data_cmd_regex :re.Pattern = re.compile(r'^fake_data\s+\-n\s*(\d+)\s*$')


class REPL(threading.Thread):

    def __init__(self, queue :Queue[List[float]], quit :threading.Event):
        self.queue :Queue[List[float]] = queue
        self.quit = quit
        super().__init__()


    def run(self):
        try:
            # Here's a primitive REPL to contol the application
            while True:
                command = input('> ')

                if command == 'help':
                    print("quit")
                    print("    Exit the application.")
                    print("fake_data -n <number>")
                    print("    Generate some data and put it in queue.")
                    continue

                if command == 'quit':
                    self.quit.set() # fire quit event to stop theads that check it.
                    break           # exit this loop, terminating this thread.

                match = fake_data_cmd_regex.match(command)
                if match:
                    n, = match.groups()
                    print(f"generating {n} fake data points...")
                    self.queue.put([random.random() for _ in range(int(n))])
                    continue 

                if command == '':
                    continue
                
                print("Unknown command. Type 'help' for help.")

        except KeyboardInterrupt:
            self.quit.set() # stop well monitor if it's running

    
        print(f"cli done")  

Finally, I have the stripchart, this takes the data from the queue and plots it. This function runs in the main thread of the program.

import matplotlib.pyplot as plt
import threading
from queue import Queue, Empty
from typing import List

def live_stripchart(queue: Queue[List[float]], quit: threading.Event):

    plt.ion()
    fig, ax = plt.subplots()
    x, y = [], []
    i = 0

    line, = ax.plot([], [], 'b-')  # Initialize an empty plot line

    while not quit.is_set():
        try:
            # grab data from the queue
            data = queue.get(block=True, timeout=0.5)

            # Update x and y
            x.extend(range(i, i + len(data)))
            y.extend(data)
            i += len(data)

            # Keep only the most recent 100 points
            if len(x) > 100:
                x = x[-100:]
                y = y[-100:]

            # autoscale the view
            line.set_data(x, y)
            ax.relim()
            ax.autoscale_view()

            # actually plot it
            plt.draw()
            plt.pause(0.1)
        except Empty:
            plt.pause(0.1) # gracefully handle timeout

    print("plot_ui done", flush=True)
    plt.close(fig)

When I run this, it mostly does what I want. The matplotlib window appears and I can add data to it with my REPL command fake_data -n 100. The problem is that the icon for the matplotlib window keeps blinking in the taskbar (this is windows 10). This makes me think something is wrong.

Here's the commands I issue in the REPL...

I get even more problems when I try to continuously update the queue with more data generated from yet another thread. That's more involved to explain, so I am just posting a question about this more simplified situation to see if this is even the right path.

Is this approach correct? Why is the icon for matplotlib blinking continuously? Is it OK to implement a REPL as I have done in it's own thread?

Using IPython you can load your script, parallelize Matplotlib (using get_ipython().run_line_magic('matplotlib', '')), and have a REPL where everything you defined in your script (e.g., t, pi and sin) is ready to be used interactively.

Here is a simple example, your more complex case may be treaed in the same way.

$ cat one.py
get_ipython().run_line_magic('matplotlib', '')
import matplotlib.pyplot as plt
from numpy import cos, linspace, pi, sin

t = linspace(0, 2*pi, 101)

plt.plot(t/pi, cos(t))
plt.xlabel('t/π')

$ ipython3-3.12 one.py -i
Python 3.12.7 (main, Oct 01 2024, 15:35:42) [GCC]
Type 'copyright', 'credits' or 'license' for more information
IPython 8.30.0 -- An enhanced Interactive Python. Type '?' for help.
Using matplotlib backend: qtagg

In [1]: plt.plot(t/pi, sin(t))
Out[1]: [<matplotlib.lines.Line2D at 0x7f1f91e32660>]

In [2]:                                                                            
Do you really want to exit ([y]/n)? 
$