Генераторы Python для эффективной обработки данных

def simple_generator():
    yield 1
    yield 2
    yield 3
 
gen = simple_generator()
print(next(gen))  # 1
print(next(gen))  # 2
print(next(gen))  # 3

def read_large_file(file_path):
    with open(file_path, 'r') as file:
        for line in file:
            yield line.strip()
 
def filter_lines(lines, pattern):
    for line in lines:
        if pattern in line:
            yield line
 
def process_matching_lines(file_path, pattern):
    lines = read_large_file(file_path)
    matching_lines = filter_lines(lines, pattern)
    for line in matching_lines:
        # Обработка найденых строк
        yield line.upper()

def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b
 
# Используем генератор для вывода первых 10 чисел
fib_gen = fibonacci()
for _ in range(10):
    print(next(fib_gen))

import itertools
 
# Бесконечные итераторы
for i in itertools.count(10, 2):  # начинаем с 10, шаг 2
    if i > 20:
        break
    print(i)  # 10, 12, 14, 16, 18, 20
 
# Комбинаторика
for combo in itertools.combinations('ABCD', 2):
    print(''.join(combo))  # AB, AC, AD, BC, BD, CD
 
# Группировка данных
animals = ['duck', 'dog', 'deer', 'cat', 'crow']
for key, group in itertools.groupby(sorted(animals), key=lambda x: x[0]):
    print(key, list(group))  # c ['cat', 'crow'], d ['deer', 'dog', 'duck']

from functools import lru_cache
 
@lru_cache(maxsize=None)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)
 
# Используем с генератором
def fib_generator(max_n):
    n = 0
    while n < max_n:
        yield fibonacci(n)
        n += 1

def is_prime(n):
    return n > 1 and all(n % i != 0 for i in range(2, int(n**0.5) + 1))
 
# Проверка без вычисления всех значений
has_prime = any(is_prime(x) for x in range(100, 110))

data = [('a', 1), ('b', 2), ('c', 3)]
# Создание словаря из генератора
d = dict((k, v*2) for k, v in data)
print(d)  # {'a': 2, 'b': 4, 'c': 6}
 
# Суммирование без создания промежуточного списка
total = sum(v for k, v in data)

def squares_generator(n):
for i in range(n):
    yield i * i
 
# Использование функции-генератора
gen = squares_generator(5)
print(type(gen))  # <class 'generator'>
for square in gen:
print(square)  # Выводит 0, 1, 4, 9, 16

# Списковое включение (загружает всё в память)
squares_list = [x*x for x in range(1000000)]
 
# Выражение-генератор (вычисляет значения по требованию)
squares_gen = (x*x for x in range(1000000))

import sys
squares_list = [i*i for i in range(10000)]
squares_gen = (i*i for i in range(10000))
 
print(sys.getsizeof(squares_list))  # ~87 Кб (зависит от версии Python)
print(sys.getsizeof(squares_gen))   # ~112 байт

gen = squares_generator(3)
print(list(gen))  # [0, 1, 4]
print(list(gen))  # [] - генератор уже исчерпан

def count_forever(start=0, step=1):
n = start
while True:
    yield n
    n += step
 
counter = count_forever(10, 2)
for _ in range(5):
print(next(counter))  # Выведет 10, 12, 14, 16, 18

def file_reader(file_path):
file = open(file_path, 'r')
try:
    for line in file:
        yield line.strip()
finally:
    file.close()  # Гарантированно выполнится, даже если генератор не исчерпан

iterator = iter(something)  # Вызывает something.__iter__()
while True:
    try:
        x = next(iterator)  # Вызывает iterator.__next__()
        # Код внутри цикла for
    except StopIteration:
        break  # Прерываем цикл, когда итератор исчерпан

class SquaresIterator:
    def __init__(self, max_number):
        self.max_number = max_number
        self.current = 0
        
    def __iter__(self):
        return self
        
    def __next__(self):
        if self.current >= self.max_number:
            raise StopIteration
        result = self.current ** 2
        self.current += 1
        return result
 
# Использование
for square in SquaresIterator(5):
    print(square)  # 0, 1, 4, 9, 16

def squares_generator(max_number):
    for i in range(max_number):
        yield i ** 2
 
# Использование точно такое же
for square in squares_generator(5):
    print(square)  # 0, 1, 4, 9, 16

squares = (x**2 for x in range(5))
# Python автоматически создаёт итератор

my_list = [1, 2, 3]
iterator = iter(my_list)  # Создаёт отдельный объект-итератор
print(my_list is iterator)  # False
 
gen = squares_generator(3)
print(gen is iter(gen))  # True - генератор сам является своим итератором

gen = squares_generator(10)
# Преобразование генератора в список (исчерпывает генератор)
squares_list = list(gen)

def nested_generator():
yield from range(3)
yield from "ABC"
 
list(nested_generator())  # [0, 1, 2, 'A', 'B', 'C']

def file_reading_generator(filename):
file = open(filename, 'r')
try:
while True:
    line = file.readline()
    if not line:
        break
    yield line.strip()
finally:
    file.close()
    print(f"Файл {filename} закрыт!")
 
