使用Matplotlib實現實時數據流可視化（animation模塊）

> 本文介紹了使用animation和pyplot模塊實現實時數據流可視化的方法 > 鑒於網上這方面資料很少，做一記錄供大家學習

先說一下自己的需求：為辣雞項目所迫，有一硬件產生實時數據流，需要采集並動態展示數據變化規律，幀數在20-50幀

一.性能較差的方法

開始我是不知道有animation這個神器的，就用set_xdata/set_ydata更新數據，pause刷新圖像

pltx = np.arange(0, 400, 10)
plty = [0 for length in range(0, 40)]
plt.ion()
fig = plt.figure()
ax = fig.add_subplot(111)
line, = ax .plot(pltx, plty)

def update()
    line.set_ydata(plty)
    plt.pause(0.001)

功能是實現了，但沒想到效率極其坑爹，8組數據+20幀的配置就拽不動了，有明顯的滯后。鬼知道它這個pause是怎么實現的。。
下面是正解:

二.FuncAnimation類

• 實現高效動畫的核心就是FuncAnimation類（需要導入matplotlib.animation），提供用於更新數據的update和獲取幀數的frames即可

看一下官方API文檔的參數解釋：

FuncAnimation(fig, func, frames=None, init_func=None, fargs=None, save_count=None, **kwargs)

fig : matplotlib.figure.Figure
The figure object that is used to get draw, resize, and any other needed events.

func : callable
The function to call at each frame. The first argument will be the next value in frames. Any additional positional arguments can be supplied via the fargs parameter.

frames : iterable, int, generator function, or None, optional
Source of data to pass func and each frame of the animation
If an iterable, then simply use the values provided. If the iterable has a length, it will override the save_count kwarg.
If an integer, then equivalent to passing range(frames)
If None, then equivalent to passing itertools.count.
In all of these cases, the values in frames is simply passed through to the user-supplied func and thus can be of any type.

init_func : callable, optional
A function used to draw a clear frame. If not given, the results of drawing from the first item in the frames sequence will be used. This function will be called once before the first frame.
If blit == True, init_func must return an iterable of artists to be re-drawn.
The required signature is:
def init_func() -> iterable_of_artists:

fargs : tuple or None, optional
Additional arguments to pass to each call to func.

save_count : int, optional
The number of values from frames to cache.

interval : number, optional
Delay between frames in milliseconds. Defaults to 200.

repeat_delay : number, optional
If the animation in repeated, adds a delay in milliseconds before repeating the animation. Defaults to None.

repeat : bool, optional
Controls whether the animation should repeat when the sequence of frames is completed. Defaults to True.

blit : bool, optional
Controls whether blitting is used to optimize drawing. Defaults to False.

說幾個常用參數

• update：

update函數接收一個frame參數作為當前幀數，可以根據幀數更新數據，也可以把幀數作為x計算y的值（函數動畫）
返回更新數據后的plot對象

• frames：

frames為每一幀調用的update函數提供幀數，可以是整數、可迭代對象或生產器函數，視具體情況而定

• interval：

刷新時間間隔，以ms為單位

• repeat:

控制當幀序列迭代完畢后是否重新迭代，默認為True

三.例子1

官方例子，貝葉斯函數動畫：

import math

import numpy as np
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation


def beta_pdf(x, a, b):
    return (x**(a-1) * (1-x)**(b-1) * math.gamma(a + b)
            / (math.gamma(a) * math.gamma(b)))


class UpdateDist(object):
    def __init__(self, ax, prob=0.5):
        self.success = 0
        self.prob = prob
        self.line, = ax.plot([], [], 'k-')
        self.x = np.linspace(0, 1, 200)
        self.ax = ax

        # Set up plot parameters
        self.ax.set_xlim(0, 1)
        self.ax.set_ylim(0, 15)
        self.ax.grid(True)

        # This vertical line represents the theoretical value, to
        # which the plotted distribution should converge.
        self.ax.axvline(prob, linestyle='--', color='black')

    def init(self):
        self.success = 0
        self.line.set_data([], [])
        return self.line,

    def __call__(self, i):
        # This way the plot can continuously run and we just keep
        # watching new realizations of the process
        if i == 0:
            return self.init()

        # Choose success based on exceed a threshold with a uniform pick
        if np.random.rand(1,) < self.prob:
            self.success += 1
        y = beta_pdf(self.x, self.success + 1, (i - self.success) + 1)
        self.line.set_data(self.x, y)
        return self.line,

# Fixing random state for reproducibility
np.random.seed(19680801)


fig, ax = plt.subplots()
ud = UpdateDist(ax, prob=0.7)
anim = FuncAnimation(fig, ud, frames=np.arange(100), init_func=ud.init,
                     interval=5, blit=True)
plt.show()

• ud對象被調用時根據i計算貝葉斯函數值
• frames是包含0~99的numpy數組，同時也是一個可迭代對象

效果如圖：

四.例子2

我的程序需要在每次新數據包發來時更新圖像，但更新數據的時間是不可預知的

• 因此，將幀間隔時間設置到一個小於數據包間隔時間的值（我設為30ms），並在update函數中判斷當前數據包是不是新的，如果是則更新圖像
• 這里給frames傳了個生成器函數進去，不斷檢測數據包是否更新，並將結果返回給update函數

關鍵代碼：

def update(frame):
    if (frame == 0):
        return line1
    del plty[0]
    plty.append((outputData[listPos - 1][3]))
    return line

def gen_function():
    global listPos
    lastPos = 0
    while (1):
        if (lastPos != listPos):
            lastPos = listPos
            yield 1
        else:
            yield 0

if (__name__ == "__main__"):
    pyplotInit()
    ani = animation.FuncAnimation(fig, update, frames=gen_function, interval=30, blit=True)
    plt.show()