""" Planificateur Full Auto pour l'orchestrateur Skull Pi. Gère les cycles automatiques chanson → accueil → IA. """ import asyncio import logging import time from typing import Optional from orchestrator.config import Config from orchestrator.state_machine import StateMachine class FullAutoScheduler: """Planificateur pour le mode Full Auto.""" def __init__( self, state_machine: StateMachine, config: Config, logger: logging.Logger ): self.state_machine = state_machine self.config = config self.logger = logger # État du scheduler self.active = False self.last_song_time = 0.0 self.cycle_in_progress = False # Configuration par défaut self.interval_song_ms = 300000 # 5 minutes self.accueil_after_song = True self.ia_after_accueil = True # Tâche asyncio self.scheduler_task: Optional[asyncio.Task] = None async def start(self) -> bool: """Démarre le scheduler Full Auto.""" try: if self.active: self.logger.info("Scheduler Full Auto déjà actif") return True # Charger la configuration await self.update_config() self.active = True self.last_song_time = time.time() # Démarrer la tâche de planification self.scheduler_task = asyncio.create_task(self._scheduler_loop()) self.logger.info("Scheduler Full Auto démarré") return True except Exception as e: self.logger.error(f"Erreur démarrage scheduler: {e}") return False async def stop(self) -> None: """Arrête le scheduler Full Auto.""" self.active = False if self.scheduler_task: self.scheduler_task.cancel() try: await self.scheduler_task except asyncio.CancelledError: pass self.scheduler_task = None self.logger.info("Scheduler Full Auto arrêté") async def update_config(self) -> None: """Met à jour la configuration depuis le fichier config.""" fullauto_config = self.config.get("fullauto", {}) self.interval_song_ms = fullauto_config.get("interval_song_ms", 300000) self.accueil_after_song = fullauto_config.get("accueil_after_song", True) self.ia_after_accueil = fullauto_config.get("ia_after_accueil", True) self.logger.debug( f"Config scheduler mise à jour: " f"interval={self.interval_song_ms}ms, " f"accueil={self.accueil_after_song}, " f"ia={self.ia_after_accueil}" ) def is_active(self) -> bool: """Retourne si le scheduler est actif.""" return self.active def get_next_song_time(self) -> Optional[float]: """Retourne le timestamp du prochain cycle chanson.""" if not self.active: return None interval_seconds = self.interval_song_ms / 1000.0 return self.last_song_time + interval_seconds def get_time_until_next_song(self) -> Optional[float]: """Retourne le nombre de secondes avant le prochain cycle.""" next_time = self.get_next_song_time() if next_time is None: return None return max(0, next_time - time.time()) async def check(self) -> None: """Vérifie s'il faut déclencher un cycle (appelé depuis la boucle principale).""" if not self.active or self.cycle_in_progress: return time_until_next = self.get_time_until_next_song() if time_until_next is not None and time_until_next <= 0: await self.trigger_cycle() async def trigger_cycle(self) -> None: """Déclenche un cycle Full Auto manuellement.""" if self.cycle_in_progress: self.logger.warning("Cycle Full Auto déjà en cours") return self.cycle_in_progress = True self.logger.info("Déclenchement cycle Full Auto") try: # Phase 1: Chanson await self._phase_song() # Phase 2: Accueil (si activé) if self.accueil_after_song: await self._phase_accueil() # Phase 3: IA (si activée) if self.ia_after_accueil: await self._phase_ia() # Marquer le temps pour le prochain cycle self.last_song_time = time.time() self.logger.info("Cycle Full Auto terminé") except Exception as e: self.logger.error(f"Erreur dans le cycle Full Auto: {e}") finally: self.cycle_in_progress = False async def _scheduler_loop(self) -> None: """Boucle principale du scheduler.""" while self.active: try: await self.check() await asyncio.sleep(1.0) # Vérifier chaque seconde except asyncio.CancelledError: break except Exception as e: self.logger.error(f"Erreur dans la boucle scheduler: {e}") await asyncio.sleep(5.0) async def _phase_song(self) -> None: """Phase chanson du cycle Full Auto.""" self.logger.info("Full Auto: Phase chanson") # Passer en mode Chanson await self.state_machine.set_mode("Chanson") # Attendre que la chanson soit finie max_wait = 300 # 5 minutes max pour une chanson waited = 0 while waited < max_wait: status = self.state_machine.get_status() if status["state"] == "Idle": break await asyncio.sleep(1) waited += 1 if waited >= max_wait: self.logger.warning("Timeout attente fin chanson") await self.state_machine.set_mode("Idle") async def _phase_accueil(self) -> None: """Phase accueil du cycle Full Auto.""" self.logger.info("Full Auto: Phase accueil") # Passer en mode Accueil await self.state_machine.set_mode("Accueil") # Attendre un moment pour détecter du mouvement wait_time = 30 # 30 secondes for _ in range(wait_time): status = self.state_machine.get_status() # Si on détecte du mouvement et qu'on passe en Talk, attendre la fin if status["state"] in ["Track", "Talk"]: await self._wait_for_idle(60) # 1 minute max break await asyncio.sleep(1) async def _phase_ia(self) -> None: """Phase IA du cycle Full Auto.""" self.logger.info("Full Auto: Phase IA") # Passer en mode IA await self.state_machine.set_mode("IA") # Attendre un moment pour détecter de l'activité vocale wait_time = 60 # 1 minute for _ in range(wait_time): status = self.state_machine.get_status() # Si on détecte activité (Listen, Think, Speak), attendre la fin if status["state"] in ["Listen", "Think", "Speak"]: await self._wait_for_idle(120) # 2 minutes max break await asyncio.sleep(1) async def _wait_for_idle(self, max_wait: int) -> None: """Attend que la machine d'états revienne en Idle.""" waited = 0 while waited < max_wait: status = self.state_machine.get_status() if status["state"] == "Idle": break await asyncio.sleep(1) waited += 1 if waited >= max_wait: self.logger.warning("Timeout attente retour Idle") await self.state_machine.force_state("Idle") def get_status(self) -> dict: """Retourne le statut du scheduler.""" return { "active": self.active, "cycle_in_progress": self.cycle_in_progress, "last_song_time": self.last_song_time, "next_song_time": self.get_next_song_time(), "time_until_next": self.get_time_until_next_song(), "config": { "interval_song_ms": self.interval_song_ms, "accueil_after_song": self.accueil_after_song, "ia_after_accueil": self.ia_after_accueil, }, }