gen = file_reading_generator("example.txt")
print(next(gen))  # Выводит первую строку
gen.close()       # Закрывает генератор и файл

def echo_generator():
result = yield
while True:
    result = yield result
 
gen = echo_generator()
next(gen)  # Первый next() для инициализации. Дойдёт до первого yield и остановится
print(gen.send("Привет"))  # Выведет "Привет"
print(gen.send(42))        # Выведет 42

def handling_exceptions_generator():
try:
while True:
    try:
        x = yield
    except ValueError:
        yield "Поймал ValueError!"
    else:
        yield f"Полученно: {x}"
finally:
    yield "Генератор завершается!"
 
g = handling_exceptions_generator()
next(g)  # Подготовка генератора
print(g.send(10))              # Выведет "Полученно: 10"
print(g.throw(ValueError))     # Выведет "Поймал ValueError!"
print(g.close())               # Выведет "Генератор завершается!" и None

from contextlib import contextmanager
 
@contextmanager
def file_opener(filename):
file = None
try:
    file = open(filename, 'r')
    yield file
except FileNotFoundError:
    yield None
finally:
    if file:
        file.close()
        print(f"Файл {filename} закрыт")
 
# Использование
with file_opener("example.txt") as f:
if f:
    print(f.readline())

def error_handling_generator(source_gen):
while True:
    try:
        item = next(source_gen)
        yield item
    except StopIteration:
        break
    except ValueError as e:
        yield f"Ошибка значения: {e}"
    except Exception as e:
        yield f"Неизвестная ошибка: {e}"
 
def problematic_generator():
yield 1
yield 2
yield int("не число")  # Вызовет ValueError
yield 4
 
# Использование
for item in error_handling_generator(problematic_generator()):
print(item)

def outer_generator():
try:
yield from inner_generator()
except Exception as e:
yield f"Перехвачено исключение: {e}"
 
def inner_generator():
yield 1
raise ValueError("Что-то пошло не так")
yield 2  # До этой строки никогда не дойдёт выполнение

def process_log_file(filepath, pattern=None):
    with open(filepath, 'r') as f:
        for line in f:
            if pattern is None or pattern in line:
                yield line.strip()
 
# Использование
for line in process_log_file('enormous_log.txt', 'ERROR'):
    process_error(line)

def read_csv(filename):
    with open(filename, 'r') as f:
        header = next(f).strip().split(',')
        for line in f:
            yield dict(zip(header, line.strip().split(',')))
 
def filter_by_column(rows, column, value):
    for row in rows:
        if row.get(column) == value:
            yield row
 
def calculate_average(rows, column):
    total = count = 0
    for row in rows:
        total += float(row.get(column, 0))
        count += 1
    return total / count if count else 0
 
# Использование
data = read_csv('sales.csv')
filtered_data = filter_by_column(data, 'region', 'North')
avg = calculate_average(filtered_data, 'amount')

def sensor_reader(sensor_id, interval):
    while True:
        yield get_sensor_data(sensor_id)
        time.sleep(interval)
 
def alert_on_threshold(readings, threshold):
    for reading in readings:
        if reading > threshold:
            yield f"Alert! Reading {reading} exceeds threshold {threshold}"
 
# Мониторинг в реальном времени
sensor_data = sensor_reader('temperature_1', 5)  # Чтение каждые 5 секунд
alerts = alert_on_threshold(sensor_data, 30)
for alert in alerts:
    send_notification(alert)

def random_user_generator():
    names = ["Алексей", "Мария", "Иван", "Анна", "Петр"]
    domains = ["gmail.com", "yandex.ru", "mail.ru", "outlook.com"]
    while True:
        name = random.choice(names)
        domain = random.choice(domains)
        yield {
            "name": name,
            "email": f"{name.lower()}{random.randint(1, 999)}@{domain}",
            "age": random.randint(18, 70)
        }
 
# Генерация тестовых пользователей
users_gen = random_user_generator()
test_users = [next(users_gen) for _ in range(100)]

def batch_requests(urls, batch_size=10):
    for i in range(0, len(urls), batch_size):
        batch = urls[i:i+batch_size]
        yield batch
 
def process_batch(url_batch):
    results = []
    for url in url_batch:
        results.append(fetch_data(url))
    return results
 
urls = ["https://api.example.com/user/1", "https://api.example.com/user/2", ...]
for batch in batch_requests(urls):
    results = process_batch(batch)
    store_results(results)

def crawl_site(start_url, max_depth=2):
    visited = set([start_url])
    queue = deque([(start_url, 0)])  # URL и глубина
    
    while queue:
        url, depth = queue.popleft()
        page = fetch_page(url)
        yield (url, extract_data(page))
        
        if depth < max_depth:
            for link in extract_links(page):
                if link not in visited:
                    visited.add(link)
                    queue.append((link, depth + 1))

import ijson
 
def json_objects_generator(filename, prefix):
    with open(filename, 'rb') as f:
        for item in ijson.items(f, prefix):
            yield item
 
# Пример использования: обработка огромного файла с твитами
tweets = json_objects_generator('massive_tweets.json', 'item')
for tweet in tweets:
    if 'python' in tweet.get('text', '').lower():
        process_python_tweet(tweet)

