リアルタイム描画したい(matplotlib.animation.FuncAnimation

 1import matplotlib.pyplot as plt
 2import numpy as np
 3from matplotlib.animation import FuncAnimation
 4
 5# データを準備する
 6x = np.linspace(0, 2 * np.pi, 100)
 7y = np.sin(x)
 8
 9# キャンバスを作成
10fig, ax = plt.subplots()
11xdata = []
12ydata = []
13line, = ax.plot(x, y)  # 初期の線を描画
14
15# アニメーションの更新関数
16def update(frame):
17    xdata.append(frame)
18    ydata.append(np.random.rand())  # ランダムな値を追加
19
20    line.set_data(xdata, ydata)  # 線のデータを更新
21    ax.relim()  # 軸の範囲を再計算
22    ax.autoscale_view()  # 軸の範囲を自動調整
23
24    return line,  # 更新された線を返す
25
26# アニメーションを作成
27ani = FuncAnimation(
28    fig,          # アニメーションを描画するFigureオブジェクト
29    update,       # 更新関数
30    frames=100,   # フレーム数
31    interval=50,  # フレーム間の時間(ミリ秒)
32    blit=True     # 描画の最適化
33)
34
35# アニメーションを表示
36plt.show()

FuncAnimationを使って、リアルタイムでグラフを更新できます。 更新関数を定義し、フレームごとに呼び出されるようにします。 フレーム数や更新間隔を指定して、アニメーションの速度や長さを調整できます。 オプションでblit=Trueを指定すると、描画の最適化が行われ、パフォーマンスが向上します。

実用的なリアルタイムモニターしたい

  1from collections import deque
  2import threading
  3import time
  4import numpy as np
  5import matplotlib.pyplot as plt
  6from matplotlib.animation import FuncAnimation
  7
  8# データバッファ
  9class DataBuffer:
 10    def __init__(self, maxlen=100):
 11        self.x = deque(maxlen=maxlen)  # 時間軸のデータを保持
 12        self.y = deque(maxlen=maxlen)  # 値のデータを保持
 13        self.lock = threading.Lock()   # スレッドセーフなロック
 14
 15    # データを追加するメソッド
 16    def push(self, x, y):
 17        with self.lock:
 18            self.x.append(x)
 19            self.y.append(y)
 20
 21    # データのスナップショットを取得するメソッド
 22    def snapshot(self):
 23        with self.lock:
 24            return self.x, self.y
 25
 26# データ生成スレッド
 27def data_generator(buffer: DataBuffer):
 28    t = 0  # 時間の初期値
 29    while True:
 30        value = np.random.rand()  # ランダムな値を生成
 31        buffer.push(t, value)
 32        t += 0.1
 33        time.sleep(0.1)  # データ生成の間隔
 34
 35def dashboard(buffer: DataBuffer):
 36
 37    # レイアウトを定義
 38    panels = [
 39        ["main", "main", "side"],
 40        ["main", "main", "side"],
 41    ]
 42
 43    # キャンバスを作成
 44    fig, axs = plt.subplot_mosaic(
 45        panels,    # 割付を指定
 46        figsize=(8, 4),   # 横長のキャンバスを作成
 47        layout="constrained",  # レイアウト調整
 48    )
 49
 50    # main: メインのグラフを初期化
 51    line, = axs['main'].plot([], [], lw=2, label="Random Data")
 52    axs['main'].set_xlim(0, 20)
 53    axs['main'].set_ylim(0, 1)
 54    axs['main'].legend()
 55
 56    # side: サイドパネルにテキストを表示
 57    txt = axs["side"].text(
 58        0.5,
 59        0.5,
 60        "",
 61        ha="center",
 62        va="center",
 63    )
 64    axs["side"].axis("off")
 65
 66    def update(frame):
 67        x, y = buffer.snapshot()
 68
 69        if len(x) == 0:
 70            return line, txt
 71
 72        # deque -> list に変換してから描画
 73        x = list(x)
 74        y = list(y)
 75
 76        # グラフを更新
 77        line.set_data(x, y)
 78
 79        # 最新の20秒分を表示
 80        axs["main"].set_xlim(max(0, x[-1] - 20), x[-1])
 81
 82        # サイドパネルのテキストを更新
 83        txt.set_text(f"Latest Value: {y[-1]:.2f}")
 84
 85        return line, txt
 86
 87    ani = FuncAnimation(
 88        fig,
 89        update,
 90        interval=100,  # 描画の更新間隔
 91    )
 92
 93    plt.show()
 94
 95if __name__ == "__main__":
 96    # データバッファを作成
 97    buffer = DataBuffer(maxlen=200)
 98
 99    # データ生成スレッドを開始
100    t = threading.Thread(
101        target=data_generator,
102        args=(buffer,),
103        daemon=True,  # メインスレッドが終了したら自動的に終了する
104    )
105    t.start()
106
107    # ダッシュボードを開始
108    dashboard(buffer)

実用的なリアルタイムモニターを作成する場合には、 データを取得する処理と、グラフを更新する処理は分けて設計する必要があります。

上記のサンプルでは、データ取得とグラフ更新の橋渡しをするために、 スレッドセーフなデータバッファクラス(DataBuffer)を定義しています。 このクラスではdequeを使って、一定数のデータを保持し、スレッド間で安全にデータをやり取りできるようにしています。

グラフ更新に遅延が生じても、データ取得には影響せず、リアルタイムモニターとしての機能が保たれます。

注釈

matplotlib.animation.FuncAnimationは、リアルタイムにグラフを更新するための強力なツールです。 しかし、複雑なダッシュボードを構築する場合には、データ管理やスレッド設計など工夫が必要になります。

運用を重視したダッシュボードを作成する場合には、 streamlitdashgradioなどウェブベースのフレームワークも検討してみるとよいです。 また、textualでターミナルベースのダッシュボードを作成することもできます。