Matplotlib：脚本等待“input()”时无法操作绘图

Question

我试图input()在用户使用标准缩放控件操作绘图后获取用户输入。例如。用户操作绘图，计算出所需的 X 值，并将其输入到命令行提示符中。

绘图可以在单独的窗口中（Spyder/Python）或内嵌（在 Jupiter Notebook 中）。

用户输入值后，脚本继续（例如，从图中请求另一个值，或对这些值进行一些计算）。

但是，当命令行等待用户输入时，我无法让绘图实际显示并做出响应。我努力了：

• plot()先声明，input()后声明。
• 使用 Python 3.6 的 Spyder（我认为），来自 MacPorts 的源代码（尽我所能更新了 Spyder）
• Spyder 通过 Python 3.7，来自 ContinuumIO 的 Anaconda 包，采用 IPython
• Jupiter Notebook 也来自 Anaconda
• 众多后端：macosx、qt 等。
• 笔记本%matplotlib, notebook,等inline​qt
• 单独的图形窗口（Spyder 和 Python）与内联图形（Jupyter Notebook）
• fig.show( block=False )以及它的变体，例如。plt.show( block=False )
• 两台不同的 MacBook（2017 款和 2010 款 MacBook Pro）

我确实通过在和语句matplotlib.pyplot.pause(0.5)之间添加一个来实际更新绘图（以前它要么是笔记本中的空白区域，要么是空白的单独图形窗口）。这是重大进展，但是一旦脚本命中语句，我就会在图形窗口上看到一个旋转的沙滩球（防止缩放等），直到我通过输入某些内容来完成语句，然后脚本完成。那时情节是互动的。plot()input()input()input()

python 控制台似乎无法同时处理多个用户交互？IE。input()是否冻结所有其他用户交互？

我已经搜索了SO、谷歌等好几天了，但还没有弄清楚这一点！这个想法是使用它作为一种“快速而肮脏”的方式从图中获取用户输入，然后再进行理论上更复杂的任务，即直接从图中获取用户点击（这必须捕捉到绘制的数据，例如数据）光标）。

Answer 1

理论

主执行线程会阻塞用户输入，从而有效地暂停包括渲染在内的所有其他操作。您可以通过在另一个线程中进行绘图并通过队列将 UI 输入传递到该线程来缓解这种情况，以便线程永远不会阻塞并保持响应。

该文档有一个关于交互式图形的重要部分，包括ipython集成。

这里有些例子：

• 使用非阻塞图：plt.show(block=False)
• 使用matplotlib.animation
• 使用更复杂的多线程和队列（有利于集成到 UI 中）

下面的一些代码来自我的一个旧项目。

input()使用示例matplotlib.animation

更新起始x位置input()，退出q。请注意，您可以在等待用户输入时缩放和平移图。plt.show()另请注意以下中非阻塞的使用mainloop()：

import queue
import numpy as np  # just used for mocking data, not necessary
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation

fig = plt.figure()
ax = fig.add_subplot(1, 1, 1)
animation_queue = queue.Queue()
update_rate_ms = 50

xdata = np.linspace(0, 2 * np.pi, 256)
ydata = np.sin(xdata)
zdata = np.cos(xdata)

def normal_plot_stuff():
    """Some run of the mill plotting."""
    ax.set_title("Example Responsive Plot")
    ax.set_xlabel("X")
    ax.set_ylabel("Y")
    ax.plot(xdata, ydata, "C0", label="sin")
    ax.plot(xdata, zdata, "C1", label="cos")
    ax.legend(loc="lower right")

def animate(_, q):
    """Define a callback function for the matplotlib animation. 
       This reads messages from the queue 'q' to adjust the plot.
    """
    while not q.empty():
        message = q.get_nowait()
        q.task_done()
        x0 = float(message)
        ax.set_xlim([x0, x0 + 5])

def mainloop():
    """The main loop"""
    _ = FuncAnimation(fig, animate, interval=update_rate_ms, fargs=(animation_queue,))
    normal_plot_stuff()
    plt.show(block=False)
    while True:
        try:
            uinput = input("Type starting X value or 'q' to quit: ")
            if uinput == "q":
                break
            animation_queue.put_nowait(float(uinput))
        except ValueError:
            print("Please enter a valid number.")

mainloop()

UI 中嵌入实时绘图的示例

当用户在文本字段中输入时，以 X 开头的窗口和窗口大小会更新。画布matplotlib与 UI 渲染相关联以实现响应能力。

"""
Imbed a live animation into a PySimpleGUI frontend.

The animation fires on a timer callback from matplotlib and renders to
a PySimpleGUI canvas (which is really just a wrapped tk canvas).
"""