usernames = json_objects_generator('users.json', 'item.username')

import xml.etree.ElementTree as ET
 
def parse_huge_xml(filename, element_tag):
    context = ET.iterparse(filename, events=('end',))
    for event, elem in context:
        if elem.tag == element_tag:
            yield elem
            # Очистка памяти после обработки элемента
            elem.clear()
    
# Обработка гигантского XML-файла с продуктами
products = parse_huge_xml('enormous_catalog.xml', 'product')
for product in products:
    if float(product.find('price').text) < 100:
        add_to_discount_list(product)

def extract_attributes(elements, attr_name):
    for elem in elements:
        yield elem.get(attr_name)
 
def match_pattern(values, pattern):
    for value in values:
        if pattern in value:
            yield value
 
# Использование
products = parse_huge_xml('catalog.xml', 'product')
product_ids = extract_attributes(products, 'id')
matching_ids = match_pattern(product_ids, 'X-')
for product_id in matching_ids:
    print(f"Найден продукт: {product_id}")

def deep_search(elem, target_tag):
    queue = deque([elem])
    while queue:
        current = queue.popleft()
        if current.tag == target_tag:
            yield current
        queue.extend(list(current))

def extract():
    data = []
    with open('source1.csv', 'r') as f1:
        data.extend([line.strip() for line in f1])
    with open('source2.csv', 'r') as f2:
        data.extend([line.strip() for line in f2])
    return data
 
def transform(data):
    result = []
    for item in data:
        # Преобразование
        result.append(transformed_item)
    return result
 
def load(data):
    with open('destination.csv', 'w') as f:
        for item in data:
            f.write(f"{item}\n")
 
data = extract()
transformed_data = transform(data)
load(transformed_data)

def extract():
    with open('source1.csv', 'r') as f1:
        for line in f1:
            yield line.strip()
    with open('source2.csv', 'r') as f2:
        for line in f2:
            yield line.strip()
 
def transform(data_gen):
    for item in data_gen:
        # Преобразование
        yield transformed_item
 
def load(data_gen):
    with open('destination.csv', 'w') as f:
        for item in data_gen:
            f.write(f"{item}\n")
 
# Композиция всего пайплайна
load(transform(extract()))

def incremental_extract(last_update_time):
    with db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM table WHERE update_time > %s", [last_update_time])
        while True:
            batch = cursor.fetchmany(1000)  # Обрабатываем по 1000 записей за раз
            if not batch:
                break
            for row in batch:
                yield row
 
def detect_changes(source_gen, target_db):
    for record in source_gen:
        target_record = get_record_from_target(record.id, target_db)
        if not target_record or has_changes(record, target_record):
            yield ('UPDATE', record)
        else:
            yield ('SKIP', record)
            
def apply_changes(changes_gen, target_db):
    stats = {'updated': 0, 'skipped': 0}
    for action, record in changes_gen:
        if action == 'UPDATE':
            update_target(record, target_db)
            stats['updated'] += 1
        else:
            stats['skipped'] += 1
    return stats

def parallel_transform(data_gen, transform_func, n_workers=4):
    queue = Queue(maxsize=n_workers * 2)
    result_queue = Queue()
    
    def worker():
        while True:
            batch = queue.get()
            if batch is None:
                queue.task_done()
                break
            result = [transform_func(item) for item in batch]
            result_queue.put(result)
            queue.task_done()
    
    workers = []
    for _ in range(n_workers):
        t = Thread(target=worker)
        t.start()
        workers.append(t)
    
    batch_size = 100
    batch = []
    for item in data_gen:
        batch.append(item)
        if len(batch) >= batch_size:
            queue.put(batch)
            batch = []
    
    if batch:  # Не забываем про остаток
        queue.put(batch)
    
    # Сообщаем воркерам о завершении
    for _ in range(n_workers):
        queue.put(None)
    
    # Ждем завершения всех задач
    queue.join()
    
    # Собираем результаты
    remaining = result_queue.qsize()
    for _ in range(remaining):
        for item in result_queue.get():
            yield item

def logging_wrapper(gen, name):
    for i, item in enumerate(gen):
        if i % 1000 == 0:
            print(f"{name}: обработано {i} элементов")
        yield item

def process_large_csv(filename, chunksize=10000):
    for chunk in pd.read_csv(filename, chunksize=chunksize):
        # Обработка чанка
        processed = some_transformation(chunk)
        yield processed

# Сервис-источник
def user_data_stream():
    user_id = 0
    while True:
        user = fetch_user_from_db(user_id)
        if not user:
            break
        yield user
        user_id += 1
 
@app.route('/stream-users')
def stream_users():
    def generate():
        for user in user_data_stream():
            yield json.dumps(user) + '\n'
    return Response(generate(), mimetype='application/x-json-stream')

def process_user_stream(url):
    response = requests.get(url, stream=True)
    for line in response.iter_lines():
        if line:
            user = json.loads(line)
            yield process_user(user)

def chunked_stream(stream_generator, chunk_size=100):
    chunk = []
    for item in stream_generator:
        chunk.append(item)
        if len(chunk) >= chunk_size:
            yield chunk
            chunk = []
    if chunk:  # Не забываем про остаток
        yield chunk

