基于Python sounddevice的异步传感器音频反馈代码优化咨询

1. 消除全局变量，封装状态 #

原代码用global start_idx管理音频相位，易引发线程安全问题且不利于扩展。建议将音频相关状态封装到类中，通过实例属性维护：

class AudioFeedback:
    def __init__(self, sensor, samplerate=None, freq=500.0):
        self.sensor = sensor
        self.samplerate = samplerate or sd.query_devices(None, 'output')['default_samplerate']
        self.freq = freq
        self.phase = 0.0  # 用相位累积替代start_idx，避免浮点溢出
        self.lock = asyncio.Lock()

    def callback(self, outdata, frames, time, status):
        if status:
            print(status, file=sys.stderr)
        # 线程安全获取最新传感器数据
        with self.sensor.lock:
            y = self.sensor.data[-1]
        
        error = abs(self.sensor.target - y)
        scale_factor = error / self.sensor.target if self.sensor.target != 0 else 0

        # 生成正弦波，用相位循环累加避免精度误差
        phase_step = 2 * np.pi * self.freq / self.samplerate
        phases = self.phase + phase_step * np.arange(frames)
        sine_wave = 0.5 * np.sin(phases).reshape(-1, 1)
        self.phase = phases[-1] % (2 * np.pi)

        if error <= self.sensor.threshold:
            outdata[:] = sine_wave
        else:
            noise = scale_factor * np.random.normal(0, 1, size=(frames, 1))
            outdata[:] = sine_wave + noise

2. 保证线程安全 #

sounddevice回调运行在独立后台线程，传感器数据读取在asyncio事件循环线程，直接访问sensor.data存在竞态风险。给Sensor类添加锁：

class Sensor(): 
    def __init__(self):
        self.start_time = timer()
        self.target = 5.
        self.threshold = 0.2
        self.sample_frequency = 40.
        self.input_period = 10
        self.data = [0]
        self.lock = asyncio.Lock()  # 线程安全锁

    async def read(self):
        while True:
            time = timer()-self.start_time
            new_data = self.input(time)
            async with self.lock:
                self.data.append(new_data)
                # 限制列表长度，避免内存泄漏
                if len(self.data) > 10:
                    self.data.pop(0)
            await asyncio.sleep(1./self.sample_frequency)

3. 优化信号生成与代码简洁性 #

• 移除自定义white函数，用np.random.normal直接生成白噪声，语义更清晰
• 把硬编码的设备ID、频率等改为可配置参数，提升灵活性
• 拆分音频流初始化逻辑，增强可读性：

async def play_audio(audio_feedback):
    event = asyncio.Event()
    stream = sd.OutputStream(
        device=None,  # 使用默认输出设备
        channels=1, 
        callback=audio_feedback.callback,
        samplerate=audio_feedback.samplerate
    )
    with stream:
        await event.wait()

4. 为复杂回调逻辑做扩展性准备 #

• 将音频效果拆分为独立方法，便于后续添加新反馈模式：

class AudioFeedback:
    # ... 初始化代码 ...
    def _generate_pure_tone(self, frames):
        phase_step = 2 * np.pi * self.freq / self.samplerate
        phases = self.phase + phase_step * np.arange(frames)
        tone = 0.5 * np.sin(phases).reshape(-1, 1)
        self.phase = phases[-1] % (2 * np.pi)
        return tone

    def _generate_noisy_tone(self, frames, scale_factor):
        pure_tone = self._generate_pure_tone(frames)
        noise = scale_factor * np.random.normal(0, 1, size=(frames, 1))
        return pure_tone + noise

    def callback(self, outdata, frames, time, status):
        try:
            with self.sensor.lock:
                y = self.sensor.data[-1]
            error = abs(self.sensor.target - y)
            scale_factor = error / self.sensor.target if self.sensor.target != 0 else 0

            if error <= self.sensor.threshold:
                outdata[:] = self._generate_pure_tone(frames)
            else:
                outdata[:] = self._generate_noisy_tone(frames, scale_factor)
        except Exception as e:
            print(f"Callback error: {e}", file=sys.stderr)
            outdata.fill(0)  # 出错时输出静音避免爆音

• 在Sensor类中添加事件通知机制（如asyncio.Event），当数据触发特定事件时直接通知音频模块调整输出。

5. 错误处理增强 #

在回调中添加异常捕获，避免单个回调崩溃导致音频流终止；用日志替代print，便于生产环境调试。

内容的提问来源于stack exchange，提问作者oweydd