Pythonでプログレスバーを自作する(tqdm/tqdm.notebook/tqdm.auto/tqdm.pandas 比較)

tqdm.tqdm・tqdm.notebook・tqdm.auto・tqdm.pandas・tqdm.contrib.concurrent.process_map と自作プログレスバーを徹底比較。キャリッジリターンの基礎からETA計算、ANSIカラー、マルチバー、`bar_format` カスタム書式、nested bar(`leave=False`/`position`)、pandas の `df.progress_apply`、`rich.progress`/`alive-progress` との使い分けまで体系的に整理。長時間計算(ベイズ最適化・GA・SA・モンテカルロ)の可視化に直結。

なぜプログレスバーを自作するのか

Pythonでプログレスバーといえば tqdm が定番ですが、自作することにもメリットがあります。

  • 外部依存をゼロにできる(pip install 不要)
  • ターミナル制御の仕組みを理解できる
  • 表示フォーマットを自由にカスタマイズできる

この記事では、キャリッジリターンの基礎から始めて、ETA表示・ANSIカラー・マルチバーまで段階的にプログレスバーを実装していきます。

キャリッジリターンの基本

プログレスバーの核心は、ターミナルの同じ行を繰り返し上書きすることです。これには キャリッジリターン \r を使います。

\r はカーソルを行の先頭に戻す制御文字です。改行せずに出力すれば、前の内容を上書きできます。

import time

for i in range(101):
    print(f"\r処理中... {i}%", end="", flush=True)
    time.sleep(0.05)
print()  # 最後に改行

ポイントは3つです。

  • \r を文字列の先頭に置く
  • end="" で改行を抑制する
  • flush=True でバッファを即時フラッシュする(これがないと表示が更新されないことがあります)

基本的なプログレスバー

数値だけでは味気ないので、バーを描画してみます。[████████░░░░░░░░] 50% (50/100) のような形式を目指します。

import sys
import time

def progress_bar(current, total, bar_length=30):
    fraction = current / total
    filled = int(bar_length * fraction)
    bar = "█" * filled + "░" * (bar_length - filled)
    percent = fraction * 100
    sys.stdout.write(f"\r[{bar}] {percent:5.1f}% ({current}/{total})")
    sys.stdout.flush()

# 使用例
total = 100
for i in range(total + 1):
    progress_bar(i, total)
    time.sleep(0.03)
print()

実行すると、以下のように表示が更新されていきます。

[█████████████████░░░░░░░░░░░░░] 56.0% (56/100)

sys.stdout.write を使っている理由は、print と違って余計な改行やスペースが入らないためです。

ETA(残り時間)の計算

長い処理では「あとどれくらいかかるのか」が知りたくなります。経過時間から残り時間を推定しましょう。

import sys
import time

def format_time(seconds):
    """秒数を HH:MM:SS 形式にフォーマットする"""
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = int(seconds % 60)
    return f"{h:02d}:{m:02d}:{s:02d}"

def progress_bar_eta(current, total, start_time, bar_length=30):
    fraction = current / total
    filled = int(bar_length * fraction)
    bar = "█" * filled + "░" * (bar_length - filled)
    percent = fraction * 100

    elapsed = time.time() - start_time
    if current > 0:
        rate = current / elapsed          # 1秒あたりの処理数
        remaining = (total - current) / rate
        eta_str = format_time(remaining)
        rate_str = f"{rate:.1f} it/s"
    else:
        eta_str = "--:--:--"
        rate_str = "-- it/s"

    sys.stdout.write(
        f"\r[{bar}] {percent:5.1f}% | ETA: {eta_str} | {rate_str}"
    )
    sys.stdout.flush()

# 使用例
total = 200
start = time.time()
for i in range(total + 1):
    progress_bar_eta(i, total, start)
    time.sleep(0.02)
print()

出力例:

[███████████████░░░░░░░░░░░░░░░] 50.0% | ETA: 00:00:02 | 49.8 it/s

current == 0 のときはゼロ除算を避けるため、ETAをプレースホルダにしています。

ANSIカラー対応

ターミナルは ANSIエスケープシーケンス を使って文字に色をつけることができます。プログレスの進捗に応じてバーの色を変えてみましょう。