def circuit_breaker(api_call, failure_threshold=3, reset_time=60):
    failures = 0
    last_failure_time = 0
    
    while True:
        try:
            if failures >= failure_threshold and time.time() - last_failure_time < reset_time:
                yield {'status': 'error', 'message': 'Circuit open'}
                continue
                
            result = api_call()
            failures = 0  # Сбрасываем счётчик при успехе
            yield {'status': 'success', 'data': result}
            
        except Exception as e:
            failures += 1
            last_failure_time = time.time()
            yield {'status': 'error', 'message': str(e)}

async def async_user_stream():
    user_id = 0
    while True:
        user = await fetch_user_async(user_id)
        if not user:
            break
        yield user
        user_id += 1

def fibonacci_iterator(n):
a, b = 0, 1
for _ in range(n):
yield a
a, b = b, a + b
 
# Использование
for num in fibonacci_iterator(10):
print(num)

def observer_pattern():
subscribers = []
value = None
while True:
    event = yield value
    if event == 'subscribe':
        subscribers.append((yield 'Введите функцию обработчик:'))
    elif event == 'notify':
        message = (yield 'Введите сообщение:')
        value = [callback(message) for callback in subscribers]
 
# Использование
def logger(message):
    print(f"Лог: {message}")
    return "Сообщение залогировано"
 
def alerter(message):
    print(f"Тревога: {message}")
    return "Предупреждение отправлено"
 
observer = observer_pattern()
next(observer)  # Инициализация генератора
observer.send('subscribe')
observer.send(logger)
observer.send('subscribe')
observer.send(alerter)
results = observer.send('notify')
message = observer.send('Важное событие!')
print(results)  # Выведет результаты от всех подписчиков

def text_processor_chain():
text = yield
while True:
    # Обработчик 1: Удаление специальных символов
    text = re.sub(r'[^\w\s]', '', text)
    
    # Обработчик 2: Приведение к нижнему регистру
    text = text.lower()
    
    # Обработчик 3: Удаление стоп-слов
    stop_words = {'и', 'в', 'на', 'с'}
    text = ' '.join(word for word in text.split() if word not in stop_words)
    
    # Возвращаем результат и получаем новый текст
    text = yield text
 
# Использование
processor = text_processor_chain()
next(processor)  # Инициализация
processed_text = processor.send("Привет, мир! Это текст с разными символами и словами.")
print(processed_text)

def strategy_pattern(strategy='sum'):
strategies = {
    'sum': lambda x, y: x + y,
    'multiply': lambda x, y: x * y,
    'max': lambda x, y: max(x, y),
    'min': lambda x, y: min(x, y)
}
current_strategy = strategies.get(strategy, strategies['sum'])
    
x, y = yield
while True:
    result = current_strategy(x, y)
    x, y = yield result
 
# Использование
calculator = strategy_pattern('multiply')
next(calculator)  # Инициализация
print(calculator.send((10, 20)))  # 200

def template_method(process_step=None, format_step=None):
if process_step is None:
    process_step = lambda x: x.upper()
if format_step is None:
    format_step = lambda x: f"[{x}]"
    
data = yield
while True:
    # Шаг 1: Обработка данных (переменный)
    processed = process_step(data)
    
    # Шаг 2: Форматирование (переменный)
    formatted = format_step(processed)
    
    # Получаем новые данные и возвращаем результат
    data = yield formatted
 
# Кастомизированный шаблонный метод
custom_template = template_method(
    process_step=lambda x: x.replace(' ', '_'),
    format_step=lambda x: f"<{x}>"
)
next(custom_template)
result = custom_template.send("Hello World")
print(result)  # <Hello_World>

def tree_visitor():
tree = yield
while True:
    result = []
    
    def visit(node, depth=0):
        if isinstance(node, dict):
            result.append("  " * depth + f"Dict({len(node)})")
            for k, v in node.items():
                result.append("  " * (depth+1) + f"Key: {k}")
                visit(v, depth+2)
        elif isinstance(node, list):
            result.append("  " * depth + f"List({len(node)})")
            for item in node:
                visit(item, depth+1)
        else:
            result.append("  " * depth + f"Value: {node}")
    
    visit(tree)
    tree = yield "\n".join(result)
 
# Использование
visitor = tree_visitor()
next(visitor)
result = visitor.send({"users": [{"name": "Alice", "age": 30}, {"name": "Bob"}]})
print(result)

def data_source():
for i in range(100):
yield i
 
def filter_even(numbers):
for num in numbers:
if num % 2 == 0:
    yield num
 
def multiply_by_three(numbers):
for num in numbers:
yield num * 3
 
def pipeline():
return multiply_by_three(filter_even(data_source()))
 
# Использование
for result in pipeline():
print(result)  # 0, 6, 12, 18, 24, ...

from itertools import tee
 
def process_with_branching(gen):
stats_gen, main_gen = tee(gen)
 
 
# Ветвь 1: Собираем статистику
total = sum(1 for _ in stats_gen)
print(f"Всего элементов: {total}")
 
# Ветвь 2: Обрабатываем данные
for item in main_gen:
process_item(item)

