401 lines
13 KiB
Python
401 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
多进程音频录音系统
|
||
基于进程隔离的音频处理架构
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import argparse
|
||
import json
|
||
import time
|
||
import subprocess
|
||
from typing import Dict, Any
|
||
|
||
# 导入LED控制器
|
||
try:
|
||
from led_controller import get_led_controller, SystemState, start_led_effects, stop_led_effects, set_led_state
|
||
LED_AVAILABLE = True
|
||
except ImportError:
|
||
LED_AVAILABLE = False
|
||
print("⚠️ LED控制器模块未找到，LED功能将被禁用")
|
||
|
||
def check_dependencies():
|
||
"""检查系统依赖"""
|
||
missing_deps = []
|
||
|
||
try:
|
||
import pyaudio
|
||
except ImportError:
|
||
missing_deps.append("pyaudio")
|
||
|
||
try:
|
||
import numpy
|
||
except ImportError:
|
||
missing_deps.append("numpy")
|
||
|
||
try:
|
||
import requests
|
||
except ImportError:
|
||
missing_deps.append("requests")
|
||
|
||
try:
|
||
import websockets
|
||
except ImportError:
|
||
missing_deps.append("websockets")
|
||
|
||
if missing_deps:
|
||
print("❌ 缺少以下依赖库:")
|
||
for dep in missing_deps:
|
||
print(f" - {dep}")
|
||
print("\n请运行以下命令安装:")
|
||
print(f"pip install {' '.join(missing_deps)}")
|
||
return False
|
||
|
||
return True
|
||
|
||
def check_nfc_dependencies():
|
||
"""检查NFC相关依赖"""
|
||
try:
|
||
# 检查nfc-list命令是否存在
|
||
result = subprocess.run(['which', 'nfc-list'], capture_output=True, text=True)
|
||
if result.returncode != 0:
|
||
print("❌ 未找到nfc-list命令")
|
||
print(" 请安装libnfc工具:")
|
||
print(" Ubuntu/Debian: sudo apt-get install libnfc-bin")
|
||
print(" macOS: brew install libnfc")
|
||
return False
|
||
print("✅ NFC工具已安装")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ NFC依赖检查失败: {e}")
|
||
return False
|
||
|
||
def check_environment():
|
||
"""检查运行环境"""
|
||
print("🔍 检查运行环境...")
|
||
|
||
# 检查Python版本
|
||
python_version = sys.version_info
|
||
if python_version.major < 3 or (python_version.major == 3 and python_version.minor < 7):
|
||
print(f"❌ Python版本过低: {python_version.major}.{python_version.minor}")
|
||
print("需要Python 3.7或更高版本")
|
||
return False
|
||
|
||
print(f"✅ Python版本: {python_version.major}.{python_version.minor}.{python_version.micro}")
|
||
|
||
# 检查操作系统
|
||
import platform
|
||
system = platform.system().lower()
|
||
print(f"✅ 操作系统: {system}")
|
||
|
||
# 检查音频设备
|
||
try:
|
||
import pyaudio
|
||
audio = pyaudio.PyAudio()
|
||
device_count = audio.get_device_count()
|
||
print(f"✅ 音频设备数量: {device_count}")
|
||
|
||
if device_count == 0:
|
||
print("❌ 未检测到音频设备")
|
||
return False
|
||
|
||
audio.terminate()
|
||
except Exception as e:
|
||
print(f"❌ 音频设备检查失败: {e}")
|
||
return False
|
||
|
||
# 检查网络连接
|
||
try:
|
||
import requests
|
||
response = requests.get("https://www.baidu.com", timeout=5)
|
||
print("✅ 网络连接正常")
|
||
except:
|
||
print("⚠️ 网络连接可能有问题，会影响在线AI功能")
|
||
|
||
# 检查API密钥
|
||
api_key = os.environ.get("ARK_API_KEY")
|
||
if api_key:
|
||
print("✅ ARK_API_KEY 已设置")
|
||
else:
|
||
print("⚠️ ARK_API_KEY 未设置，大语言模型功能将被禁用")
|
||
print(" 请运行: export ARK_API_KEY='your_api_key_here'")
|
||
|
||
return True
|
||
|
||
def list_characters():
|
||
"""列出可用角色"""
|
||
characters_dir = os.path.join(os.path.dirname(__file__), "characters")
|
||
|
||
if not os.path.exists(characters_dir):
|
||
print("❌ 角色目录不存在")
|
||
return
|
||
|
||
characters = []
|
||
for file in os.listdir(characters_dir):
|
||
if file.endswith('.json'):
|
||
character_name = file[:-5]
|
||
config_file = os.path.join(characters_dir, file)
|
||
|
||
try:
|
||
with open(config_file, 'r', encoding='utf-8') as f:
|
||
config = json.load(f)
|
||
name = config.get('name', character_name)
|
||
desc = config.get('description', '无描述')
|
||
characters.append(f"{character_name}: {name} - {desc}")
|
||
except:
|
||
characters.append(f"{character_name}: 配置文件读取失败")
|
||
|
||
if characters:
|
||
print("🎭 可用角色列表:")
|
||
for char in characters:
|
||
print(f" - {char}")
|
||
else:
|
||
print("❌ 未找到任何角色配置文件")
|
||
|
||
def create_sample_config():
|
||
"""创建示例配置文件"""
|
||
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,
|
||
"format": "paInt16"
|
||
},
|
||
"recording": {
|
||
"min_duration": 2.0,
|
||
"max_duration": 30.0,
|
||
"silence_threshold": 3.0,
|
||
"pre_record_duration": 2.0
|
||
},
|
||
"processing": {
|
||
"enable_asr": True,
|
||
"enable_llm": True,
|
||
"enable_tts": True,
|
||
"character": "libai",
|
||
"max_tokens": 50
|
||
},
|
||
"detection": {
|
||
"zcr_min": 2400,
|
||
"zcr_max": 12000,
|
||
"consecutive_silence_count": 30,
|
||
"max_zcr_history": 30
|
||
},
|
||
"playback": {
|
||
"buffer_size": 1000,
|
||
"show_progress": True,
|
||
"progress_interval": 100,
|
||
"chunk_size": 512
|
||
}
|
||
}
|
||
|
||
config_file = "config.json"
|
||
try:
|
||
with open(config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(config, f, indent=2, ensure_ascii=False)
|
||
print(f"✅ 示例配置文件已创建: {config_file}")
|
||
except Exception as e:
|
||
print(f"❌ 创建配置文件失败: {e}")
|
||
|
||
def main():
|
||
"""主函数"""
|
||
parser = argparse.ArgumentParser(
|
||
description='多进程音频录音系统',
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
使用示例:
|
||
python multiprocess_recorder.py # 使用默认角色
|
||
python multiprocess_recorder.py -c zhubajie # 指定角色
|
||
python multiprocess_recorder.py -l # 列出角色
|
||
python multiprocess_recorder.py --create-config # 创建配置文件
|
||
python multiprocess_recorder.py --enable-nfc # 启用NFC角色切换
|
||
"""
|
||
)
|
||
|
||
parser.add_argument('--character', '-c', type=str, default='libai',
|
||
help='选择角色 (默认: libai)')
|
||
parser.add_argument('--list-characters', '-l', action='store_true',
|
||
help='列出所有可用角色')
|
||
parser.add_argument('--config', type=str,
|
||
help='配置文件路径')
|
||
parser.add_argument('--create-config', action='store_true',
|
||
help='创建示例配置文件')
|
||
parser.add_argument('--check-env', action='store_true',
|
||
help='检查运行环境')
|
||
parser.add_argument('--verbose', '-v', action='store_true',
|
||
help='详细输出')
|
||
parser.add_argument('--enable-nfc', action='store_true',
|
||
help='启用NFC角色切换功能')
|
||
parser.add_argument('--enable-led', action='store_true',
|
||
help='启用LED灯光效果')
|
||
parser.add_argument('--led-count', type=int, default=10,
|
||
help='LED数量 (默认: 10)')
|
||
parser.add_argument('--led-brightness', type=int, default=100,
|
||
help='LED亮度 (默认: 100)')
|
||
parser.add_argument('--led-pin', type=int, default=10,
|
||
help='LED GPIO引脚 (默认: 10)')
|
||
|
||
args = parser.parse_args()
|
||
|
||
# 显示欢迎信息
|
||
print("🚀 多进程音频录音系统")
|
||
print("=" * 60)
|
||
|
||
# 检查依赖
|
||
if not check_dependencies():
|
||
sys.exit(1)
|
||
|
||
# 创建配置文件
|
||
if args.create_config:
|
||
create_sample_config()
|
||
return
|
||
|
||
# 检查环境
|
||
if args.check_env:
|
||
check_environment()
|
||
return
|
||
|
||
# 列出角色
|
||
if args.list_characters:
|
||
list_characters()
|
||
return
|
||
|
||
# 检查characters目录
|
||
characters_dir = os.path.join(os.path.dirname(__file__), "characters")
|
||
if not os.path.exists(characters_dir):
|
||
print(f"⚠️ 角色目录不存在: {characters_dir}")
|
||
print("请确保characters目录存在并包含角色配置文件")
|
||
|
||
# 检查指定角色
|
||
character_file = os.path.join(characters_dir, f"{args.character}.json")
|
||
if not os.path.exists(character_file):
|
||
print(f"⚠️ 角色文件不存在: {character_file}")
|
||
print(f"可用角色:")
|
||
list_characters()
|
||
return
|
||
|
||
print(f"🎭 当前角色: {args.character}")
|
||
print("🎯 系统特点:")
|
||
print(" - 多进程架构：输入输出完全隔离")
|
||
print(" - 零切换延迟：无需音频设备重置")
|
||
print(" - 实时响应：并行处理录音和播放")
|
||
print(" - 智能检测：基于ZCR的语音识别")
|
||
print(" - 流式TTS：实时音频生成和播放")
|
||
print(" - 角色扮演：支持多种AI角色")
|
||
print("=" * 60)
|
||
|
||
# 显示使用说明
|
||
print("📖 使用说明:")
|
||
print(" - 检测到语音自动开始录音")
|
||
print(" - 持续静音3秒自动结束录音")
|
||
print(" - 录音完成后自动处理和播放")
|
||
print(" - 按 Ctrl+C 退出")
|
||
print("=" * 60)
|
||
|
||
# 加载配置
|
||
config = None
|
||
if args.config:
|
||
try:
|
||
with open(args.config, 'r', encoding='utf-8') as f:
|
||
config = json.load(f)
|
||
print(f"📋 加载配置文件: {args.config}")
|
||
except Exception as e:
|
||
print(f"⚠️ 配置文件加载失败: {e}")
|
||
print("使用默认配置")
|
||
|
||
try:
|
||
# 导入控制系统
|
||
from control_system import ControlSystem
|
||
|
||
# 创建控制系统
|
||
control_system = ControlSystem(config)
|
||
|
||
# 设置角色
|
||
control_system.config['processing']['character'] = args.character
|
||
|
||
# 设置日志级别
|
||
if args.verbose:
|
||
control_system.config['system']['log_level'] = "DEBUG"
|
||
|
||
# 启用NFC功能（如果指定）- 在系统启动前启用，确保校准完成后能立即开始NFC检测
|
||
nfc_enabled = False
|
||
if args.enable_nfc:
|
||
print("📱 检查NFC依赖...")
|
||
if not check_nfc_dependencies():
|
||
print("❌ NFC依赖检查失败，无法启用NFC功能")
|
||
else:
|
||
print("📱 启用NFC角色切换功能...")
|
||
control_system.enable_nfc()
|
||
nfc_enabled = True
|
||
|
||
# 启用LED功能（如果指定）
|
||
led_enabled = False
|
||
if args.enable_led:
|
||
if LED_AVAILABLE:
|
||
print("💡 启用LED灯光效果...")
|
||
print(f" LED数量: {args.led_count}")
|
||
print(f" LED亮度: {args.led_brightness}")
|
||
print(f" LED引脚: GPIO{args.led_pin}")
|
||
|
||
try:
|
||
# 启动LED系统
|
||
success = start_led_effects(
|
||
led_count=args.led_count,
|
||
brightness=args.led_brightness,
|
||
led_pin=args.led_pin
|
||
)
|
||
if success:
|
||
led_enabled = True
|
||
print("✅ LED系统已启动")
|
||
|
||
# 设置初始状态为启动效果
|
||
set_led_state(SystemState.STARTUP)
|
||
else:
|
||
print("❌ LED系统启动失败")
|
||
except Exception as e:
|
||
print(f"❌ LED系统初始化失败: {e}")
|
||
else:
|
||
print("⚠️ LED功能不可用，请检查硬件和驱动")
|
||
|
||
# 启动系统（自动校准，但根据NFC状态决定是否启动监听）
|
||
print("🚀 启动系统（包含自动校准）...")
|
||
control_system.start(auto_calibration=True, auto_monitoring=not nfc_enabled)
|
||
|
||
# 如果启用了LED，设置系统状态为等待NFC或空闲监听
|
||
if led_enabled:
|
||
if nfc_enabled:
|
||
set_led_state(SystemState.WAITING_NFC)
|
||
else:
|
||
set_led_state(SystemState.IDLE_MONITORING)
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n👋 用户中断")
|
||
except Exception as e:
|
||
print(f"❌ 系统启动失败: {e}")
|
||
if args.verbose:
|
||
import traceback
|
||
traceback.print_exc()
|
||
finally:
|
||
# 清理LED系统
|
||
if LED_AVAILABLE and led_enabled:
|
||
try:
|
||
print("💡 关闭LED系统...")
|
||
set_led_state(SystemState.SHUTDOWN)
|
||
time.sleep(1) # 给LED一些时间显示关闭效果
|
||
stop_led_effects()
|
||
print("✅ LED系统已关闭")
|
||
except Exception as e:
|
||
print(f"❌ 关闭LED系统失败: {e}")
|
||
|
||
print("👋 系统退出")
|
||
|
||
if __name__ == "__main__":
|
||
main() |