""" Gestionnaire de configuration pour l'orchestrateur Skull Pi. Synchronise les configurations entre fichiers JSON et MQTT retained. """ import json import logging import os from pathlib import Path from typing import Any, Dict, Optional class Config: """Gestionnaire de configuration centralisé.""" def __init__(self): self.logger = logging.getLogger("orchestrator.config") # Chemins de configuration self.config_dir = Path("/opt/Skull/config") self.config_dir.mkdir(parents=True, exist_ok=True) # Configuration par défaut self.default_config = { "mqtt": { "broker": "localhost", "port": 1883, "username": None, "password": None, }, "api": {"port": 8080, "host": "0.0.0.0"}, "vision": { "motion_threshold": 0.3, "face_confidence": 0.7, "tracking_enabled": True, }, "motion": { "pan_limits": [-90, 90], "tilt_limits": [-45, 45], "jaw_limits": [0, 30], "speed": 50, "estop_enabled": True, }, "accueil": { "message": "Bonjour ! Je suis Skull Pi.", "enabled": True, "timeout": 10, }, "song": { "directory": "/opt/Skull/songs", "current_file": "", "jaw_sync": True, "volume": 0.8, }, "ia": { "model": "local", "system_prompt": "Tu es Skull Pi, un assistant vocal amical.", "max_tokens": 150, "temperature": 0.7, }, "fullauto": { "enabled": False, "interval_song_ms": 300000, # 5 minutes "accueil_after_song": True, "ia_after_accueil": True, "sleep_mode": False, }, "audio": { "input_device": "default", "sample_rate": 16000, "vad_threshold": 0.5, }, "health": { "check_interval": 30, "service_timeout": 60, "restart_attempts": 3, }, } # Configuration active self._config: Dict[str, Any] = {} self._load_config() def _load_config(self) -> None: """Charge la configuration depuis les fichiers.""" self._config = self.default_config.copy() # Charger les fichiers de config existants for config_file in self.config_dir.glob("*.json"): try: section_name = config_file.stem with open(config_file, "r", encoding="utf-8") as f: section_config = json.load(f) if section_name in self._config: self._config[section_name].update(section_config) else: self._config[section_name] = section_config self.logger.debug(f"Config {section_name} chargée") except Exception as e: self.logger.error(f"Erreur chargement config {config_file}: {e}") # Variables d'environnement self._load_env_overrides() def _load_env_overrides(self) -> None: """Applique les surcharges par variables d'environnement.""" env_mappings = { "SKULL_MQTT_BROKER": ("mqtt", "broker"), "SKULL_MQTT_PORT": ("mqtt", "port"), "SKULL_MQTT_USERNAME": ("mqtt", "username"), "SKULL_MQTT_PASSWORD": ("mqtt", "password"), "SKULL_API_PORT": ("api", "port"), "SKULL_LOG_LEVEL": ("logging", "level"), } for env_var, (section, key) in env_mappings.items(): value = os.getenv(env_var) if value is not None: if section not in self._config: self._config[section] = {} # Conversion de type if key == "port": value = int(value) elif value.lower() in ("true", "false"): value = value.lower() == "true" self._config[section][key] = value self.logger.debug(f"Override env {env_var}: {section}.{key} = {value}") def get(self, section: str, key: Optional[str] = None, default: Any = None) -> Any: """Récupère une valeur de configuration.""" if key is None: return self._config.get(section, default) section_config = self._config.get(section, {}) return section_config.get(key, default) def set(self, section: str, key: str, value: Any) -> bool: """Définit une valeur de configuration.""" try: if section not in self._config: self._config[section] = {} self._config[section][key] = value # Sauvegarder self._save_section(section) self.logger.info(f"Config mise à jour: {section}.{key} = {value}") return True except Exception as e: self.logger.error(f"Erreur mise à jour config: {e}") return False def update_section(self, section: str, config: Dict[str, Any]) -> bool: """Met à jour une section complète.""" try: if section not in self._config: self._config[section] = {} self._config[section].update(config) # Sauvegarder self._save_section(section) self.logger.info(f"Section {section} mise à jour") return True except Exception as e: self.logger.error(f"Erreur mise à jour section {section}: {e}") return False def _save_section(self, section: str) -> None: """Sauvegarde une section dans son fichier.""" config_file = self.config_dir / f"{section}.json" try: with open(config_file, "w", encoding="utf-8") as f: json.dump(self._config[section], f, indent=2, ensure_ascii=False) except Exception as e: self.logger.error(f"Erreur sauvegarde {config_file}: {e}") def get_all(self) -> Dict[str, Any]: """Retourne toute la configuration.""" return self._config.copy() def get_section_list(self) -> list[str]: """Retourne la liste des sections disponibles.""" return list(self._config.keys()) def reset_section(self, section: str) -> bool: """Remet une section aux valeurs par défaut.""" try: if section in self.default_config: self._config[section] = self.default_config[section].copy() self._save_section(section) self.logger.info(f"Section {section} remise aux valeurs par défaut") return True else: self.logger.error(f"Section {section} inconnue") return False except Exception as e: self.logger.error(f"Erreur reset section {section}: {e}") return False def validate_section( self, section: str, config: Dict[str, Any] ) -> tuple[bool, list[str]]: """Valide la configuration d'une section.""" errors = [] try: if section == "motion": # Validation des limites moteurs if "pan_limits" in config: limits = config["pan_limits"] if not (isinstance(limits, list) and len(limits) == 2): errors.append("pan_limits doit être [min, max]") elif limits[0] >= limits[1]: errors.append("pan_limits: min doit être < max") if "tilt_limits" in config: limits = config["tilt_limits"] if not (isinstance(limits, list) and len(limits) == 2): errors.append("tilt_limits doit être [min, max]") elif limits[0] >= limits[1]: errors.append("tilt_limits: min doit être < max") elif section == "fullauto": # Validation Full Auto if "interval_song_ms" in config: interval = config["interval_song_ms"] if not isinstance(interval, int) or interval < 1000: errors.append("interval_song_ms doit être >= 1000") elif section == "vision": # Validation vision if "motion_threshold" in config: threshold = config["motion_threshold"] if not isinstance(threshold, (int, float)) or not ( 0 <= threshold <= 1 ): errors.append("motion_threshold doit être entre 0 et 1") return len(errors) == 0, errors except Exception as e: return False, [f"Erreur validation: {e}"] def sync_from_mqtt(self, topic: str, payload: str) -> bool: """Synchronise depuis un message MQTT retained.""" try: # Format: skull/config/section if not topic.startswith("skull/config/"): return False section = topic.replace("skull/config/", "") config_data = json.loads(payload) # Valider valid, errors = self.validate_section(section, config_data) if not valid: self.logger.error(f"Config MQTT invalide pour {section}: {errors}") return False # Appliquer return self.update_section(section, config_data) except Exception as e: self.logger.error(f"Erreur sync MQTT {topic}: {e}") return False def get_mqtt_topics(self) -> Dict[str, str]: """Retourne les topics MQTT pour chaque section.""" topics = {} for section in self._config: topics[f"skull/config/{section}"] = json.dumps(self._config[section]) return topics