def cached_generator(gen_func):
cache = {}
def wrapper(*args, **kwargs):
    key = (args, frozenset(kwargs.items()))
    if key not in cache:
        cache[key] = list(gen_func(*args, [B]kwargs))
    for item in cache[key]:
        yield item
return wrapper
 
@cached_generator
def expensive_computation(n):
print(f"Вычисляем для {n}...")
for i in range(n):
yield i [/B] 2
 
# Первый вызов выполнит вычисления
for x in expensive_computation(3):
print(x)  # Выведет "Вычисляем для 3..." и 0, 1, 4
 
# Повторный вызов использует кэш
for x in expensive_computation(3):
print(x)  # Выведет только 0, 1, 4 без сообщения о вычислении

def moving_average(data, window_size=3):
window = deque(maxlen=window_size)
for x in data:
window.append(x)
if len(window) == window_size:
    yield sum(window) / window_size

async def async_data_fetcher(urls):
for url in urls:
response = await aiohttp.get(url)
data = await response.json()
yield data
 
async def process_async_data():
urls = ["https://api.example.com/data/1", "https://api.example.com/data/2"]
async for data in async_data_fetcher(urls):
print(f"Получено {len(data)} записей")

@contextmanager
def managed_generator(gen):
try:
yield gen
finally:
gen.close()  # Гарантированное закрытие генератора
 
with managed_generator(file_reading_generator("huge_file.txt")) as gen:
for line in gen:
if "ERROR" in line:
    break  # Даже при досрочном выходе ресурсы будут освобождены

def batch_fetch_generator(ids, batch_size=100):
for i in range(0, len(ids), batch_size):
    batch = ids[i:i+batch_size]
    results = db.fetch_many(batch)  # Одна эффективная база данных
    for result in results:
        yield result

def fork_generator(source_gen, n):
buffers = [[] for _ in range(n)]
consumers = [0] * n
active = n
    
while active > 0:
    try:
        item = next(source_gen)
        for i in range(n):
            if consumers[i] < len(buffers[i]):
                buffers[i].append(item)
    except StopIteration:
        active = 0
        for i in range(n):
            if consumers[i] < len(buffers[i]):
                active += 1
    
    yield [buffer[consumers[i]] if consumers[i] < len(buffer) else None
           for i, buffer in enumerate(buffers)]
    
    for i in range(n):
        if consumers[i] < len(buffers[i]):
            consumers[i] += 1

def echo_coroutine():
    print("Сопрограмма запущена")
    while True:
        value = yield
        print(f"Получено: {value}")
 
coro = echo_coroutine()
next(coro)  # Подготовка сопрограммы, первый yield
coro.send("Hello")  # Выведет "Получено: Hello"
coro.send(42)       # Выведет "Получено: 42"

def coroutine(func):
    def wrapper(*args, **kwargs):
        gen = func(*args, **kwargs)
        next(gen)  # Прайминг
        return gen
    return wrapper
 
@coroutine
def improved_echo():
    while True:
        value = yield
        print(f"Эхо: {value}")
 
echo = improved_echo()  # Уже готова к работе
echo.send("Python")     # Выведет "Эхо: Python"

@coroutine
def filter_lines(pattern, target):
    while True:
        line = yield
        if pattern in line:
            target.send(line)
 
@coroutine
def printer():
    while True:
        line = yield
        print(f"Найдено: {line}")
 
# Создаем конвейер
output = printer()
filtered = filter_lines("ERROR", output)
 
# Используем
filtered.send("INFO: Всё в порядке")    # Ничего не выведет
filtered.send("ERROR: Критическая ошибка")  # Выведет "Найдено: ERROR: Критическая ошибка"

@coroutine
def data_processor():
    try:
        while True:
            data = yield
            # Обработка данных
            print(f"Обработано: {data}")
    except GeneratorExit:
        print("Сопрограмма завершила работу")
        # Освобождение ресурсов
 
processor = data_processor()
processor.send("Data 1")
processor.close()  # Выведет "Сопрограмма завершила работу"

@coroutine
def error_handling_coroutine():
    try:
        while True:
            try:
                value = yield
                print(f"Обрабатываем: {value}")
            except ValueError as e:
                print(f"Перехвачена ошибка значения: {e}")
    finally:
        print("Освобождаем ресурсы")
 
handler = error_handling_coroutine()
handler.send(10)
handler.throw(ValueError, "Некорректные данные")  # Сопрограмма перехватит это исключение
handler.close()  # Финализация

@coroutine
def router():
    targets = {
        'info': [],
        'warning': [],
        'error': []
    }
    try:
        while True:
            message = yield
            level = message.get('level', 'info')
            for target in targets.get(level, []):
                target.send(message)
    except GeneratorExit:
        # Закрываем все подписанные сопрограммы
        for subscribers in targets.values():
            for subscriber in subscribers:
                subscriber.close()
 
@coroutine
def subscribe_to(router, level):
    subscribers = []
    # Регистрируем сопрограмму в маршрутизаторе
    router.targets[level].append(subscribers)
    try:
        while True:
            message = yield
            # Обработка сообщения определённого уровня
            print(f"[{level.upper()}] {message['text']}")
    except GeneratorExit:
        # Удаляем сопрограмму из маршрутизатора
        router.targets[level].remove(subscribers)

