""" 配置管理器模块 负责管理语音控制插件的配置 """ import time import json import os from typing import Dict, Any, List, Optional import threading class VoiceConfigManager: """ 语音配置管理器 负责管理语音控制插件的配置 """ def __init__(self, plugin): """ 初始化配置管理器 Args: plugin: 语音控制插件实例 """ self.plugin = plugin self.enabled = False self.initialized = False # 配置文件路径 self.config_path = "/home/hello/EG/plugins/user/voice_control/config/voice_config.json" # 默认配置 self.default_config = { "general": { "enable_voice_control": True, "enable_voice_feedback": True, "language": "zh-CN", "wake_word": "小艾", "enable_wake_word": True, "continuous_listening": False, "silence_timeout": 3.0 }, "recognition": { "language": "zh-CN", "sample_rate": 16000, "channels": 1, "sensitivity": 0.5, "enable_keyword_spotting": True, "enable_continuous_recognition": True, "silence_threshold": 0.01, "speech_threshold": 0.02, "max_recording_duration": 10.0 }, "synthesis": { "language": "zh-CN", "voice": "default", "speed": 1.0, "volume": 1.0, "pitch": 1.0, "enable_ssml": False }, "commands": { "enable_fuzzy_matching": True, "fuzzy_threshold": 0.7, "enable_context_commands": True, "context_timeout": 30.0, "enable_chained_commands": True, "max_chain_length": 5, "enable_confirmation": True, "confirmation_threshold": 0.9 }, "audio": { "input_device": "default", "output_device": "default", "sample_rate": 22050, "channels": 1, "enable_echo_cancellation": True, "enable_noise_reduction": True, "enable_auto_gain": True }, "nlp": { "primary_language": "zh-CN", "enable_intent_recognition": True, "enable_entity_extraction": True, "enable_sentiment_analysis": True, "confidence_threshold": 0.6 }, "events": { "enable_event_queue": True, "max_queue_size": 100, "enable_event_filtering": True, "filter_duplicate_events": True, "duplicate_time_window": 1.0 } } # 当前配置 self.current_config = {} # 配置变更监听器 self.config_listeners = {} # 配置备份 self.config_backup = {} # 配置统计 self.config_stats = { "loads": 0, "saves": 0, "changes": 0, "errors": 0, "backups": 0 } # 线程锁 self.config_lock = threading.RLock() # 回调函数 self.config_callbacks = { "config_loaded": [], "config_saved": [], "config_changed": [], "config_error": [] } # 时间戳记录 self.last_load_time = 0.0 self.last_save_time = 0.0 self.last_change_time = 0.0 print("✓ 语音配置管理器已创建") def initialize(self) -> bool: """ 初始化配置管理器 Returns: 是否初始化成功 """ try: # 创建配置目录 config_dir = os.path.dirname(self.config_path) if not os.path.exists(config_dir): os.makedirs(config_dir) # 加载配置 self.load_config() # 如果配置为空，使用默认配置 if not self.current_config: self.current_config = self.default_config.copy() self.save_config() 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 print("✓ 语音配置管理器已启用") return True except Exception as e: print(f"✗ 语音配置管理器启用失败: {e}") import traceback traceback.print_exc() return False def disable(self): """禁用配置管理器""" try: self.enabled = False print("✓ 语音配置管理器已禁用") except Exception as e: print(f"✗ 语音配置管理器禁用失败: {e}") import traceback traceback.print_exc() def finalize(self): """清理配置管理器资源""" try: self.disable() self.config_listeners.clear() self.config_callbacks.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() # 配置管理器不需要频繁更新 pass except Exception as e: print(f"✗ 语音配置管理器更新失败: {e}") import traceback traceback.print_exc() def load_config(self) -> bool: """ 加载配置 Returns: 是否加载成功 """ try: with self.config_lock: if os.path.exists(self.config_path): with open(self.config_path, 'r', encoding='utf-8') as f: self.current_config = json.load(f) else: # 使用默认配置 self.current_config = self.default_config.copy() self.config_stats["loads"] += 1 self.last_load_time = time.time() # 触发配置加载回调 self._trigger_config_callback("config_loaded", { "config": self.current_config, "timestamp": self.last_load_time }) print("✓ 配置加载完成") return True except Exception as e: print(f"✗ 配置加载失败: {e}") self.config_stats["errors"] += 1 # 触发配置错误回调 self._trigger_config_callback("config_error", { "error": str(e), "operation": "load" }) return False def save_config(self) -> bool: """ 保存配置 Returns: 是否保存成功 """ try: with self.config_lock: # 创建配置目录 config_dir = os.path.dirname(self.config_path) if not os.path.exists(config_dir): os.makedirs(config_dir) # 保存配置 with open(self.config_path, 'w', encoding='utf-8') as f: json.dump(self.current_config, f, ensure_ascii=False, indent=2) self.config_stats["saves"] += 1 self.last_save_time = time.time() # 触发配置保存回调 self._trigger_config_callback("config_saved", { "config": self.current_config, "timestamp": self.last_save_time }) print("✓ 配置保存完成") return True except Exception as e: print(f"✗ 配置保存失败: {e}") self.config_stats["errors"] += 1 # 触发配置错误回调 self._trigger_config_callback("config_error", { "error": str(e), "operation": "save" }) return False def get_config(self, section: str = None, key: str = None, default: Any = None) -> Any: """ 获取配置值 Args: section: 配置节名称 key: 配置键名称 default: 默认值 Returns: 配置值 """ try: with self.config_lock: if section is None: return self.current_config.copy() elif key is None: return self.current_config.get(section, {}).copy() else: return self.current_config.get(section, {}).get(key, default) except Exception as e: print(f"✗ 配置获取失败: {e}") return default def set_config(self, section: str, key: str = None, value: Any = None, save: bool = True) -> bool: """ 设置配置值 Args: section: 配置节名称 key: 配置键名称 value: 配置值 save: 是否保存到文件 Returns: 是否设置成功 """ try: with self.config_lock: # 备份当前配置 self._backup_config() if key is None: # 设置整个节 if isinstance(value, dict): self.current_config[section] = value else: print(f"✗ 配置值必须是字典类型: {section}") return False else: # 设置特定键值 if section not in self.current_config: self.current_config[section] = {} old_value = self.current_config[section].get(key) self.current_config[section][key] = value # 触发配置变更回调 self._trigger_config_callback("config_changed", { "section": section, "key": key, "old_value": old_value, "new_value": value, "timestamp": time.time() }) self.config_stats["changes"] += 1 self.last_change_time = time.time() # 保存配置 if save: return self.save_config() else: return True except Exception as e: print(f"✗ 配置设置失败: {e}") self.config_stats["errors"] += 1 # 触发配置错误回调 self._trigger_config_callback("config_error", { "error": str(e), "operation": "set" }) return False def update_config(self, updates: Dict[str, Any], save: bool = True) -> bool: """ 批量更新配置 Args: updates: 更新字典 save: 是否保存到文件 Returns: 是否更新成功 """ try: with self.config_lock: # 备份当前配置 self._backup_config() old_values = {} # 应用更新 for section, section_data in updates.items(): if isinstance(section_data, dict): if section not in self.current_config: self.current_config[section] = {} for key, value in section_data.items(): old_values[f"{section}.{key}"] = self.current_config[section].get(key) self.current_config[section][key] = value else: old_values[section] = self.current_config.get(section) self.current_config[section] = section_data self.config_stats["changes"] += 1 self.last_change_time = time.time() # 触发配置变更回调 self._trigger_config_callback("config_changed", { "updates": updates, "old_values": old_values, "timestamp": self.last_change_time }) # 保存配置 if save: return self.save_config() else: return True except Exception as e: print(f"✗ 配置批量更新失败: {e}") self.config_stats["errors"] += 1 # 触发配置错误回调 self._trigger_config_callback("config_error", { "error": str(e), "operation": "update" }) return False def reset_config(self, section: str = None) -> bool: """ 重置配置 Args: section: 要重置的配置节，None表示重置所有配置 Returns: 是否重置成功 """ try: with self.config_lock: # 备份当前配置 self._backup_config() if section is None: # 重置所有配置 self.current_config = self.default_config.copy() elif section in self.default_config: # 重置特定节 self.current_config[section] = self.default_config[section].copy() else: print(f"✗ 无效的配置节: {section}") return False self.config_stats["changes"] += 1 self.last_change_time = time.time() # 触发配置变更回调 self._trigger_config_callback("config_changed", { "operation": "reset", "section": section, "timestamp": self.last_change_time }) # 保存配置 return self.save_config() except Exception as e: print(f"✗ 配置重置失败: {e}") self.config_stats["errors"] += 1 # 触发配置错误回调 self._trigger_config_callback("config_error", { "error": str(e), "operation": "reset" }) return False def _backup_config(self): """备份当前配置""" try: self.config_backup = json.loads(json.dumps(self.current_config)) self.config_stats["backups"] += 1 print("✓ 配置已备份") except Exception as e: print(f"✗ 配置备份失败: {e}") def restore_config(self) -> bool: """ 恢复配置备份 Returns: 是否恢复成功 """ try: with self.config_lock: if self.config_backup: self.current_config = json.loads(json.dumps(self.config_backup)) self.config_stats["changes"] += 1 self.last_change_time = time.time() # 触发配置变更回调 self._trigger_config_callback("config_changed", { "operation": "restore", "timestamp": self.last_change_time }) # 保存配置 return self.save_config() else: print("✗ 没有可用的配置备份") return False except Exception as e: print(f"✗ 配置恢复失败: {e}") self.config_stats["errors"] += 1 # 触发配置错误回调 self._trigger_config_callback("config_error", { "error": str(e), "operation": "restore" }) return False def validate_config(self) -> bool: """ 验证配置 Returns: 配置是否有效 """ try: # 这里可以添加配置验证逻辑 # 例如：检查必需的配置项是否存在，值是否在有效范围内等 # 检查基本结构 required_sections = ["general", "recognition", "synthesis", "commands"] for section in required_sections: if section not in self.current_config: print(f"✗ 缺少必需的配置节: {section}") return False # 检查关键配置值范围 general = self.current_config.get("general", {}) if "language" in general and general["language"] not in ["zh-CN", "en-US", "ja-JP"]: print("✗ 无效的语言设置") return False recognition = self.current_config.get("recognition", {}) if "sample_rate" in recognition and recognition["sample_rate"] not in [8000, 16000, 22050, 44100, 48000]: print("✗ 无效的采样率设置") return False synthesis = self.current_config.get("synthesis", {}) if "speed" in synthesis and not (0.5 <= synthesis["speed"] <= 2.0): print("✗ 语速设置超出有效范围 (0.5-2.0)") return False if "volume" in synthesis and not (0.0 <= synthesis["volume"] <= 1.0): print("✗ 音量设置超出有效范围 (0.0-1.0)") return False print("✓ 配置验证通过") return True except Exception as e: print(f"✗ 配置验证失败: {e}") return False def export_config(self, file_path: str) -> bool: """ 导出配置到文件 Args: file_path: 导出文件路径 Returns: 是否导出成功 """ try: with self.config_lock: with open(file_path, 'w', encoding='utf-8') as f: json.dump(self.current_config, f, ensure_ascii=False, indent=2) print(f"✓ 配置已导出到: {file_path}") return True except Exception as e: print(f"✗ 配置导出失败: {e}") return False def import_config(self, file_path: str) -> bool: """ 从文件导入配置 Args: file_path: 导入文件路径 Returns: 是否导入成功 """ try: with self.config_lock: # 备份当前配置 self._backup_config() with open(file_path, 'r', encoding='utf-8') as f: imported_config = json.load(f) # 验证导入的配置 temp_manager = VoiceConfigManager(self.plugin) temp_manager.current_config = imported_config if not temp_manager.validate_config(): print("✗ 导入的配置无效") return False # 应用导入的配置 self.current_config = imported_config self.config_stats["changes"] += 1 self.last_change_time = time.time() # 触发配置变更回调 self._trigger_config_callback("config_changed", { "operation": "import", "source": file_path, "timestamp": self.last_change_time }) # 保存配置 return self.save_config() except Exception as e: print(f"✗ 配置导入失败: {e}") self.config_stats["errors"] += 1 # 触发配置错误回调 self._trigger_config_callback("config_error", { "error": str(e), "operation": "import" }) return False def get_default_config(self) -> Dict[str, Any]: """ 获取默认配置 Returns: 默认配置字典 """ return self.default_config.copy() def get_config_stats(self) -> Dict[str, int]: """ 获取配置统计信息 Returns: 统计信息字典 """ return self.config_stats.copy() def reset_config_stats(self): """重置配置统计信息""" try: self.config_stats = { "loads": 0, "saves": 0, "changes": 0, "errors": 0, "backups": 0 } print("✓ 配置统计信息已重置") except Exception as e: print(f"✗ 配置统计信息重置失败: {e}") def register_config_listener(self, section: str, key: str, listener: callable) -> bool: """ 注册配置监听器 Args: section: 配置节名称 key: 配置键名称 listener: 监听器函数 Returns: 是否注册成功 """ try: listener_key = f"{section}.{key}" if listener_key not in self.config_listeners: self.config_listeners[listener_key] = [] self.config_listeners[listener_key].append(listener) print(f"✓ 配置监听器已注册: {listener_key}") return True except Exception as e: print(f"✗ 配置监听器注册失败: {e}") return False def unregister_config_listener(self, section: str, key: str, listener: callable) -> bool: """ 注销配置监听器 Args: section: 配置节名称 key: 配置键名称 listener: 监听器函数 Returns: 是否注销成功 """ try: listener_key = f"{section}.{key}" if listener_key in self.config_listeners: if listener in self.config_listeners[listener_key]: self.config_listeners[listener_key].remove(listener) print(f"✓ 配置监听器已注销: {listener_key}") return True print(f"✗ 配置监听器不存在: {listener_key}") return False except Exception as e: print(f"✗ 配置监听器注销失败: {e}") return False def _trigger_config_callback(self, callback_type: str, data: Dict[str, Any]): """ 触发配置回调 Args: callback_type: 回调类型 data: 回调数据 """ try: if callback_type in self.config_callbacks: for callback in self.config_callbacks[callback_type]: try: callback(data) except Exception as e: print(f"✗ 配置回调执行失败: {e}") except Exception as e: print(f"✗ 配置回调触发失败: {e}") def register_config_callback(self, callback_type: str, callback: callable): """ 注册配置回调 Args: callback_type: 回调类型 callback: 回调函数 """ try: if callback_type in self.config_callbacks: self.config_callbacks[callback_type].append(callback) print(f"✓ 配置回调已注册: {callback_type}") else: print(f"✗ 无效的回调类型: {callback_type}") except Exception as e: print(f"✗ 配置回调注册失败: {e}") def unregister_config_callback(self, callback_type: str, callback: callable): """ 注销配置回调 Args: callback_type: 回调类型 callback: 回调函数 """ try: if callback_type in self.config_callbacks: if callback in self.config_callbacks[callback_type]: self.config_callbacks[callback_type].remove(callback) print(f"✓ 配置回调已注销: {callback_type}") except Exception as e: print(f"✗ 配置回调注销失败: {e}") def apply_config_to_modules(self): """ 将配置应用到各个模块 """ try: if not self.enabled: return # 应用配置到语音管理器 if self.plugin.voice_manager: general_config = self.get_config("general") self.plugin.voice_manager.set_voice_config(general_config) # 应用配置到语音识别器 if self.plugin.speech_recognizer: recognition_config = self.get_config("recognition") self.plugin.speech_recognizer.set_recognition_config(recognition_config) # 应用配置到语音合成器 if self.plugin.speech_synthesizer: synthesis_config = self.get_config("synthesis") self.plugin.speech_synthesizer.set_synthesis_config(synthesis_config) # 应用配置到命令处理器 if self.plugin.command_processor: commands_config = self.get_config("commands") self.plugin.command_processor.set_command_config(commands_config) # 应用配置到音频管理器 if self.plugin.audio_manager: audio_config = self.get_config("audio") self.plugin.audio_manager.set_audio_config(audio_config) # 应用配置到NLP处理器 if self.plugin.nlp_processor: nlp_config = self.get_config("nlp") self.plugin.nlp_processor.set_nlp_config(nlp_config) # 应用配置到事件处理器 if self.plugin.event_handler: events_config = self.get_config("events") self.plugin.event_handler.set_event_config(events_config) print("✓ 配置已应用到所有模块") except Exception as e: print(f"✗ 配置应用失败: {e}") import traceback traceback.print_exc()