zonza/engine.py

"""Moteur de dictée — capture, transcription, collage, son.

Aucune dépendance UI : importable par le contrôleur et l'app.
"""
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",
}

_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):
    """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).
    """
    if not text:
        return
    pyperclip.copy(text)
    libre = attendre(
        lambda: not modificateurs_enfonces(),
        CONFIG["attente_modificateurs_s"],
        0.03,
    )
    if not libre:
        print("[collage] modificateurs encore enfoncés — collage tenté quand même")
    time.sleep(CONFIG["delai_presse_papier_s"])  # laisser le presse-papier se mettre à jour
    with _kbd.pressed(Key.cmd):
        _kbd.press("v")
        _kbd.release("v")
    print("[collage] ok")


class Recorder:
    """Capture le micro en mémoire (16 kHz mono float32). Start/stop par le contrôleur.

    on_block : callback optionnel appelé à chaque bloc audio avec le tableau numpy du bloc
    (utilisé pour alimenter l'animation en temps réel).
    """

    def __init__(self, sample_rate, on_block=None, ouvrir_flux=None):
        self.sample_rate = sample_rate
        self.on_block = on_block
        self._frames = []
        self._stream = None
        # ouvrir_flux : injectable pour les tests (aucun peripherique requis).
        self._ouvrir_flux = ouvrir_flux or (lambda **kw: sd.InputStream(**kw))

    def _callback(self, indata, frames, time_info, status):
        block = indata.copy()
        self._frames.append(block)
        if self.on_block is not None:
            self.on_block(block.flatten())

    def _liberer(self):
        """Libère le flux courant. Ne lève jamais, ne bloque pas durablement.

        On DÉTACHE la référence avant d'agir : même si la libération échoue, le
        prochain démarrage ne doit pas retomber sur un flux mort. Et on utilise
        `abort()` plutôt que `stop()` — ce dernier draine les tampons et peut se
        bloquer sur un périphérique accaparé, ce qui laissait le micro tenu pour
        toujours (2026-08-22 : trois dictées muettes d'affilée).
        """
        flux, self._stream = self._stream, None
        if flux is None:
            return
        for action in ("abort", "close"):
            try:
                getattr(flux, action)()
            except Exception as exc:
                print(f"[audio] {action}() du flux précédent a échoué : {exc}")

    def start(self):
        # un flux orphelin d'un essai precedent tiendrait le micro : on libere d'abord
        self._liberer()
        self._frames = []
        self._stream = self._ouvrir_flux(
            samplerate=self.sample_rate,
            channels=1,
            dtype="float32",
            callback=self._callback,
        )
        self._stream.start()

    def stop(self):
        """Arrête le flux et renvoie l'audio concaténé (numpy 1-D float32)."""
        frames, self._frames = self._frames, []
        self._liberer()
        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", ""))