Pythonジェネレータの基本と使い方: 遅延評価で省メモリ処理

# 0, 1, 4, 9, ... と平方数を必要な分だけ順次生成するジェネレータ
def squares(limit):
    # limit個ぶんの平方数を遅延生成
    for n in range(limit):
        # 値を1つ返して関数の状態を一時停止
        yield n * n

# 使い方例: forループで消費
for v in squares(5):
    print(v)

# return版: すべての結果を一括で返す(メモリ消費が大きい)
def build_squares_list(limit):
    # 全要素をリストに格納してから返す
    return [n * n for n in range(limit)]

# yield版: 必要になったときに1つずつ返す(省メモリ)
def gen_squares(limit):
    for n in range(limit):
        # 値を1つずつ遅延生成
        yield n * n

# 使い方比較
print("return版:", build_squares_list(5))  # すぐにリストができる
print("yield版:", list(gen_squares(5)))    # 消費するときに生成される

def count_up():
    print("start")       # 最初のnext()でここから実行
    yield 1              # 値1を返して停止
    print("resume A")    # 次のnext()でここから再開
    yield 2
    print("resume B")
    yield 3
    print("end")         # 次のnext()でStopIteration

g = count_up()
print(next(g))  # 最初の値: 1
print(next(g))  # 次の値: 2
print(next(g))  # 次の値: 3

# 次を取り出そうとすると要素が尽きてStopIteration
try:
    print(next(g))
except StopIteration:
    print("StopIteration発生")

def gen123():
    yield 1
    yield 2
    yield 3

# forは内部でStopIterationを処理する
for x in gen123():
    print(x)
print("forループ終了")  # 例外処理は不要

# 0〜4の平方を遅延生成するジェネレータ式
g = (x * x for x in range(5))

# 1つずつ取り出す
print(next(g))
print(next(g))

# 残りを合計(消費しながら計算)
print(sum(g))

import sys

N = 100_000

# リスト内包表記: すぐに10万要素のリストを作る
lst = [i * 2 for i in range(N)]
# ジェネレータ式: 必要時に1つずつ取り出す
gen = (i * 2 for i in range(N))

print("listの概算サイズ(bytes):", sys.getsizeof(lst))
print("generatorの概算サイズ(bytes):", sys.getsizeof(gen))

# 両者の結果は同じだが、作り方が異なる
print("listの最初の3要素:", lst[:3])

# generatorはスライスできないため、nextで取り出す
g_copy = (i * 2 for i in range(3))
print("generatorの最初の3要素:", [next(g_copy) for _ in range(3)])

# 1〜20のうち偶数だけ2乗し、そのうち100以上のものを合計
nums = range(1, 21)
evens = (n for n in nums if n % 2 == 0)              # フィルタ
squares = (n * n for n in evens)                     # マップ
large = (x for x in squares if x >= 100)             # さらにフィルタ
print(sum(large))  # 全体を展開せず合計

# map/filterを使う場合(どちらも遅延)
from math import sqrt
roots = map(sqrt, filter(lambda x: x % 3 == 0, range(1, 50)))
# 最初の5個だけ確認
import itertools
print(list(itertools.islice(roots, 5)))

import io

def read_lines(stream):
    # 行末の改行を取り除きながら1行ずつ返す
    for line in stream:
        yield line.rstrip("\n")

# デモ用の「大きな」テキスト(実際は小さい)
data = "\n".join(f"line {i}" for i in range(1, 8))
fake_file = io.StringIO(data)

# 最初の3行だけ処理して途中で打ち切る例
for i, line in enumerate(read_lines(fake_file), start=1):
    print(f"{i}: {line}")
    if i == 3:
        break

import io

def read_lines(stream):
    # 入力を1行ずつ読み出す
    for line in stream:
        yield line.rstrip("\n")

def parse_csv(lines, sep=","):
    # "a,b,c" -> ["a","b","c"] のように分割
    for line in lines:
        yield [part.strip() for part in line.split(sep)]

def filter_status(rows, status="ERROR"):
    # 2列目(インデックス1)がERRORの行のみ通す
    for row in rows:
        if len(row) >= 2 and row[1] == status:
            yield row

def extract_message(rows):
    # 3列目(インデックス2)のメッセージだけ取り出す
    for row in rows:
        if len(row) >= 3:
            yield row[2]

# デモ用CSV風データ: "timestamp,level,message"
raw = "\n".join([
    "2024-01-01,INFO,Started",
    "2024-01-01,ERROR,Dropped connection",
    "2024-01-01,INFO,Heartbeat",
    "2024-01-01,ERROR,Timeout",
])
fake_csv = io.StringIO(raw)

# パイプライン: read -> parse -> filter(ERROR) -> extract message
messages = extract_message(filter_status(parse_csv(read_lines(fake_csv))))
for m in messages:
    print(m)

import itertools

def naturals(start=1):
    # 無限に自然数を生成
    n = start
    while True:
        yield n
        n += 1

# 最初の10個だけ取り出す
first_ten = list(itertools.islice(naturals(), 10))
print(first_ten)

# フィボナッチの無限列を例示
def fib():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

print(list(itertools.islice(fib(), 8)))

def resourceful():
    print("open resource")
    try:
        for i in range(5):
            yield i
    finally:
        # g.close()やガベージコレクション時に呼ばれる
        print("cleanup resource")

g = resourceful()
print(next(g))   # 0を取得
print(next(g))   # 1を取得
g.close()        # ここで後始末が走る

# close後にnextするとStopIteration
try:
    print(next(g))
except StopIteration:
    print("closed")

# コンテキストマネージャと併用する安全なパターン
def read_numbers(path):
    # ファイルを安全に開いて1行ずつ整数化して返す
    with open(path, "rt", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                # 整数に変換できない場合はスキップ
                try:
                    yield int(line)
                except ValueError:
                    # ログ出力などに切り替え可
                    continue

# デモ環境ではファイルを用意できないため、この関数の使い方だけ示します。
print("read_numbersは遅延で整数を返し、ファイルは自動クローズされます。")