# Fichier: python_cheats/cheatsheets/generators.txt
# Cheatsheet Générateurs Python - Guide Complet


[OK] INTRODUCTION AUX GÉNÉRATEURS

# Les générateurs sont des itérateurs paresseux (lazy iterators)
# Ils génèrent des valeurs à la volée au lieu de les stocker en mémoire

# Avantages:
# [OK] Économie de mémoire (ne stocke qu'une valeur à la fois)
# [OK] Performance pour grandes séquences
# [OK] Peut représenter séquences infinies
# [OK] Code plus lisible et élégant
# [X] Ne peut parcourir qu'une seule fois
# [X] Pas d'accès aléatoire (pas d'indexation)
# [X] Pas de len()


[OK] CRÉER UN GÉNÉRATEUR AVEC YIELD

# Fonction générateur basique
def count_up_to(n):
    """Compte de 1 à n"""
    i = 1
    while i <= n:
        yield i
        i += 1

# Utilisation
gen = count_up_to(5)
print(next(gen))  # 1
print(next(gen))  # 2

# Parcourir avec for
for num in count_up_to(5):
    print(num)  # 1, 2, 3, 4, 5

# Différence avec return
def regular_function():
    return [1, 2, 3]  # Crée liste complète en mémoire

def generator_function():
    yield 1           # Génère valeur par valeur
    yield 2
    yield 3

# Yield multiple
def multiple_yields():
    yield 'a'
    yield 'b'
    yield 'c'
    return "Done"     # StopIteration avec valeur

gen = multiple_yields()
print(list(gen))      # ['a', 'b', 'c']


[OK] EXPRESSIONS GÉNÉRATRICES

# Similaire aux list comprehensions mais avec ()

# List comprehension (crée liste complète)
squares_list = [x**2 for x in range(10)]
print(type(squares_list))  # <class 'list'>

# Expression génératrice (lazy)
squares_gen = (x**2 for x in range(10))
print(type(squares_gen))   # <class 'generator'>

# Utilisation
for square in squares_gen:
    print(square)

# Avec conditions
even_squares = (x**2 for x in range(10) if x % 2 == 0)

# Nested
matrix_gen = ((i, j) for i in range(3) for j in range(3))

# Conversion en liste
result = list((x**2 for x in range(5)))  # [0, 1, 4, 9, 16]

# Avec fonctions
def process(x):
    return x * 2

processed = (process(x) for x in range(5))

# Dans fonction (pas besoin de double parenthèses)
sum_of_squares = sum(x**2 for x in range(10))
max_value = max(x**2 for x in range(10))


[OK] MÉTHODES DES GÉNÉRATEURS

# next() - Obtenir prochaine valeur
def counter():
    n = 0
    while True:
        yield n
        n += 1

gen = counter()
print(next(gen))        # 0
print(next(gen))        # 1
print(next(gen))        # 2

# StopIteration quand épuisé
def finite_gen():
    yield 1
    yield 2

g = finite_gen()
next(g)  # 1
next(g)  # 2
next(g)  # StopIteration

# send() - Envoyer valeur au générateur
def echo_generator():
    while True:
        received = yield
        print(f"Reçu: {received}")

gen = echo_generator()
next(gen)              # Amorcer le générateur
gen.send("Hello")      # Reçu: Hello
gen.send("World")      # Reçu: World

# send() avec retour
def accumulator():
    total = 0
    while True:
        value = yield total
        if value is not None:
            total += value

acc = accumulator()
next(acc)              # 0 (amorcer)
print(acc.send(10))    # 10
print(acc.send(5))     # 15
print(acc.send(3))     # 18

# throw() - Lever exception dans générateur
def error_handler():
    try:
        while True:
            value = yield
            print(f"Reçu: {value}")
    except ValueError:
        print("ValueError attrapée!")
        yield "Recovered"

gen = error_handler()
next(gen)
gen.send(42)
gen.throw(ValueError)  # ValueError attrapée!

# close() - Fermer générateur
def closable():
    try:
        while True:
            yield "Running"
    finally:
        print("Nettoyage!")

gen = closable()
next(gen)
gen.close()            # Nettoyage!


[OK] GÉNÉRATEURS INFINIS

# Compteur infini
def infinite_counter(start=0):
    n = start
    while True:
        yield n
        n += 1

# Utilisation avec islice
from itertools import islice
gen = infinite_counter()
first_10 = list(islice(gen, 10))  # [0, 1, 2, ..., 9]

# Cycle infini
def cycle(iterable):
    """Cycle indéfiniment sur iterable"""
    while True:
        for item in iterable:
            yield item

colors = cycle(['red', 'green', 'blue'])
# Utiliser avec take
def take(n, iterable):
    return list(islice(iterable, n))

print(take(7, colors))  # ['red', 'green', 'blue', 'red', ...]

# Répétition infinie
def repeat(value):
    """Répète valeur indéfiniment"""
    while True:
        yield value

ones = repeat(1)
print(take(5, ones))    # [1, 1, 1, 1, 1]

# Fibonacci infini
def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

fib = fibonacci()
print(take(10, fib))    # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]


[OK] GÉNÉRATEURS AVEC ÉTAT

# Générateur avec mémoire
def running_average():
    total = 0
    count = 0
    while True:
        value = yield total / count if count > 0 else 0
        if value is not None:
            total += value
            count += 1

