🐳

リアルタイム可視化でラグを解消した話 - スライディングウィンドウ最適化テクニック

に公開
バッチ処理
リアルタイム
センサー
iq
tech

背景

ミリ波センサーを使ったバイタルサインモニタリングシステムを開発していた際、リアルタイム処理で深刻な問題に直面しました。高フレームレートでデータを取得した際、画面表示が数秒遅れてしまい、実用に耐えないシステムになっていたのです。

この問題は致命的です。

この記事で説明すること

本記事では、リアルタイム信号処理におけるラグ問題を解決するための実践的な手法を解説します：

  1. スライディングウィンドウ方式の基礎理論

    • 窓幅とオーバーラップの考え方
    • 定常性vs非定常性のトレードオフ

  2. ラグ発生の3つの主要原因

    • フレームレートの問題
    • 表示更新の重い処理
    • 可視化ライブラリの性能差

  3. 段階的解決アプローチ

    • 最も効果的な「表示間引き処理」
    • PyQtGraphによる劇的な高速化
    • 実装時の注意点とトラブルシューティング

実際にリアルタイム処理で数秒のラグを50ms以下に短縮した経験を基に、具体的で実用的な解決策を提示します。

スライディングウィンドウ方式の基礎

スライディングウィンドウとは

スライディングウィンドウ方式は、連続的なデータストリームを固定サイズの「窓（ウィンドウ）」で切り取りながら処理する手法です。新しいデータが到着するたびに、窓を少しずつスライドさせて解析を継続します。

from collections import deque
import numpy as np

class SlidingWindowProcessor:
    def __init__(self, window_size=50, overlap_ratio=0.5):
        self.window_size = window_size
        self.overlap_size = int(window_size * overlap_ratio)
        self.step_size = window_size - self.overlap_size
        
        # 高速なバッファ（dequeを使用）
        self.buffer = deque(maxlen=window_size)
    
    def process_new_sample(self, sample):
        self.buffer.append(sample)
        
        # ウィンドウが満杯になったら処理実行
        if len(self.buffer) == self.window_size:
            window_data = np.array(list(self.buffer))
            
            # 信号処理（FFT、移動平均など）
            result = self._analyze_window(window_data)
            
            # オーバーラップ処理：step_size分だけ削除
            for _ in range(self.step_size):
                self.buffer.popleft()
            
            return result
        return None
    
    def _analyze_window(self, data):
        # 実際の解析処理
        fft_result = np.fft.rfft(data)
        peak_freq = np.argmax(np.abs(fft_result[1:])) + 1
        return {'peak_frequency': peak_freq, 'mean': np.mean(data)}

ウィンドウサイズの選択とトレードオフ

スライディングウィンドウ方式における最も重要な設計パラメータはウィンドウサイズです。

リアルタイム処理でのラグ問題と解決策

リアルタイム処理システムでは、以下の要因でラグが発生します：

  1. 処理時間の累積: フレーム処理時間 > フレーム間隔
  2. グラフ描画オーバーヘッド: GUI更新の重い処理
  3. メモリ管理: バッファの非効率な操作
  4. ライブラリ性能: 可視化ライブラリの描画速度

解決策1: フレームレート調整

フレームレートを下げることで処理負荷を軽減できますが、バイタルサイン解析では精度が犠牲になるため推奨しません。

import time
from collections import deque

class FrameRateController:
    def __init__(self, target_fps=20):
        self.target_fps = target_fps
        self.frame_interval = 1.0 / target_fps
        self.last_frame_time = time.time()
    
    def wait_for_next_frame(self):
        """フレームレート制御"""
        elapsed = time.time() - self.last_frame_time
        if elapsed < self.frame_interval:
            time.sleep(self.frame_interval - elapsed)
        self.last_frame_time = time.time()

# フレームレート vs 解析精度
# 20Hz: 呼吸解析に最適（ナイキスト定理: 2 × 最大呼吸周波数）
# 10Hz: 精度低下（高周波成分の見逃し）
# 5Hz:  実用困難（エイリアシング発生）