主なカラーコードは以下の通りです。

コード
\033[91m赤(明るい)
\033[93m黄(明るい)
\033[92m緑(明るい)
\033[0mリセット
import os
import sys
import time

def get_color(fraction):
    """進捗率に応じて色を返す(赤→黄→緑)"""
    if fraction < 0.33:
        return "\033[91m"  # 赤
    elif fraction < 0.66:
        return "\033[93m"  # 黄
    else:
        return "\033[92m"  # 緑

RESET = "\033[0m"

def progress_bar_color(current, total, start_time):
    try:
        terminal_width = os.get_terminal_size().columns
    except OSError:
        terminal_width = 80

    # バー以外の部分の文字数を概算して、バーの長さを決定
    suffix = f" {current/total*100:5.1f}% | ETA: 00:00:00 | 000.0 it/s"
    bar_length = max(10, terminal_width - len(suffix) - 4)  # [] と余白分

    fraction = current / total
    filled = int(bar_length * fraction)
    color = get_color(fraction)
    bar = color + "█" * filled + RESET + "░" * (bar_length - filled)

    elapsed = time.time() - start_time
    if current > 0:
        rate = current / elapsed
        remaining = (total - current) / rate
        eta_str = format_time(remaining)
        rate_str = f"{rate:.1f} it/s"
    else:
        eta_str = "--:--:--"
        rate_str = "-- it/s"

    percent = fraction * 100
    sys.stdout.write(f"\r[{bar}] {percent:5.1f}% | ETA: {eta_str} | {rate_str}")
    sys.stdout.flush()

# format_time は前のセクションと同じ

total = 150
start = time.time()
for i in range(total + 1):
    progress_bar_color(i, total, start)
    time.sleep(0.02)
print()

os.get_terminal_size() でターミナルの幅を取得し、バーの長さを動的に調整しています。ウィンドウをリサイズしても表示が崩れにくくなります。

マルチバーの実装

複数のタスクを同時に進める場合、バーを縦に並べて表示したくなります。これには ANSIカーソル移動 を使います。

  • \033[{n}A : カーソルをn行上に移動
  • \033[{n}B : カーソルをn行下に移動

仕組みはシンプルです。複数行を出力した後、カーソルを先頭まで戻して上書きします。

import sys
import time
import random

def format_time(seconds):
    h = int(seconds // 3600)
    m = int((seconds % 3600) // 60)
    s = int(seconds % 60)
    return f"{h:02d}:{m:02d}:{s:02d}"

def get_color(fraction):
    if fraction < 0.33:
        return "\033[91m"
    elif fraction < 0.66:
        return "\033[93m"
    else:
        return "\033[92m"

RESET = "\033[0m"

def render_bar(label, current, total, start_time, bar_length=25):
    """1本のバーの文字列を生成して返す"""
    fraction = current / total if total > 0 else 0
    filled = int(bar_length * fraction)
    color = get_color(fraction)
    bar = color + "█" * filled + RESET + "░" * (bar_length - filled)

    elapsed = time.time() - start_time
    if current > 0:
        rate = current / elapsed
        remaining = (total - current) / rate
        eta_str = format_time(remaining)
    else:
        eta_str = "--:--:--"

    percent = fraction * 100
    return f"{label}: [{bar}] {percent:5.1f}% ({current}/{total}) ETA: {eta_str}"

def multi_progress(tasks):
    """複数タスクのプログレスバーを表示する"""
    n = len(tasks)
    start_times = [time.time() for _ in range(n)]
    progress = [0] * n
    totals = [t["total"] for t in tasks]
    labels = [t["label"] for t in tasks]

    # 初回表示: 空行を確保
    for i in range(n):
        print(render_bar(labels[i], 0, totals[i], start_times[i]))

    while any(progress[i] < totals[i] for i in range(n)):
        # カーソルをn行上に戻す
        sys.stdout.write(f"\033[{n}A")

        for i in range(n):
            if progress[i] < totals[i]:
                # タスクごとに異なるスピードで進む
                step = random.randint(1, 3)
                progress[i] = min(progress[i] + step, totals[i])
            line = render_bar(labels[i], progress[i], totals[i], start_times[i])
            # 行末まで消去してから改行(前の表示のゴミを消す)
            sys.stdout.write(f"\r{line}\033[K\n")

        sys.stdout.flush()
        time.sleep(0.1)

# 使用例
tasks = [
    {"label": "Download ", "total": 100},
    {"label": "Extract  ", "total": 80},
    {"label": "Install  ", "total": 120},
]
multi_progress(tasks)

実行すると、ターミナル上で以下のような表示になります。

マルチプログレスバーの表示例

\033[K は行末までを消去するエスケープシーケンスです。バーの長さが前回より短くなった場合に、ゴミ文字が残るのを防ぎます。

まとめ:ProgressBarクラス

ここまでの要素を1つのクラスにまとめます。

import os
import sys
import time

class ProgressBar:
    def __init__(self, total, label="Progress", bar_length=None, color=True):
        self.total = total
        self.label = label
        self.color = color
        self.current = 0
        self.start_time = None

        if bar_length is None:
            try:
                self.bar_length = max(10, os.get_terminal_size().columns - 60)
            except OSError:
                self.bar_length = 30
        else:
            self.bar_length = bar_length

    def _get_color(self, fraction):
        if not self.color:
            return ""
        if fraction < 0.33:
            return "\033[91m"
        elif fraction < 0.66:
            return "\033[93m"
        return "\033[92m"

    def _format_time(self, seconds):
        h = int(seconds // 3600)
        m = int((seconds % 3600) // 60)
        s = int(seconds % 60)
        return f"{h:02d}:{m:02d}:{s:02d}"

    def update(self, n=1):
        if self.start_time is None:
            self.start_time = time.time()

        self.current = min(self.current + n, self.total)
        fraction = self.current / self.total
        filled = int(self.bar_length * fraction)
        color = self._get_color(fraction)
        reset = "\033[0m" if self.color else ""
        bar = color + "█" * filled + reset + "░" * (self.bar_length - filled)

        elapsed = time.time() - self.start_time
        if self.current > 0:
            rate = self.current / elapsed
            remaining = (self.total - self.current) / rate
            eta_str = self._format_time(remaining)
            rate_str = f"{rate:.1f} it/s"
        else:
            eta_str = "--:--:--"
            rate_str = "-- it/s"

        percent = fraction * 100
        line = f"\r{self.label}: [{bar}] {percent:5.1f}% | ETA: {eta_str} | {rate_str}"
        sys.stdout.write(line + "\033[K")
        sys.stdout.flush()

    def finish(self):
        self.update(0)
        print()

# 使用例
bar = ProgressBar(total=100, label="Training")
for i in range(100):
    time.sleep(0.03)
    bar.update()
bar.finish()

出力例:

Training: [██████████████████████████████] 100.0% | ETA: 00:00:00 | 32.8 it/s

このクラスは約60行で、tqdm の基本機能をカバーしています。必要に応じてコンテキストマネージャ(__enter__ / __exit__)やイテラブルのラッパーを追加すれば、さらに便利に使えます。

tqdm 実戦パターン:tqdm.tqdm / tqdm.notebook / tqdm.auto

自作はターミナル制御の理解には最適ですが、本番コードでは tqdm を素直に使うほうが安全です。tqdm には実行環境ごとに最適な実装が複数用意されています。

# 標準: ターミナル向け
from tqdm import tqdm
for x in tqdm(range(1000), desc="train"):
    ...

# Jupyter Notebook 向け(HTML ベースのバー)
from tqdm.notebook import tqdm as tqdm_nb
for x in tqdm_nb(range(1000), desc="epoch"):
    ...

# 環境を自動判定(CLI/Notebook/IPython のどれでも最適化)
from tqdm.auto import tqdm

ライブラリやスクリプトを書く側であれば from tqdm.auto import tqdm を採用するのが安全です。Jupyter 実行時は tqdm.notebook 相当、ターミナル実行時は標準実装に自動で切り替わります。

bar_format でカスタム書式を作る

tqdmbar_format 引数は Python の str.format スタイルで表示テンプレートを完全に制御できます。実験ログを構造化したいときに特に有用です。

from tqdm import tqdm

fmt = "{l_bar}{bar:30}{r_bar} | loss={postfix[0]:.4f}"
with tqdm(range(200), bar_format=fmt, postfix=[0.0]) as pbar:
    for step in pbar:
        loss = 1.0 / (step + 1)
        pbar.postfix[0] = loss
        pbar.update(0)  # 表示だけ更新

主なプレースホルダ:

プレースホルダ内容
{l_bar}説明 + パーセンテージ
{bar:N}幅 N のバー本体
{r_bar}カウント + 経過 + ETA + レート
{n_fmt} / {total_fmt}現在値 / 合計(K, M フォーマット可)
{rate_fmt}「12.5it/s」「800ms/it」など
{elapsed} / {remaining}経過時間 / 残り時間
{postfix}set_postfix(loss=0.12, acc=0.98) の内容

set_postfix を使うと、学習中の指標を tqdm の右側にリアルタイム表示できます。

for x in (pbar := tqdm(range(N))):
    pbar.set_postfix(loss=f"{loss:.3f}", acc=f"{acc:.3f}")

nested bar:leave=False と position

巣状ループ(外側=エポック、内側=バッチ)は positionleave=False の組み合わせが定番です。

from tqdm import tqdm

for epoch in tqdm(range(EPOCHS), desc="epoch", position=0):
    for batch in tqdm(range(N_BATCH), desc="batch",
                      position=1, leave=False):
        ...

leave=False を内側に指定すると、完了時にその行が消えるため画面が散らかりません。マルチプロセスの場合は tqdm.set_lock(RLock()) でロックを共有し、position を被らせないようにすると行衝突を防げます。

tqdm.contrib.concurrent.process_map:並列処理に直結

並列計算の進捗表示は本記事で扱う他の長時間計算(ベイズ最適化・遺伝的アルゴリズム・シミュレーテッドアニーリング・モンテカルロ)と組み合わせると効きます。tqdm.contrib.concurrentProcessPoolExecutor を包む高水準 API を提供します。

from tqdm.contrib.concurrent import process_map, thread_map

def heavy(seed):
    # 重い計算(例:1試行のモンテカルロ)
    ...
    return result

# CPU バウンド(プロセス並列)
results = process_map(heavy, range(10_000), max_workers=8, chunksize=20)

# I/O バウンド(スレッド並列)
results = thread_map(fetch, urls, max_workers=32)

chunksize を 1 にすると進捗の粒度は細かいですが overhead が増えます。1 万件規模なら chunksize=20〜100 が現実的なバランスです。

pandas 連携:df.progress_apply

tqdm.pandas() を一度呼ぶだけで、DataFrame.apply / groupby.apply / Series.mapprogress_apply / progress_map メソッドが生えます。

import pandas as pd
from tqdm import tqdm
tqdm.pandas(desc="feature")

df["price_log"] = df["price"].progress_apply(lambda x: math.log1p(x))

df.groupby("user_id").progress_apply(extract_features)

データ加工パイプラインで「いつ終わるのか」が読めない状況を一行で解消できるので、データ前処理スクリプトには常に入れておくと便利です。

tqdm vs rich.progress vs alive-progress 比較

最近は rich 系の代替も普及しています。用途別の使い分けを表にまとめます。

観点tqdmrich.progressalive-progress
依存追加サイズ軽量(純 Python、依存ほぼなし)中程度(rich 一式が必要)軽量
マルチバーposition 手動管理Progress コンテキストでネイティブ対応スピナー併用が容易
カラーリングANSI 直書きで自由スタイル DSL([bold red] 等)でリッチアニメーション豊富
Jupyter 対応tqdm.notebook で HTML 描画rich.jupyter で対応限定的(CLI 推奨)
pandas 連携tqdm.pandas() で簡単公式連携なし(手動ラップ)なし
並列処理ヘルパtqdm.contrib.concurrentProgress.add_task を手動更新なし
ログとの混在注意が必要(行が乱れがち)Console.log と統合可補助 print あり
用途ML/データ処理の標準TUI を作る・凝った CLI単一処理の楽しい可視化

実務での目安:

  • 大量データの ETL や ML 学習tqdm.auto + tqdm.pandas + process_map
  • 配信ツール・自作 CLI で見た目を整えたいrich.progress
  • 個人プロジェクトで遊び心が欲しいalive-progress

関連記事

参考