Compare commits
8 Commits
332afede0e
...
7661cbce36
| Author | SHA1 | Date | |
|---|---|---|---|
| 7661cbce36 | |||
| dcec1e61d0 | |||
| 3c3ea4444d | |||
| 5f83e4c6a3 | |||
| 734bba31da | |||
| 5c4477fe1d | |||
| 434944def6 | |||
| 54479d9af9 |
30
app.py
30
app.py
@ -12,6 +12,7 @@ native est de toute façon minuscule pour nos besoins : une icône, deux items d
|
||||
Zonza vit dans la barre de menu, jamais dans le Dock : c'est le LSUIElement de
|
||||
l'Info.plist (posé par build_app.sh) qui s'en charge.
|
||||
"""
|
||||
import os
|
||||
import threading
|
||||
|
||||
from objc import super
|
||||
@ -28,8 +29,10 @@ from PyObjCTools import AppHelper
|
||||
from pynput import keyboard
|
||||
|
||||
from controller import ZonzaController
|
||||
from core import CONFIG, acquire_single_instance_lock, validate_config
|
||||
from pulse_window import PulseWindow
|
||||
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 = {
|
||||
"idle": "🎙️",
|
||||
@ -41,15 +44,15 @@ _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
|
||||
self._item = None
|
||||
self._sounds_item = None
|
||||
self.pulse = PulseWindow.alloc().init()
|
||||
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
|
||||
@ -120,7 +123,7 @@ class ZonzaApp(NSObject):
|
||||
def _warmup_async(self):
|
||||
def run():
|
||||
try:
|
||||
self.controller.transcriber.warmup()
|
||||
self.controller.transcripteur.warmup()
|
||||
except Exception as e:
|
||||
print(f"[erreur] warmup modèle : {e}")
|
||||
threading.Thread(target=run, daemon=True).start()
|
||||
@ -128,13 +131,22 @@ class ZonzaApp(NSObject):
|
||||
|
||||
def main():
|
||||
# une seule instance a la fois : sinon deux modeles en RAM et la meme
|
||||
# hotkey captee deux fois (cf acquire_single_instance_lock).
|
||||
lock = acquire_single_instance_lock()
|
||||
# hotkey captee deux fois (cf systeme.acquerir_verrou_instance).
|
||||
systeme = choisir_systeme()
|
||||
lock = systeme.acquerir_verrou_instance(
|
||||
os.path.join(systeme.dossier_donnees(), "instance.lock")
|
||||
)
|
||||
if lock is None:
|
||||
print("[garde] une autre instance de Zonza tourne deja — arret.")
|
||||
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.
|
||||
@ -145,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
|
||||
|
||||
@ -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", ""))
|
||||
65
contrats.py
Normal file
65
contrats.py
Normal file
@ -0,0 +1,65 @@
|
||||
"""Les trois frontières de Zonza.
|
||||
|
||||
Chaque contrat est délibérément minuscule : ce sont les seuls points par lesquels
|
||||
le noyau parle au monde extérieur. Tout ce qui dépend d'un système d'exploitation,
|
||||
d'une bibliothèque d'affichage ou d'un moteur de transcription vit DERRIÈRE l'un
|
||||
d'eux — jamais dans `core.py`, `controller.py` ni `audio_level.py`.
|
||||
"""
|
||||
from typing import IO, Protocol
|
||||
|
||||
# numpy entre dans ce fichier volontairement minuscule car c'est déjà une
|
||||
# dépendance dure du projet, et le type du signal audio fait partie du
|
||||
# contrat : un `Any` implicite laisserait passer un mauvais type sans que
|
||||
# rien ne bronche.
|
||||
import numpy as np
|
||||
|
||||
|
||||
class Transcripteur(Protocol):
|
||||
"""Transforme un signal audio en texte."""
|
||||
|
||||
def warmup(self) -> None:
|
||||
"""Charge le modèle une fois, pour que la première dictée ne traîne pas."""
|
||||
|
||||
def transcrire(self, audio: np.ndarray) -> str:
|
||||
"""Rend le texte d'un tableau numpy float32 mono."""
|
||||
|
||||
|
||||
class Overlay(Protocol):
|
||||
"""La fenêtre flottante affichée pendant la dictée."""
|
||||
|
||||
def show(self) -> None:
|
||||
"""Affiche la fenêtre et démarre l'animation. Thread principal."""
|
||||
|
||||
def hide(self) -> None:
|
||||
"""Masque la fenêtre et arrête l'animation. Thread principal."""
|
||||
|
||||
def set_level(self, level: float) -> None:
|
||||
"""Pousse un niveau audio dans [0, 1]. Appelable depuis un thread de fond."""
|
||||
|
||||
def set_status(self, text: str) -> None:
|
||||
"""Texte affiché dans la bulle. Appelable depuis un thread de fond."""
|
||||
|
||||
def rearmer_garde(self, duree_s: float) -> None:
|
||||
"""Remplace la durée de vie maximale de la fenêtre. Thread principal."""
|
||||
|
||||
|
||||
class Systeme(Protocol):
|
||||
"""Les appels qui diffèrent d'un système d'exploitation à l'autre."""
|
||||
|
||||
def jouer_son(self, nom: str) -> None:
|
||||
"""Joue « start », « done » ou « error » sans bloquer."""
|
||||
|
||||
def modificateurs_enfonces(self) -> bool:
|
||||
"""True si une touche modificatrice est physiquement enfoncée."""
|
||||
|
||||
def raccourci_defaut(self) -> str:
|
||||
"""Raccourci global au format pynput, propre à la plateforme."""
|
||||
|
||||
def chemin_log(self) -> str:
|
||||
"""Chemin absolu du journal."""
|
||||
|
||||
def dossier_donnees(self) -> str:
|
||||
"""Dossier des données de l'application (verrou d'instance, réglages)."""
|
||||
|
||||
def acquerir_verrou_instance(self, chemin: str) -> IO | None:
|
||||
"""Verrou exclusif non bloquant ; rend le fichier, ou None s'il est pris."""
|
||||
@ -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.transcripteur = transcripteur
|
||||
self.systeme = systeme
|
||||
self.recording = False
|
||||
self._lock = threading.Lock()
|
||||
self._smoothed_level = 0.0
|
||||
@ -42,6 +43,16 @@ class ZonzaController:
|
||||
self._blocs_recus = 0 # compteur pour détecter un micro muet
|
||||
self._veille_silence = None
|
||||
|
||||
def _jouer_son(self, nom):
|
||||
"""Joue un son, sauf si l'utilisateur les a coupés par le menu.
|
||||
|
||||
La décision vit ICI et non dans la couche système : celle-ci doit rester
|
||||
bête, sinon chaque nouvelle plateforme devra repenser à ce garde-fou.
|
||||
"""
|
||||
if not self.config["sounds"]:
|
||||
return
|
||||
self.systeme.jouer_son(nom)
|
||||
|
||||
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
|
||||
@ -68,7 +79,7 @@ class ZonzaController:
|
||||
self.recording = True
|
||||
self.pulse.show()
|
||||
self.pulse.set_status("Je vous écoute…")
|
||||
play_sound("start")
|
||||
self._jouer_son("start")
|
||||
self.on_state("recording")
|
||||
self._arm_safety_timer()
|
||||
self._armer_veille_silence()
|
||||
@ -88,7 +99,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._jouer_son("error")
|
||||
self.run_on_main(lambda: self.pulse.set_status("Aucun son — micro pris ailleurs ?"))
|
||||
|
||||
def _ouvrir_flux(self):
|
||||
@ -97,7 +108,7 @@ class ZonzaController:
|
||||
self.recorder.start()
|
||||
except Exception as e:
|
||||
print(f"[erreur] démarrage enregistrement : {e}")
|
||||
play_sound("error")
|
||||
self._jouer_son("error")
|
||||
self.recording = False
|
||||
self._cancel_safety_timer()
|
||||
self.run_on_main(self.pulse.hide)
|
||||
@ -143,22 +154,22 @@ class ZonzaController:
|
||||
principal (2026-08-21)."""
|
||||
try:
|
||||
audio = self.recorder.stop()
|
||||
play_sound("done")
|
||||
self._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.transcripteur.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._jouer_son("error")
|
||||
except Exception as e:
|
||||
print(f"[erreur] arrêt ou transcription : {e}")
|
||||
play_sound("error")
|
||||
self._jouer_son("error")
|
||||
finally:
|
||||
self.run_on_main(self.pulse.hide)
|
||||
self.on_state("idle")
|
||||
|
||||
31
core.py
31
core.py
@ -3,8 +3,6 @@
|
||||
Tout ce qui est ici est testable sans micro, sans clavier, sans modèle.
|
||||
"""
|
||||
|
||||
import fcntl
|
||||
import os
|
||||
import re
|
||||
|
||||
CONFIG = {
|
||||
@ -196,33 +194,4 @@ def clean_transcript(text):
|
||||
return re.sub(r"\s+", " ", text).strip()
|
||||
|
||||
|
||||
LOCK_PATH = os.path.expanduser("~/Library/Application Support/Zonza/instance.lock")
|
||||
|
||||
|
||||
def acquire_single_instance_lock(path=LOCK_PATH):
|
||||
"""Prend un verrou exclusif non bloquant ; renvoie le fichier, ou None si une
|
||||
autre instance le detient deja.
|
||||
|
||||
Pourquoi : deux bundles Zonza de meme identifiant (dont un fantome reste dans
|
||||
la Corbeille mais toujours enregistre aupres de LaunchServices) pouvaient
|
||||
tourner en meme temps. Chacun chargeait son propre modele MLX-Whisper et
|
||||
captait la MEME hotkey globale : un appui, deux transcriptions, machine
|
||||
saturee. Purger le fantome corrige l'incident ; ce verrou corrige la classe.
|
||||
|
||||
Le handle doit rester vivant tant que l'app tourne : le verrou tombe a la
|
||||
fermeture du fichier, et le noyau le libere si le processus meurt brutalement
|
||||
(donc pas de verrou orphelin apres un plantage).
|
||||
"""
|
||||
parent = os.path.dirname(path)
|
||||
if parent:
|
||||
os.makedirs(parent, exist_ok=True)
|
||||
handle = open(path, "a")
|
||||
try:
|
||||
fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except OSError:
|
||||
handle.close()
|
||||
return None
|
||||
return handle
|
||||
|
||||
|
||||
CONFIG["initial_prompt"] = build_initial_prompt()
|
||||
|
||||
11
interface/__init__.py
Normal file
11
interface/__init__.py
Normal file
@ -0,0 +1,11 @@
|
||||
"""Choisit l'implémentation d'overlay selon la plateforme."""
|
||||
import sys
|
||||
|
||||
|
||||
def classe_overlay_pour(plateforme=None):
|
||||
"""Rend la CLASSE d'overlay (pas une instance : Cocoa exige `alloc().init()`)."""
|
||||
plateforme = plateforme or sys.platform
|
||||
if plateforme.startswith("darwin"):
|
||||
from interface.cocoa.overlay import PulseWindow
|
||||
return PulseWindow
|
||||
raise NotImplementedError(f"Aucune interface pour : {plateforme}")
|
||||
0
interface/cocoa/__init__.py
Normal file
0
interface/cocoa/__init__.py
Normal file
62
moteur/__init__.py
Normal file
62
moteur/__init__.py
Normal file
@ -0,0 +1,62 @@
|
||||
"""Choisit le moteur de transcription d'après le matériel."""
|
||||
import ctypes
|
||||
import platform
|
||||
import sys
|
||||
|
||||
|
||||
def nom_moteur_pour(plateforme, machine, cuda_disponible):
|
||||
"""Décide du moteur. Fonction PURE : aucune interrogation de la machine ici,
|
||||
ce qui rend les quatre configurations testables depuis un seul poste."""
|
||||
if plateforme.startswith("darwin") and machine in ("arm64", "aarch64"):
|
||||
return "mlx"
|
||||
return "faster-gpu" if cuda_disponible else "faster-cpu"
|
||||
|
||||
|
||||
def detecter_moteur():
|
||||
"""Interroge la machine, puis délègue la décision à `nom_moteur_pour`."""
|
||||
return nom_moteur_pour(sys.platform, platform.machine(), _cuda_disponible())
|
||||
|
||||
|
||||
# Le nom de la bibliothèque partagée CUDA diffère radicalement par plateforme ET
|
||||
# par version : `libcudart.so` (parfois suffixé `.12`, `.11.0`...) sous Linux/macOS,
|
||||
# mais `cudart64_12.dll`, `cudart64_110.dll`... sous Windows, où le numéro de version
|
||||
# est ENCODÉ DANS LE NOM du fichier. `ctypes.util.find_library` ne résout pas ces DLL
|
||||
# versionnées de façon fiable — il cherche un nom canonique qui n'existe pas pour
|
||||
# CUDA sous Windows. On essaie donc une liste explicite de noms candidats, du plus
|
||||
# récent au plus ancien, et on s'arrête au premier qui charge.
|
||||
_NOMS_CUDART_WINDOWS = [
|
||||
"cudart64_12.dll",
|
||||
"cudart64_110.dll",
|
||||
"cudart64_101.dll",
|
||||
"cudart64_100.dll",
|
||||
]
|
||||
_NOMS_CUDART_UNIX = [
|
||||
"libcudart.so",
|
||||
"libcudart.so.12",
|
||||
"libcudart.so.11.0",
|
||||
]
|
||||
|
||||
|
||||
def _cuda_disponible(plateforme=None, charger=None):
|
||||
"""Sonde la présence du runtime CUDA en tentant de charger sa bibliothèque
|
||||
partagée. `plateforme` et `charger` sont injectables : ça rend la sonde
|
||||
testable avec une doublure, sans dépendre d'une vraie machine CUDA ni d'un
|
||||
vrai Windows. Ne lève jamais — dans le doute, on rend False."""
|
||||
plateforme = plateforme if plateforme is not None else sys.platform
|
||||
charger = charger if charger is not None else ctypes.CDLL
|
||||
noms = _NOMS_CUDART_WINDOWS if plateforme.startswith("win") else _NOMS_CUDART_UNIX
|
||||
for nom in noms:
|
||||
try:
|
||||
charger(nom)
|
||||
return True
|
||||
except Exception:
|
||||
continue
|
||||
return False
|
||||
|
||||
|
||||
def choisir_transcripteur(nom, modele, langue, amorce):
|
||||
"""Rend l'implémentation du contrat Transcripteur portant ce nom."""
|
||||
if nom == "mlx":
|
||||
from moteur.mlx import TranscripteurMLX
|
||||
return TranscripteurMLX(modele, langue, amorce)
|
||||
raise NotImplementedError(f"Moteur non pris en charge à cette étape : {nom}")
|
||||
27
moteur/mlx.py
Normal file
27
moteur/mlx.py
Normal file
@ -0,0 +1,27 @@
|
||||
"""Transcription par MLX-Whisper — GPU Apple Silicon uniquement."""
|
||||
import numpy as np
|
||||
|
||||
from core import clean_transcript
|
||||
|
||||
|
||||
class TranscripteurMLX:
|
||||
"""Implémentation du contrat Transcripteur pour Apple Silicon."""
|
||||
|
||||
def __init__(self, modele, langue, amorce):
|
||||
self.modele = modele
|
||||
self.langue = langue
|
||||
self.amorce = amorce
|
||||
|
||||
def warmup(self):
|
||||
"""Charge le modèle une fois, sur une seconde de silence."""
|
||||
self.transcrire(np.zeros(16000, dtype="float32"))
|
||||
|
||||
def transcrire(self, audio):
|
||||
import mlx_whisper
|
||||
resultat = mlx_whisper.transcribe(
|
||||
audio,
|
||||
path_or_hf_repo=self.modele,
|
||||
language=self.langue,
|
||||
initial_prompt=self.amorce,
|
||||
)
|
||||
return clean_transcript(resultat.get("text", ""))
|
||||
15
systeme/__init__.py
Normal file
15
systeme/__init__.py
Normal file
@ -0,0 +1,15 @@
|
||||
"""Choisit la couche système selon la plateforme."""
|
||||
import sys
|
||||
|
||||
|
||||
def choisir_systeme(plateforme=None):
|
||||
"""Rend l'implémentation du contrat Systeme pour cette machine.
|
||||
|
||||
`plateforme` est injectable pour tester les branches absentes de la machine
|
||||
de développement.
|
||||
"""
|
||||
plateforme = plateforme or sys.platform
|
||||
if plateforme.startswith("darwin"):
|
||||
from systeme.macos import SystemeMacOS
|
||||
return SystemeMacOS()
|
||||
raise NotImplementedError(f"Plateforme non prise en charge : {plateforme}")
|
||||
54
systeme/macos.py
Normal file
54
systeme/macos.py
Normal file
@ -0,0 +1,54 @@
|
||||
"""Appels propres à macOS. Rien ici ne doit être importé directement par le noyau."""
|
||||
import fcntl
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
_SONS = {
|
||||
"start": "/System/Library/Sounds/Pop.aiff",
|
||||
"done": "/System/Library/Sounds/Glass.aiff",
|
||||
"error": "/System/Library/Sounds/Basso.aiff",
|
||||
}
|
||||
|
||||
# Masque des modificateurs (ctrl, shift, alt, cmd) dans les drapeaux Quartz.
|
||||
_MASQUE_MODIFICATEURS = 0x000F0000
|
||||
|
||||
|
||||
class SystemeMacOS:
|
||||
"""Implémentation du contrat Systeme pour macOS."""
|
||||
|
||||
def jouer_son(self, nom):
|
||||
chemin = _SONS.get(nom)
|
||||
if not chemin:
|
||||
return
|
||||
try:
|
||||
subprocess.Popen(["afplay", chemin])
|
||||
except Exception:
|
||||
pass # le son ne doit jamais casser la dictée
|
||||
|
||||
def modificateurs_enfonces(self):
|
||||
try:
|
||||
from Quartz import CGEventSourceFlagsState
|
||||
return bool(CGEventSourceFlagsState(1) & _MASQUE_MODIFICATEURS)
|
||||
except Exception:
|
||||
return False # dans le doute, ne jamais bloquer le collage
|
||||
|
||||
def raccourci_defaut(self):
|
||||
return "<cmd>+<ctrl>+d"
|
||||
|
||||
def chemin_log(self):
|
||||
return os.path.expanduser("~/Library/Logs/Zonza.log")
|
||||
|
||||
def dossier_donnees(self):
|
||||
return os.path.expanduser("~/Library/Application Support/Zonza")
|
||||
|
||||
def acquerir_verrou_instance(self, chemin):
|
||||
parent = os.path.dirname(chemin)
|
||||
if parent:
|
||||
os.makedirs(parent, exist_ok=True)
|
||||
fichier = open(chemin, "a")
|
||||
try:
|
||||
fcntl.flock(fichier, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except OSError:
|
||||
fichier.close()
|
||||
return None
|
||||
return fichier
|
||||
35
tests/test_contrats.py
Normal file
35
tests/test_contrats.py
Normal file
@ -0,0 +1,35 @@
|
||||
"""Une suite unique, rejouée contre TOUTES les implémentations d'un contrat.
|
||||
|
||||
Le jour où une implémentation oublie une méthode, c'est ici que ça rougit — y
|
||||
compris pour les plateformes absentes de la machine de développement.
|
||||
"""
|
||||
import inspect
|
||||
|
||||
import pytest
|
||||
|
||||
from contrats import Overlay, Systeme, Transcripteur
|
||||
|
||||
|
||||
@pytest.mark.parametrize("contrat, methodes", [
|
||||
(Transcripteur, ["warmup", "transcrire"]),
|
||||
(Overlay, ["show", "hide", "set_level", "set_status", "rearmer_garde"]),
|
||||
(Systeme, ["jouer_son", "modificateurs_enfonces", "raccourci_defaut",
|
||||
"chemin_log", "dossier_donnees", "acquerir_verrou_instance"]),
|
||||
])
|
||||
def test_le_contrat_declare_exactement_ses_methodes(contrat, methodes):
|
||||
declarees = [n for n, _ in inspect.getmembers(contrat, inspect.isfunction)
|
||||
if not n.startswith("_")]
|
||||
assert sorted(declarees) == sorted(methodes)
|
||||
|
||||
# Ce sont les docstrings de MÉTHODE qui portent l'essentiel du contrat
|
||||
# (la plage [0, 1] de set_level, l'affinité thread, la sémantique de
|
||||
# retour) : une docstring vidée ici doit rougir, pas seulement celle
|
||||
# de la classe.
|
||||
for nom in methodes:
|
||||
methode = getattr(contrat, nom)
|
||||
assert methode.__doc__, f"{contrat.__name__}.{nom} doit dire à quoi il sert"
|
||||
|
||||
|
||||
def test_les_contrats_sont_documentes():
|
||||
for contrat in (Transcripteur, Overlay, Systeme):
|
||||
assert contrat.__doc__, f"{contrat.__name__} doit dire à quoi il sert"
|
||||
16
tests/test_interface.py
Normal file
16
tests/test_interface.py
Normal file
@ -0,0 +1,16 @@
|
||||
"""La fabrique rend un overlay conforme au contrat, sans ouvrir de fenêtre."""
|
||||
import pytest
|
||||
|
||||
from interface import classe_overlay_pour
|
||||
|
||||
|
||||
def test_l_overlay_macos_declare_tout_le_contrat():
|
||||
classe = classe_overlay_pour("darwin")
|
||||
for methode in ("show", "hide", "set_level", "set_status", "rearmer_garde"):
|
||||
assert callable(getattr(classe, methode, None)), f"{methode} manque"
|
||||
|
||||
|
||||
def test_une_plateforme_inconnue_est_refusee_clairement():
|
||||
with pytest.raises(NotImplementedError) as e:
|
||||
classe_overlay_pour("haiku-os")
|
||||
assert "haiku-os" in str(e.value)
|
||||
@ -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:
|
||||
|
||||
66
tests/test_moteur.py
Normal file
66
tests/test_moteur.py
Normal file
@ -0,0 +1,66 @@
|
||||
"""La fabrique choisit le moteur d'après le matériel — décision pure, testable
|
||||
depuis n'importe quelle machine."""
|
||||
import pytest
|
||||
|
||||
from moteur import choisir_transcripteur, nom_moteur_pour
|
||||
|
||||
|
||||
@pytest.mark.parametrize("plateforme, machine, cuda, attendu", [
|
||||
("darwin", "arm64", False, "mlx"),
|
||||
("darwin", "x86_64", False, "faster-cpu"),
|
||||
("win32", "AMD64", True, "faster-gpu"),
|
||||
("win32", "AMD64", False, "faster-cpu"),
|
||||
])
|
||||
def test_la_decision_suit_le_materiel(plateforme, machine, cuda, attendu):
|
||||
assert nom_moteur_pour(plateforme, machine, cuda) == attendu
|
||||
|
||||
|
||||
def test_apple_silicon_prend_mlx_meme_sans_cuda():
|
||||
assert nom_moteur_pour("darwin", "arm64", cuda_disponible=False) == "mlx"
|
||||
|
||||
|
||||
def test_la_fabrique_rend_un_transcripteur_complet():
|
||||
t = choisir_transcripteur(nom="mlx", modele="mlx-community/whisper-medium",
|
||||
langue="fr", amorce="")
|
||||
assert callable(t.warmup) and callable(t.transcrire)
|
||||
|
||||
|
||||
def test_un_moteur_inconnu_est_refuse_clairement():
|
||||
with pytest.raises(NotImplementedError) as e:
|
||||
choisir_transcripteur(nom="vosk", modele="x", langue="fr", amorce="")
|
||||
assert "vosk" in str(e.value)
|
||||
|
||||
|
||||
# --- Sonde CUDA : le nom de la bibliothèque partagée diffère par OS et par version.
|
||||
# Ces tests injectent une doublure de chargeur pour rester exécutables depuis
|
||||
# n'importe quelle machine, sans jamais dépendre d'une vraie installation CUDA.
|
||||
from moteur import _cuda_disponible
|
||||
|
||||
|
||||
def test_cuda_reconnue_sur_windows_via_dll_versionnee():
|
||||
def charger(nom):
|
||||
if nom == "cudart64_12.dll":
|
||||
return object()
|
||||
raise OSError(f"introuvable : {nom}")
|
||||
assert _cuda_disponible(plateforme="win32", charger=charger) is True
|
||||
|
||||
|
||||
def test_cuda_absente_sur_windows_si_aucune_dll_ne_charge():
|
||||
def charger(nom):
|
||||
raise OSError(f"introuvable : {nom}")
|
||||
assert _cuda_disponible(plateforme="win32", charger=charger) is False
|
||||
|
||||
|
||||
def test_cuda_reconnue_sur_linux_darwin_via_so():
|
||||
def charger(nom):
|
||||
if nom == "libcudart.so":
|
||||
return object()
|
||||
raise OSError(f"introuvable : {nom}")
|
||||
assert _cuda_disponible(plateforme="linux", charger=charger) is True
|
||||
assert _cuda_disponible(plateforme="darwin", charger=charger) is True
|
||||
|
||||
|
||||
def test_cuda_sonde_ne_leve_jamais_meme_si_le_chargeur_explose():
|
||||
def charger(nom):
|
||||
raise RuntimeError("chargeur casse, pas seulement OSError")
|
||||
assert _cuda_disponible(plateforme="win32", charger=charger) is False
|
||||
@ -13,7 +13,7 @@ bug — quel que soit le nombre de bundles présents, la 2e instance renonce.
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
from core import acquire_single_instance_lock
|
||||
from systeme import choisir_systeme
|
||||
|
||||
|
||||
def _tmp_lock():
|
||||
@ -21,31 +21,31 @@ def _tmp_lock():
|
||||
|
||||
|
||||
def test_premiere_instance_obtient_le_verrou():
|
||||
handle = acquire_single_instance_lock(_tmp_lock())
|
||||
handle = choisir_systeme().acquerir_verrou_instance(_tmp_lock())
|
||||
assert handle is not None
|
||||
|
||||
|
||||
def test_deuxieme_instance_est_refusee():
|
||||
path = _tmp_lock()
|
||||
premier = acquire_single_instance_lock(path)
|
||||
premier = choisir_systeme().acquerir_verrou_instance(path)
|
||||
assert premier is not None
|
||||
assert acquire_single_instance_lock(path) is None
|
||||
assert choisir_systeme().acquerir_verrou_instance(path) is None
|
||||
|
||||
|
||||
def test_verrou_repris_apres_liberation():
|
||||
path = _tmp_lock()
|
||||
premier = acquire_single_instance_lock(path)
|
||||
premier = choisir_systeme().acquerir_verrou_instance(path)
|
||||
premier.close()
|
||||
assert acquire_single_instance_lock(path) is not None
|
||||
assert choisir_systeme().acquerir_verrou_instance(path) is not None
|
||||
|
||||
|
||||
def test_cree_le_dossier_parent_manquant():
|
||||
path = _tmp_lock()
|
||||
assert not os.path.isdir(os.path.dirname(path))
|
||||
acquire_single_instance_lock(path)
|
||||
choisir_systeme().acquerir_verrou_instance(path)
|
||||
assert os.path.isdir(os.path.dirname(path))
|
||||
|
||||
|
||||
def test_deux_chemins_differents_ne_se_genent_pas():
|
||||
assert acquire_single_instance_lock(_tmp_lock()) is not None
|
||||
assert acquire_single_instance_lock(_tmp_lock()) is not None
|
||||
assert choisir_systeme().acquerir_verrou_instance(_tmp_lock()) is not None
|
||||
assert choisir_systeme().acquerir_verrou_instance(_tmp_lock()) is not None
|
||||
|
||||
@ -50,14 +50,39 @@ class FauxRecorder:
|
||||
def stop(self): return self.audio
|
||||
|
||||
|
||||
def _ctrl(recorder, taches=None):
|
||||
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 "<cmd>+<ctrl>+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, config=None):
|
||||
pulse = FauxPulse()
|
||||
c = ZonzaController(CONFIG, pulse, lancer_tache=taches)
|
||||
systeme = FauxSysteme()
|
||||
c = ZonzaController(config or CONFIG, pulse, FauxTranscripteur(), systeme, 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)
|
||||
@ -101,3 +126,24 @@ def test_un_enregistrement_trop_court_referme_aussi():
|
||||
for t in differees:
|
||||
t()
|
||||
assert not pulse.visible
|
||||
|
||||
|
||||
def test_le_menu_sons_joue_un_son_quand_actives():
|
||||
differees = []
|
||||
c, pulse = _ctrl(FauxRecorder(np.ones(16000, dtype="float32") * 0.1), taches=differees.append)
|
||||
c.toggle(); c.toggle()
|
||||
for t in differees:
|
||||
t()
|
||||
assert c.systeme.sons, "un cycle demarrage/arret doit produire au moins un son"
|
||||
|
||||
|
||||
def test_le_menu_sons_coupe_le_son_quand_desactive():
|
||||
differees = []
|
||||
config = dict(CONFIG, sounds=False)
|
||||
c, pulse = _ctrl(
|
||||
FauxRecorder(np.ones(16000, dtype="float32") * 0.1), taches=differees.append, config=config
|
||||
)
|
||||
c.toggle(); c.toggle()
|
||||
for t in differees:
|
||||
t()
|
||||
assert c.systeme.sons == [], "sounds=False doit couper tous les sons"
|
||||
|
||||
39
tests/test_systeme.py
Normal file
39
tests/test_systeme.py
Normal file
@ -0,0 +1,39 @@
|
||||
"""La fabrique rend une implémentation conforme au contrat Systeme."""
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
from systeme import choisir_systeme
|
||||
|
||||
|
||||
def test_la_fabrique_rend_une_implementation_complete():
|
||||
s = choisir_systeme()
|
||||
for methode in ("jouer_son", "modificateurs_enfonces", "raccourci_defaut",
|
||||
"chemin_log", "dossier_donnees", "acquerir_verrou_instance"):
|
||||
assert callable(getattr(s, methode)), f"{methode} manque"
|
||||
|
||||
|
||||
def test_une_plateforme_inconnue_est_refusee_clairement():
|
||||
with pytest.raises(NotImplementedError) as e:
|
||||
choisir_systeme(plateforme="haiku-os")
|
||||
assert "haiku-os" in str(e.value)
|
||||
|
||||
|
||||
def test_le_raccourci_macos_est_celui_d_aujourd_hui():
|
||||
assert choisir_systeme(plateforme="darwin").raccourci_defaut() == "<cmd>+<ctrl>+d"
|
||||
|
||||
|
||||
def test_le_verrou_refuse_une_seconde_prise():
|
||||
s = choisir_systeme()
|
||||
chemin = os.path.join(tempfile.mkdtemp(), "sous-dossier", "z.lock")
|
||||
# Référence gardée : sinon CPython ramasse l'objet fichier dès la fin de
|
||||
# l'assert (refcount à 0), ce qui referme le fichier et relâche le verrou
|
||||
# AVANT la seconde prise — le test passerait pour la mauvaise raison.
|
||||
verrou = s.acquerir_verrou_instance(chemin)
|
||||
assert verrou is not None
|
||||
assert s.acquerir_verrou_instance(chemin) is None
|
||||
|
||||
|
||||
def test_aucun_modificateur_enfonce_pendant_les_tests():
|
||||
assert choisir_systeme().modificateurs_enfonces() is False
|
||||
Loading…
Reference in New Issue
Block a user