""" 配置管理器模块 负责管理触觉反馈系统的配置和用户偏好设置 """ import time import json import os from typing import Dict, Any, List, Optional from pathlib import Path class HapticConfigManager: """ 触觉配置管理器 负责管理触觉反馈系统的配置和用户偏好设置 """ def __init__(self, plugin): """ 初始化触觉配置管理器 Args: plugin: 触觉反馈系统插件实例 """ self.plugin = plugin self.enabled = False self.initialized = False # 配置文件路径 self.config_path = "/home/hello/EG/plugins/user/haptic_feedback_system/config/haptic_config.json" # 默认配置 self.default_config = { 'global': { 'enabled': True, 'intensity_scale': 1.0, 'frequency_scale': 1.0, 'duration_scale': 1.0, 'master_volume': 1.0, 'enable_device_detection': True, 'auto_enable_new_devices': True }, 'devices': { 'default_intensity': 0.8, 'default_frequency': 50.0, 'max_concurrent_effects': 16, 'crossfade_time': 0.1, 'device_timeout': 30.0 }, 'effects': { 'enable_preprocessing': True, 'enable_postprocessing': True, 'max_effect_duration': 10.0, 'effect_smoothing': 0.1, 'enable_custom_effects': True }, 'physics': { 'enable_collision_feedback': True, 'enable_force_feedback': True, 'enable_friction_feedback': True, 'collision_intensity_scale': 1.0, 'force_intensity_scale': 1.0, 'friction_intensity_scale': 1.0 }, 'audio': { 'enable_audio_haptics': True, 'enable_beat_detection': True, 'audio_analysis_rate': 30.0, 'frequency_bands': 8, 'audio_intensity_scale': 1.0 }, 'events': { 'enable_event_processing': True, 'enable_event_suppression': True, 'enable_event_combination': True, 'event_queue_size': 100, 'event_suppression_time': 0.1 }, 'profiles': { 'current_profile': 'default', 'profiles': { 'default': { 'name': 'Default', 'description': 'Default haptic profile', 'settings': {} }, 'cinematic': { 'name': 'Cinematic', 'description': 'Enhanced haptic feedback for cinematic experiences', 'settings': { 'intensity_scale': 1.2, 'frequency_scale': 1.1 } }, 'gaming': { 'name': 'Gaming', 'description': 'Optimized haptic feedback for gaming', 'settings': { 'intensity_scale': 1.0, 'response_time': 0.01 } }, 'subtle': { 'name': 'Subtle', 'description': 'Subtle haptic feedback for casual use', 'settings': { 'intensity_scale': 0.5, 'frequency_scale': 0.8 } } } }, 'user_preferences': { 'enable_haptics': True, 'preferred_devices': [], 'blocked_devices': [], 'custom_mappings': {}, 'sensitivity': 1.0, 'comfort_mode': False } } # 当前配置 self.current_config = self.default_config.copy() # 配置变更监听器 self.config_listeners = {} # 配置验证规则 self.validation_rules = { 'global.intensity_scale': (0.0, 2.0), 'global.frequency_scale': (0.1, 5.0), 'global.duration_scale': (0.1, 5.0), 'global.master_volume': (0.0, 1.0), 'devices.default_intensity': (0.0, 1.0), 'devices.default_frequency': (1.0, 100.0), 'devices.max_concurrent_effects': (1, 64), 'devices.crossfade_time': (0.01, 1.0), 'devices.device_timeout': (1.0, 300.0), 'audio.audio_analysis_rate': (1.0, 120.0), 'audio.frequency_bands': (1, 32) } # 配置历史 self.config_history = [] self.max_history_size = 50 # 统计信息 self.stats = { 'config_loads': 0, 'config_saves': 0, 'config_changes': 0, 'validation_errors': 0 } print("✓ 触觉配置管理器已创建") def initialize(self) -> bool: """ 初始化触觉配置管理器 Returns: 是否初始化成功 """ try: # 创建配置目录 config_dir = Path(self.config_path).parent config_dir.mkdir(parents=True, exist_ok=True) # 加载配置 self.load_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_history.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 # 定期保存配置（如果需要） pass except Exception as e: print(f"✗ 触觉配置管理器更新失败: {e}") import traceback traceback.print_exc() def load_config(self) -> bool: """ 加载配置 Returns: 是否加载成功 """ try: if os.path.exists(self.config_path): with open(self.config_path, 'r', encoding='utf-8') as f: loaded_config = json.load(f) # 合并配置（保留默认配置结构） self._merge_config(self.current_config, loaded_config) print(f"✓ 配置已从 {self.config_path} 加载") else: print("⚠ 配置文件不存在，使用默认配置") self.stats['config_loads'] += 1 return True except Exception as e: print(f"✗ 配置加载失败: {e}") return False def save_config(self) -> bool: """ 保存配置 Returns: 是否保存成功 """ try: with open(self.config_path, 'w', encoding='utf-8') as f: json.dump(self.current_config, f, indent=2, ensure_ascii=False) self.stats['config_saves'] += 1 print(f"✓ 配置已保存到 {self.config_path}") return True except Exception as e: print(f"✗ 配置保存失败: {e}") return False def _merge_config(self, base_config: Dict[str, Any], new_config: Dict[str, Any]): """ 合并配置 Args: base_config: 基础配置 new_config: 新配置 """ try: for key, value in new_config.items(): if key in base_config and isinstance(base_config[key], dict) and isinstance(value, dict): self._merge_config(base_config[key], value) else: base_config[key] = value except Exception as e: print(f"✗ 配置合并失败: {e}") def get_config_value(self, key_path: str, default_value: Any = None) -> Any: """ 获取配置值 Args: key_path: 配置键路径（如 'global.intensity_scale'） default_value: 默认值 Returns: 配置值 """ try: keys = key_path.split('.') config = self.current_config for key in keys: if isinstance(config, dict) and key in config: config = config[key] else: return default_value return config except Exception as e: print(f"✗ 配置值获取失败: {e}") return default_value def set_config_value(self, key_path: str, value: Any) -> bool: """ 设置配置值 Args: key_path: 配置键路径 value: 配置值 Returns: 是否设置成功 """ try: # 验证配置值 if not self._validate_config_value(key_path, value): print(f"✗ 配置值验证失败: {key_path} = {value}") self.stats['validation_errors'] += 1 return False # 保存到历史记录 old_value = self.get_config_value(key_path) self._add_to_history(key_path, old_value, value) # 设置新值 keys = key_path.split('.') config = self.current_config for key in keys[:-1]: if key not in config or not isinstance(config[key], dict): config[key] = {} config = config[key] config[keys[-1]] = value # 触发配置变更回调 self._trigger_config_change(key_path, old_value, value) self.stats['config_changes'] += 1 print(f"✓ 配置已更新: {key_path} = {value}") return True except Exception as e: print(f"✗ 配置值设置失败: {e}") return False def _validate_config_value(self, key_path: str, value: Any) -> bool: """ 验证配置值 Args: key_path: 配置键路径 value: 配置值 Returns: 是否验证通过 """ try: if key_path in self.validation_rules: rule = self.validation_rules[key_path] if isinstance(rule, tuple) and len(rule) == 2: min_val, max_val = rule if isinstance(value, (int, float)): return min_val <= value <= max_val # 可以添加更多验证规则 return True # 默认通过验证 except Exception as e: print(f"✗ 配置值验证异常: {e}") return False def _add_to_history(self, key_path: str, old_value: Any, new_value: Any): """ 添加到配置历史 Args: key_path: 配置键路径 old_value: 旧值 new_value: 新值 """ try: history_entry = { 'key_path': key_path, 'old_value': old_value, 'new_value': new_value, 'timestamp': time.time() } self.config_history.append(history_entry) # 保持历史记录大小 if len(self.config_history) > self.max_history_size: self.config_history.pop(0) except Exception as e: print(f"✗ 配置历史记录失败: {e}") def reset_to_default(self) -> bool: """ 重置为默认配置 Returns: 是否重置成功 """ try: self.current_config = self.default_config.copy() print("✓ 配置已重置为默认值") return True except Exception as e: print(f"✗ 配置重置失败: {e}") return False def get_current_config(self) -> Dict[str, Any]: """ 获取当前配置 Returns: 当前配置字典 """ return self.current_config.copy() def get_default_config(self) -> Dict[str, Any]: """ 获取默认配置 Returns: 默认配置字典 """ return self.default_config.copy() def export_config(self, file_path: str) -> bool: """ 导出配置到文件 Args: file_path: 导出文件路径 Returns: 是否导出成功 """ try: with open(file_path, 'w', encoding='utf-8') as f: json.dump(self.current_config, f, indent=2, ensure_ascii=False) 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 open(file_path, 'r', encoding='utf-8') as f: imported_config = json.load(f) self._merge_config(self.current_config, imported_config) print(f"✓ 配置已从 {file_path} 导入") return True except Exception as e: print(f"✗ 配置导入失败: {e}") return False def apply_profile(self, profile_name: str) -> bool: """ 应用配置预设 Args: profile_name: 预设名称 Returns: 是否应用成功 """ try: profiles = self.current_config.get('profiles', {}).get('profiles', {}) if profile_name not in profiles: print(f"✗ 预设不存在: {profile_name}") return False profile = profiles[profile_name] settings = profile.get('settings', {}) # 应用预设设置 for key, value in settings.items(): self.set_config_value(key, value) # 更新当前预设 self.set_config_value('profiles.current_profile', profile_name) print(f"✓ 预设已应用: {profile_name}") return True except Exception as e: print(f"✗ 预设应用失败: {e}") return False def create_profile(self, profile_name: str, settings: Dict[str, Any], name: str = None, description: str = None) -> bool: """ 创建配置预设 Args: profile_name: 预设名称 settings: 预设设置 name: 显示名称 description: 描述 Returns: 是否创建成功 """ try: profiles = self.current_config.setdefault('profiles', {}).setdefault('profiles', {}) profiles[profile_name] = { 'name': name or profile_name, 'description': description or f'Custom profile: {profile_name}', 'settings': settings } print(f"✓ 预设已创建: {profile_name}") return True except Exception as e: print(f"✗ 预设创建失败: {e}") return False def delete_profile(self, profile_name: str) -> bool: """ 删除配置预设 Args: profile_name: 预设名称 Returns: 是否删除成功 """ try: profiles = self.current_config.get('profiles', {}).get('profiles', {}) if profile_name in profiles: del profiles[profile_name] print(f"✓ 预设已删除: {profile_name}") return True else: print(f"✗ 预设不存在: {profile_name}") return False except Exception as e: print(f"✗ 预设删除失败: {e}") return False def get_available_profiles(self) -> List[Dict[str, Any]]: """ 获取可用预设列表 Returns: 预设列表 """ try: profiles = self.current_config.get('profiles', {}).get('profiles', {}) result = [] for name, profile in profiles.items(): result.append({ 'name': name, 'display_name': profile.get('name', name), 'description': profile.get('description', ''), 'settings_count': len(profile.get('settings', {})) }) return result except Exception as e: print(f"✗ 预设列表获取失败: {e}") return [] def _trigger_config_change(self, key_path: str, old_value: Any, new_value: Any): """ 触发配置变更回调 Args: key_path: 配置键路径 old_value: 旧值 new_value: 新值 """ try: # 通知监听器 if key_path in self.config_listeners: for callback in self.config_listeners[key_path]: try: callback(key_path, old_value, new_value) except Exception as e: print(f"✗ 配置变更回调执行失败: {e}") # 通知全局监听器 if '*' in self.config_listeners: for callback in self.config_listeners['*']: try: callback(key_path, old_value, new_value) except Exception as e: print(f"✗ 全局配置变更回调执行失败: {e}") except Exception as e: print(f"✗ 配置变更回调触发失败: {e}") def register_config_listener(self, key_path: str, callback: callable): """ 注册配置变更监听器 Args: key_path: 配置键路径（'*' 表示监听所有变更） callback: 回调函数 """ try: if key_path not in self.config_listeners: self.config_listeners[key_path] = [] self.config_listeners[key_path].append(callback) print(f"✓ 配置监听器已注册: {key_path}") except Exception as e: print(f"✗ 配置监听器注册失败: {e}") def unregister_config_listener(self, key_path: str, callback: callable): """ 注销配置变更监听器 Args: key_path: 配置键路径 callback: 回调函数 """ try: if key_path in self.config_listeners: if callback in self.config_listeners[key_path]: self.config_listeners[key_path].remove(callback) print(f"✓ 配置监听器已注销: {key_path}") except Exception as e: print(f"✗ 配置监听器注销失败: {e}") def get_stats(self) -> Dict[str, int]: """ 获取统计信息 Returns: 统计信息字典 """ return self.stats.copy() def reset_stats(self): """重置统计信息""" try: self.stats = { 'config_loads': 0, 'config_saves': 0, 'config_changes': 0, 'validation_errors': 0 } print("✓ 配置管理器统计信息已重置") except Exception as e: print(f"✗ 配置管理器统计信息重置失败: {e}")