import queue
import numpy as np  # just used for mocking data, not necessary
import PySimpleGUI as sg  # used just for example
import matplotlib
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg  # used just for example

matplotlib.use("TkAgg")


fig = plt.figure()
ax = fig.add_subplot(1, 1, 1)
animation_queue = queue.Queue()
update_rate_ms = 50

xdata = np.linspace(0, 2 * np.pi, 256)
ydata = np.sin(xdata)
zdata = np.cos(xdata)


def animate(_, q):
    """Define a callback function for the matplotlib animation."""
    message = None
    while not q.empty():
        message = q.get_nowait()
        q.task_done()
    if not message:  # ignore empty UI events
        return

    ax.clear()
    if message[1]["sin"]:  # if SIN enable checkbox is checked
        ax.plot(xdata, ydata, "C0", label="sin")
        ax.legend(loc="lower right")
    if message[1]["cos"]:  # if COS enable checkbox is checked
        ax.plot(xdata, zdata, "C1", label="cos")
        ax.legend(loc="lower right")

    x0 = float(message[1]["x_start"])
    size = float(message[1]["w_size"])
    ax.set_xlim([x0, x0 + size])
    ax.set_title("Example Responsive Plot")
    ax.set_xlabel("X")
    ax.set_ylabel("Y")


layout = [
    [
        sg.Text("Start X:"),
        sg.Input(size=(5, 0), default_text=0, key="x_start"),
        sg.Text("Window Size:"),
        sg.Input(size=(10, 0), default_text=6.28, key="w_size"),
        sg.Button("Exit"),
    ],
    [
        sg.Frame(
            title="SIN",
            relief=sg.RELIEF_SUNKEN,
            layout=[
                [sg.Checkbox("Enabled", default=True, key="sin", enable_events=True)],
            ],
        ),
        sg.Frame(
            title="COS",
            relief=sg.RELIEF_SUNKEN,
            layout=[
                [sg.Checkbox("Enabled", default=True, key="cos", enable_events=True)],
            ],
        ),
    ],
    [sg.Canvas(key="-CANVAS-")],
]


def plot_setup():
    """MUST maintain this order: define animation, plt.draw(), setup
    window with finalize=True, then create, draw and pack the TkAgg
    canvas.
    """
    _ = FuncAnimation(fig, animate, interval=update_rate_ms, fargs=(animation_queue,))
    plt.draw()
    window = sg.Window(
        "Responsive Plot Example",
        layout,
        font="18",
        element_justification="center",
        finalize=True,
    )
    # tie matplotlib renderer to pySimpleGui canvas
    canvas = FigureCanvasTkAgg(fig, window["-CANVAS-"].TKCanvas)
    canvas.draw()
    canvas.get_tk_widget().pack(side="top", fill="both", expand=1)
    return window


def mainloop():
    """Main GUI loop. Reads events and sends them to a queue for processing."""
    window = plot_setup()
    while True:
        event, values = window.read(timeout=update_rate_ms)
        if event in ("Exit", None):
            break
        animation_queue.put_nowait([event, values])
    window.close()


mainloop()

实时数据流示例

window具体来说，请注意，您可以在 UI 顶部的字段中输入不同的值，并且绘图会立即更新，而不会阻塞/滞后。底部的 ADC 控件对于本示例来说毫无意义，但它们确实演示了将 UI 数据传递到绘图线程的更多方法。

"""
Imbed a live animation into a PySimpleGUI frontend, with extra plotting
and sensor control.

Live sensor data gets read from a separate thread and is converted to
PSI using calibration coefficients from a file.

The animation fires on a timer callback from matplotlib and renders to
a PySimpleGUI canvas (which is really just a wrapped tk canvas).
"""

import time
import queue
import random
import threading
from datetime import datetime
import numpy as np  # just used for mocking data, not necessary
import PySimpleGUI as sg
import matplotlib
import matplotlib.pyplot as plt
from matplotlib import animation
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg

matplotlib.use("TkAgg")


fig = plt.figure()
ax = fig.add_subplot(1, 1, 1)
animation_queue = queue.Queue()  # to pass GUI events to animation
raw_data_queue = queue.Queue()  # to pass raw data to main thread
update_rate_ms = 50  # refresh time in ms
ts, adc0, adc1 = [], [], []  # live data containers


def get_sensors(msg):
    """Return the names of the currently selected sensors from the GUI."""
    names = np.array(["A", "B", "C"])
    s0 = [msg[2], msg[3], msg[4]]  # adc0 sensor
    s1 = [msg[6], msg[7], msg[8]]  # adc1 sensor
    return (names[s0][0], names[s1][0])  # boolean index to the names


