Python迭代器和生成器

class CountDown:
    def __init__(self, start):
        self.start = start

    def __iter__(self):
        return self

    def __next__(self):
        if self.start <= 0:
            raise StopIteration
        self.start -= 1
        return self.start + 1

# 使用自定义迭代器
counter = CountDown(5)
for num in counter:
    print(num)  # 输出：5, 4, 3, 2, 1
                    

# 列表推导式（一次性生成所有元素）
squares_list = [x**2 for x in range(1000000)]  # 占用大量内存

# 生成器表达式（按需生成元素）
squares_gen = (x**2 for x in range(1000000))   # 内存高效

# 使用生成器表达式
for square in squares_gen:
    if square > 100:
        break
    print(square)
                    

使用生成器处理大型日志文件，避免一次性将整个文件加载到内存中：

def parse_log_file(file_path):
    """分析大型日志文件的生成器"""
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            # 解析每行日志
            if '[ERROR]' in line:
                timestamp = line[0:19]
                error_msg = line[line.find('[ERROR]')+7:].strip()
                yield {
                    'timestamp': timestamp,
                    'level': 'ERROR',
                    'message': error_msg
                }

# 使用示例
for error_log in parse_log_file('app.log'):
    print(f"时间: {error_log['timestamp']}")
    print(f"错误信息: {error_log['message']}\n")
                    

使用生成器创建数据处理管道，实现ETL（提取、转换、加载）过程：

def extract_data(csv_file):
    """从CSV文件提取数据"""
    with open(csv_file, 'r', encoding='utf-8') as f:
        for line in f:
            yield line.strip().split(',')

def transform_data(rows):
    """转换数据：计算学生平均分"""
    for row in rows:
        if len(row) >= 4:  # 假设格式：姓名,语文,数学,英语
            name = row[0]
            scores = [float(score) for score in row[1:]]
            average = sum(scores) / len(scores)
            yield (name, average)

def load_data(results):
    """加载数据：将结果写入新文件"""
    with open('results.txt', 'w', encoding='utf-8') as f:
        for name, avg in results:
            f.write(f"{name}: {avg:.2f}\n")
            yield f"{name}的平均分处理完成"

# 使用示例
data = extract_data('scores.csv')
transformed = transform_data(data)
for status in load_data(transformed):
    print(status)
                    

使用生成器处理大量图片文件，实现批量图片处理：

from PIL import Image
import os

def image_processor(directory):
    """处理目录中的所有图片"""
    for filename in os.listdir(directory):
        if filename.endswith(('.png', '.jpg', '.jpeg')):
            image_path = os.path.join(directory, filename)
            try:
                with Image.open(image_path) as img:
                    # 调整图片大小
                    resized = img.resize((800, 600))
                    # 保存处理后的图片
                    new_path = os.path.join(directory, f'processed_{filename}')
                    resized.save(new_path)
                    yield f"已处理: {filename}"
            except Exception as e:
                yield f"处理失败 {filename}: {str(e)}"

# 使用示例
for status in image_processor('images/'):
    print(status)
                    

使用生成器实现实时数据监控和处理：

import time
from datetime import datetime

def sensor_data_simulator():
    """模拟传感器数据流"""
    while True:
        yield {
            'timestamp': datetime.now(),
            'temperature': random.uniform(20, 30),
            'humidity': random.uniform(40, 60)
        }
        time.sleep(1)  # 模拟每秒产生一个数据点

def alert_monitor(data_stream, temp_threshold=28):
    """监控温度异常"""
    for data in data_stream:
        if data['temperature'] > temp_threshold:
            yield {
                'time': data['timestamp'],
                'alert': f"温度过高: {data['temperature']:.1f}°C"
            }

def data_logger(alert_stream):
    """记录警报信息"""
    with open('temperature_alerts.log', 'a', encoding='utf-8') as f:
        for alert in alert_stream:
            log_entry = f"[{alert['time']}] {alert['alert']}\n"
            f.write(log_entry)
            yield "警报已记录"

# 使用示例
sensor_data = sensor_data_simulator()
alerts = alert_monitor(sensor_data)
for status in data_logger(alerts):
    print(status)
    if input("继续监控？(y/n): ").lower() != 'y':
        break