Python进阶指南之无限序列生成器的实践指南

def my_generator():
    yield 1
    yield 2
    yield 3

gen = my_generator()
print(next(gen))  # 1
print(next(gen))  # 2
print(next(gen))  # 3

gen_expr = (x * 2 for x in range(5))
print(list(gen_expr))  # [0, 2, 4, 6, 8]

def infinite_sequence():
    num = 1
    while True:
        yield num
        num += 1

# 测试
seq = infinite_sequence()
print("前 10 个数：")
for i in range(10):
    print(next(seq), end=" ")

def infinite_sequence(start=1, step=1):
    """
    生成一个无限序列，从 start 开始，每次增加 step
    
    Args:
        start: 起始值，默认为 1
        step: 步长，默认为 1
    """
    current = start
    while True:
        yield current
        current += step

# 测试
print("从 5 开始，步长为 3 的序列（前 8 项）：")
seq = infinite_sequence(start=5, step=3)
for i in range(8):
    print(next(seq), end=" ")

def fibonacci_generator():
    a, b = 0, 1
    while True:
        if a > 1000:
            break  # 终止条件
        yield a
        a, b = b, a + b

# 使用
print("斐波那契数列（小于 1000）：")
fib = fibonacci_generator()
for num in fib:
    print(num, end=" ")

import random
import time

def sensor_data_stream():
    """模拟每秒发送一次温度数据"""
    timestamp = 0
    while True:
        temp = round(random.uniform(20.0, 30.0), 1)
        yield {
            "timestamp": timestamp,
            "temperature": temp,
            "unit": "°C"
        }
        timestamp += 1
        time.sleep(1)  # 模拟真实延迟

# 使用：只获取前 5 条数据
print("📊 实时传感器数据（前 5 条）：")
data_gen = sensor_data_stream()
for _ in range(5):
    reading = next(data_gen)
    print(f"时间戳 {reading['timestamp']}: {reading['temperature']}°C")

def gen():
    yield 1
    yield 2

g = gen()
print(g.__iter__ is g)  # True
print(hasattr(g, '__next__'))  # True

def controlled_counter():
    value = 0
    while True:
        try:
            increment = yield value
            if increment is not None:
                value += increment
            else:
                value += 1
        except Exception as e:
            print(f"错误：{e}")
            break

# 使用
counter = controlled_counter()
print(next(counter))  # 0
print(next(counter))  # 1
print(counter.send(5))  # 6
counter.throw(ValueError, "强制中断")

class InfiniteSequence:
    def __init__(self, start=1, step=1):
        self.start = start
        self.step = step
        self.current = start

    def __iter__(self):
        return self

    def __next__(self):
        value = self.current
        self.current += self.step
        return value

    def reset(self):
        self.current = self.start

# 测试
print("🔁 无限序列类测试：")
seq = InfiniteSequence(start=10, step=2)
for i in range(6):
    print(next(seq), end=" ")

import time

def generate_list(n):
    return [i for i in range(n)]

def generate_generator(n):
    return (i for i in range(n))

# 测试
n = 1_000_000

# 列表方式
start_time = time.time()
lst = generate_list(n)
list_time = time.time() - start_time
print(f"✅ 列表生成耗时：{list_time:.4f} 秒")

# 生成器方式
start_time = time.time()
gen = generate_generator(n)
gen_time = time.time() - start_time
print(f"🚀 生成器初始化耗时：{gen_time:.4f} 秒")

# 内存占用对比（简化示意）
print(f"🧠 列表占用内存 ≈ {len(lst) * 28} 字节")  # 大致估算
print(f"💡 生成器几乎不占内存")

import unittest

class TestInfiniteSequence(unittest.TestCase):

    def test_natural_numbers(self):
        seq = infinite_sequence(start=1, step=1)
        for i in range(1, 6):
            self.assertEqual(next(seq), i)

    def test_custom_start_step(self):
        seq = infinite_sequence(start=10, step=2)
        expected = [10, 12, 14, 16, 18]
        actual = [next(seq) for _ in range(5)]
        self.assertEqual(actual, expected)

    def test_reset_with_class(self):
        seq = InfiniteSequence(start=5, step=1)
        first = [next(seq) for _ in range(3)]
        seq.reset()
        second = [next(seq) for _ in range(3)]
        self.assertEqual(first, [5, 6, 7])
        self.assertEqual(second, [5, 6, 7])

if __name__ == "__main__":
    unittest.main(argv=[''], exit=False, verbosity=2)

count = 0
for item in infinite_sequence():
    if count >= 10:
        break
    print(item)
    count += 1