avg = running_average()
next(avg)              # 0 (amorcer)
print(avg.send(10))    # 10.0
print(avg.send(20))    # 15.0
print(avg.send(30))    # 20.0

# Générateur avec historique
def with_history():
    history = []
    while True:
        value = yield history.copy()
        if value is not None:
            history.append(value)

hist = with_history()
next(hist)
print(hist.send(1))    # [1]
print(hist.send(2))    # [1, 2]
print(hist.send(3))    # [1, 2, 3]

# Machine à états
def state_machine():
    state = "START"
    while True:
        if state == "START":
            action = yield state
            state = "RUNNING" if action == "run" else "START"
        elif state == "RUNNING":
            action = yield state
            state = "STOPPED" if action == "stop" else "RUNNING"
        elif state == "STOPPED":
            action = yield state
            state = "START" if action == "reset" else "STOPPED"

sm = state_machine()
print(next(sm))        # START
print(sm.send("run"))  # RUNNING
print(sm.send("stop")) # STOPPED


[OK] GÉNÉRATEURS CHAÎNÉS

# Chaîner plusieurs générateurs
def numbers(n):
    for i in range(n):
        yield i

def squares(gen):
    for num in gen:
        yield num ** 2

def even_only(gen):
    for num in gen:
        if num % 2 == 0:
            yield num

# Pipeline
result = even_only(squares(numbers(10)))
print(list(result))  # [0, 4, 16, 36, 64]

# yield from (Python 3.3+)
def chain_generators(gen1, gen2):
    yield from gen1
    yield from gen2

g1 = (x for x in range(3))
g2 = (x for x in range(3, 6))
combined = chain_generators(g1, g2)
print(list(combined))  # [0, 1, 2, 3, 4, 5]

# Aplatir structure imbriquée
def flatten(nested):
    for item in nested:
        if isinstance(item, (list, tuple)):
            yield from flatten(item)
        else:
            yield item

nested_list = [1, [2, 3, [4, 5]], 6, [7, [8, 9]]]
print(list(flatten(nested_list)))  # [1, 2, 3, 4, 5, 6, 7, 8, 9]

# Délégation avec yield from
def sub_generator():
    yield "A"
    yield "B"
    return "Done with sub"

def delegating_generator():
    result = yield from sub_generator()
    yield f"Sub returned: {result}"

gen = delegating_generator()
print(list(gen))  # ['A', 'B', 'Sub returned: Done with sub']


[OK] GÉNÉRATEURS POUR TRAITEMENT DE FICHIERS

# Lecture ligne par ligne (efficace mémoire)
def read_large_file(file_path):
    """Lit fichier ligne par ligne"""
    with open(file_path, 'r') as f:
        for line in f:
            yield line.strip()

# Utilisation
for line in read_large_file('large_file.txt'):
    process(line)

# Lecture par blocs
def read_in_chunks(file_path, chunk_size=1024):
    """Lit fichier par blocs"""
    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

for chunk in read_in_chunks('data.bin'):
    process_chunk(chunk)

# Filtrer lignes
def filter_lines(file_path, keyword):
    """Génère lignes contenant keyword"""
    with open(file_path, 'r') as f:
        for line in f:
            if keyword in line:
                yield line.strip()

# Parser CSV
def parse_csv(file_path):
    """Parse CSV ligne par ligne"""
    with open(file_path, 'r') as f:
        header = next(f).strip().split(',')
        for line in f:
            values = line.strip().split(',')
            yield dict(zip(header, values))

for row in parse_csv('data.csv'):
    print(row)

# Logs en temps réel
def tail_file(file_path):
    """Suit fichier en temps réel (comme tail -f)"""
    import time
    with open(file_path, 'r') as f:
        f.seek(0, 2)  # Aller à la fin
        while True:
            line = f.readline()
            if line:
                yield line.strip()
            else:
                time.sleep(0.1)

# Utilisation
for log_line in tail_file('app.log'):
    if 'ERROR' in log_line:
        alert(log_line)


[OK] GÉNÉRATEURS POUR TRAITEMENT DE DONNÉES

# Pipeline de transformation
def load_data(filename):
    """Charge données"""
    with open(filename) as f:
        for line in f:
            yield line.strip()

def parse_line(lines):
    """Parse chaque ligne"""
    for line in lines:
        fields = line.split(',')
        yield {'name': fields[0], 'age': int(fields[1])}

def filter_adults(records):
    """Filtre adultes"""
    for record in records:
        if record['age'] >= 18:
            yield record

def format_output(records):
    """Formate sortie"""
    for record in records:
        yield f"{record['name']} ({record['age']} ans)"

# Pipeline complet
pipeline = format_output(
    filter_adults(
        parse_line(
            load_data('people.csv')
        )
    )
)

for result in pipeline:
    print(result)

# Traitement par lots (batching)
def batch(iterable, size):
    """Groupe éléments par lots"""
    from itertools import islice
    iterator = iter(iterable)
    while True:
        batch_items = list(islice(iterator, size))
        if not batch_items:
            break
        yield batch_items

numbers = range(10)
for batch_group in batch(numbers, 3):
    print(batch_group)  # [0, 1, 2], [3, 4, 5], [6, 7, 8], [9]

# Fenêtre glissante (sliding window)
def window(iterable, size):
    """Fenêtre glissante de taille size"""
    from collections import deque
    it = iter(iterable)
    win = deque(islice(it, size), maxlen=size)
    if len(win) == size:
        yield tuple(win)
    for item in it:
        win.append(item)
        yield tuple(win)