def data_collection_thread(data_queue):
    """Simulate some live streamed data that and put it on a queue."""
    t = 0
    while True:
        t += 1
        x = np.sin(np.pi * t / 112) * 12000 - 10000
        y = random.randrange(-23000, 3000)
        line = f"{t}:{x}:{y}"
        data_queue.put(line)
        time.sleep(0.001)


def process_data(data_queue, message, t, x, y):
    """Consume and process the data from the live streamed data queue."""
    while not data_queue.empty():
        line = data_queue.get()
        try:
            t0, v0, v1 = line.split(":")
            t.append(float(t0))
            x.append(float(v0))
            y.append(float(v1))
        except ValueError:
            pass  # ignore bad data
        data_queue.task_done()
    try:  # truncate to appropriate window size
        n = int(message[0])
        return t[-n:], x[-n:], y[-n:]
    except (ValueError, TypeError):
        return t, x, y  # don't truncate if there is a bad window size


# draws live plot on a timer callback
def animate(_, q):
    # get last message on event queue
    message = None
    while not q.empty():
        message = q.get_nowait()
        q.task_done()

    # plot last n datapoints
    try:
        n = int(message[1][0])  # parse window size
        adc0_window = adc0[-n:]
        adc1_window = adc1[-n:]
        ts_window = [i for i in range(len(adc0_window))]
        ax.clear()
        if message[1][1]:  # if adc0 enable checkbox is checked
            ax.plot(ts_window, adc0_window, "C0", label="adc0")
            ax.legend(loc="lower right")
        if message[1][5]:  # if adc0 enable checkbox is checked
            ax.plot(ts_window, adc1_window, "C1", label="adc1")
            ax.legend(loc="lower right")
        ax.set_title("Live Sensor Readings")
        ax.set_xlabel("Time (ms)")
        ax.set_ylabel("Pressure (psi)")

        # save displayed data
        if message[0] == "Save":
            basename = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
            plt.savefig(basename + ".png")
    except (ValueError, TypeError):
        pass  # ignore poorly formatted messages from the GUI


layout = [
    [  # row 1, some control buttons
        sg.Text("Window Size (ms):"),
        sg.Input(size=(5, 0), default_text=100),
        sg.Button("Start"),
        sg.Button("Pause"),
        sg.Button("Save"),
        sg.Button("Exit"),
    ],
    [sg.Canvas(key="-CANVAS-")],  # row 2, the animation
    [  # row 3, some frames for the ADC options
        sg.Frame(
            title="ADC 0",
            relief=sg.RELIEF_SUNKEN,
            layout=[
                [sg.Checkbox("Enabled", default=True)],
                [
                    sg.Radio("Sensor A", 1, default=True),
                    sg.Radio("Sensor B", 1),
                    sg.Radio("Sensor C", 1),
                ],
            ],
        ),
        sg.Frame(
            title="ADC 1",
            relief=sg.RELIEF_SUNKEN,
            layout=[
                [sg.Checkbox("Enabled", default=True)],
                [
                    sg.Radio("Sensor A", 2),
                    sg.Radio("Sensor B", 2, default=True),
                    sg.Radio("Sensor C", 2),
                ],
            ],
        ),
    ],
]

# MUST maintain this order: define animation, plt.draw(), setup window
# with finalize=True, then create, draw and pack the TkAgg canvas
ani = animation.FuncAnimation(
    fig, animate, interval=update_rate_ms, fargs=(animation_queue,)
)
plt.draw()  # must call plot.draw() to start the animation
window = sg.Window(
    "Read Pressure Sensors",
    layout,
    finalize=True,
    element_justification="center",
    font="18",
)

# tie matplotlib renderer to pySimpleGui canvas
canvas = FigureCanvasTkAgg(fig, window["-CANVAS-"].TKCanvas)
canvas.draw()
canvas.get_tk_widget().pack(side="top", fill="both", expand=1)

# kick off data collection thred
threading.Thread(
    target=data_collection_thread, args=(raw_data_queue,), daemon=True
).start()
data_collection_enable = True

# main event loop for GUI
while True:
    event, values = window.read(timeout=update_rate_ms)
    # check for button events
    if event in ("Exit", None):
        break
    if event == "Start":
        data_collection_enable = True
    if event == "Pause":
        data_collection_enable = False
    # send GUI events to animation
    animation_queue.put_nowait((event, values))
    # process data when not paused
    if data_collection_enable:
        ts, adc0, adc1 = process_data(raw_data_queue, values, ts, adc0, adc1)
    else:  # if paused, throw away live data
        while not raw_data_queue.empty():
            raw_data_queue.get()
            raw_data_queue.task_done()

window.close()