robot-project/wake_word_module/aec_processor.py

"""
AEC回音消除处理器
支持Windows和Linux平台的系统级回音消除功能
"""

import platform
from typing import Optional

import numpy as np


class AECProcessor:
    """音频回声消除处理器（Windows/Linux系统级AEC）"""

    def __init__(self, sample_rate: int = 16000, channels: int = 1):
        """
        初始化AEC处理器

        Args:
            sample_rate: 采样率
            channels: 声道数
        """
        # 平台信息
        self._platform = platform.system().lower()
        self._is_windows = self._platform == "windows"
        self._is_linux = self._platform == "linux"

        # 音频参数
        self.sample_rate = sample_rate
        self.channels = channels

        # 状态标志
        self._is_initialized = False
        self._is_closing = False

    async def initialize(self):
        """初始化AEC处理器"""
        try:
            if self._is_windows or self._is_linux:
                # Windows 和 Linux 平台使用系统级AEC，无需额外处理
                print(f"{self._platform.capitalize()} 平台使用系统级回声消除")
                self._is_initialized = True
                return  print("AEC处理器初始化完成")
            else:
                print(f"当前平台 {self._platform} 暂不支持AEC功能")
                print("提示: 本模块仅支持Windows和Linux平台")
                self._is_initialized = True
                return  print("AEC处理器初始化失败")
           

        except Exception as e:
            print(f"AEC处理器初始化失败: {e}")
            await self.close()
            raise

    def process_audio(self, capture_audio: np.ndarray) -> np.ndarray:
        """处理音频帧，应用AEC

        Args:
            capture_audio: 麦克风采集的音频数据 (16kHz, int16)

        Returns:
            处理后的音频数据
        """
        if not self._is_initialized:
            return capture_audio

        # Windows 和 Linux 平台直接返回原始音频（系统级处理）
        if self._is_windows or self._is_linux:
            # 系统级AEC已自动处理，无需额外处理
            return capture_audio

        # 其他平台直接返回原始音频
        return capture_audio

    async def close(self):
        """关闭AEC处理器"""
        if self._is_closing:
            return

        self._is_closing = True
        print("开始关闭AEC处理器...")

        try:
            # Windows/Linux平台无需清理额外资源
            print("AEC处理器已关闭（系统级AEC无需额外清理）")

            self._is_initialized = False

        except Exception as e:
            print(f"关闭AEC处理器时发生错误: {e}")

    @property
    def is_initialized(self) -> bool:
        """返回是否已初始化"""
        return self._is_initialized

    @property
    def platform(self) -> str:
        """返回当前平台"""
        return self._platform