解決策2: 表示間引き処理

最も効果的で実装が簡単な解決策は、表示更新を間引くことです。

class OptimizedProcessor:
    def __init__(self, display_interval=4):
        """
        Args:
            display_interval: 4なら4フレームに1回表示更新
        """
        self.display_interval = display_interval
        self.frame_count = 0
    
    def process_frame(self, new_data):
        # === データ処理（毎フレーム実行） ===
        analysis_result = self._analyze_data(new_data)  # 必須処理
        
        # === 表示更新判定===
        should_update_display = (self.frame_count % self.display_interval == 0)
        
        display_data = None
        if should_update_display:
            display_data = self._prepare_display(analysis_result)  # 重い処理
        
        self.frame_count += 1
        
        return {
            'analysis_result': analysis_result,
            'display_data': display_data,
            'should_update': should_update_display
        }
    
    def _analyze_data(self, data):
        """軽量な解析処理（毎フレーム実行）"""
        return {'mean': np.mean(data), 'peak': np.max(data)}
    
    def _prepare_display(self, result):
        """重い表示準備処理（間引きして実行）"""
        # グラフデータの準備、FFT計算など
        return result

効果:

  • フレームレート20Hz、間引き4フレーム → 有効表示レート5Hz
  • 処理負荷: 約75%削減
  • 体感ラグ: ほぼゼロ（200ms以下）

解決策3: PyQtGraphによる高速可視化

PyQtGraphの基本設定

import pyqtgraph as pg
from PyQt5 import QtWidgets, QtCore
import numpy as np

# 高速化設定
pg.setConfigOptions(
    antialias=False,    # アンチエイリアス無効（2-3倍高速化）
    useOpenGL=True      # OpenGL使用（5-10倍高速化）
)

class FastRealtimeWidget(QtWidgets.QMainWindow):
    def __init__(self):
        super().__init__()
        
        # プロットウィジェット作成
        self.plot_widget = pg.PlotWidget(title='高速リアルタイムプロット')
        self.setCentralWidget(self.plot_widget)
        
        # カーブオブジェクト（再利用で高速化）
        self.curve = self.plot_widget.plot(pen=pg.mkPen('cyan', width=2))
        
        # タイマー（リアルタイム更新用）
        self.timer = QtCore.QTimer()
        self.timer.timeout.connect(self.update_plot)
        self.timer.start(50)  # 20 FPS
    
    def update_plot(self):
        """高速プロット更新"""
        # データ生成（実際にはセンサーデータ）
        x = np.linspace(0, 10, 100)
        y = np.sin(x + time.time())
        
        # setDataは非常に高速（既存カーブの再利用）
        self.curve.setData(x, y)

# 使用例
app = QtWidgets.QApplication([])
window = FastRealtimeWidget()
window.show()
app.exec()

PyQtGraph使用上の注意点

よくあるエラーと対処法

# エラー1: OpenGL関連エラー
"""
RuntimeError: OpenGL context could not be created
"""
# 対処法
pg.setConfigOptions(useOpenGL=False)  # OpenGL無効化

# エラー2: Qt バージョン競合
# 対処法: 自動検出とフォールバック
try:
    from PyQt5 import QtWidgets, QtCore
except ImportError:
    from PyQt6 import QtWidgets, QtCore

# エラー3: メモリリーク
# 対処法: 適切なバッファ管理
self.data_buffer = deque(maxlen=1000)  # maxlenで自動制限

# エラー4: GUI フリーズ
# 対処法: 処理時間制限
def safe_update(self):
    start_time = time.time()
    # 重い処理
    if time.time() - start_time > 0.016:  # 16ms制限
        return  # 処理を中断

実装時

  • カーブの再利用: plot()ではなくsetData()を使用
  • OpenGL有効化: 可能な限り有効にして高速化
  • バッファサイズ制限: メモリリーク防止
  • エラーハンドリング: 堅牢性の確保
  • 処理時間測定: 性能ボトルネックの特定