for w in window(range(5), 3):
    print(w)  # (0,1,2), (1,2,3), (2,3,4)

# Agrégation par clé
def group_by(iterable, key_func):
    """Groupe éléments par clé"""
    from itertools import groupby
    sorted_items = sorted(iterable, key=key_func)
    for key, group in groupby(sorted_items, key_func):
        yield key, list(group)

people = [
    {'name': 'Alice', 'city': 'Paris'},
    {'name': 'Bob', 'city': 'Lyon'},
    {'name': 'Charlie', 'city': 'Paris'},
]

for city, group in group_by(people, lambda x: x['city']):
    print(f"{city}: {[p['name'] for p in group]}")


[OK] GÉNÉRATEURS ASYNCHRONES (async/await)

# Générateur asynchrone (Python 3.6+)
async def async_counter(n):
    """Compteur asynchrone"""
    for i in range(n):
        await asyncio.sleep(0.1)  # Opération async
        yield i

# Utilisation avec async for
async def main():
    async for num in async_counter(5):
        print(num)

# Lancer
import asyncio
asyncio.run(main())

# Expression génératrice asynchrone
async def async_squares(n):
    return [x**2 async for x in async_counter(n)]

# Itérateur asynchrone personnalisé
class AsyncRange:
    def __init__(self, n):
        self.n = n
        self.i = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.i >= self.n:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)
        self.i += 1
        return self.i - 1

# Utilisation
async def use_async_range():
    async for num in AsyncRange(5):
        print(num)


[OK] PERFORMANCE ET MÉMOIRE

# Comparaison liste vs générateur
import sys

# Liste (tout en mémoire)
numbers_list = [x**2 for x in range(1000000)]
print(sys.getsizeof(numbers_list))  # ~8 MB

# Générateur (minimal)
numbers_gen = (x**2 for x in range(1000000))
print(sys.getsizeof(numbers_gen))   # ~128 bytes

# Benchmark temps
import time

def time_it(func):
    start = time.time()
    result = func()
    end = time.time()
    return end - start, result

# Liste
def sum_with_list():
    return sum([x**2 for x in range(1000000)])

# Générateur
def sum_with_generator():
    return sum(x**2 for x in range(1000000))

list_time, _ = time_it(sum_with_list)
gen_time, _ = time_it(sum_with_generator)

print(f"Liste: {list_time:.4f}s")
print(f"Générateur: {gen_time:.4f}s")

# Générateur pour grandes données
def process_large_dataset(filename):
    """Traite dataset sans charger en mémoire"""
    total = 0
    count = 0
    
    for line in read_large_file(filename):
        value = float(line)
        total += value
        count += 1
    
    return total / count if count > 0 else 0


[OK] PATTERNS AVANCÉS

# Générateur avec contexte
from contextlib import contextmanager

@contextmanager
def managed_generator():
    print("Setup")
    try:
        yield "resource"
    finally:
        print("Cleanup")

with managed_generator() as resource:
    print(f"Using {resource}")

# Générateur récursif
def tree_traversal(node):
    """Parcours arbre en profondeur"""
    yield node.value
    for child in node.children:
        yield from tree_traversal(child)

# Coroutine avec générateur
def coroutine_example():
    """Coroutine basique"""
    print("Coroutine démarrée")
    try:
        while True:
            value = yield
            print(f"Reçu: {value}")
    except GeneratorExit:
        print("Coroutine terminée")

# Décorateur pour amorcer coroutine
def coroutine(func):
    def wrapper(*args, **kwargs):
        gen = func(*args, **kwargs)
        next(gen)
        return gen
    return wrapper

@coroutine
def printer():
    while True:
        value = yield
        print(value)

p = printer()
p.send("Hello")  # Pas besoin de next() d'abord

# Générateur avec __iter__ et __next__
class CustomGenerator:
    def __init__(self, n):
        self.n = n
        self.current = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.current >= self.n:
            raise StopIteration
        self.current += 1
        return self.current ** 2

gen = CustomGenerator(5)
for value in gen:
    print(value)  # 1, 4, 9, 16, 25


[OK] ITERTOOLS AVEC GÉNÉRATEURS

from itertools import *

# count - compteur infini
counter = count(start=10, step=2)
print(take(5, counter))  # [10, 12, 14, 16, 18]

# cycle - cycle sur iterable
colors = cycle(['R', 'G', 'B'])
print(take(7, colors))  # ['R', 'G', 'B', 'R', 'G', 'B', 'R']

# repeat - répète valeur
repeated = repeat('A', 3)
print(list(repeated))  # ['A', 'A', 'A']

# chain - chaîne iterables
chained = chain([1, 2], [3, 4], [5, 6])
print(list(chained))  # [1, 2, 3, 4, 5, 6]

# compress - filtre avec sélecteur
data = ['A', 'B', 'C', 'D']
selector = [1, 0, 1, 0]
print(list(compress(data, selector)))  # ['A', 'C']

# dropwhile - drop jusqu'à condition fausse
numbers = [1, 3, 5, 2, 4, 6]
print(list(dropwhile(lambda x: x < 5, numbers)))  # [5, 2, 4, 6]

# takewhile - prend jusqu'à condition fausse
print(list(takewhile(lambda x: x < 5, numbers)))  # [1, 3]

