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

import time

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

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)

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

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()

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)

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

# 標準: ターミナル向け
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 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)  # 表示だけ更新

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

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):
        ...

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)

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)

import time

N = 1_000_000

def trivial_step(i, total):
    return total + (i * i) % 97

def baseline(n):
    total = 0
    for i in range(n):
        total = trivial_step(i, total)
    return total

t0 = time.perf_counter()
baseline(N)
t1 = time.perf_counter()
print(f"baseline: {t1 - t0:.3f}s")  # -> baseline: 0.065s

from tqdm import tqdm

def run(n, batch):
    total = 0
    with tqdm(total=n, mininterval=0, miniters=1) as pbar:
        count = 0
        for i in range(n):
            total += (i * i) % 97
            count += 1
            if count >= batch:
                pbar.update(count)
                count = 0
        if count:
            pbar.update(count)
    return total

import multiprocessing as mp
import time
from tqdm import tqdm

def worker_naive(worker_id, n=40):
    # バグ：全ワーカーが position=0（デフォルト）のまま => 同じ行を奪い合う
    for i in tqdm(range(n), desc=f"worker-{worker_id}", position=0, mininterval=0, miniters=1):
        time.sleep(0.005)

if __name__ == "__main__":
    procs = [mp.Process(target=worker_naive, args=(i,)) for i in range(4)]
    for p in procs:
        p.start()
    for p in procs:
        p.join()

\rworker-0:   0%|          | 0/40 [00:00<?, ?it/s]\rworker-2:   0%|          | 0/40 [00:00<?, ?it/s]\rworker-1:   0%|          | 0/40 [00:00<?, ?it/s]\rworker-3:   0%|          | 0/40 [00:00<?, ?it/s]\rworker-0:   2%|...

import multiprocessing as mp
import time
from tqdm import tqdm

def worker_fixed(worker_id, lock, n=40):
    # 修正：position=worker_id で自分専用の行を確保し、
    # 親プロセスから渡されたロックを共有する
    tqdm.set_lock(lock)
    for i in tqdm(
        range(n), desc=f"worker-{worker_id}", position=worker_id,
        mininterval=0, miniters=1,
    ):
        time.sleep(0.005)

if __name__ == "__main__":
    lock = mp.RLock()
    procs = [mp.Process(target=worker_fixed, args=(i, lock)) for i in range(4)]
    for p in procs:
        p.start()
    for p in procs:
        p.join()

\rworker-0:   0%|          | 0/40 [00:00<?, ?it/s]\n\rworker-1:   0%|          | 0/40 [00:00<?, ?it/s]\x1b[A\n\n\n\rworker-3: ...

import sys
from tqdm import tqdm

# tqdm は自動判定してくれないので、自分で isatty() を見て明示的に disable する
is_interactive = sys.stdout.isatty()

for x in tqdm(range(1000), disable=not is_interactive, mininterval=0.5):
    ...

# あるいは disable=None にすると「ファイルにリダイレクトされたら自動的に無効化」される
for x in tqdm(range(1000), disable=None):
    ...