リアルタイム描画したい(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は、リアルタイムにグラフを更新するための強力なツールです。
しかし、複雑なダッシュボードを構築する場合には、データ管理やスレッド設計など工夫が必要になります。
運用を重視したダッシュボードを作成する場合には、
streamlitやdash、gradioなどウェブベースのフレームワークも検討してみるとよいです。
また、textualでターミナルベースのダッシュボードを作成することもできます。