開発時の注意点：バッチ処理からリアルタイム処理への段階的移行

実際のリアルタイム信号処理システム開発において最も重要なポイントは、いきなりリアルタイム処理から始めないことです。アルゴリズムの検証と性能最適化を同時に行うのは骨が折れます（ポキッ）。

段階的開発アプローチ

Phase 1: バッチ処理による基盤構築

まず最初に、CSVファイルなどに保存されたデータを使ったバッチ処理版を作成します。この段階では、リアルタイム性を一切考慮せず、アルゴリズムの正確性と再現性の確保に集中します。

バッチ処理版では、センサーデータをCSVファイルに保存し、それを読み込んで一括解析を行います。この方法の最大の利点は、全く同じデータに対して何度でも同じ処理を実行できることです。バグの原因特定、アルゴリズムの改良、パラメータの調整など、すべてが確実に再現可能な環境で行えます。

また、バッチ処理では処理時間を気にする必要がないため、デバッグ出力を豊富に含めたり、詳細な中間結果を保存したりできます。これにより、アルゴリズムの動作を深く理解し、問題点を特定することが容易になります。

Phase 2: パラメータ最適化

バッチ処理版が完成したら、次にパラメータの最適化を行います。スライディングウィンドウのサイズ、オーバーラップ率、フィルタの設定値など、システムの性能を左右する重要なパラメータを体系的に調整します。

バッチ処理環境では、異なるパラメータセットでの一括テストが効率的に実行できます。例えば、ウィンドウサイズを30、50、100フレームで比較したり、オーバーラップ率を0.5、0.75、0.9で検証したりといった網羅的な評価が可能です。

この段階では、複数の患者データや様々なノイズ環境でのテストデータを用意し、それぞれでの解析精度を定量的に評価します。

Phase 3: リアルタイム処理への移行

パラメータが最適化されて、バッチ処理での動作が完全に検証された後、ようやくリアルタイム処理版の開発に移行します。この段階では、アルゴリズム自体は一切変更せず、バッチ版で使用していた解析ロジックをそのまま継承します。

リアルタイム版で新たに追加するのは、フレームバッファの管理、タイミング制御、表示更新の最適化といった、純粋にリアルタイム処理に特化した機能のみです。コアとなる信号解析アルゴリズムは既に検証済みなので、安心して性能最適化に集中できます。

この段階で初めて、表示間引き処理やPyQtGraphによる高速化といった、この記事で説明した最適化手法を適用します。アルゴリズムの品質は保証されているため、性能向上の効果が正確に測定でき、問題が発生した場合の原因も特定しやすくなります。

実装時のワークフロー

実際の開発では、まずセンサーから取得したデータをCSVファイルとして複数パターン保存します。正常な呼吸パターン、異常なパターン、ノイズが多い環境でのデータなど、様々な条件でのテストデータセットを構築します。

次に、これらのCSVデータを使ってバッチ処理版を開発し、各テストデータでの解析結果を検証します。結果はすべてCSVファイルとして保存します。

パラメータ最適化では、性能評価指標を明確に定義し、グリッドサーチやベイズ最適化といった体系的手法を適用します。最適なパラメータが決定されたら、JSON形式の設定ファイルとして保存し、バージョン管理システムで管理します。

最後のリアルタイム移行では、バッチ版の設定ファイルを読み込み、同じパラメータを使用することで、バッチ処理での検証結果がそのままリアルタイム版に継承されることを保証します。

このような段階的アプローチにより、アルゴリズムの品質を犠牲にすることなく、効率的にリアルタイム性能の最適化を実現できます。

まとめ

リアルタイム信号処理におけるラグ解消のポイントを整理します：

解決策の効果比較

  1. 第一段階: 表示間引き処理の実装（最もコスパが良い）
  2. 第二段階: PyQtGraphへの移行（劇的な性能向上）
  3. 最適化: スレッド化とメモリ管理の改善

Discussion