Ralph Mayola
22080eb997
Reconstruction propre de l'ancien projet Dictée. Bundle id neuf (cloud.mrtechlab.zonza), chemin de build unique vers /Applications (plus jamais de builds /tmp enregistrés dans LaunchServices). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
113 lines
3.2 KiB
Python
113 lines
3.2 KiB
Python
"""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, 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
|
|
|
|
|
|
def inject_text(text):
|
|
"""Copie le texte dans le presse-papier puis colle avec Cmd+V dans l'app active."""
|
|
if not text:
|
|
return
|
|
pyperclip.copy(text)
|
|
time.sleep(0.05) # laisser le presse-papier se mettre à jour
|
|
with _kbd.pressed(Key.cmd):
|
|
_kbd.press("v")
|
|
_kbd.release("v")
|
|
|
|
|
|
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):
|
|
self.sample_rate = sample_rate
|
|
self.on_block = on_block
|
|
self._frames = []
|
|
self._stream = None
|
|
|
|
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 start(self):
|
|
self._frames = []
|
|
self._stream = sd.InputStream(
|
|
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)."""
|
|
if self._stream is not None:
|
|
self._stream.stop()
|
|
self._stream.close()
|
|
self._stream = None
|
|
if not self._frames:
|
|
return np.zeros(0, dtype="float32")
|
|
return np.concatenate(self._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,
|
|
)
|
|
return clean_transcript(result.get("text", ""))
|