From dcec1e61d0c757aab9b663e1e18852867c3b27e2 Mon Sep 17 00:00:00 2001 From: Ralph Mayola Date: Sun, 23 Aug 2026 18:19:25 +0200 Subject: [PATCH] Le controleur passe par les contrats, la duplication disparait --- app.py | 13 ++++-- engine.py => capture.py | 80 +++++------------------------------ controller.py | 25 +++++------ tests/test_liberation_flux.py | 2 +- tests/test_statut_bulle.py | 28 +++++++++++- 5 files changed, 61 insertions(+), 87 deletions(-) rename engine.py => capture.py (56%) diff --git a/app.py b/app.py index 00c43e0..5d58370 100644 --- a/app.py +++ b/app.py @@ -31,6 +31,7 @@ from pynput import keyboard from controller import ZonzaController from core import CONFIG, validate_config from interface import classe_overlay_pour +from moteur import choisir_transcripteur, detecter_moteur from systeme import choisir_systeme _STATE_TITLES = { @@ -43,7 +44,7 @@ _STATE_TITLES = { class ZonzaApp(NSObject): """Délégué applicatif : status item + menu + hotkey + contrôleur.""" - def init(self): + def initWithTranscripteur_systeme_(self, transcripteur, systeme): self = super().init() if self is None: return None @@ -51,7 +52,7 @@ class ZonzaApp(NSObject): self._sounds_item = None self.pulse = classe_overlay_pour().alloc().init() self.controller = ZonzaController( - CONFIG, self.pulse, on_state=self._set_state, + CONFIG, self.pulse, transcripteur, systeme, on_state=self._set_state, run_on_main=AppHelper.callAfter, ) self._listener = None @@ -140,6 +141,12 @@ def main(): return main._lock = lock # garder le verrou vivant tant que le process vit validate_config(CONFIG) + transcripteur = choisir_transcripteur( + nom=detecter_moteur(), + modele=CONFIG["model"], + langue=CONFIG["language"], + amorce=CONFIG["initial_prompt"], + ) # si l'Accessibilité n'est pas accordée, macOS affiche sa popup officielle # (« Zonza souhaite contrôler cet ordinateur ») et ajoute l'app à la liste — # bien plus fiable qu'un ajout manuel via « + » dans les Réglages. @@ -150,7 +157,7 @@ def main(): trusted = AXIsProcessTrustedWithOptions({kAXTrustedCheckOptionPrompt: True}) print(f"[diag] accessibilité accordée : {bool(trusted)}") app = NSApplication.sharedApplication() - delegate = ZonzaApp.alloc().init() + delegate = ZonzaApp.alloc().initWithTranscripteur_systeme_(transcripteur, systeme) delegate.setup() # référence gardée pour toute la vie du processus (sinon GC → icône disparaît) main._delegate = delegate diff --git a/engine.py b/capture.py similarity index 56% rename from engine.py rename to capture.py index d9f8375..22dec48 100644 --- a/engine.py +++ b/capture.py @@ -1,66 +1,35 @@ -"""Moteur de dictée — capture, transcription, collage, son. +"""Capture audio et collage du texte transcrit. -Aucune dépendance UI : importable par le contrôleur et l'app. +Aucune dépendance UI : importable par le contrôleur et l'app. La transcription +et les appels système propres à une plateforme vivent désormais ailleurs +(`moteur/`, `systeme/`) — ce module ne garde que ce qui n'a pas de frontière +de plateforme : la capture du micro et le collage du résultat. """ -import subprocess import time -import mlx_whisper import numpy as np import pyperclip import sounddevice as sd from pynput.keyboard import Controller, Key -from core import CONFIG, attendre, clean_transcript - -SOUNDS = { - "start": "/System/Library/Sounds/Pop.aiff", - "done": "/System/Library/Sounds/Glass.aiff", - "error": "/System/Library/Sounds/Basso.aiff", -} +from core import CONFIG, attendre _kbd = Controller() -def play_sound(name): - """Joue un son système macOS sans bloquer. Silencieux si CONFIG['sounds'] est False.""" - if not CONFIG["sounds"]: - return - path = SOUNDS.get(name) - if not path: - return - try: - subprocess.Popen(["afplay", path]) - except Exception: - pass # le son ne doit jamais casser le daemon - - -# Masque des modificateurs (ctrl, shift, alt, cmd) dans les drapeaux Quartz. -_MASQUE_MODIFICATEURS = 0x000F0000 - - -def modificateurs_enfonces(): - """True si au moins un modificateur physique est actuellement enfoncé.""" - try: - from Quartz import CGEventSourceFlagsState - return bool(CGEventSourceFlagsState(1) & _MASQUE_MODIFICATEURS) - except Exception: - return False # dans le doute, ne jamais bloquer le collage - - -def inject_text(text): +def inject_text(text, systeme): """Copie le texte dans le presse-papier puis colle avec Cmd+V dans l'app active. On attend d'abord que l'utilisateur ait relâché ses touches : s'il tient encore - Cmd+Ctrl+D quand on envoie Cmd+V, le système reçoit Cmd+Ctrl+V et l'application - cible ne colle rien — sans la moindre erreur de notre côté (constaté le - 2026-08-21 : deux dictées transcrites, jamais collées). + la hotkey quand on envoie Cmd+V, le système reçoit une combinaison différente et + l'application cible ne colle rien — sans la moindre erreur de notre côté + (constaté le 2026-08-21). """ if not text: return pyperclip.copy(text) libre = attendre( - lambda: not modificateurs_enfonces(), + lambda: not systeme.modificateurs_enfonces(), CONFIG["attente_modificateurs_s"], 0.03, ) @@ -131,30 +100,3 @@ class Recorder: if not frames: return np.zeros(0, dtype="float32") return np.concatenate(frames, axis=0).flatten() - - -class Transcriber: - """Transcrit l'audio en français avec MLX-Whisper. - - Note : mlx_whisper.transcribe charge/met en cache le modèle au premier appel. - Le premier lancement télécharge le modèle depuis Hugging Face. - """ - - def __init__(self, model, language): - self.model = model - self.language = language - - def warmup(self): - """Force le téléchargement/chargement du modèle au démarrage (1 fois).""" - silence = np.zeros(16000, dtype="float32") - self.transcribe(silence) - - def transcribe(self, audio): - """Transcrit l'audio (numpy float32) et renvoie le texte nettoyé.""" - result = mlx_whisper.transcribe( - audio, - path_or_hf_repo=self.model, - language=self.language, - initial_prompt=CONFIG["initial_prompt"], - ) - return clean_transcript(result.get("text", "")) diff --git a/controller.py b/controller.py index 8794c4d..02587aa 100644 --- a/controller.py +++ b/controller.py @@ -11,15 +11,15 @@ contexte de l'app). La transcription, elle, part sur un thread de fond. import threading from audio_level import compute_rms, shape_level +from capture import Recorder, inject_text 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): + def __init__(self, config, pulse_window, transcripteur, systeme, 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) @@ -34,7 +34,8 @@ class ZonzaController: 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.transcriber = transcripteur + self.systeme = systeme self.recording = False self._lock = threading.Lock() self._smoothed_level = 0.0 @@ -68,7 +69,7 @@ class ZonzaController: self.recording = True self.pulse.show() self.pulse.set_status("Je vous écoute…") - play_sound("start") + self.systeme.jouer_son("start") self.on_state("recording") self._arm_safety_timer() self._armer_veille_silence() @@ -88,7 +89,7 @@ class ZonzaController: 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.systeme.jouer_son("error") self.run_on_main(lambda: self.pulse.set_status("Aucun son — micro pris ailleurs ?")) def _ouvrir_flux(self): @@ -97,7 +98,7 @@ class ZonzaController: self.recorder.start() except Exception as e: print(f"[erreur] démarrage enregistrement : {e}") - play_sound("error") + self.systeme.jouer_son("error") self.recording = False self._cancel_safety_timer() self.run_on_main(self.pulse.hide) @@ -143,22 +144,22 @@ class ZonzaController: principal (2026-08-21).""" try: audio = self.recorder.stop() - play_sound("done") + self.systeme.jouer_son("done") if not is_long_enough(len(audio), self.config["sample_rate"], self.config["min_duration_s"]): return - text = self.transcriber.transcribe(audio) + text = self.transcriber.transcrire(audio) if not text: return print(f"[texte] {text}") try: - inject_text(text) + inject_text(text, self.systeme) except Exception as e: print(f"[erreur] collage échoué ({e}) — texte dans le presse-papier, fais Cmd+V") - play_sound("error") + self.systeme.jouer_son("error") except Exception as e: print(f"[erreur] arrêt ou transcription : {e}") - play_sound("error") + self.systeme.jouer_son("error") finally: self.run_on_main(self.pulse.hide) self.on_state("idle") diff --git a/tests/test_liberation_flux.py b/tests/test_liberation_flux.py index 9ded53c..4d9991b 100644 --- a/tests/test_liberation_flux.py +++ b/tests/test_liberation_flux.py @@ -11,7 +11,7 @@ draine les tampons et peut bloquer), et sans jamais lever. """ import numpy as np -from engine import Recorder +from capture import Recorder class FauxFlux: diff --git a/tests/test_statut_bulle.py b/tests/test_statut_bulle.py index e4f62ab..3b6321c 100644 --- a/tests/test_statut_bulle.py +++ b/tests/test_statut_bulle.py @@ -50,14 +50,38 @@ class FauxRecorder: def stop(self): return self.audio +class FauxTranscripteur: + def warmup(self): pass + def transcrire(self, audio): return "bonjour" + + +class FauxSysteme: + def __init__(self): self.sons = [] + def jouer_son(self, nom): self.sons.append(nom) + def modificateurs_enfonces(self): return False + def raccourci_defaut(self): return "++d" + def chemin_log(self): return "/tmp/zonza-test.log" + def dossier_donnees(self): return "/tmp/zonza-test" + def acquerir_verrou_instance(self, chemin): return object() + + def _ctrl(recorder, taches=None): pulse = FauxPulse() - c = ZonzaController(CONFIG, pulse, lancer_tache=taches) + c = ZonzaController(CONFIG, pulse, FauxTranscripteur(), FauxSysteme(), lancer_tache=taches) c.recorder = recorder - c.transcriber = type("T", (), {"transcribe": staticmethod(lambda a: "bonjour")})() return c, pulse +def test_le_controleur_n_importe_aucun_module_de_plateforme(): + """Le noyau ne doit connaître que des contrats. Si cette assertion tombe, une + dépendance de plateforme s'est glissée dans le chemin portable.""" + import controller + source = open(controller.__file__, encoding="utf-8").read() + for interdit in ("AppKit", "Foundation", "objc", "Quartz", "mlx_whisper", + "afplay", "from engine import"): + assert interdit not in source, f"{interdit} n'a rien à faire dans controller.py" + + def test_demarrer_ne_touche_pas_l_audio_sur_le_thread_principal(): differees = [] c, pulse = _ctrl(RecorderQuiBloque(), taches=differees.append)