如何使用 matplotlib.animation 从动态增长的列表中连续流式传输数据点以进行实时绘图?
How to continuously stream data points from dynamically growing list using matplotlib.animation for real time graphing?
我想使用 python 的 matplotlib.animation 模块根据我的数据实时绘制一条线。我的代码的概述是我正在计算一个分数并将其存储在列表("my_average")中,这是 y 坐标。分数将始终介于 -1 和 +1 之间。此外,我希望我的 x 坐标是我的列表“(len(my_average))”的长度。例如;列表收到一个分数 x 坐标将是列表的长度,因此 1 和 y 坐标将是分数,列表收到第二个分数,列表图(1,分数 1)(2,分数 2)等。我无法显示图表并需要我的这部分代码的帮助。如果可能的话,我不想从 csv 文件中读取列表,而是直接从内存中读取列表,并且仍然能够查看历史记录中的先前数据点。
代码如下:
from tweepy.streaming import StreamListener
from tweepy import OAuthHandler
from tweepy import Stream
import json
from textblob import TextBlob
import matplotlib.pyplot as plt
import matplotlib.animation as animation
import time
import numpy
# Variables that contains the user credentials to access Twitter API
access_token =
access_token_secret =
consumer_key =
consumer_secret =
# This is a basic listener that just prints received tweets to stdout.
my_list = [] #creates empty list
my_average = []
class StdOutListener(StreamListener):
def on_data(self, data):
json_load = json.loads(data)
texts = json_load['text'] # string
#print(texts)
wiki = TextBlob(texts)
r = wiki.sentiment.polarity
my_list.append(r)
#drop zero in list
if 0 in my_list: my_list.remove(0)
print (my_list)
#calculate average
average = numpy.mean(my_list)
b = my_average.append(average)
#drop "nan" from list
if 'nan' in my_average: my_average.remove('nan')
print "average", my_average
fig = plt.figure()
ax = plt.axes(xlim=(0, 10), ylim=(0, -10))
line, = ax.plot([], [], lw=2)
def init():
line.set_data([], [])
return line,
# animation function. This is called sequentially
def animate(i):
x = (len(my_average))
y = (my_average)
line.set_data(x, y)
return line,
anim = animation.FuncAnimation(fig, animate, init_func=init,frames=100, interval=20, blit=True)
plt.show()
return True
def on_error(self, status):
print(status)
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
stream = Stream(auth, StdOutListener())
# This line filter Twitter Streams to capture data by the keywords: 'python', 'javascript', 'ruby'
stream.filter(track=['USD/CAD', 'Dollar', 'Loonie' ], languages=['en'])
提前谢谢你。
我没有 Twitter 密钥来测试你的完整代码,但是这里有一个代码示例,你可以修改它来制作动画情节。这种方法的缺点是您需要终止进程才能停止动画。此示例绘制了平均值和原始数据,以供说明。
import numpy
from pylab import *
import time
class StdOutListener():
def __init__(self):
self.start_time = time.time()
self.x = []
self.y = []
self.my_average = []
self.line_actual, = plot(self.x, self.y) # line stores a Line2D we can update
self.line_average, = plot(self.x, self.my_average) # line stores a Line2D we can update
def on_data(self, new_value):
time_delta = time.time() - self.start_time # on our x axis we store time since start
self.x.append(time_delta)
self.y.append(new_value)
self.my_average.append(numpy.mean(self.y))
self.line_actual.set_data(self.x, self.y)
self.line_average.set_data(self.x, self.my_average)
ylim([min(self.y), max(self.y)]) # update axes to fit the data
xlim([min(self.x), max(self.x)])
draw() # redraw the plot
ion() # ion() allows matplotlib to update animations.
out_listener = StdOutListener()
for i in range(1000):
out_listener.on_data(i + numpy.random.randint(-5,5))
我想使用 python 的 matplotlib.animation 模块根据我的数据实时绘制一条线。我的代码的概述是我正在计算一个分数并将其存储在列表("my_average")中,这是 y 坐标。分数将始终介于 -1 和 +1 之间。此外,我希望我的 x 坐标是我的列表“(len(my_average))”的长度。例如;列表收到一个分数 x 坐标将是列表的长度,因此 1 和 y 坐标将是分数,列表收到第二个分数,列表图(1,分数 1)(2,分数 2)等。我无法显示图表并需要我的这部分代码的帮助。如果可能的话,我不想从 csv 文件中读取列表,而是直接从内存中读取列表,并且仍然能够查看历史记录中的先前数据点。
代码如下:
from tweepy.streaming import StreamListener
from tweepy import OAuthHandler
from tweepy import Stream
import json
from textblob import TextBlob
import matplotlib.pyplot as plt
import matplotlib.animation as animation
import time
import numpy
# Variables that contains the user credentials to access Twitter API
access_token =
access_token_secret =
consumer_key =
consumer_secret =
# This is a basic listener that just prints received tweets to stdout.
my_list = [] #creates empty list
my_average = []
class StdOutListener(StreamListener):
def on_data(self, data):
json_load = json.loads(data)
texts = json_load['text'] # string
#print(texts)
wiki = TextBlob(texts)
r = wiki.sentiment.polarity
my_list.append(r)
#drop zero in list
if 0 in my_list: my_list.remove(0)
print (my_list)
#calculate average
average = numpy.mean(my_list)
b = my_average.append(average)
#drop "nan" from list
if 'nan' in my_average: my_average.remove('nan')
print "average", my_average
fig = plt.figure()
ax = plt.axes(xlim=(0, 10), ylim=(0, -10))
line, = ax.plot([], [], lw=2)
def init():
line.set_data([], [])
return line,
# animation function. This is called sequentially
def animate(i):
x = (len(my_average))
y = (my_average)
line.set_data(x, y)
return line,
anim = animation.FuncAnimation(fig, animate, init_func=init,frames=100, interval=20, blit=True)
plt.show()
return True
def on_error(self, status):
print(status)
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
stream = Stream(auth, StdOutListener())
# This line filter Twitter Streams to capture data by the keywords: 'python', 'javascript', 'ruby'
stream.filter(track=['USD/CAD', 'Dollar', 'Loonie' ], languages=['en'])
提前谢谢你。
我没有 Twitter 密钥来测试你的完整代码,但是这里有一个代码示例,你可以修改它来制作动画情节。这种方法的缺点是您需要终止进程才能停止动画。此示例绘制了平均值和原始数据,以供说明。
import numpy
from pylab import *
import time
class StdOutListener():
def __init__(self):
self.start_time = time.time()
self.x = []
self.y = []
self.my_average = []
self.line_actual, = plot(self.x, self.y) # line stores a Line2D we can update
self.line_average, = plot(self.x, self.my_average) # line stores a Line2D we can update
def on_data(self, new_value):
time_delta = time.time() - self.start_time # on our x axis we store time since start
self.x.append(time_delta)
self.y.append(new_value)
self.my_average.append(numpy.mean(self.y))
self.line_actual.set_data(self.x, self.y)
self.line_average.set_data(self.x, self.my_average)
ylim([min(self.y), max(self.y)]) # update axes to fit the data
xlim([min(self.x), max(self.x)])
draw() # redraw the plot
ion() # ion() allows matplotlib to update animations.
out_listener = StdOutListener()
for i in range(1000):
out_listener.on_data(i + numpy.random.randint(-5,5))