实现生成器链式调用：正弦波生成器功能扩展问题咨询

报错的核心原因是你的add_noise方法返回的是Python原生生成器对象，而非带有to_pandas等方法的自定义类实例，导致链式调用到to_pandas时找不到对应属性。

要实现类似TensorFlow Dataset的链式调用，核心是用自定义类封装生成器逻辑，所有处理方法都返回该类的新实例，而非原生生成器。这样既能保证链式调用的连贯性，又能实现高度可扩展性。

1. 通用Dataset类实现 #

这个类封装生成器，并提供通用的处理、输出方法，所有操作都返回类实例以支持链式调用：

import pandas as pd
import numpy as np
from typing import Generator, Callable

class CustomDataset:
    def __init__(self, generator: Generator):
        self.generator = generator

    # 通用转换接口，接收处理函数，返回新的Dataset实例
    def _transform(self, func: Callable) -> 'CustomDataset':
        def transformed_gen():
            for time, val in self.generator:
                yield func(time, val)
        return CustomDataset(transformed_gen())

    # 给数值添加噪声，time值保持不变
    def add_noise(self, mean: float = 0, std: float = 1) -> 'CustomDataset':
        def add_noise_func(time, val):
            noise = np.random.normal(mean, std)
            return time, val + noise
        return self._transform(add_noise_func)

    # 打乱数据（生成器流式特性决定需用缓冲区实现近似打乱）
    def shuffle(self, buffer_size: int = 1000) -> 'CustomDataset':
        buffer = []
        # 先填充缓冲区
        try:
            for _ in range(buffer_size):
                buffer.append(next(self.generator))
        except StopIteration:
            pass
        
        def shuffle_gen():
            np.random.shuffle(buffer)
            for item in buffer:
                yield item
            # 流式处理剩余数据，每加入一个就打乱缓冲区并输出
            for item in self.generator:
                buffer.append(item)
                np.random.shuffle(buffer)
                yield buffer.pop()
        return CustomDataset(shuffle_gen())

    # 转换为Pandas DataFrame
    def to_pandas(self) -> pd.DataFrame:
        data = list(self.generator)
        return pd.DataFrame(data, columns=['time', 'value'])

2. 独立的生成器函数 #

将数据生成逻辑与处理逻辑分离，更换生成器时无需修改Dataset类：

# 正弦波生成器
def sine_wave_generator(num_samples: int, freq: float = 1.0) -> Generator:
    for i in range(num_samples):
        time = i / 100.0
        sine_val = np.sin(2 * np.pi * freq * time)
        yield time, sine_val

# 对数生成器示例（直接复用所有处理方法）
def log_generator(num_samples: int) -> Generator:
    for i in range(1, num_samples + 1):
        time = i / 10.0
        log_val = np.log(time)
        yield time, log_val

3. 链式调用示例 #

# 正弦波生成器链式调用
ds = CustomDataset(sine_wave_generator(1000))
noisy_sine_df = ds.add_noise(mean=0, std=0.1).shuffle().to_pandas()
print(noisy_sine_df.head())

# 对数生成器直接复用处理方法
log_ds = CustomDataset(log_generator(100))
noisy_log_df = log_ds.add_noise(std=0.05).to_pandas()
print(noisy_log_df.head())

1. 分离职责：生成器函数负责纯数据生成，Dataset类负责通用处理逻辑，符合开闭原则，更换生成器无需修改处理代码。
2. 统一返回实例：所有处理方法都返回CustomDataset实例，保证链式调用的连贯性，避免返回原生生成器丢失方法。
3. 流式优先：除shuffle这类必须缓存的操作，尽量保持流式生成，避免一次性加载大量数据到内存，适配大规模数据场景。
4. 类型提示：使用typing模块添加类型提示，提升代码可读性和维护性。

• 生成器只能遍历一次，每次链式调用后的新实例是独立的可遍历对象，不要重复使用同一个实例多次遍历。
• shuffle的buffer_size需根据数据规模调整：值越大打乱效果越好，但内存占用越高。

内容的提问来源于stack exchange，提问作者J Agustin Barrachina