robot-project/wake_word_module/test.py

"""
语音唤醒模块测试脚本
测试各个组件的基本功能
"""

import asyncio
import sys
from pathlib import Path

# 添加模块路径
# sys.path.insert(0, str(Path(__file__).parent))
sys.path.insert(0, str(Path(__file__).parent.parent))
from wake_word_module import WakeWordDetector, AudioCapture, WakeWordConfig


def test_config():
    """测试配置类"""
    print("\n" + "=" * 60)
    print("测试 1: 配置类")
    print("=" * 60)

    try:
        config = WakeWordConfig(
            model_path="models",
            sample_rate=16000,
            keywords_threshold=0.2,
        )

        print(f"✅ 配置创建成功")
        print(f"   采样率: {config.sample_rate}Hz")
        print(f"   阈值: {config.keywords_threshold}")
        print(f"   冷却时间: {config.detection_cooldown}s")

        # 尝试验证（如果模型文件存在）
        try:
            config.validate()
            print(f"✅ 配置验证通过")
        except FileNotFoundError as e:
            print(f"⚠️  配置验证跳过（模型文件不存在）: {e}")

        return True

    except Exception as e:
        print(f"❌ 配置测试失败: {e}")
        return False


def test_audio_capture():
    """测试音频采集器"""
    print("\n" + "=" * 60)
    print("测试 2: 音频采集器")
    print("=" * 60)

    try:
        # 列出设备
        print("\n列出可用设备:")
        AudioCapture.list_devices()

        # 创建采集器
        audio_capture = AudioCapture(
            sample_rate=16000,
            channels=1,
            device_id=None,
        )

        print(f"✅ 音频采集器创建成功")
        print(f"   采样率: {audio_capture.sample_rate}Hz")
        print(f"   声道数: {audio_capture.channels}")
        print(f"   帧大小: {audio_capture.frame_size}")

        return True

    except Exception as e:
        print(f"❌ 音频采集器测试失败: {e}")
        return False


async def test_detector():
    """测试检测器"""
    print("\n" + "=" * 60)
    print("测试 3: 检测器")
    print("=" * 60)

    try:
        config = WakeWordConfig(
            model_path="models",
            sample_rate=16000,
        )

        # 尝试创建检测器
        try:
            detector = WakeWordDetector(config)
            print(f"✅ 检测器创建成功")
            print(f"   配置: {detector.config.model_path}")

            # 测试回调设置
            detector.on_detected(lambda r, t: print(f"检测: {r}"))
            detector.on_error(lambda e: print(f"错误: {e}"))
            print(f"✅ 回调设置成功")

            return True

        except FileNotFoundError as e:
            print(f"⚠️  检测器创建跳过（模型文件不存在）: {e}")
            return True  # 不算失败，只是缺少模型文件

        except Exception as e:
            print(f"❌ 检测器测试失败: {e}")
            return False

    except Exception as e:
        print(f"❌ 检测器测试失败: {e}")
        return False


async def test_integration():
    """集成测试"""
    print("\n" + "=" * 60)
    print("测试 4: 集成测试 - 唤醒词灵敏度测试")
    print("=" * 60)

    try:
        config = WakeWordConfig(
            model_path="models",
            sample_rate=16000,
        )

        audio_capture = AudioCapture(sample_rate=16000, channels=1)

        try:
            detector = WakeWordDetector(config)
            
            # 唤醒词检测统计
            detection_count = 0
            detection_times = []
            
            def on_detected(result, *args):
                nonlocal detection_count, detection_times
                detection_count += 1
                import time
                detection_times.append(time.time())
                # 尝试获取打分信息
                score = "未知"
                if hasattr(result, 'score'):
                    score = result.score
                print(f"🎉 检测到唤醒词! (次数: {detection_count}, 打分: {score})")

            detector.on_detected(on_detected)

            # 连接
            audio_capture.add_audio_listener(detector)
            print(f"✅ 音频采集器和检测器连接成功")

            # 启动
            await audio_capture.start()
            await detector.start()
            print(f"✅ 系统启动成功")

            # 运行 60 秒进行测试
            test_duration = 60
            print(f"\n运行 {test_duration} 秒进行唤醒词灵敏度测试...")
            print("请在这段时间内说出唤醒词，测试系统的检测效果")
            
            import time
            start_time = time.time()
            
            while time.time() - start_time < test_duration:
                await asyncio.sleep(0.1)
                if detection_count > 0:
                    # 每检测到一次唤醒词，显示剩余时间
                    remaining = max(0, test_duration - (time.time() - start_time))
                    print(f"剩余测试时间: {remaining:.1f} 秒")

            # 停止
            await detector.stop()
            await audio_capture.stop()
            print(f"✅ 系统停止成功")

            # 输出测试结果
            print(f"\n测试结果:")
            print(f"- 测试时长: {test_duration} 秒")
            print(f"- 检测到唤醒词次数: {detection_count}")
            
            if detection_count > 0:
                # 计算检测间隔
                intervals = []
                for i in range(1, len(detection_times)):
                    intervals.append(detection_times[i] - detection_times[i-1])
                
                if intervals:
                    avg_interval = sum(intervals) / len(intervals)
                    print(f"- 平均检测间隔: {avg_interval:.2f} 秒")
                
                print("✅ 唤醒词灵敏度测试完成")
            else:
                print("⚠️  未检测到唤醒词，请尝试调整唤醒词或提高音量")

            return True

        except FileNotFoundError as e:
            print(f"⚠️  集成测试跳过（模型文件不存在）: {e}")
            return True

        except Exception as e:
            print(f"❌ 集成测试失败: {e}")
            # 确保清理
            try:
                await audio_capture.stop()
            except:
                pass
            return False

    except Exception as e:
        print(f"❌ 集成测试失败: {e}")
        return False


async def main():
    """主测试函数"""
    print("\n" + "=" * 60)
    print("语音唤醒模块测试")
    print("=" * 60)

    results = []

    # 测试 1: 配置类
    results.append(("配置类", test_config()))

    # 测试 2: 音频采集器
    results.append(("音频采集器", test_audio_capture()))

    # 测试 3: 检测器
    results.append(("检测器", await test_detector()))

    # 测试 4: 集成测试
    results.append(("集成测试", await test_integration()))

    # 汇总结果
    print("\n" + "=" * 60)
    print("测试结果汇总")
    print("=" * 60)

    for name, result in results:
        status = "✅ 通过" if result else "❌ 失败"
        print(f"{name:20s} {status}")

    # 统计
    passed = sum(1 for _, r in results if r)
    total = len(results)
    print(f"\n总计: {passed}/{total} 测试通过")

    return all(r for _, r in results)


if __name__ == "__main__":
    try:
        success = asyncio.run(main())
        sys.exit(0 if success else 1)
    except KeyboardInterrupt:
        print("\n\n测试被中断")
        sys.exit(1)
    except Exception as e:
        print(f"\n❌ 测试异常: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)