# filterfalse - inverse de filter
numbers = range(10)
print(list(filterfalse(lambda x: x % 2 == 0, numbers)))  # [1,3,5,7,9]

# islice - slice pour itérables
print(list(islice(count(), 5)))  # [0, 1, 2, 3, 4]
print(list(islice(count(), 2, 7)))  # [2, 3, 4, 5, 6]

# tee - duplique itérateur
gen1, gen2 = tee(range(3), 2)
print(list(gen1))  # [0, 1, 2]
print(list(gen2))  # [0, 1, 2]

# zip_longest - zip avec padding
from itertools import zip_longest
a = [1, 2, 3]
b = ['a', 'b']
print(list(zip_longest(a, b, fillvalue='X')))  # [(1,'a'), (2,'b'), (3,'X')]

# combinations - combinaisons
print(list(combinations([1, 2, 3], 2)))  # [(1,2), (1,3), (2,3)]

# permutations - permutations
print(list(permutations([1, 2, 3], 2)))  # [(1,2), (1,3), (2,1), ...]

# product - produit cartésien
print(list(product([1, 2], ['a', 'b'])))  # [(1,'a'), (1,'b'), (2,'a'), (2,'b')]

# accumulate - accumulation
print(list(accumulate([1, 2, 3, 4])))  # [1, 3, 6, 10]

# groupby - groupe par clé
data = [('A', 1), ('A', 2), ('B', 3), ('B', 4)]
for key, group in groupby(data, lambda x: x[0]):
    print(key, list(group))


[OK] GÉNÉRATEURS POUR ALGORITHMES

# Nombres premiers
def primes():
    """Génère nombres premiers indéfiniment"""
    yield 2
    primes_found = [2]
    candidate = 3
    while True:
        is_prime = True
        for prime in primes_found:
            if prime * prime > candidate:
                break
            if candidate % prime == 0:
                is_prime = False
                break
        if is_prime:
            primes_found.append(candidate)
            yield candidate
        candidate += 2

# Utilisation
prime_gen = primes()
first_10_primes = [next(prime_gen) for _ in range(10)]
print(first_10_primes)  # [2, 3, 5, 7, 11, 13, 17, 19, 23, 29]

# Permutations (implémentation manuelle)
def permute(items):
    """Génère permutations"""
    if len(items) <= 1:
        yield items
    else:
        for i, item in enumerate(items):
            rest = items[:i] + items[i+1:]
            for perm in permute(rest):
                yield [item] + perm

print(list(permute([1, 2, 3])))

# Quick sort avec générateur
def quick_sort(arr):
    """Quick sort lazy"""
    if len(arr) <= 1:
        yield from arr
    else:
        pivot = arr[0]
        smaller = [x for x in arr[1:] if x <= pivot]
        larger = [x for x in arr[1:] if x > pivot]
        yield from quick_sort(smaller)
        yield pivot
        yield from quick_sort(larger)

print(list(quick_sort([3, 1, 4, 1, 5, 9, 2, 6])))

# Parcours graphe (BFS)
def bfs(graph, start):
    """Parcours largeur"""
    from collections import deque
    visited = set()
    queue = deque([start])
    
    while queue:
        node = queue.popleft()
        if node not in visited:
            visited.add(node)
            yield node
            queue.extend(graph.get(node, []))

graph = {
    'A': ['B', 'C'],
    'B': ['D', 'E'],
    'C': ['F'],
    'D': [], 'E': [], 'F': []
}
print(list(bfs(graph, 'A')))


[OK] DEBUGGING ET INTROSPECTION

# Inspecter générateur
def sample_gen():
    yield 1
    yield 2
    yield 3

gen = sample_gen()

# État du générateur
import inspect
print(inspect.isgenerator(gen))           # True
print(inspect.isgeneratorfunction(sample_gen))  # True

# gi_frame, gi_code, gi_running
print(gen.gi_frame)   # Frame object
print(gen.gi_code)    # Code object
print(gen.gi_running) # False

# Wrapper de debugging
def debug_generator(gen):
    """Wrapper pour debugger générateur"""
    for i, value in enumerate(gen):
        print(f"[{i}] Yielding: {value}")
        yield value

wrapped = debug_generator(sample_gen())
list(wrapped)

# Compter valeurs générées
def count_items(gen):
    """Compte items sans consommer générateur"""
    gen1, gen2 = tee(gen)
    count = sum(1 for _ in gen1)
    return count, gen2

# Logger générateur
class LoggedGenerator:
    def __init__(self, gen):
        self.gen = gen
        self.count = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        self.count += 1
        value = next(self.gen)
        print(f"Item {self.count}: {value}")
        return value

logged = LoggedGenerator(range(3))
list(logged)


[OK] BONNES PRATIQUES

# 1. Nommer clairement les générateurs
# [OK] Bon
def even_numbers(n):
    for i in range(n):
        if i % 2 == 0:
            yield i

# [X] Mauvais
def gen(n):
    for i in range(n):
        if i % 2 == 0:
            yield i

# 2. Documenter générateurs
def fibonacci(n):
    """
    Génère suite Fibonacci jusqu'à n termes.
    
    Args:
        n: Nombre de termes à générer
    
    Yields:
        int: Prochain nombre Fibonacci
    
    Example:
        >>> list(fibonacci(5))
        [0, 1, 1, 2, 3]
    """
    a, b = 0, 1
    for _ in range(n):
        yield a
        a, b = b, a + b

