我一直在搜索stackoverflow，以寻找在将数据流式传输到交互式mpl图形时的解决方案。我已经接近了下面的示例，但是当我绘制这个时，无论我对代码进行什么修改，我都无法与图形窗口进行交互。我缺少什么概念可以让我将数据流式传输到此窗口并拖动窗口？如您所见，它会打开并开始绘图，但是如果我抓住框架将其移动到不同的位置，它会崩溃并且大部分时间都卡在了原地。
我正在使用 Python 2.7、Paycharm 和 Windows 10。
这是我正在使用的一个有用的示例。 来自文档的示例
问题是因为我使用的是 plt.show() 吗？我的测试代码由 3 个文件组成，一个数据生成器、一个数据使用者（绘图仪）和一个顶级文件（测试台），用于实例化和启动数据生成器和绘图模块。我只是将 62 字节的正弦波数据附加到数组的末尾并绘制它，因此它看起来像是在滚动。
测试台：NB_DataGen -> NB_Plotter（接收 62 字节的数据和绘图）。
模块 1：数据绘图模块
# This is the no blitting data plot module built out as a threaded module.
#
#
# Notes:
# 1. Bug in frame rate code
# 2. Going to try to remove queue from plotter and just have a direct access call for
# direct writes to plot. Queue seems to be bogging down window and I can't drag it
# around.
#
try:
import Queue as queue
except:
import queue
import numpy as np
from matplotlib import pyplot as plt
import time
import threading
import matplotlib
print(matplotlib.__version__)
class BlitManager:
def __init__(self, canvas, animated_artists=()):
"""
Parameters
----------
canvas : FigureCanvasAgg
The canvas to work with, this only works for sub-classes of the Agg
canvas which have the `~FigureCanvasAgg.copy_from_bbox` and
`~FigureCanvasAgg.restore_region` methods.
animated_artists : Iterable[Artist]
List of the artists to manage
"""
self.canvas = canvas
self._bg = None
self._artists = []
for a in animated_artists:
self.add_artist(a)
# grab the background on every draw
self.cid = canvas.mpl_connect("draw_event", self.on_draw)
def on_draw(self, event):
"""Callback to register with 'draw_event'."""
cv = self.canvas
if event is not None:
if event.canvas != cv:
raise RuntimeError
self._bg = cv.copy_from_bbox(cv.figure.bbox)
self._draw_animated()
def add_artist(self, art):
"""
Add an artist to be managed.
Parameters
----------
art : Artist
The artist to be added. Will be set to 'animated' (just
to be safe). *art* must be in the figure associated with
the canvas this class is managing.
"""
if art.figure != self.canvas.figure:
raise RuntimeError
art.set_animated(True)
self._artists.append(art)
def _draw_animated(self):
"""Draw all of the animated artists."""
fig = self.canvas.figure
for a in self._artists:
fig.draw_artist(a)
def update(self):
"""Update the screen with animated artists."""
cv = self.canvas
fig = cv.figure
# paranoia in case we missed the draw event,
if self._bg is None:
self.on_draw(None)
else:
# restore the background
cv.restore_region(self._bg)
# draw all of the animated artists
self._draw_animated()
# update the GUI state
cv.blit(fig.bbox)
# let the GUI event loop process anything it has to do
cv.flush_events()
#
# Main Class
#
class NB_Plotter4(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.i = 0
# thread loop flag
self.thread_event = threading.Event()
self.thread_event.set() # set thread by default
# create plot objects
self.fig = plt.figure()
self.ax1 = self.fig.add_subplot(1,1,1)
self.ax1.grid()
self.line, = self.ax1.plot([], lw=3)
self.text = self.ax1.text(0.8, 0.5, "")
self.x_new = np.linspace(0.0, 200.0, num=1000)
self.y_total = np.empty(1000, dtype=float)
#set limits
self.ax1.set_xlim(self.x_new.min(), self.x_new.max())
self.ax1.set_ylim([-1.1, 1.1])
# start timer for frame counter
self.t_start = time.time()
#self.bm = BlitManager(self.fig.canvas, [self.line, self.text])
self.bm = BlitManager(self.fig.canvas, [self.line])
plt.show(block=False)
plt.pause(0.1)
#
# main thread loop
#
def Write(self, data):
# need to grab y-data here from queue
self.y_ndata = data
self.y_total = np.concatenate([self.y_total[62:], self.y_ndata])
self.i = self.i + 1
#
# Over-ride thread run method
#
def run(self):
while self.thread_event.is_set():
self.line.set_data(self.x_new, self.y_total)
tx = 'Mean Frame Rate:\n {fps:.5f}FPS'.format(fps=((self.i + 1) / (time.time() - self.t_start)))
self.text.set_text(tx)
self.bm.update()
模块 2：数据生成模块
# This is the no blitting data gen module. This module is intended to produce data
# in 62 byte blocks resulting in sine wave data blocks. This is initally meant to
# spoof my DR500 USB payload size so I can' drive some real time plotting.
#
#
# Notes:
#
#
#
try:
import Queue as queue
except:
import queue
import numpy as np
import threading
import time
#
# Main Class
#
# For the 62 byte class the rounding in the x vector produces some small errors. This shouldn't
# be a big problem since the resolution is so high.
#
class NB_DataGen2(threading.Thread):
def __init__(self, Plotter):
threading.Thread.__init__(self)
self.y_data = np.empty(62, dtype=float)
self.x_data = np.linspace(0.0, np.pi/62.0, num=62)
self.offset_val = self.x_data[1]
self.Plotter_Handle = Plotter
self.inc_cnt = 0.0
self.first_it_flag = 0 # first iteration flag
# thread loop flag
self.thread_event = threading.Event()
self.thread_event.set() # set thread by default
#
# Produce 62 byte packet of sine wave data
# - produce next 62 byte chunk of sine wave data
def Get62ByteSine(self, debug=False):
# hit for iterations > 0
if(self.first_it_flag > 0):
# gen
self.x_data = self.x_data + (np.pi / 62.0) + self.offset_val
self.y_data = np.sin(self.x_data)
if(debug == True):
print(self.y_data)
return self.y_data
# hit for iterations < 1 -> (first iteration)
else:
# first iteration
self.x_data = self.x_data
self.y_data = np.sin(self.x_data)
if (debug == True):
print(self.y_data)
self.inc_cnt = self.inc_cnt + 1.0
# flip first run flag
self.first_it_flag = 1
return self.y_data
#
# Ignore / Not in use
#
# Used to check the error from the last value in one block (62 byte block)
# and the first value of the next block. the difference in the two should
# match the offset value roughly. Enable print funcitons in the above
# Get62ByteSine function for this to work.
#
def CheckError(self):
self.Get62ByteSine(True)
self.Get62ByteSine(True)
self.Get62ByteSine(True)
# print offset
print(self.offset_val)
#
# Kill thread
#
def KillThread(self):
self.thread_event.clear()
#
# main thread loop
#
def run(self):
while self.thread_event.is_set():
self.Plotter_Handle.Write(self.Get62ByteSine())
time.sleep(1)
模块 3：顶级测试台
# This is the no blitting test bench top module
#
#
# Notes:
#
#
#
from NB_DataGen2 import NB_DataGen2
from NB_Plotter4 import NB_Plotter4
#
# Testbench Class
#
class NB_TestBench(object):
def __init__(self):
# create data/plot objects (DUTs) - obj's under test
self.Plotter = NB_Plotter4()
self.DataGen = NB_DataGen2(self.Plotter)
def Start(self):
self.DataGen.start()
self.DataGen.isDaemon()
self.Plotter.start()
self.Plotter.isDaemon()
# Run test bench
NB = NB_TestBench()
NB.Start()
长话短说 - 我正在尝试运行此代码来绘制传入数据并能够拖动窗口或通常通过鼠标与之交互。有谁看到我哪里出错了？