如何实时更新 matplotlib 图并拥有 REPL？

Question

我现在的情况是，我有一个带有一些简单命令的原始 REPL。这些命令生成数据并将数据放入线程安全队列中。然后我有一个函数从队列中获取数据并将其绘制在 matplotlib 图中。

我最终想要做的是能够有一个 REPL 命令来创建一个线程，该线程在启动时生成数据并将数据放入队列中。然后，数据从队列到达时将被绘制成 matplotlib 绘图。我无法理解如何实现这一点。

我的第一个想法是在主线程中将 REPL 实现为 while 循环，然后让 matplotlib 的函数在单独的线程中运行。不幸的是，matplotlib 似乎需要在应用程序的主线程中运行。所以我无法追求那个。

我尝试的下一件事是将 REPL 放入其自己的线程中，并在主线程中保留一个 matplotlib 函数。这是我的 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()

REPL 由一个在单独线程中运行的简单 while 循环组成。我感觉这是一个问题。是否可以使用在单独线程中从键盘输入获取数据的输入语句？

¯\_(ツ)_/¯

，但看起来好像有效。

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")  

最后，我有了条形图，它从队列中获取数据并绘制它。该函数在程序的主线程中运行。

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)

当我运行它时，它主要完成我想要的事情。 matplotlib 窗口出现，我可以使用 REPL 命令

fake_data -n 100

向其中添加数据。问题是 matplotlib 窗口的图标在任务栏中不断闪烁（这是 Windows 10）。 这让我觉得有些不对劲。

这是我在 REPL 中发出的命令...

当我尝试使用另一个线程生成的更多数据不断更新队列时，我会遇到更多问题。这需要更多的解释，所以我只是发布一个关于这个更简化的情况的问题，看看这是否是正确的路径。

这种做法正确吗？ 为什么 matplotlib 的图标不断闪烁？ 像我在自己的线程中所做的那样实现 REPL 可以吗？

Answer 1

使用 IPython，您可以加载脚本、parallelize Matplotlib（使用

get_ipython().run_line_magic('matplotlib', '')

），并拥有一个 REPL，其中您在脚本中定义的所有内容（例如，

t

、

pi

和

sin

）都已准备就绪交互使用。

这是一个简单的例子，您更复杂的情况可以用同样的方式处理。

$ 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)? 
$