# 3. Gérer cleanup avec try/finally
def resource_generator():
    """Générateur avec nettoyage"""
    resource = acquire_resource()
    try:
        while True:
            yield resource.read()
    finally:
        resource.close()

# 4. Éviter stockage inutile
# [X] Mauvais
def bad_gen(n):
    results = []
    for i in range(n):
        results.append(i**2)
    for result in results:
        yield result

# [OK] Bon
def good_gen(n):
    for i in range(n):
        yield i**2

# 5. Utiliser yield from pour déléguer
# [X] Mauvais
def chain_bad(iter1, iter2):
    for item in iter1:
        yield item
    for item in iter2:
        yield item

# [OK] Bon
def chain_good(iter1, iter2):
    yield from iter1
    yield from iter2

# 6. Amorcer coroutines explicitement
def coroutine_func():
    value = yield
    # ...

gen = coroutine_func()
next(gen)  # Amorcer avant send()

# 7. Gérer StopIteration
def safe_next(gen, default=None):
    """next() sûr avec valeur par défaut"""
    try:
        return next(gen)
    except StopIteration:
        return default

# 8. Convertir en liste si réutilisation
gen = (x**2 for x in range(5))
# Si besoin de réutiliser
data = list(gen)

# 9. Utiliser itertools pour opérations communes
from itertools import islice, chain
# Au lieu d'implémenter manuellement

# 10. Tester limites
def test_generator():
    gen = sample_gen()
    assert next(gen) == expected_first
    assert list(gen) == expected_rest


[OK] ERREURS COURANTES

# Erreur 1: Réutiliser générateur épuisé
gen = (x for x in range(3))
list(gen)  # [0, 1, 2]
list(gen)  # [] - épuisé!

# Solution: Recréer ou convertir en liste
data = list(x for x in range(3))

# Erreur 2: Oublier d'amorcer coroutine
def printer():
    while True:
        value = yield
        print(value)

p = printer()
p.send("Hello")  # TypeError! Oublié next()

# Solution
p = printer()
next(p)
p.send("Hello")  # OK

# Erreur 3: Confusion send() vs next()
gen = sample_gen()
gen.send(None)  # OK (équivalent à next)
gen.send(42)    # Erreur si yield pas assigné

# Erreur 4: Return dans générateur Python 2
def old_gen():
    yield 1
    return 2  # SyntaxError en Python 2

# Erreur 5: Modifier itérable pendant iteration
def dangerous():
    items = [1, 2, 3]
    for item in items:
        items.append(item * 2)  # Boucle infinie!
        yield item

# Solution: Copier ou utiliser while
def safe():
    items = [1, 2, 3]
    for item in items.copy():
        yield item

# Erreur 6: Oublier yield
def not_a_generator():
    return x**2 for x in range(5)  # Retourne generator, pas itère

# Solution
def is_a_generator():
    for x in range(5):
        yield x**2

# Erreur 7: Mélanger yield et return avec valeur
def confused():
    yield 1
    yield 2
    return [3, 4]  # Pas ce que vous pensez!
# return met valeur dans StopIteration.value, pas yield

# Erreur 8: Accès indexé
gen = (x for x in range(10))
gen[5]  # TypeError! Pas d'indexation

# Solution: Convertir ou utiliser islice
from itertools import islice
value = next(islice(gen, 5, 6))  # 6ème élément


[OK] CAS D'USAGE RÉELS

# === Pipeline ETL (Extract, Transform, Load) ===
def extract_data(source):
    """Extrait données depuis source"""
    with open(source) as f:
        for line in f:
            yield line.strip()

def transform_data(lines):
    """Transforme données"""
    for line in lines:
        if line and not line.startswith('#'):
            parts = line.split(',')
            yield {
                'id': int(parts[0]),
                'name': parts[1],
                'value': float(parts[2])
            }

def filter_data(records, min_value):
    """Filtre données"""
    for record in records:
        if record['value'] >= min_value:
            yield record

def load_data(records, destination):
    """Charge données"""
    with open(destination, 'w') as f:
        for record in records:
            f.write(f"{record}\n")
            yield record

# Pipeline complet
pipeline = load_data(
    filter_data(
        transform_data(
            extract_data('input.csv')
        ),
        min_value=100
    ),
    'output.txt'
)

# Exécuter pipeline
for record in pipeline:
    print(f"Processed: {record}")


# === Web Scraping avec pagination ===
def scrape_pages(base_url, max_pages=None):
    """Scrape pages avec pagination"""
    import requests
    from bs4 import BeautifulSoup
    
    page = 1
    while max_pages is None or page <= max_pages:
        url = f"{base_url}?page={page}"
        response = requests.get(url)
        
        if response.status_code != 200:
            break
            
        soup = BeautifulSoup(response.content, 'html.parser')
        items = soup.find_all('div', class_='item')
        
        if not items:
            break
            
        for item in items:
            yield {
                'title': item.find('h2').text,
                'price': item.find('span', class_='price').text
            }
        
        page += 1

# Utilisation
for product in scrape_pages('https://example.com/products', max_pages=5):
    print(product)


# === Streaming API responses ===
def stream_api_data(api_url, chunk_size=100):
    """Stream données depuis API par chunks"""
    import requests
    
    offset = 0
    while True:
        response = requests.get(
            api_url,
            params={'offset': offset, 'limit': chunk_size}
        )
        data = response.json()
        
        if not data['results']:
            break
            
        for item in data['results']:
            yield item
            
        offset += chunk_size

