161 lines
5.2 KiB
Python
161 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
主进程控制示例
|
|
演示如何从主进程控制输入进程的校准和监听功能
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import time
|
|
import json
|
|
from control_system import ControlSystem
|
|
|
|
def main():
|
|
"""主函数 - 演示主进程控制"""
|
|
print("🚀 主进程控制示例")
|
|
print("=" * 60)
|
|
|
|
# 1. 创建控制系统
|
|
print("📦 创建控制系统...")
|
|
config = {
|
|
'system': {
|
|
'max_queue_size': 1000,
|
|
'process_timeout': 30,
|
|
'heartbeat_interval': 1.0,
|
|
'log_level': "INFO"
|
|
},
|
|
'audio': {
|
|
'sample_rate': 16000,
|
|
'channels': 1,
|
|
'chunk_size': 1024
|
|
},
|
|
'recording': {
|
|
'min_duration': 2.0,
|
|
'max_duration': 30.0,
|
|
'silence_threshold': 3.0
|
|
},
|
|
'processing': {
|
|
'enable_asr': True,
|
|
'enable_llm': True,
|
|
'enable_tts': True,
|
|
'character': 'libai'
|
|
}
|
|
}
|
|
|
|
control_system = ControlSystem(config)
|
|
|
|
# 2. 启动系统
|
|
print("🔧 启动系统（包含自动校准和监听）...")
|
|
# 使用新的自动启动功能
|
|
control_system.start(auto_calibration=True, auto_monitoring=True)
|
|
|
|
print("✅ 系统已启动")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
# 3. 验证自动启动状态
|
|
print("\n🔍 验证自动启动状态...")
|
|
|
|
calibration_status = control_system.get_calibration_status()
|
|
if calibration_status:
|
|
print(f" 校准状态: {calibration_status}")
|
|
if calibration_status['calibrating']:
|
|
print(" ⚠️ 校准仍在进行中...")
|
|
else:
|
|
print(" ✅ 校准已完成")
|
|
else:
|
|
print(" 无法获取校准状态")
|
|
|
|
monitoring_status = control_system.get_monitoring_status()
|
|
if monitoring_status:
|
|
print(f" 监听状态: {monitoring_status}")
|
|
if monitoring_status['enabled']:
|
|
print(" ✅ 监听已启用")
|
|
print(f" 音频流状态: {'活跃' if monitoring_status['audio_stream_active'] else '非活跃'}")
|
|
else:
|
|
print(" ❌ 监听未启用")
|
|
else:
|
|
print(" 无法获取监听状态")
|
|
|
|
# 如果校准未完成，等待一下
|
|
if calibration_status and calibration_status['calibrating']:
|
|
print("\n⏱️ 等待校准完成...")
|
|
if control_system.wait_for_calibration_complete(timeout=15):
|
|
print("✅ 校准完成")
|
|
else:
|
|
print("⚠️ 校准超时")
|
|
|
|
# 如果监听未启用，尝试启用
|
|
if monitoring_status and not monitoring_status['enabled']:
|
|
print("\n🎯 尝试启用监听...")
|
|
success = control_system.start_monitoring()
|
|
if success:
|
|
print("✅ 监听已启用")
|
|
else:
|
|
print("❌ 监听启用失败")
|
|
|
|
# 8. 运行一段时间进行演示
|
|
print("\n🎙️ 系统正在运行...")
|
|
print("💡 请说话测试语音检测和录音功能")
|
|
print("📊 可以看到实时状态更新")
|
|
print("⏱️ 运行30秒后自动停止...")
|
|
|
|
# 启动控制循环（简化版）
|
|
start_time = time.time()
|
|
while time.time() - start_time < 30:
|
|
# 检查状态
|
|
try:
|
|
control_system.check_events()
|
|
control_system.display_status()
|
|
time.sleep(0.1)
|
|
except KeyboardInterrupt:
|
|
print("\n👋 用户中断")
|
|
break
|
|
|
|
# 9. 测试停止监听功能
|
|
print("\n🎯 测试停止监听功能...")
|
|
success = control_system.stop_monitoring()
|
|
if success:
|
|
print("✅ 停止监听命令发送成功")
|
|
else:
|
|
print("❌ 停止监听命令发送失败")
|
|
|
|
# 10. 验证停止状态
|
|
print("\n🔍 验证停止状态...")
|
|
time.sleep(1) # 等待状态更新
|
|
monitoring_status = control_system.get_monitoring_status()
|
|
if monitoring_status and not monitoring_status['enabled']:
|
|
print("✅ 监听已停止")
|
|
else:
|
|
print("❌ 监听未正确停止")
|
|
|
|
# 11. 测试重新启动监听
|
|
print("\n🎯 测试重新启动监听...")
|
|
success = control_system.start_monitoring()
|
|
if success:
|
|
print("✅ 监听重新启动成功")
|
|
time.sleep(1)
|
|
monitoring_status = control_system.get_monitoring_status()
|
|
if monitoring_status and monitoring_status['enabled']:
|
|
print("✅ 监听状态正常")
|
|
else:
|
|
print("❌ 监听重新启动失败")
|
|
|
|
print("\n🎉 演示完成！")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n👋 用户中断")
|
|
except Exception as e:
|
|
print(f"❌ 运行错误: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
# 清理
|
|
print("\n🛑 清理系统...")
|
|
control_system.shutdown()
|
|
print("✅ 系统已关闭")
|
|
|
|
if __name__ == "__main__":
|
|
main() |