from typing import Iterator, Generator, Iterable, TypeVar
T = TypeVar('T')
def fibonacci_generator(n: int) -> Generator[int, None, None]:
a, b = 0, 1
for _ in range(n):
yield a
a, b = b, a + b
def read_lines_lazy(filename: str) -> Generator[str, None, None]:
with open(filename, 'r') as f:
for line in f:
yield line.rstrip('\n')
def batch_iterator(items: Iterable[T], size: int) -> Generator[list[T], None, None]:
batch: list[T] = []
for item in items:
batch.append(item)
if len(batch) == size:
yield batch
batch = []
if batch:
yield batch
def flatten(nested: Iterable[Iterable[T]]) -> Generator[T, None, None]:
for sublist in nested:
for item in sublist:
yield item
def unique_iter(items: Iterable[T]) -> Generator[T, None, None]:
seen: set[T] = set()
for item in items:
if item not in seen:
seen.add(item)
yield item
def cycle(items: Iterable[T]) -> Generator[T, None, None]:
saved: list[T] = []
for item in items:
saved.append(item)
yield item
while True:
for item in saved:
yield item
def range_with_step(start: int, stop: int, step: int) -> Generator[int, None, None]:
current = start
while current < stop:
yield current
current += step
def send_example() -> Generator[int, int, None]:
received = yield 1
print(f"Received: {received}")
received = yield 2
print(f"Received: {received}")
yield 3
def pipeline_map(items: Iterable[T], func: Callable[[T], R]) -> Generator[R, None, None]:
for item in items:
yield func(item)
def pipeline_filter(items: Iterable[T], predicate: Callable[[T], bool]) -> Generator[T, None, None]:
for item in items:
if predicate(item):
yield item
def accumulate(items: Iterable[T], func: Callable[[T, T], T]) -> Generator[T, None, None]:
it = iter(items)
try:
total = next(it)
except StopIteration:
return
yield total
for item in it:
total = func(total, item)
yield total
def windowed(items: Iterable[T], n: int) -> Generator[tuple[T, ...], None, None]:
from collections import deque
it = iter(items)
window: deque[T] = deque(maxlen=n)
for _ in range(n):
try:
window.append(next(it))
except StopIteration:
return
yield tuple(window)
for item in it:
window.append(item)
yield tuple(window)
from typing import Callable
R = TypeVar('R')