# Utilisation
for item in stream_api_data('https://api.example.com/data'):
    process_item(item)


# === Monitoring temps réel ===
def monitor_system(interval=1):
    """Monitor ressources système"""
    import psutil
    import time
    
    while True:
        yield {
            'cpu': psutil.cpu_percent(),
            'memory': psutil.virtual_memory().percent,
            'disk': psutil.disk_usage('/').percent,
            'timestamp': time.time()
        }
        time.sleep(interval)

# Utilisation
for stats in monitor_system(interval=5):
    if stats['cpu'] > 80:
        alert(f"High CPU: {stats['cpu']}%")


# === Traitement vidéo frame par frame ===
def process_video_frames(video_path):
    """Traite vidéo frame par frame"""
    import cv2
    
    cap = cv2.VideoCapture(video_path)
    
    try:
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            yield frame
    finally:
        cap.release()

# Utilisation
for frame in process_video_frames('video.mp4'):
    processed = apply_filter(frame)
    display(processed)


# === Configuration dynamique ===
def config_loader(config_file):
    """Charge config avec hot-reload"""
    import json
    import time
    
    last_modified = 0
    current_config = {}
    
    while True:
        import os
        mtime = os.path.getmtime(config_file)
        
        if mtime > last_modified:
            with open(config_file) as f:
                current_config = json.load(f)
            last_modified = mtime
            
        yield current_config
        time.sleep(1)


# === Queue consumer ===
def queue_consumer(queue, timeout=1):
    """Consomme messages d'une queue"""
    import queue as q
    
    while True:
        try:
            message = queue.get(timeout=timeout)
            yield message
            queue.task_done()
        except q.Empty:
            continue

# Utilisation avec threading
import queue
import threading

msg_queue = queue.Queue()

def producer():
    for i in range(10):
        msg_queue.put(f"Message {i}")

threading.Thread(target=producer).start()

for message in queue_consumer(msg_queue):
    process_message(message)


# === Rate limiting ===
def rate_limited(iterable, calls_per_second=10):
    """Limite débit d'un itérable"""
    import time
    
    interval = 1.0 / calls_per_second
    last_call = 0
    
    for item in iterable:
        now = time.time()
        elapsed = now - last_call
        
        if elapsed < interval:
            time.sleep(interval - elapsed)
            
        last_call = time.time()
        yield item

# Utilisation
urls = ['url1', 'url2', 'url3', ...]
for url in rate_limited(urls, calls_per_second=5):
    fetch(url)


# === Chunked file upload ===
def chunk_file(file_path, chunk_size=1024*1024):
    """Découpe fichier en chunks pour upload"""
    import os
    
    file_size = os.path.getsize(file_path)
    
    with open(file_path, 'rb') as f:
        chunk_num = 0
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
                
            yield {
                'chunk_num': chunk_num,
                'data': chunk,
                'total_chunks': (file_size + chunk_size - 1) // chunk_size
            }
            chunk_num += 1

# Utilisation
for chunk in chunk_file('large_file.zip', chunk_size=5*1024*1024):
    upload_chunk(chunk)


[OK] GÉNÉRATEURS VS ALTERNATIVES

# === Générateurs vs Listes ===

# Liste: Tout en mémoire immédiatement
def squares_list(n):
    return [x**2 for x in range(n)]

# Générateur: Lazy evaluation
def squares_gen(n):
    return (x**2 for x in range(n))

# Quand utiliser liste:
# - Petites collections
# - Besoin accès multiple
# - Besoin indexation
# - Besoin len()

# Quand utiliser générateur:
# - Grandes collections
# - Un seul passage
# - Mémoire limitée
# - Calculs coûteux


# === Générateurs vs Itérateurs ===

# Itérateur (classe)
class SquaresIterator:
    def __init__(self, n):
        self.n = n
        self.i = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.i >= self.n:
            raise StopIteration
        result = self.i ** 2
        self.i += 1
        return result

# Générateur (plus simple)
def squares_generator(n):
    for i in range(n):
        yield i ** 2

# Générateur généralement préféré sauf si:
# - Besoin méthodes personnalisées
# - État complexe
# - Héritage nécessaire


# === Générateurs vs map/filter ===

# map
mapped = map(lambda x: x**2, range(10))
# Équivalent générateur
gen = (x**2 for x in range(10))

# filter
filtered = filter(lambda x: x % 2 == 0, range(10))
# Équivalent générateur
gen = (x for x in range(10) if x % 2 == 0)

# Générateurs plus lisibles pour logique complexe
# map/filter OK pour opérations simples


[OK] OPTIMISATIONS

# 1. Éviter calculs répétés
# [X] Mauvais
def slow_gen(n):
    for i in range(n):
        yield expensive_calculation(i)  # Appelé à chaque iteration

# [OK] Bon - cache si possible
from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_calculation(x):
    return x ** 10

# 2. Utiliser yield from au lieu de boucle
# [X] Mauvais
def chain_slow(*iterables):
    for iterable in iterables:
        for item in iterable:
            yield item

# [OK] Bon
def chain_fast(*iterables):
    for iterable in iterables:
        yield from iterable

# 3. Générateur pour filter + map
# [X] Mauvais
result = map(transform, filter(predicate, data))

# [OK] Bon (une seule passe)
result = (transform(x) for x in data if predicate(x))

# 4. islice au lieu de next() répété
# [X] Mauvais
gen = count()
for _ in range(1000000):
    next(gen)
