""" 音频管理器模块 负责管理音频输入输出设备 """ 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}")