標準入出力のバッファリングとパイプでの挙動

# test.py
import time

for i in range(10):
    print(f"line {i}")
    time.sleep(1)

# 端末に直接出力: 1 秒ごとに表示される
python test.py

# パイプ経由: 最後にまとめて表示される
python test.py | cat

import sys

# 標準出力が端末かどうかを確認
print(sys.stdout.isatty())  # 端末なら True、パイプなら False

import sys
import time

for i in range(10):
    print(f"line {i}", flush=True)  # Python 3.3+
    time.sleep(1)

# または
for i in range(10):
    print(f"line {i}")
    sys.stdout.flush()
    time.sleep(1)

# バッファなしモードで実行
python -u test.py | cat

# 環境変数でも設定可能
PYTHONUNBUFFERED=1 python test.py | cat

import sys

# stdout は行バッファリング（端末）またはフルバッファリング（パイプ）
# stderr は常にバッファなし

print("stdout message")      # バッファリングされる可能性
print("stderr message", file=sys.stderr)  # 即座に出力

# パイプのバッファサイズを確認
cat /proc/sys/fs/pipe-max-size
# 1048576 (1MB、最大値)

# 行バッファリングに変更
stdbuf -oL python test.py | cat

# バッファなしに変更
stdbuf -o0 python test.py | cat

# -o: stdout, -e: stderr, -i: stdin
# L: 行バッファ, 0: バッファなし, サイズ指定も可

# logger.py - ログを生成するスクリプト
import time
import sys

for i in range(100):
    print(f"[{i}] Processing...", flush=True)
    time.sleep(0.1)

# リアルタイムで grep
python logger.py | grep --line-buffered "Processing"

# リアルタイムでタイムスタンプ付与
python logger.py | while IFS= read -r line; do
    echo "$(date +%H:%M:%S) $line"
done