752 lines
23 KiB
Python
752 lines
23 KiB
Python
"""
|
||
音频管理器模块
|
||
负责管理音频输入输出设备
|
||
"""
|
||
|
||
import time
|
||
from typing import Dict, Any, List, Optional
|
||
import threading
|
||
|
||
class AudioManager:
|
||
"""
|
||
音频管理器
|
||
负责管理音频输入输出设备
|
||
"""
|
||
|
||
def __init__(self, plugin):
|
||
"""
|
||
初始化音频管理器
|
||
|
||
Args:
|
||
plugin: 语音控制插件实例
|
||
"""
|
||
self.plugin = plugin
|
||
self.enabled = False
|
||
self.initialized = False
|
||
|
||
# 音频设备列表
|
||
self.audio_devices = {
|
||
'input': {},
|
||
'output': {}
|
||
}
|
||
|
||
# 当前设备配置
|
||
self.current_devices = {
|
||
'input': None,
|
||
'output': None
|
||
}
|
||
|
||
# 音频流管理
|
||
self.audio_streams = {
|
||
'input': None,
|
||
'output': None
|
||
}
|
||
|
||
# 音频配置
|
||
self.audio_config = {
|
||
'sample_rate': 16000,
|
||
'channels': 1,
|
||
'format': 'int16',
|
||
'input_buffer_size': 1024,
|
||
'output_buffer_size': 1024,
|
||
'latency': 'low',
|
||
'enable_echo_cancellation': True,
|
||
'enable_noise_reduction': True,
|
||
'enable_auto_gain': True
|
||
}
|
||
|
||
# 音频状态
|
||
self.audio_state = {
|
||
'input_level': 0.0,
|
||
'output_level': 0.0,
|
||
'is_recording': False,
|
||
'is_playing': False,
|
||
'last_input_time': 0.0,
|
||
'last_output_time': 0.0
|
||
}
|
||
|
||
# 设备统计
|
||
self.device_stats = {
|
||
'devices_found': 0,
|
||
'input_devices': 0,
|
||
'output_devices': 0,
|
||
'devices_connected': 0,
|
||
'devices_disconnected': 0,
|
||
'data_transferred': 0
|
||
}
|
||
|
||
# 线程管理
|
||
self.monitor_thread = None
|
||
self.running = False
|
||
|
||
# 回调函数
|
||
self.audio_callbacks = {
|
||
'device_connected': [],
|
||
'device_disconnected': [],
|
||
'input_level_changed': [],
|
||
'output_level_changed': [],
|
||
'recording_started': [],
|
||
'recording_stopped': [],
|
||
'playback_started': [],
|
||
'playback_stopped': []
|
||
}
|
||
|
||
# 时间戳记录
|
||
self.last_device_scan = 0.0
|
||
self.last_level_update = 0.0
|
||
|
||
print("✓ 音频管理器已创建")
|
||
|
||
def initialize(self) -> bool:
|
||
"""
|
||
初始化音频管理器
|
||
|
||
Returns:
|
||
是否初始化成功
|
||
"""
|
||
try:
|
||
# 扫描音频设备
|
||
self._scan_audio_devices()
|
||
|
||
# 初始化默认设备
|
||
self._initialize_default_devices()
|
||
|
||
self.initialized = True
|
||
print("✓ 音频管理器初始化完成")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"✗ 音频管理器初始化失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
def enable(self) -> bool:
|
||
"""
|
||
启用音频管理器
|
||
|
||
Returns:
|
||
是否启用成功
|
||
"""
|
||
try:
|
||
if not self.initialized:
|
||
print("✗ 音频管理器未初始化")
|
||
return False
|
||
|
||
self.enabled = True
|
||
self.running = True
|
||
|
||
# 启动监控线程
|
||
self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
|
||
self.monitor_thread.start()
|
||
|
||
print("✓ 音频管理器已启用")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"✗ 音频管理器启用失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
def disable(self):
|
||
"""禁用音频管理器"""
|
||
try:
|
||
self.enabled = False
|
||
self.running = False
|
||
|
||
# 等待监控线程结束
|
||
if self.monitor_thread and self.monitor_thread.is_alive():
|
||
self.monitor_thread.join(timeout=2.0)
|
||
|
||
# 关闭所有音频流
|
||
self._close_all_streams()
|
||
|
||
print("✓ 音频管理器已禁用")
|
||
|
||
except Exception as e:
|
||
print(f"✗ 音频管理器禁用失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
def finalize(self):
|
||
"""清理音频管理器资源"""
|
||
try:
|
||
self.disable()
|
||
self.audio_callbacks.clear()
|
||
self.audio_devices.clear()
|
||
self.initialized = False
|
||
print("✓ 音频管理器资源已清理")
|
||
|
||
except Exception as e:
|
||
print(f"✗ 音频管理器资源清理失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
def update(self, dt: float):
|
||
"""
|
||
更新音频管理器状态
|
||
|
||
Args:
|
||
dt: 时间增量
|
||
"""
|
||
try:
|
||
if not self.enabled:
|
||
return
|
||
|
||
current_time = time.time()
|
||
|
||
# 定期扫描设备
|
||
if current_time - self.last_device_scan > 5.0: # 每5秒扫描一次
|
||
self._scan_audio_devices()
|
||
self.last_device_scan = current_time
|
||
|
||
self.last_level_update = current_time
|
||
|
||
except Exception as e:
|
||
print(f"✗ 音频管理器更新失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
def _scan_audio_devices(self):
|
||
"""扫描音频设备"""
|
||
try:
|
||
# 模拟设备扫描
|
||
# 在实际实现中，这里会调用平台特定的API来发现音频设备
|
||
|
||
# 模拟输入设备
|
||
input_devices = {
|
||
0: {
|
||
'id': 0,
|
||
'name': '系统默认麦克风',
|
||
'type': 'input',
|
||
'channels': 1,
|
||
'sample_rates': [16000, 44100, 48000],
|
||
'is_default': True,
|
||
'connected': True
|
||
},
|
||
1: {
|
||
'id': 1,
|
||
'name': 'USB 麦克风',
|
||
'type': 'input',
|
||
'channels': 2,
|
||
'sample_rates': [16000, 44100, 48000],
|
||
'is_default': False,
|
||
'connected': True
|
||
},
|
||
2: {
|
||
'id': 2,
|
||
'name': '蓝牙耳机麦克风',
|
||
'type': 'input',
|
||
'channels': 1,
|
||
'sample_rates': [16000, 44100],
|
||
'is_default': False,
|
||
'connected': True
|
||
}
|
||
}
|
||
|
||
# 模拟输出设备
|
||
output_devices = {
|
||
0: {
|
||
'id': 0,
|
||
'name': '系统默认扬声器',
|
||
'type': 'output',
|
||
'channels': 2,
|
||
'sample_rates': [16000, 44100, 48000],
|
||
'is_default': True,
|
||
'connected': True
|
||
},
|
||
1: {
|
||
'id': 1,
|
||
'name': 'USB 音响',
|
||
'type': 'output',
|
||
'channels': 2,
|
||
'sample_rates': [16000, 44100, 48000, 96000],
|
||
'is_default': False,
|
||
'connected': True
|
||
},
|
||
2: {
|
||
'id': 2,
|
||
'name': '蓝牙耳机',
|
||
'type': 'output',
|
||
'channels': 2,
|
||
'sample_rates': [16000, 44100],
|
||
'is_default': False,
|
||
'connected': True
|
||
}
|
||
}
|
||
|
||
# 更新设备列表
|
||
self.audio_devices['input'] = input_devices
|
||
self.audio_devices['output'] = output_devices
|
||
|
||
# 更新统计信息
|
||
self.device_stats['devices_found'] = len(input_devices) + len(output_devices)
|
||
self.device_stats['input_devices'] = len(input_devices)
|
||
self.device_stats['output_devices'] = len(output_devices)
|
||
|
||
print(f"✓ 音频设备扫描完成: {len(input_devices)}个输入设备, {len(output_devices)}个输出设备")
|
||
|
||
except Exception as e:
|
||
print(f"✗ 音频设备扫描失败: {e}")
|
||
|
||
def _initialize_default_devices(self):
|
||
"""初始化默认设备"""
|
||
try:
|
||
# 查找默认输入设备
|
||
for device_id, device_info in self.audio_devices['input'].items():
|
||
if device_info.get('is_default', False):
|
||
self.current_devices['input'] = device_id
|
||
break
|
||
|
||
# 查找默认输出设备
|
||
for device_id, device_info in self.audio_devices['output'].items():
|
||
if device_info.get('is_default', False):
|
||
self.current_devices['output'] = device_id
|
||
break
|
||
|
||
print("✓ 默认音频设备初始化完成")
|
||
|
||
except Exception as e:
|
||
print(f"✗ 默认音频设备初始化失败: {e}")
|
||
|
||
def _monitor_loop(self):
|
||
"""监控循环线程"""
|
||
try:
|
||
monitor_interval = 0.1 # 100ms
|
||
|
||
while self.running:
|
||
if self.enabled:
|
||
self._update_audio_levels()
|
||
self._check_device_status()
|
||
|
||
time.sleep(monitor_interval)
|
||
|
||
except Exception as e:
|
||
print(f"✗ 音频监控循环失败: {e}")
|
||
|
||
def _update_audio_levels(self):
|
||
"""更新音频级别"""
|
||
try:
|
||
current_time = time.time()
|
||
|
||
# 模拟输入级别更新
|
||
if self.audio_state['is_recording']:
|
||
import random
|
||
new_input_level = random.uniform(0.0, 1.0)
|
||
if abs(new_input_level - self.audio_state['input_level']) > 0.01:
|
||
self.audio_state['input_level'] = new_input_level
|
||
self._trigger_audio_callback('input_level_changed', {
|
||
'level': new_input_level,
|
||
'timestamp': current_time
|
||
})
|
||
|
||
# 模拟输出级别更新
|
||
if self.audio_state['is_playing']:
|
||
import random
|
||
new_output_level = random.uniform(0.0, 1.0)
|
||
if abs(new_output_level - self.audio_state['output_level']) > 0.01:
|
||
self.audio_state['output_level'] = new_output_level
|
||
self._trigger_audio_callback('output_level_changed', {
|
||
'level': new_output_level,
|
||
'timestamp': current_time
|
||
})
|
||
|
||
self.last_level_update = current_time
|
||
|
||
except Exception as e:
|
||
print(f"✗ 音频级别更新失败: {e}")
|
||
|
||
def _check_device_status(self):
|
||
"""检查设备状态"""
|
||
try:
|
||
# 在实际实现中，这里会检查设备连接状态
|
||
# 目前只是模拟
|
||
pass
|
||
except Exception as e:
|
||
print(f"✗ 设备状态检查失败: {e}")
|
||
|
||
def _close_all_streams(self):
|
||
"""关闭所有音频流"""
|
||
try:
|
||
self.audio_streams['input'] = None
|
||
self.audio_streams['output'] = None
|
||
print("✓ 所有音频流已关闭")
|
||
except Exception as e:
|
||
print(f"✗ 音频流关闭失败: {e}")
|
||
|
||
def get_input_devices(self) -> Dict[int, Dict[str, Any]]:
|
||
"""
|
||
获取输入设备列表
|
||
|
||
Returns:
|
||
输入设备字典
|
||
"""
|
||
return self.audio_devices['input'].copy()
|
||
|
||
def get_output_devices(self) -> Dict[int, Dict[str, Any]]:
|
||
"""
|
||
获取输出设备列表
|
||
|
||
Returns:
|
||
输出设备字典
|
||
"""
|
||
return self.audio_devices['output'].copy()
|
||
|
||
def get_all_devices(self) -> Dict[str, Dict[int, Dict[str, Any]]]:
|
||
"""
|
||
获取所有设备列表
|
||
|
||
Returns:
|
||
所有设备字典
|
||
"""
|
||
return {
|
||
'input': self.audio_devices['input'].copy(),
|
||
'output': self.audio_devices['output'].copy()
|
||
}
|
||
|
||
def set_input_device(self, device_id: int) -> bool:
|
||
"""
|
||
设置输入设备
|
||
|
||
Args:
|
||
device_id: 设备ID
|
||
|
||
Returns:
|
||
是否设置成功
|
||
"""
|
||
try:
|
||
if device_id in self.audio_devices['input']:
|
||
old_device = self.current_devices['input']
|
||
self.current_devices['input'] = device_id
|
||
|
||
print(f"✓ 输入设备已设置: {self.audio_devices['input'][device_id]['name']}")
|
||
|
||
# 触发设备变更回调
|
||
self._trigger_audio_callback('device_connected', {
|
||
'device_type': 'input',
|
||
'device_id': device_id,
|
||
'device_name': self.audio_devices['input'][device_id]['name'],
|
||
'old_device_id': old_device
|
||
})
|
||
|
||
return True
|
||
else:
|
||
print(f"✗ 输入设备不存在: ID {device_id}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"✗ 输入设备设置失败: {e}")
|
||
return False
|
||
|
||
def set_output_device(self, device_id: int) -> bool:
|
||
"""
|
||
设置输出设备
|
||
|
||
Args:
|
||
device_id: 设备ID
|
||
|
||
Returns:
|
||
是否设置成功
|
||
"""
|
||
try:
|
||
if device_id in self.audio_devices['output']:
|
||
old_device = self.current_devices['output']
|
||
self.current_devices['output'] = device_id
|
||
|
||
print(f"✓ 输出设备已设置: {self.audio_devices['output'][device_id]['name']}")
|
||
|
||
# 触发设备变更回调
|
||
self._trigger_audio_callback('device_connected', {
|
||
'device_type': 'output',
|
||
'device_id': device_id,
|
||
'device_name': self.audio_devices['output'][device_id]['name'],
|
||
'old_device_id': old_device
|
||
})
|
||
|
||
return True
|
||
else:
|
||
print(f"✗ 输出设备不存在: ID {device_id}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"✗ 输出设备设置失败: {e}")
|
||
return False
|
||
|
||
def get_current_input_device(self) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
获取当前输入设备
|
||
|
||
Returns:
|
||
当前输入设备信息或None
|
||
"""
|
||
try:
|
||
device_id = self.current_devices['input']
|
||
if device_id is not None and device_id in self.audio_devices['input']:
|
||
return self.audio_devices['input'][device_id]
|
||
return None
|
||
except Exception as e:
|
||
print(f"✗ 当前输入设备获取失败: {e}")
|
||
return None
|
||
|
||
def get_current_output_device(self) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
获取当前输出设备
|
||
|
||
Returns:
|
||
当前输出设备信息或None
|
||
"""
|
||
try:
|
||
device_id = self.current_devices['output']
|
||
if device_id is not None and device_id in self.audio_devices['output']:
|
||
return self.audio_devices['output'][device_id]
|
||
return None
|
||
except Exception as e:
|
||
print(f"✗ 当前输出设备获取失败: {e}")
|
||
return None
|
||
|
||
def start_recording(self) -> bool:
|
||
"""
|
||
开始录音
|
||
|
||
Returns:
|
||
是否开始成功
|
||
"""
|
||
try:
|
||
if not self.enabled:
|
||
print("✗ 音频管理器未启用")
|
||
return False
|
||
|
||
# 模拟开始录音
|
||
self.audio_state['is_recording'] = True
|
||
self.audio_state['last_input_time'] = time.time()
|
||
|
||
# 触发录音开始回调
|
||
self._trigger_audio_callback('recording_started', {
|
||
'device_id': self.current_devices['input'],
|
||
'timestamp': time.time()
|
||
})
|
||
|
||
print("✓ 录音已开始")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"✗ 录音开始失败: {e}")
|
||
return False
|
||
|
||
def stop_recording(self) -> bool:
|
||
"""
|
||
停止录音
|
||
|
||
Returns:
|
||
是否停止成功
|
||
"""
|
||
try:
|
||
self.audio_state['is_recording'] = False
|
||
|
||
# 触发录音停止回调
|
||
self._trigger_audio_callback('recording_stopped', {
|
||
'device_id': self.current_devices['input'],
|
||
'timestamp': time.time()
|
||
})
|
||
|
||
print("✓ 录音已停止")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"✗ 录音停止失败: {e}")
|
||
return False
|
||
|
||
def start_playback(self) -> bool:
|
||
"""
|
||
开始播放
|
||
|
||
Returns:
|
||
是否开始成功
|
||
"""
|
||
try:
|
||
if not self.enabled:
|
||
print("✗ 音频管理器未启用")
|
||
return False
|
||
|
||
# 模拟开始播放
|
||
self.audio_state['is_playing'] = True
|
||
self.audio_state['last_output_time'] = time.time()
|
||
|
||
# 触发播放开始回调
|
||
self._trigger_audio_callback('playback_started', {
|
||
'device_id': self.current_devices['output'],
|
||
'timestamp': time.time()
|
||
})
|
||
|
||
print("✓ 播放已开始")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"✗ 播放开始失败: {e}")
|
||
return False
|
||
|
||
def stop_playback(self) -> bool:
|
||
"""
|
||
停止播放
|
||
|
||
Returns:
|
||
是否停止成功
|
||
"""
|
||
try:
|
||
self.audio_state['is_playing'] = False
|
||
|
||
# 触发播放停止回调
|
||
self._trigger_audio_callback('playback_stopped', {
|
||
'device_id': self.current_devices['output'],
|
||
'timestamp': time.time()
|
||
})
|
||
|
||
print("✓ 播放已停止")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"✗ 播放停止失败: {e}")
|
||
return False
|
||
|
||
def get_audio_level(self, stream_type: str) -> float:
|
||
"""
|
||
获取音频级别
|
||
|
||
Args:
|
||
stream_type: 流类型('input'或'output')
|
||
|
||
Returns:
|
||
音频级别(0.0-1.0)
|
||
"""
|
||
try:
|
||
if stream_type == 'input':
|
||
return self.audio_state['input_level']
|
||
elif stream_type == 'output':
|
||
return self.audio_state['output_level']
|
||
else:
|
||
print(f"✗ 无效的流类型: {stream_type}")
|
||
return 0.0
|
||
except Exception as e:
|
||
print(f"✗ 音频级别获取失败: {e}")
|
||
return 0.0
|
||
|
||
def set_audio_config(self, config: Dict[str, Any]) -> bool:
|
||
"""
|
||
设置音频配置
|
||
|
||
Args:
|
||
config: 配置字典
|
||
|
||
Returns:
|
||
是否设置成功
|
||
"""
|
||
try:
|
||
self.audio_config.update(config)
|
||
print(f"✓ 音频配置已更新: {self.audio_config}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"✗ 音频配置设置失败: {e}")
|
||
return False
|
||
|
||
def get_audio_config(self) -> Dict[str, Any]:
|
||
"""
|
||
获取音频配置
|
||
|
||
Returns:
|
||
配置字典
|
||
"""
|
||
return self.audio_config.copy()
|
||
|
||
def is_recording(self) -> bool:
|
||
"""
|
||
检查是否正在录音
|
||
|
||
Returns:
|
||
是否正在录音
|
||
"""
|
||
return self.audio_state['is_recording']
|
||
|
||
def is_playing(self) -> bool:
|
||
"""
|
||
检查是否正在播放
|
||
|
||
Returns:
|
||
是否正在播放
|
||
"""
|
||
return self.audio_state['is_playing']
|
||
|
||
def _trigger_audio_callback(self, callback_type: str, data: Dict[str, Any]):
|
||
"""
|
||
触发音频回调
|
||
|
||
Args:
|
||
callback_type: 回调类型
|
||
data: 回调数据
|
||
"""
|
||
try:
|
||
if callback_type in self.audio_callbacks:
|
||
for callback in self.audio_callbacks[callback_type]:
|
||
try:
|
||
callback(data)
|
||
except Exception as e:
|
||
print(f"✗ 音频回调执行失败: {e}")
|
||
except Exception as e:
|
||
print(f"✗ 音频回调触发失败: {e}")
|
||
|
||
def register_audio_callback(self, callback_type: str, callback: callable):
|
||
"""
|
||
注册音频回调
|
||
|
||
Args:
|
||
callback_type: 回调类型
|
||
callback: 回调函数
|
||
"""
|
||
try:
|
||
if callback_type in self.audio_callbacks:
|
||
self.audio_callbacks[callback_type].append(callback)
|
||
print(f"✓ 音频回调已注册: {callback_type}")
|
||
else:
|
||
print(f"✗ 无效的回调类型: {callback_type}")
|
||
except Exception as e:
|
||
print(f"✗ 音频回调注册失败: {e}")
|
||
|
||
def unregister_audio_callback(self, callback_type: str, callback: callable):
|
||
"""
|
||
注销音频回调
|
||
|
||
Args:
|
||
callback_type: 回调类型
|
||
callback: 回调函数
|
||
"""
|
||
try:
|
||
if callback_type in self.audio_callbacks:
|
||
if callback in self.audio_callbacks[callback_type]:
|
||
self.audio_callbacks[callback_type].remove(callback)
|
||
print(f"✓ 音频回调已注销: {callback_type}")
|
||
except Exception as e:
|
||
print(f"✗ 音频回调注销失败: {e}")
|
||
|
||
def get_device_stats(self) -> Dict[str, int]:
|
||
"""
|
||
获取设备统计信息
|
||
|
||
Returns:
|
||
统计信息字典
|
||
"""
|
||
return self.device_stats.copy()
|
||
|
||
def reset_device_stats(self):
|
||
"""重置设备统计信息"""
|
||
try:
|
||
self.device_stats = {
|
||
'devices_found': 0,
|
||
'input_devices': 0,
|
||
'output_devices': 0,
|
||
'devices_connected': 0,
|
||
'devices_disconnected': 0,
|
||
'data_transferred': 0
|
||
}
|
||
print("✓ 音频管理器统计信息已重置")
|
||
except Exception as e:
|
||
print(f"✗ 音频管理器统计信息重置失败: {e}") |