概要

前回記事でローパスフィルターを使用した進捗予測を行いました。
しかし実際にはいくつか問題があります。

不均一な更新時間

前回のローパスフィルター計算では、均一な更新間隔を前提としたパラメータalphaを設定しました。
更新間隔を考慮しない式なので、更新回数が多いものほど強く影響します。

実際には処理時間が不均一なので、処理時間が短いものほど更新回数が多く強く影響し、処理時間が長いものほど更新回数が少なくなるので、弱い影響力となります。

早い話が、処理時間の長いものは反映されずらく、処理時間が短く見積もられがちということです。

解決策

処理時間に応じて重み（という表現が正しいかわかりませんが）を変える必要があります。
係数alphaを処理時間に応じて変えてみましょう。

alphaの換算

処理時間の間隔dtに応じて、alphaの値を変形（換算）してあげましょう。

\alpha(dt) = 1 - \left(1 - \alpha_{\mathrm{static}}\right)^{\,dt}

pythonで計算すると
alpha_dt = 1 - (1 - alpha_static) ** dt
となります。

ローパスフィルターの計算式

y[k] = \alpha\, x[k] + (1-\alpha)\, y[k-1]

pythonで計算すると
smoothed_speed = alpha * speed + (1 - alpha) * smoothed_speed
ですね。これは前回記事と同じになります。

さきほどのalpha換算式を組み合わせると、以下のようになります。
ここでalpha_staticが元のalpha値、alphaが換算後のalpha値です。

alpha = 1 - (1 - alpha_static) ** dt
smoothed_speed = alpha * speed + (1 - alpha) * smoothed_speed

時定数

ついでにalphaだと感覚的に設定ができないので、今回から時定数τを導入してみます。
alphaからの換算は以下の式です。

\tau = -\frac{\Delta t}{\ln(1-\alpha)}

pythonで計算すると
tau = -delta_t / math.log(1 - alpha)
ですね。
今回は時定数τを秒単位で扱いたいので、delta_t=1とします。

下図は時定数τ=1でのグラフです。
あるステップ入力を与えたとき、1秒後に63.2%に到達します。
さらにその2.3倍の時間(=2.3秒)で約90%に到達します。

時定数を30秒に設定すれば、90%に到達するのは69秒後になります。
これはalpha=0.032に相当します。
時定数のほうが直感的でわかりやすいですね。

話をまとめると、
「時定数はalphaと相互に換算可能なパラメータで、時間単位で指定できるもの」
と考えてください。

時定数でのローパスフィルターの計算式

さきほどの時定数→alpha換算と、処理時間dtに応じた換算を一気に計算します。

\alpha(dt) = 1 - \exp\left(-\frac{dt}{\tau_{\mathrm{static}}}\right)

pythonだとこうですね。
alpha = 1 - math.exp(-dt / tau_static)
tau_staticは時定数になります。

alpha = 1 - math.exp(-dt / tau_static)
smoothed_speed = alpha * speed + (1 - alpha) * smoothed_speed

あるいはワンライナーで以下のようにも書けます。

smoothed_speed = (1 - math.exp(-dt/tau)) * speed + math.exp(-dt/tau) * smoothed_speed

プログラム実装

display_progress.py

import time
import math