async def async_range(start, stop):
for i in range(start, stop):
    await asyncio.sleep(0.1)  # Имитация I/O-операции
    yield i
 
async def main():
async for num in async_range(0, 5):
    print(num)
 
asyncio.run(main())

async def fetch_batch(urls):
async with aiohttp.ClientSession() as session:
    for url in urls:
        async with session.get(url) as response:
            data = await response.json()
            yield data
 
async def process_data():
batch_size = 0
async for data in fetch_batch(URLS):
    batch_size += sys.getsizeof(data)
    if batch_size > MAX_MEMORY:
        # Обрабатываем и высвобождаем память
        process_batch()
        batch_size = 0

import aioitertools as ait
 
async def process_pipeline():
source = fetch_batch(urls)
# Фильтруем только успешные ответы
filtered = ait.filter(lambda x: x.get('status') == 'success', source)
# Преобразуем данные
mapped = ait.map(transform_data, filtered)
# Обрабатываем результаты
async for result in mapped:
    store_result(result)

def process_large_dataframe(filename):
total_sum = 0
for chunk in pd.read_csv(filename, chunksize=10000):
    # Выполняем вычисления над каждым чанком
    chunk_sum = chunk['value'].sum()
    total_sum += chunk_sum
    # Чанк автоматически удаляется сборщиком мусора
return total_sum

