robot-project/wake_word_module/example.py

"""
语音唤醒模块使用示例（支持AEC回音消除）
演示如何使用独立的语音唤醒模块，包括AEC功能
"""

import asyncio
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from wake_word_module import WakeWordDetector, AudioCapture, WakeWordConfig, AECProcessor


async def main():
    """主函数"""
    print("=" * 60)
    print("语音唤醒模块示例（支持AEC）")
    print("=" * 60)

    # 1. 列出可用音频设备
    print("\n1. 列出可用音频设备:")
    AudioCapture.list_devices()

    # 2. 创建配置
    print("\n2. 创建语音唤醒配置:")
    # 假设模型文件在 models 目录下
    model_path = Path("models")

    if not model_path.exists():
        print(f"错误: 模型目录不存在: {model_path}")
        print("请确保以下文件存在于模型目录中:")
        print("encoder.onnx")
        print("decoder.onnx")
        print("joiner.onnx")
        print("tokens.txt")
        print("keywords.txt")
        return 

    config = WakeWordConfig(
        model_path=str(model_path),
        sample_rate=16000,
        num_threads=4,
        provider="cpu",
        keywords_score=1.8,
        keywords_threshold=0.2,
        detection_cooldown=1.5,
    )

    print(f"   模型路径: {config.model_path}")
    print(f"   采样率: {config.sample_rate}Hz")
    print(f"   检测阈值: {config.keywords_threshold}")

    # 3. 创建音频采集器（启用AEC）
    print("\n3. 创建音频采集器:")
    print("   是否启用AEC回音消除: 是")
    audio_capture = AudioCapture(
        sample_rate=16000,
        channels=1,
        device_id=None,  # 自动选择设备
        enable_aec=True,  # 启用AEC回音消除
    )

    # 4. 创建唤醒词检测器
    print("\n4. 创建唤醒词检测器:")
    detector = WakeWordDetector(config)

    # 5. 设置回调函数
    def on_wake_word_detected(result, full_text):
        """唤醒词检测回调"""
        print("\n" + "=" * 60)
        print("🎉 检测到唤醒词!唤醒成功!✅")
        print(f"   结果: {result}")
        print(f"   完整文本: {full_text}")
        print("=" * 60 + "\n")

    def on_error(error):
        """错误回调"""
        print(f"❌ 唤醒词检测错误: {error}")

    detector.on_detected(on_wake_word_detected)
    detector.on_error(on_error)

    # 6. 连接音频采集器和检测器
    print("\n5. 连接音频采集器和检测器:")
    audio_capture.add_audio_listener(detector)

    # 7. 启动系统
    print("\n6. 启动系统:")
    try:
        await audio_capture.start()
        await detector.start()

        print("\n✅ 语音唤醒系统已启动!")
        print("   AEC回音消除: 已启用")
        print("   请说出唤醒词进行测试...")
        print("   按 Ctrl+C 停止程序\n")

        # 保持运行
        while True:
            await asyncio.sleep(1)

    except KeyboardInterrupt:
        print("\n\n收到停止信号...")

    except Exception as e:
        print(f"\n❌ 发生错误: {e}")

    finally:
        # 8. 清理资源
        print("\n7. 清理资源:")
        await detector.stop()
        await audio_capture.stop()
        print("✅ 资源已清理完毕")


async def main_without_aec():
    """不使用AEC的示例"""
    print("=" * 60)
    print("语音唤醒模块示例（不使用AEC）")
    print("=" * 60)

    # 1. 列出可用音频设备
    print("\n1. 列出可用音频设备:")
    AudioCapture.list_devices()

    # 2. 创建配置
    print("\n2. 创建语音唤醒配置:")
    model_path = Path("models")

    if not model_path.exists():
        print(f"错误: 模型目录不存在: {model_path}")
        return

    config = WakeWordConfig(
        model_path=str(model_path),
        sample_rate=16000,
    )

    # 3. 创建音频采集器（不启用AEC）
    print("\n3. 创建音频采集器:")
    print("   是否启用AEC回音消除: 否")
    audio_capture = AudioCapture(
        sample_rate=16000,
        channels=1,
        device_id=None,
        enable_aec=False,  # 不启用AEC
    )

    # 4. 创建唤醒词检测器
    detector = WakeWordDetector(config)

    # 5. 设置回调
    def on_wake_word_detected(result, full_text):
        print(f"\n🎉 检测到唤醒词: {result},唤醒成功!✅")
        print(f"   完整文本: {full_text}")

    detector.on_detected(on_wake_word_detected)

    # 6. 连接并启动
    audio_capture.add_audio_listener(detector)
    await audio_capture.start()
    await detector.start()

    print("\n✅ 语音唤醒系统已启动! (不使用AEC)")
    print("   请说出唤醒词进行测试...")
    print("   按 Ctrl+C 停止程序\n")

    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        print("\n\n收到停止信号...")
    finally:
        await detector.stop()
        await audio_capture.stop()
        print("✅ 资源已清理完毕")


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="语音唤醒模块示例")
    parser.add_argument(
        "--no-aec",
        action="store_true",
        help="不使用AEC回音消除"
    )

    args = parser.parse_args()

    try:
        if args.no_aec:
            asyncio.run(main_without_aec())
        else:
            asyncio.run(main())
    except KeyboardInterrupt:
        print("\n程序已退出")
        sys.exit(0)