class ProgressTracker:
    """
    処理の進捗を管理し、進行状況（進捗、処理速度、ETA）を表示するクラス。
    """
    def __init__(self, total_num, tau=5):
        """
        コンストラクタ

        :param total_num: 処理する総数
        :param tau: LPFの時定数 秒

        alpha（平滑化係数）:
        alpha が 大きい → 最新の値（limited_speed）を重視する。
        alpha が 小さい → 過去の値（smoothed_speed）を重視する。
        """
        self.total_num = total_num
        self.progress_count = 0
        self.start_time = time.time()
        self.previous_time = self.start_time  # 前回の更新時刻
        self.smoothed_speed = None  # LPFで平滑化された速度
        self.current_speed = 0
        self.average_speed = 0

        self.task_speeds = []  # 各タスクの処理速度
        self.weights = []      # 各タスク処理時間

        self.eta_seconds  = 0 # ETA 残り時間_秒
        self.diff_time = 0 # 差分時間_秒
        self.total_time = 0 # かかった時間_秒
        self.last_value = 0.0

        self.tau = tau #時定数_秒

    def update_smooth_value(self,dt,speed):
        """
        時間差 dt と時定数 τ を使って新たなフィルタ値を計算する。

        :param dt: 最終計測時刻と現在時刻の差（秒）
        :param speed: 現在の速度などの値
        :return: 更新されたフィルタ値
        """
        alpha = 1 - math.exp(-dt / self.tau)
        new_value = alpha * speed + (1 - alpha) * self.last_value
        self.last_value = new_value

        return new_value

    def update_progress(self):
        """
        進捗を1つ更新し、現在の進捗、速度、ETAを表示します。
        """
        self.progress_count += 1

        current_time = time.time()
        self.elapsed_time = current_time - self.start_time     # 開始してからの経過時間
        self.diff_time = current_time - self.previous_time  # 前回の更新からの経過時間
        self.previous_time = current_time  # 現在の時間を記録



        # 差分で速度を計算
        self.current_speed = 1 / self.diff_time if self.diff_time > 0 else 0  # 処理１件あたりの速度

        # LPFを使用して速度を平滑化
        if self.smoothed_speed is None:
            self.smoothed_speed = self.current_speed
        else:
            self.smoothed_speed = self.update_smooth_value(self.diff_time,self.current_speed)

        self.task_speeds.append(self.diff_time)
        self.weights.append(self.smoothed_speed)

        self.average_speed = self.smoothed_speed

        # 残りの処理数とETA（予測完了時間）
        remaining = self.total_num - self.progress_count
        self.eta_seconds = remaining / self.average_speed if self.average_speed > 0 else -1

        self.display_progress()

    def display_progress(self):
        """
        現在の進捗、速度、ETAを計算して表示します。
        """
        formatted_eta = self.format_time(self.eta_seconds)

        # 進捗率と速度のフォーマット
        progress_percentage = (self.progress_count / self.total_num) * 100
        formatted_speed = f"{self.average_speed:.2f} ops/sec"
        formatted_raw_speed = f"{self.current_speed:.2f} ops/sec"

        #プログレスバー
        bar_length = 40
        filled_length = int(bar_length * self.progress_count / self.total_num)
        bar = "█" * filled_length + "-" * (bar_length - filled_length)

        # プログレスを1行ですべて表示
        print(
            f"\rProgress: {self.progress_count}/{self.total_num} ({progress_percentage:.1f}%) | "
            f"speed: {formatted_raw_speed} | Avg speed: {formatted_speed} | ETA: {formatted_eta} ",
            f"[{bar}]",
            end='',
            flush=True
        )

        if self.progress_count >= self.total_num:
            # 処理完了後にトータル時間を表示
            print(f"\n処理完了！ トータル時間: {self.total_elapsed_time()}")


    @staticmethod
    def format_time(seconds):
        """
        秒を時:分:秒の形式にフォーマットします。

        :param seconds: 秒数
        :return: フォーマットされた時間
        """
        if seconds < 0:
            return "N/A"
        hours, remainder = divmod(int(seconds), 3600)
        minutes, seconds = divmod(remainder, 60)
        return f"{hours:02}:{minutes:02}:{seconds:02}"

    def total_elapsed_time(self):
        """
        処理開始から現在までの経過時間を計算しフォーマットする。
        """
        self.total_time = time.time() - self.start_time
        return self.format_time(self.total_time)


# サンプル使用例
if __name__ == "__main__":
    total_operations = 100  # 処理の総数
    tracker = ProgressTracker(total_operations, tau=3)  # LPFと速度リミットの設定

    # サンプル処理（処理ごとに時間がかかる処理を模倣）
    for i in range(total_operations):
        if i > 20:
            time.sleep(0.1 + (i * 0.02))  # 処理時間が徐々に増加
        else:
            time.sleep(0.001)  # 序盤は速い
        tracker.update_progress()

比較

前回記事と比較してみます。

緑が実際の経過時間、青が表示された推定ETAです。

1.差分+LPF(前回手法） alpha=0.3

前回の解説での方法です。

2.差分+LPF(今回手法）時定数2.8秒（alpha=0.3相当）

あまり変わったようには見えませんが、60step,70stepの高速処理の影響がほとんど出ていませんね。
また最初の20stepまではフィルタの立ち上がりの遅さが原因で、非常に長い時間が表示されています。

おわりに

数時間、数千stepの処理を行う場合は時定数tauを100秒などの極端な値にしたほうが予測が安定すると思います。
しかしあまりtauを大きくしすぎると、予測の立ち上がりが遅くなります。
全体の10%を処理するまでは時定数を1/10にする。みたいに可変させても良さそうですね。

これでひとまずはローパスフィルターを使った進捗表示の解説は終了です。

それではまた。