def preprocess_dataset(input_file, output_file):
with open(output_file, 'w') as f:
    # Запись заголовка
    f.write('processed_feature1,processed_feature2,target
')
    
    for chunk in pd.read_csv(input_file, chunksize=5000):
        # Предобработка
        processed = transform_features(chunk)
        
        # Запись в файл без загрузки всего датасета в память
        processed.to_csv(f, header=False, index=False)

def pandas_transformer(dataframes, transform_func):
for df in dataframes:
    yield transform_func(df)
 
# Использование
csv_reader = pd.read_csv('large_file.csv', chunksize=10000)
normalized = pandas_transformer(csv_reader, normalize_dataframe)
filtered = pandas_transformer(normalized, lambda df: df[df['value'] > 0])

def timeseries_simulator(df, time_col='timestamp'):
df = df.sort_values(time_col)
for _, row in df.iterrows():
    yield row.to_dict()
    time.sleep(0.01)  # Имитация потока данных в реальном времени

def log_yields(func):
def wrapper(*args, **kwargs):
    gen = func(*args, **kwargs)
    for value in gen:
        print(f"Генератор {func.__name__} выдал: {value}")
        yield value
return wrapper
 
@log_yields
def fibonacci(n):
a, b = 0, 1
for _ in range(n):
    yield a
    a, b = b, a + b
 
# Использование
for num in fibonacci(5):
pass  # Вывод логов для каждого значения

def reusable(generator_func):
def wrapper(*args, **kwargs):
    generator_args = args
    generator_kwargs = kwargs
    
    gen = generator_func(*generator_args, **generator_kwargs)
    cache = []
    
    while True:
        try:
            value = next(gen)
            cache.append(value)
            yield value
        except StopIteration:
            if not cache:
                break
            # Если генератор исчерпан, начинаем выдавать значения из кэша
            for item in cache:
                yield item
            # Создаём новый генератор для следующего цикла
            gen = generator_func(*generator_args, **generator_kwargs)
            cache = []
                
return wrapper

def transform_yields(transform_func):
def decorator(generator_func):
    def wrapper(*args, **kwargs):
        for item in generator_func(*args, **kwargs):
            yield transform_func(item)
    return wrapper
return decorator
 
@transform_yields(lambda x: x * 2)
def count_to(n):
for i in range(n):
    yield i
 
# Все значения будут умножены на 2
print(list(count_to(5)))  # [0, 2, 4, 6, 8]

def throttle(delay=0.1):
def decorator(gen_func):
    def wrapper(*args, **kwargs):
        for item in gen_func(*args, **kwargs):
            time.sleep(delay)  # Пауза между выдачей значений
            yield item
    return wrapper
return decorator
 
@throttle(0.5)
def sensor_simulator():
while True:
    yield random.random() * 100  # Имитация показаний датчика

def with_file(file_path, mode='r'):
def decorator(gen_func):
    def wrapper(*args, **kwargs):
        with open(file_path, mode) as file:
            yield from gen_func(file, *args, **kwargs)
    return wrapper
return decorator
 
@with_file('data.txt')
def process_lines(file):
for line in file:
    yield line.strip().upper()

def stage(name):
def decorator(gen_func):
    gen_func.stage_name = name
    return gen_func
return decorator
 
@stage("Извлечение")
def extract(source):
# Извлечение данных
yield from source
 
@stage("Трансформация")
def transform(data):
for item in data:
    # Преобразование
    yield transformed_item
 
@stage("Загрузка")
def load(data):
for item in data:
    # Загрузка в базу
    store(item)
    yield item
 
# Можно создать DSL для сборки конвеера
pipeline = Pipeline()
pipeline.add(extract)
pipeline.add(transform)
pipeline.add(load)
pipeline.run(source_data)

from concurrent.futures import ThreadPoolExecutor
 
def expensive_operation(x):
    time.sleep(0.5)  # Имитация тяжёлых вычислений
    return x * x
 
def parallel_map(func, generator, max_workers=4):
    with ThreadPoolExecutor(max_workers) as executor:
        yield from executor.map(func, generator)
 
# Использование
numbers = (i for i in range(20))
results = parallel_map(expensive_operation, numbers)
for result in results:
    print(result)

from queue import Queue
from threading import Thread
 
def producer(generator, queue, num_items):
    for item in generator:
        queue.put(item)
    for _ in range(num_consumers):
        queue.put(None)  # Сигнал о завершении
 
def consumer(queue, results):
    while True:
        item = queue.get()
        if item is None:
            break
        results.append(process_item(item))
        queue.task_done()
 
# Настройка конвейера
queue = Queue(maxsize=100)
results = []
producer_thread = Thread(target=producer, args=(source_generator(), queue, 3))
consumer_threads = [Thread(target=consumer, args=(queue, results)) for _ in range(3)]
 
# Запуск
producer_thread.start()
for t in consumer_threads:
    t.start()

def merge_generators(*generators):
    queues = [Queue() for _ in generators]
    results = Queue()
    
    def producer(gen, q):
        for item in gen:
            q.put(item)
        q.put(None)  # Сигнал завершения
    
    def consumer():
        active = len(queues)
        while active > 0:
            for i, q in enumerate(queues):
                try:
                    item = q.get_nowait()
                    if item is None:
                        active -= 1
                    else:
                        results.put(process_item(item))
                except Empty:
                    pass
        results.put(None)
    
    # Запуск потоков
    for i, gen in enumerate(generators):
        Thread(target=producer, args=(gen, queues[i])).start()
    Thread(target=consumer).start()
    
    # Возвращаем результаты через генератор
    while True:
        item = results.get()
        if item is None:
            break
        yield item

def parallel_generator(max_workers=4, chunk_size=10):
    def decorator(gen_func):
        def wrapper(*args, **kwargs):
            source_gen = gen_func(*args, **kwargs)
            
            # Разбиваем на чанки для эффективности
            def get_chunk():
                chunk = []
                for _ in range(chunk_size):
                    try:
                        chunk.append(next(source_gen))
                    except StopIteration:
                        break
                return chunk
            
            with ThreadPoolExecutor(max_workers) as executor:
                while True:
                    chunk = get_chunk()
                    if not chunk:
                        break
                    futures = [executor.submit(process_item, item) for item in chunk]
                    for future in as_completed(futures):
                        yield future.result()
                        
        return wrapper
    return decorator

def parse_log_entries(log_directory):
for filename in os.listdir(log_directory):
    if not filename.endswith('.log'):
        continue
    with open(os.path.join(log_directory, filename), 'r') as f:
        for line in f:
            # Парсинг строки лога
            if line.strip():
                try:
                    entry = json.loads(line)
                    yield entry
                except json.JSONDecodeError:
                    continue
 
def analyze_errors(entries):
for entry in entries:
    if entry.get('level') == 'ERROR':
        # Анализ ошибки
        yield {
            'timestamp': entry.get('timestamp'),
            'service': entry.get('service'),
            'message': entry.get('message'),
            'severity': calculate_severity(entry)
        }
 
# Полный конвейер анализа
logs = parse_log_entries('/var/log/application')
errors = analyze_errors(logs)
severe_issues = filter(lambda e: e['severity'] > 8, errors)
 
# Генерируем отчёт
for issue in severe_issues:
    add_to_report(issue)

def device_data_receiver(mqtt_client):
queue = Queue()
mqtt_client.on_message = lambda client, userdata, msg: queue.put(msg.payload)
 
while True:
    if not queue.empty():
        data = queue.get()
        try:
            decoded = json.loads(data)
            yield decoded
        except Exception:
            continue
    await asyncio.sleep(0.01)
 
def anomaly_detector(data_stream, thresholds):
for data_point in data_stream:
    device_id = data_point.get('device_id')
    readings = data_point.get('readings', {})
    
    if device_id in thresholds:
        for metric, threshold in thresholds[device_id].items():
            if metric in readings and readings[metric] > threshold:
                yield {
                    'device_id': device_id,
                    'metric': metric,
                    'value': readings[metric],
                    'threshold': threshold,
                    'timestamp': data_point.get('timestamp')
                }

def market_data_generator(start_date, end_date=None):
current = start_date
today = datetime.now().date()
end = end_date if end_date else today
    
while current <= end:
    filename = f'market_data_{current.strftime("%Y-%m-%d")}.csv'
    if os.path.exists(filename):
        with open(filename, 'r') as f:
            # Пропускаем заголовок
            next(f)
            for line in f:
                yield parse_market_data(line)
    current += timedelta(days=1)
 
def strategy_backtester(data_stream, strategy_func):
portfolio = init_portfolio()
for tick in data_stream:
    signal = strategy_func(tick, portfolio)
    if signal:
        execute_trade(portfolio, signal, tick)
    yield {
        'timestamp': tick['timestamp'],
        'portfolio_value': calculate_portfolio_value(portfolio, tick),
        'signal': signal
    }

def read_fastq(filename):
with gzip.open(filename, 'rt') if filename.endswith('.gz') else open(filename, 'r') as f:
    while True:
        header = f.readline().strip()
        if not header:
            break
        sequence = f.readline().strip()
        f.readline()  # Пропускаем "+"
        quality = f.readline().strip()
        
        yield {
            'header': header,
            'sequence': sequence,
            'quality': quality
        }
 
def quality_filter(reads, min_quality=30):
for read in reads:
    avg_quality = sum(ord(q) - 33 for q in read['quality']) / len(read['quality'])
    if avg_quality >= min_quality:
        yield read

import inspect
 
def my_generator():
    x = 0
    while x < 5:
        x += 1
        yield x
 
gen = my_generator()
print(next(gen))  # 1
print(inspect.getgeneratorstate(gen))  # 'GEN_SUSPENDED'

print(next(gen))  # 2
print(inspect.getgeneratorlocals(gen))  # {'x': 2}

def debug_generator(func):
    def wrapper(*args, **kwargs):
        gen = func(*args, **kwargs)
        while True:
            try:
                value = next(gen)
                print(f"Генератор {func.__name__} вернул: {value}")
                yield value
            except StopIteration:
                print(f"Генератор {func.__name__} завершился")
                break
    return wrapper
 
@debug_generator
def problematic_generator():
    for i in range(5):
        if i == 3:
            i = i / (i - 3)  # Ошибка деления на ноль
        yield i

def memory_profiler_decorator(gen_func):
    def wrapper(*args, **kwargs):
        import tracemalloc
        tracemalloc.start()
        snapshot1 = tracemalloc.take_snapshot()
        
        gen = gen_func(*args, **kwargs)
        for value in gen:
            yield value
        
        snapshot2 = tracemalloc.take_snapshot()
        top_stats = snapshot2.compare_to(snapshot1, 'lineno')
        print("[ Потребление памяти ]")
        for stat in top_stats[:5]:
            print(stat)
    return wrapper

def profile_generator(gen, sample_size=1000):
    import time
    total_time = 0
    items_processed = 0
    start_time = time.time()
    
    try:
        for _ in range(sample_size):
            next(gen)
            items_processed += 1
            if items_processed % 100 == 0:
                curr_time = time.time() - start_time
                print(f"Обработано {items_processed} элементов за {curr_time:.2f} секунд")
                total_time = curr_time
    except StopIteration:
        pass
    
    print(f"Итог: {items_processed} элементов за {total_time:.2f} секунд")
    print(f"Среднее время на элемент: {(total_time/items_processed)*1000:.4f} мс")

from contextlib import contextmanager
 
@contextmanager
def catch_generator_errors(gen_name):
    try:
        yield
    except Exception as e:
        print(f"Ошибка в генераторе {gen_name}: {e}")
        raise

def debug_with_pdb():
    for i in range(10):
        # Встроенное условие для активации отладчика
        if i > 5 and i % 2 == 0:
            import pdb; pdb.set_trace()
        yield i

def analyze_transactions():
    # Загружаем все транзакции в память
    transactions = database.fetch_all_transactions()
    
    # Фильтруем
    suspicious = [t for t in transactions if is_suspicious(t)]
    
    # Группируем по клиентам
    by_customer = {}
    for t in suspicious:
        if t.customer_id not in by_customer:
            by_customer[t.customer_id] = []
        by_customer[t.customer_id].append(t)
    
    # Анализируем паттерны и создаём отчёты
    reports = []
    for customer_id, customer_transactions in by_customer.items():
        report = analyze_patterns(customer_id, customer_transactions)
        reports.append(report)
    
    return reports

def analyze_transactions():
    # Используем генератор вместо загрузки всех данных
    transactions = database.stream_transactions()
    
    # Фильтруем на лету
    suspicious = (t for t in transactions if is_suspicious(t))
    
    # Группировка требует полного прохода по данным
    # Здесь генераторы не спасут, но можно оптимизировать по-другому
    by_customer = {}
    for t in suspicious:
        by_customer.setdefault(t.customer_id, []).append(t)
    
    # Генерируем отчёты по одному
    for customer_id, customer_transactions in by_customer.items():
        yield analyze_patterns(customer_id, customer_transactions)

def stream_transactions(batch_size=1000):
    offset = 0
    while True:
        batch = fetch_transactions_batch(offset, batch_size)
        if not batch:
            break
        for transaction in batch:
            yield transaction
        offset += batch_size

def parallel_analyze(customer_batches, max_workers=4):
    with ThreadPoolExecutor(max_workers) as executor:
        futures = []
        for batch in customer_batches:
            future = executor.submit(analyze_batch, batch)
            futures.append(future)
        
        for future in as_completed(futures):
            yield future.result()

def compatibility_wrapper(generator_func):
    def wrapper(*args, **kwargs):
        return list(generator_func(*args, **kwargs))
    return wrapper
 
@compatibility_wrapper
def legacy_api_method():
    # Внутри используем генераторы
    for item in process_data():
        yield item