value = next(gen)

# [OK] Bon
gen = count()
value = next(islice(gen, 1000000, None))

# 5. tee pour duplication unique
# [X] Mauvais
data = list(expensive_gen())  # Matérialise tout
process1(iter(data))
process2(iter(data))

# [OK] Bon
gen1, gen2 = tee(expensive_gen())
process1(gen1)
process2(gen2)


[OK] TESTING GÉNÉRATEURS

import unittest
from itertools import islice

class TestGenerators(unittest.TestCase):
    
    def test_basic_generator(self):
        """Test générateur basique"""
        def sample_gen():
            yield 1
            yield 2
            yield 3
        
        result = list(sample_gen())
        self.assertEqual(result, [1, 2, 3])
    
    def test_infinite_generator(self):
        """Test générateur infini"""
        def counter():
            n = 0
            while True:
                yield n
                n += 1
        
        result = list(islice(counter(), 5))
        self.assertEqual(result, [0, 1, 2, 3, 4])
    
    def test_generator_with_state(self):
        """Test générateur avec état"""
        def accumulator():
            total = 0
            while True:
                value = yield total
                if value is not None:
                    total += value
        
        acc = accumulator()
        next(acc)
        self.assertEqual(acc.send(10), 10)
        self.assertEqual(acc.send(5), 15)
    
    def test_generator_exception(self):
        """Test gestion exceptions"""
        def error_gen():
            try:
                yield 1
                yield 2
                raise ValueError("Error")
                yield 3
            except ValueError:
                yield "Caught"
        
        result = list(error_gen())
        self.assertEqual(result, [1, 2, "Caught"])
    
    def test_generator_cleanup(self):
        """Test cleanup avec finally"""
        cleanup_called = []
        
        def cleanup_gen():
            try:
                yield 1
                yield 2
            finally:
                cleanup_called.append(True)
        
        gen = cleanup_gen()
        next(gen)
        gen.close()
        self.assertTrue(cleanup_called)
    
    def test_yield_from(self):
        """Test yield from"""
        def inner():
            yield 1
            yield 2
            return "Done"
        
        def outer():
            result = yield from inner()
            yield result
        
        self.assertEqual(list(outer()), [1, 2, "Done"])


# Testing avec pytest
import pytest

def test_generator_values():
    """Test valeurs générées"""
    gen = (x**2 for x in range(5))
    assert next(gen) == 0
    assert next(gen) == 1
    assert next(gen) == 4

def test_generator_exhaustion():
    """Test épuisement générateur"""
    gen = (x for x in range(2))
    list(gen)  # Épuise générateur
    with pytest.raises(StopIteration):
        next(gen)

def test_generator_send():
    """Test send()"""
    def echo():
        while True:
            value = yield
            yield value * 2
    
    gen = echo()
    next(gen)
    assert gen.send(5) == 10


[OK] PROFILING ET MESURE

# Mesurer temps d'exécution
import time
import sys

def profile_generator(gen_func, n=1000000):
    """Profile générateur"""
    # Temps création
    start = time.time()
    gen = gen_func(n)
    creation_time = time.time() - start
    
    # Temps consommation
    start = time.time()
    for _ in gen:
        pass
    consumption_time = time.time() - start
    
    # Mémoire
    gen = gen_func(n)
    memory = sys.getsizeof(gen)
    
    return {
        'creation_time': creation_time,
        'consumption_time': consumption_time,
        'memory': memory
    }

# Comparaison
def list_version(n):
    return [x**2 for x in range(n)]

def gen_version(n):
    return (x**2 for x in range(n))

# Profile avec cProfile
import cProfile

def test_function():
    gen = (x**2 for x in range(1000000))
    return sum(gen)

cProfile.run('test_function()')

# Memory profiling
from memory_profiler import profile

@profile
def memory_test():
    # Liste
    big_list = [x**2 for x in range(1000000)]
    
    # Générateur
    big_gen = (x**2 for x in range(1000000))
    sum(big_gen)


[OK] EXEMPLES CRÉATIFS

# === Générateur de mots de passe ===
def password_generator(length=12, count=None):
    """Génère mots de passe aléatoires"""
    import random
    import string
    
    chars = string.ascii_letters + string.digits + string.punctuation
    generated = 0
    
    while count is None or generated < count:
        password = ''.join(random.choice(chars) for _ in range(length))
        yield password
        generated += 1

# Utilisation
for pwd in password_generator(16, count=5):
    print(pwd)


# === Générateur de données de test ===
def fake_users(count=None):
    """Génère utilisateurs fake"""
    import random
    from itertools import count as counter
    
    first_names = ['Alice', 'Bob', 'Charlie', 'Diana']
    last_names = ['Smith', 'Johnson', 'Williams', 'Brown']
    
    for i in counter(1):
        if count is not None and i > count:
            break
            
        yield {
            'id': i,
            'name': f"{random.choice(first_names)} {random.choice(last_names)}",
            'email': f"user{i}@example.com",
            'age': random.randint(18, 80)
        }

# Utilisation
for user in fake_users(10):
    print(user)


# === Générateur de séquences mathématiques ===
def collatz(n):
    """Suite de Collatz"""
    yield n
    while n != 1:
        if n % 2 == 0:
            n = n // 2
        else:
            n = 3 * n + 1
        yield n

print(list(collatz(10)))  # [10, 5, 16, 8, 4, 2, 1]


