"""Chef d'orchestre : machine à états idle <-> recording. Branche le micro sur l'animation (niveau RMS → cercle) pendant l'enregistrement, puis lance la transcription sur un thread de fond et colle le texte. Ne meurt jamais sur une erreur : tout est attrapé et ramène à idle (fenêtre fermée). Les méthodes touchant l'UI (PulseWindow show/hide) doivent être invoquées sur le thread principal — c'est le rôle de l'appelant (app via la hotkey, qui s'exécute déjà dans le contexte de l'app). La transcription, elle, part sur un thread de fond. """ import threading from audio_level import compute_rms, shape_level from core import entree_audio_muette, is_long_enough from engine import Recorder, Transcriber, inject_text, play_sound class ZonzaController: """Machine à états pilotée par toggle(). on_state : callback(str) pour l'UI (menu).""" def __init__(self, config, pulse_window, on_state=None, run_on_main=None, lancer_tache=None): self.config = config self.pulse = pulse_window self.on_state = on_state or (lambda s: None) # run_on_main : marshale un appel sur le thread principal Cocoa (requis pour l'UI). # Par défaut exécute directement (utile pour les tests sans boucle Cocoa). self.run_on_main = run_on_main or (lambda fn: fn()) # lancer_tache : execute un travail HORS du thread principal. Tout ce qui # touche au peripherique audio passe par la — PortAudio peut bloquer # indefiniment si une autre application accapare le micro, et un blocage # sur le thread principal gele TOUTE l'interface Cocoa (halo compris). self.lancer_tache = lancer_tache or ( lambda fn: threading.Thread(target=fn, daemon=True).start() ) self.recorder = Recorder(config["sample_rate"], on_block=self._on_audio_block) self.transcriber = Transcriber(config["model"], config["language"]) self.recording = False self._lock = threading.Lock() self._smoothed_level = 0.0 self._safety_timer = None # arrêt auto si l'enregistrement traîne trop self._blocs_recus = 0 # compteur pour détecter un micro muet self._veille_silence = None def _on_audio_block(self, block): # thread audio : RMS brut → mise en forme (gain + courbe) → lissage attack/release # → push vers la fenêtre (thread-safe). Le lissage évite les saccades : montée # rapide quand on parle, retombée douce dans les silences. self._blocs_recus += 1 target = shape_level(compute_rms(block), self.config["level_gain"]) if target > self._smoothed_level: coef = self.config["level_attack"] else: coef = self.config["level_release"] self._smoothed_level += (target - self._smoothed_level) * coef self.pulse.set_level(self._smoothed_level) def toggle(self): with self._lock: if not self.recording: self._start() else: self._stop_and_process() def _start(self): self._smoothed_level = 0.0 self._blocs_recus = 0 self.recording = True self.pulse.show() self.pulse.set_status("Je vous écoute…") play_sound("start") self.on_state("recording") self._arm_safety_timer() self._armer_veille_silence() self.lancer_tache(self._ouvrir_flux) def _armer_veille_silence(self): """Prévient si aucun son n'arrive : le micro est pris par une autre app.""" if self._veille_silence is not None: self._veille_silence.cancel() self._veille_silence = threading.Timer( self.config["delai_detection_silence_s"], self._verifier_entree_audio ) self._veille_silence.daemon = True self._veille_silence.start() def _verifier_entree_audio(self): if not entree_audio_muette(self._blocs_recus, self.recording): return print("[micro] aucun son reçu — le micro est probablement pris par une autre application") play_sound("error") self.run_on_main(lambda: self.pulse.set_status("Aucun son — micro pris ailleurs ?")) def _ouvrir_flux(self): """Ouvre le micro. Hors thread principal : peut bloquer (cf __init__).""" try: self.recorder.start() except Exception as e: print(f"[erreur] démarrage enregistrement : {e}") play_sound("error") self.recording = False self._cancel_safety_timer() self.run_on_main(self.pulse.hide) self.on_state("idle") def _arm_safety_timer(self): # sécurité : si l'enregistrement n'est pas arrêté manuellement après max_duration_s, # on l'arrête automatiquement → la fenêtre du cercle ne peut jamais rester figée. self._cancel_safety_timer() self._safety_timer = threading.Timer( self.config["max_duration_s"], self._safety_stop ) self._safety_timer.daemon = True self._safety_timer.start() def _cancel_safety_timer(self): if self._safety_timer is not None: self._safety_timer.cancel() self._safety_timer = None if self._veille_silence is not None: self._veille_silence.cancel() self._veille_silence = None def _safety_stop(self): # déclenché par le timer (thread de fond) → l'arrêt touche l'UI (pulse.hide), donc # on marshale toggle() sur le thread principal Cocoa. print("[sécurité] durée max atteinte → arrêt auto de l'enregistrement") self.run_on_main(self.toggle) def _stop_and_process(self): """Appelé depuis le thread principal : ne touche PAS au périphérique audio.""" self.recording = False self._cancel_safety_timer() self.pulse.set_status("Transcription…") self.pulse.rearmer_garde(self.config["duree_max_transcription_s"]) self.on_state("transcribing") self.lancer_tache(self._fermer_et_transcrire) def _fermer_et_transcrire(self): """Ferme le flux, transcrit, colle. Hors thread principal : `recorder.stop()` peut bloquer si le micro a été accaparé par une autre application — c'est précisément ce qui figeait le halo quand cet appel vivait sur le thread principal (2026-08-21).""" try: audio = self.recorder.stop() play_sound("done") if not is_long_enough(len(audio), self.config["sample_rate"], self.config["min_duration_s"]): return text = self.transcriber.transcribe(audio) if not text: return print(f"[texte] {text}") try: inject_text(text) except Exception as e: print(f"[erreur] collage échoué ({e}) — texte dans le presse-papier, fais Cmd+V") play_sound("error") except Exception as e: print(f"[erreur] arrêt ou transcription : {e}") play_sound("error") finally: self.run_on_main(self.pulse.hide) self.on_state("idle")