python - 如何使用 matplotlib.animation 从动态增长的列表中连续流式传输数据点以进行实时绘图？

Question

我想使用 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'])

提前致谢。

Answer 1

我没有 Twitter key 来测试你的完整代码，但是这里有一个代码示例，你可以修改它来制作动画情节。这种方法的缺点是您需要终止进程才能停止动画。此示例绘制了平均值和原始数据，以供说明。

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