def triangular_numbers():
    """Nombres triangulaires"""
    n = 1
    total = 0
    while True:
        total += n
        yield total
        n += 1

# 1, 3, 6, 10, 15, 21, ...
print(list(islice(triangular_numbers(), 10)))


# === Générateur de fractales ===
def mandelbrot_points(width, height, max_iter=100):
    """Génère points de Mandelbrot"""
    for y in range(height):
        for x in range(width):
            zx, zy = x / width * 3.5 - 2.5, y / height * 2 - 1
            c = complex(zx, zy)
            z = 0
            
            for i in range(max_iter):
                if abs(z) > 2:
                    break
                z = z*z + c
            
            yield (x, y, i)


# === Générateur de parsing ===
def parse_log_stream(log_generator):
    """Parse logs en temps réel"""
    import re
    
    pattern = re.compile(
        r'(?P<timestamp>\S+) (?P<level>\w+) (?P<message>.*)'
    )
    
    for line in log_generator:
        match = pattern.match(line)
        if match:
            yield match.groupdict()


# === Générateur de retry ===
def retry(func, max_attempts=3, delay=1):
    """Retry avec backoff exponentiel"""
    import time
    
    for attempt in range(1, max_attempts + 1):
        try:
            result = func()
            yield {'success': True, 'result': result, 'attempt': attempt}
            break
        except Exception as e:
            if attempt == max_attempts:
                yield {'success': False, 'error': str(e), 'attempt': attempt}
            else:
                time.sleep(delay * (2 ** (attempt - 1)))
                yield {'success': False, 'retry': True, 'attempt': attempt}


[OK] RESOURCES ET DOCUMENTATION

# Documentation officielle Python:
# https://docs.python.org/3/howto/functional.html#generators
# https://docs.python.org/3/library/itertools.html
# https://docs.python.org/3/reference/expressions.html#yield-expressions

# PEPs pertinents:
# PEP 255 - Simple Generators
# PEP 342 - Coroutines via Enhanced Generators
# PEP 380 - Syntax for Delegating to a Subgenerator (yield from)
# PEP 525 - Asynchronous Generators

# Articles et tutoriels:
# - Real Python: Introduction to Python Generators
# - David Beazley: Generator Tricks for Systems Programmers
# - Jeff Knupp: Improve Your Python: 'yield' and Generators

# Livres:
# - "Fluent Python" by Luciano Ramalho (Chapitre 14)
# - "Python Cookbook" by David Beazley (Chapitre 4)

# Outils:
# - more-itertools: Extension d'itertools
# - toolz: Functional programming utilities


[OK] AIDE-MÉMOIRE RAPIDE

# Créer générateur
def gen():
    yield value

# Expression génératrice
gen = (expr for item in iterable)

# Méthodes
next(gen)           # Prochaine valeur
gen.send(value)     # Envoyer valeur
gen.throw(exc)      # Lever exception
gen.close()         # Fermer

# yield from
yield from iterable # Déléguer

# Async
async def agen():
    yield value

# async for
async for item in agen():
    process(item)

# Itertools essentiels
from itertools import *
count()             # Compteur infini
cycle()             # Cycle infini
repeat()            # Répétition
islice()            # Slice
chain()             # Chaîner
tee()               # Dupliquer

# Pattern pipeline
result = func3(func2(func1(source)))

# Pattern coroutine
@coroutine
def coro():
    while True:
        value = yield
        process(value)

# Pattern cleanup
try:
    while True:
        yield value
finally:
    cleanup()


[OK] QUIZ ET EXERCICES

# Exercice 1: Implémenter range() comme générateur
def my_range(start, stop=None, step=1):
    """Votre implémentation"""
    pass

# Exercice 2: Générateur de carrés parfaits
def perfect_squares(limit):
    """Génère carrés parfaits jusqu'à limit"""
    pass

# Exercice 3: Générateur qui alterne deux séquences
def alternate(seq1, seq2):
    """Alterne entre deux séquences"""
    pass

# Exercice 4: Générateur de chunks
def chunks(iterable, size):
    """Découpe iterable en chunks de taille size"""
    pass

# Exercice 5: Générateur de merge (comme merge sort)
def merge(gen1, gen2):
    """Merge deux générateurs triés"""
    pass

# Solutions dans commentaire ci-dessous:
"""
# Solution 1
def my_range(start, stop=None, step=1):
    if stop is None:
        stop = start
        start = 0
    current = start
    while (step > 0 and current < stop) or (step < 0 and current > stop):
        yield current
        current += step

# Solution 2
def perfect_squares(limit):
    n = 1
    while n * n <= limit:
        yield n * n
        n += 1

# Solution 3
def alternate(seq1, seq2):
    iter1, iter2 = iter(seq1), iter(seq2)
    while True:
        try:
            yield next(iter1)
            yield next(iter2)
        except StopIteration:
            break

# Solution 4
def chunks(iterable, size):
    from itertools import islice
    iterator = iter(iterable)
    while True:
        chunk = list(islice(iterator, size))
        if not chunk:
            break
        yield chunk

# Solution 5
def merge(gen1, gen2):
    val1, val2 = next(gen1, None), next(gen2, None)
    while val1 is not None or val2 is not None:
        if val2 is None or (val1 is not None and val1 <= val2):
            yield val1
            val1 = next(gen1, None)
        else:
            yield val2
            val2 = next(gen2, None)
"""