Compare commits

...
17 Commits
Author SHA1 Message Date
rmayolaandClaude Opus 5.5 2a1cd6373a Vocabulaire : Higgsfield
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-25 08:57:21 +02:00
rmayolaandClaude Opus 5.5 8d316a3e0b OpenRouter : gemini-3.8-flash, delai reseau porte a 15 s
3.5-flash-lite ajoutait une phrase non dite ; 3.1-flash-lite ecrivait
GitHub pour Gitea. 3.8-flash est juste a chaque essai (4-9 s).
3.8-flash-lite-tts est refuse par OpenRouter : modele texte -> voix.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-25 08:47:01 +02:00
rmayolaandClaude Opus 5.5 50b12795d3 Moteur OpenRouter (google/gemini-3.5-flash-lite), par defaut
Meme consigne, meme encodage, meme repli local que le moteur Gemini direct ;
seuls l'adresse, l'authentification (Bearer) et le format de requete
(input_audio, compatible OpenAI) changent. Cle : trousseau zonza-openrouter,
puis OPENROUTER_API_KEY. Un 200 portant un objet error est traite comme un
echec, pas comme une transcription vide.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-25 08:35:11 +02:00
rmayolaandClaude Opus 5.5 9be2ea030b Moteur Gemini (API) avec repli local automatique
L'audio part chez Google via gemini-3.1-flash-lite (la variante -tts ne
transcrit pas : elle produit de la voix). Cle lue dans le trousseau
(service zonza-gemini) puis GEMINI_API_KEY. Au moindre echec (cle, reseau,
quota, delai de 12 s), la dictee repasse par le moteur local.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-25 00:46:09 +02:00
rmayolaandClaude Opus 5 5eedda0467 install.sh refuse de detourner une Zonza deja installee
build_app.sh grave le chemin du dossier courant dans le binaire de
/Applications/Zonza.app. Lancer install.sh depuis une COPIE detournait donc
l'application en service vers cette copie, changeait son cdhash, et REVOQUAIT
silencieusement les autorisations macOS deja accordees.

Constate le 2026-08-23 en testant le script depuis un clone temporaire : rattrape
de justesse, avant que build_app.sh ne s'execute. Le script compare desormais la
cible gravee a son propre dossier, refuse de reconstruire s'ils different, et
explique pourquoi. ZONZA_REMPLACER=1 pour passer outre volontairement.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-23 20:09:44 +02:00
rmayolaandClaude Opus 5 901d8f695a Scripts d'installation, et requirements avec marqueurs de plateforme
requirements.txt installait mlx-whisper sans condition : l'installation aurait
echoue sur Windows, ou ce paquet n'existe pas. Et pyobjc n'y figurait PAS DU TOUT
alors que l'overlay macOS en depend — une installation neuve sur un autre Mac
aurait produit une Zonza qui ne demarre pas. Les marqueurs retiennent 13 paquets
sur Apple Silicon, 12 sur Mac Intel, 8 sur Windows.

install.sh construit l'app et lance le diagnostic. install.ps1 pose le lancement
au demarrage via pythonw (sans quoi une console resterait ouverte) et lance le
diagnostic. Ce dernier porte un avertissement : il n'a jamais tourne sur Windows.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-23 20:08:20 +02:00
rmayola 96c1379621 Lanceur Qt, assemblage partage, et masque des modificateurs corrige 2026-08-23 20:05:07 +02:00
rmayolaandClaude Opus 5 bfafc6a48c Le masque des modificateurs couvre enfin Command, et jamais Caps Lock
Deux defauts trouves en rendant la sonde injectable.

1. Le masque etait ecrit 0x000F0000 A LA MAIN. Il couvrait Shift, Control et Alt
   mais PAS Command (0x00100000) — pourtant la touche du raccourci de Zonza.
   Maintenir Cmd seul passait donc inapercu. Le masque est desormais COMPOSE
   depuis les constantes nommees de Quartz : un masque calcule de tete se trompe
   en silence, compose il ne peut plus.

2. Le test lisait le CLAVIER PHYSIQUE (`modificateurs_enfonces() is False`). Son
   resultat dependait des mains de l'utilisateur pendant la suite — un echec
   inreproductible a ete observe ce jour, puis 17 executions vertes. La sonde est
   rendue injectable, comme SystemeWindows l'avait deja, et les quatre
   modificateurs sont eprouves un par un. Un test verifie que la sonde reelle rend
   un booleen sans lever, SANS affirmer sa valeur.

Le repli hors Quartz excluait aussi le verrouillage majuscule : actif, il aurait
fait attendre 2 s avant CHAQUE collage, en permanence.

195 tests.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-23 20:04:52 +02:00
rmayolaandClaude Opus 5 aa63e94239 Un lanceur Qt pour Windows, et un assemblage partage
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-23 20:00:03 +02:00
rmayolaandClaude Opus 5 3e269dfb5e Le diagnostic eprouve le micro au debit reel de l'application
16000 etait code en dur dans trois endroits de diagnostic.py alors que
CONFIG["sample_rate"] existe. La valeur coincide aujourd'hui, donc rien ne
cassait — mais si ce reglage changeait, le diagnostic aurait teste le micro et la
synthese a un debit different de celui du pipeline reel, et aurait cesse
SILENCIEUSEMENT d'etre representatif. Un diagnostic qui ne mesure pas ce qu'on
croit est pire qu'aucun diagnostic.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-23 19:32:08 +02:00
rmayola 205424f68a Ajoute la couche Windows et la commande de diagnostic 2026-08-23 19:26:12 +02:00
rmayolaandClaude Opus 5 99376a8e12 Le test exerce enfin l'annulation de la garde, dans les deux interfaces
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-23 19:16:41 +02:00
rmayolaandClaude Opus 5 82e0e16393 Ajoute l'overlay Qt, pour les plateformes sans Cocoa
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-23 19:05:25 +02:00
rmayolaandClaude Opus 5 d7680227db Spec : remplacer mes estimations par les mesures reelles
L'estimation annoncait 15 a 30 s pour medium sur CPU et l'ecartait d'avance. La
mesure dit 4,4 s — et ce modele est PLUS JUSTE que le moteur GPU en service.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-23 18:45:42 +02:00
rmayolaandClaude Opus 5 2ccb614649 A chaque moteur son modele, et un forcage manuel
app.py passait CONFIG["model"] — un depot MLX — quel que soit le moteur detecte.
Sur une machine sans Apple Silicon, faster-whisper aurait recu un depot qu'il ne
sait pas charger. Le defaut etait invisible ici, la detection rendant toujours
"mlx" sur ce Mac : aucun test ne pouvait le voir.

MODELES_PAR_MOTEUR associe chaque moteur a son depot. Les valeurs viennent de la
mesure du jour (trois echantillons, amorce active, CPU Apple Silicon) :

  faster-whisper base   0,52 s  10/13 termes techniques   142 Mo
  faster-whisper small  1,42 s  11/13                     464 Mo
  faster-whisper medium 4,40 s  13/13                    1460 Mo
  MLX medium (GPU)      0,75 s  11/13                     489 Mo

Le GPU vise donc la precision (medium, 13/13, il absorbe le cout) ; le CPU vise
la latence (small, 11/13 — AUTANT que le MLX utilise aujourd'hui — pour 1,4 s).
`medium` sur CPU serait plus juste, mais ces 4,4 s sont celles d'un CPU Apple
Silicon : sur un CPU Intel ou Windows, le meme modele peut demander trois a quatre
fois plus. La commande de diagnostic mesurera la machine reelle et pourra
promouvoir `medium`.

CONFIG["model"] passe a None et devient un FORCAGE manuel qui prime sur la table,
au lieu d'un reglage mort qu'on modifie sans effet.

115 tests. Comportement inchange sur ce Mac : moteur mlx, modele identique.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-23 18:45:14 +02:00
rmayolaandClaude Opus 5 6974a3406c Ajoute le moteur faster-whisper, pour les machines sans Apple Silicon
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-23 18:38:09 +02:00
rmayolaandClaude Opus 5 5cf8f448ed Un module plat par plateforme, pour les trois frontieres
interface/cocoa/overlay.py etait un sous-paquet quand systeme/macos.py et
moteur/mlx.py sont des modules plats. Rien n'expliquait la difference : la couche
Windows aurait eu a deviner lequel des deux motifs imiter, et aurait reproduit
l'incoherence dans un sens ou dans l'autre.

interface/cocoa.py est donc mis a plat, contenu inchange (diff vide), et la regle
est ecrite dans interface/__init__.py : un module plat par plateforme, sous-paquet
seulement le jour ou une plateforme exige plusieurs fichiers.

105 tests inchanges.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-23 18:33:35 +02:00
30 changed files with 2983 additions and 44 deletions
+43 -25
View File
@@ -12,7 +12,19 @@ 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 sys
# --diagnostic doit s'évaluer AVANT tout import Cocoa/pyobjc : sur une machine
# qui ne les a pas (aujourd'hui aucune Windows n'est disponible pour ce
# portage, mais c'est la raison d'être de cette commande), c'est le seul
# moyen d'éprouver la chaîne. La sortie se fait donc ICI, au chargement du
# module — jamais dans main(), qui serait déjà trop tard pour éviter
# `from AppKit import ...` ci-dessous.
if "--diagnostic" in sys.argv:
from diagnostic import diagnostiquer, mettre_en_forme
print(mettre_en_forme(diagnostiquer()))
sys.exit(0)
import threading
from objc import super
@@ -28,11 +40,8 @@ from Foundation import NSObject, NSTimer
from PyObjCTools import AppHelper
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
from assemblage import acquerir_verrou, construire
from core import CONFIG
_STATE_TITLES = {
"idle": "🎙️",
@@ -42,19 +51,21 @@ _STATE_TITLES = {
class ZonzaApp(NSObject):
"""Délégué applicatif : status item + menu + hotkey + contrôleur."""
"""Délégué applicatif : status item + menu + hotkey.
def initWithTranscripteur_systeme_(self, transcripteur, systeme):
Le contrôleur (donc le transcripteur, l'overlay et le système) vient déjà
assemblé d'`assemblage.construire()` — voir `main()`. Ce délégué ne fait
plus QUE ce qui est propre à Cocoa : le NSStatusItem, le menu, et le
marshalage `AppHelper.callAfter` vers le thread principal.
"""
def initWithController_(self, controller):
self = super().init()
if self is None:
return None
self._item = None
self._sounds_item = None
self.pulse = classe_overlay_pour().alloc().init()
self.controller = ZonzaController(
CONFIG, self.pulse, transcripteur, systeme, on_state=self._set_state,
run_on_main=AppHelper.callAfter,
)
self.controller = controller
self._listener = None
return self
@@ -131,22 +142,18 @@ class ZonzaApp(NSObject):
def main():
# une seule instance a la fois : sinon deux modeles en RAM et la meme
# hotkey captee deux fois (cf systeme.acquerir_verrou_instance).
# hotkey captee deux fois (cf systeme.acquerir_verrou_instance). La prise
# du verrou reste ICI (pas dans assemblage.construire) : c'est au lanceur
# de décider quoi faire d'un refus, avant même de construire quoi que ce
# soit d'autre.
from systeme import choisir_systeme
systeme = choisir_systeme()
lock = systeme.acquerir_verrou_instance(
os.path.join(systeme.dossier_donnees(), "instance.lock")
)
lock = acquerir_verrou(systeme)
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.
@@ -156,8 +163,19 @@ def main():
)
trusted = AXIsProcessTrustedWithOptions({kAXTrustedCheckOptionPrompt: True})
print(f"[diag] accessibilité accordée : {bool(trusted)}")
app = NSApplication.sharedApplication()
delegate = ZonzaApp.alloc().initWithTranscripteur_systeme_(transcripteur, systeme)
# Dépendance circulaire : `on_state` doit marshaler vers `delegate._apply_title`,
# mais `delegate` a besoin du contrôleur déjà construit pour naître. Cassée
# avec une case remplie APRÈS coup, comme dans lanceur_qt.py — la fermeture
# ne lit `case["delegate"]` qu'au moment de l'appel, jamais à la construction.
case = {}
assemblage = construire(
run_on_main=AppHelper.callAfter,
on_state=lambda s: case["delegate"]._set_state(s),
)
delegate = ZonzaApp.alloc().initWithController_(assemblage["controller"])
case["delegate"] = delegate
delegate.setup()
# référence gardée pour toute la vie du processus (sinon GC → icône disparaît)
main._delegate = delegate
+75
View File
@@ -0,0 +1,75 @@
"""Assemblage commun aux deux lanceurs (`app.py` Cocoa, `lanceur_qt.py` Qt/Windows).
Avant ce module, `app.py` était le SEUL endroit qui savait choisir le système,
prendre le verrou d'instance, construire le transcripteur et l'overlay, puis
câbler le tout dans un `ZonzaController`. Un second lanceur qui recopierait
cette séquence créerait deux copies vouées à diverger en silence — exactement
le défaut déjà résorbé entre `engine.py` et `systeme/macos.py` (cf journal du
projet). Les deux lanceurs appellent donc les deux fonctions d'ici ; ce qui
reste propre à chacun (boucle d'événements, icône de barre, menu) reste dans
`app.py` et `lanceur_qt.py`.
Tout est injectable (`plateforme`, `run_on_main`...) pour rester testable sans
jamais ouvrir de fenêtre ni de micro : la construction du transcripteur charge
un vrai modèle, donc les tests injectent un `moteur.detecter_moteur`/
`moteur.choisir_transcripteur` doublés plutôt que de laisser tourner le vrai.
"""
import os
import moteur
from controller import ZonzaController
from core import CONFIG, modele_retenu, validate_config
from interface import classe_overlay_pour
from systeme import choisir_systeme
def acquerir_verrou(systeme, chemin_verrou=None):
"""Prend le verrou d'instance unique ; rend le handle, ou None si pris.
Une seule instance à la fois : sinon deux modèles en RAM et la même
hotkey captée deux fois (cf systeme.acquerir_verrou_instance). Le chemin
est injectable : les tests et les essais manuels d'un lanceur doivent
pouvoir viser un fichier qui n'est PAS celui de l'instance déjà en cours
de l'utilisateur.
"""
chemin = chemin_verrou or os.path.join(systeme.dossier_donnees(), "instance.lock")
return systeme.acquerir_verrou_instance(chemin)
def construire(plateforme=None, run_on_main=None, on_state=None,
detecter_moteur=None, choisir_transcripteur=None):
"""Construit le système, le transcripteur, l'overlay et le contrôleur.
Ne prend PAS le verrou (c'est le rôle d'`acquerir_verrou`, appelée à part
par le lanceur, qui décide seul de ce qu'il fait d'un refus). Ne valide
la config qu'une fois, ici, pour que les deux lanceurs en bénéficient
sans avoir à y penser.
`detecter_moteur`/`choisir_transcripteur` sont injectables : ils
permettent aux tests de construire l'assemblage entier sans charger un
vrai modèle de transcription (opération lourde, et qui échouerait sans
matériel/poids téléchargés).
"""
detecter_moteur = detecter_moteur or moteur.detecter_moteur
choisir_transcripteur = choisir_transcripteur or moteur.choisir_transcripteur
validate_config(CONFIG)
systeme = choisir_systeme(plateforme)
moteur_nom = detecter_moteur()
transcripteur = choisir_transcripteur(
nom=moteur_nom,
modele=modele_retenu(moteur_nom, CONFIG["model"]),
langue=CONFIG["language"],
amorce=CONFIG["initial_prompt"],
)
overlay = classe_overlay_pour(plateforme)()
controller = ZonzaController(
CONFIG, overlay, transcripteur, systeme, on_state=on_state,
run_on_main=run_on_main,
)
return {
"systeme": systeme,
"transcripteur": transcripteur,
"overlay": overlay,
"controller": controller,
}
+58 -1
View File
@@ -6,7 +6,14 @@ Tout ce qui est ici est testable sans micro, sans clavier, sans modèle.
import re
CONFIG = {
"model": "mlx-community/whisper-medium",
# Moteur de transcription. None = detection materielle (mlx, faster-cpu,
# faster-gpu, tous locaux). "gemini" (API Google directe) ou "openrouter"
# (meme modele, via OpenRouter) = l'audio quitte le Mac, avec le moteur
# local en repli si l'API ne repond pas (cf moteur/gemini.py).
"moteur": "openrouter",
# Forcage manuel du modele. None = on suit MODELES_PAR_MOTEUR, qui choisit
# d'apres le moteur detecte. Renseigner cette cle prime sur la table.
"model": None,
"language": "fr",
"hotkey": "<cmd>+<ctrl>+d", # Cmd+Ctrl+D (libre : ni macOS ni VS Code ne l'utilisent)
"sounds": True,
@@ -63,6 +70,8 @@ VOCABULAIRE = [
"Gitea", "GitHub", "Hetzner", "Hostinger", "iCloud", "Clerk",
# paiement, mail, automatisation
"Stripe", "PayPal", "Shopify", "Brevo", "n8n", "Airtable", "Yabetoo",
# IA et medias
"Higgsfield",
# projets
"FacturPro", "Forescale", "Reflow", "CiteGain", "Faramaya", "Soko",
"Moubote", "Yelessa", "Zonza", "MR TECH LAB",
@@ -163,6 +172,54 @@ def duree_vie_overlay_coherente(config):
return config["duree_max_overlay_s"] >= config["max_duration_s"] + MARGE_TRANSCRIPTION_S
# --- modele par moteur ----------------------------------------------------
# Chaque moteur a son propre format de depot : MLX lit `mlx-community/*`,
# faster-whisper lit `Systran/faster-whisper-*`. Passer l'un a l'autre echoue.
#
# Mesure du 2026-08-23, trois echantillons, amorce de vocabulaire active,
# sur le CPU d'un Apple Silicon :
# faster-whisper base 0,52 s 10/13 termes techniques 142 Mo
# faster-whisper small 1,42 s 11/13 464 Mo
# faster-whisper medium 4,40 s 13/13 1460 Mo
# MLX medium (GPU) 0,75 s 11/13 489 Mo
#
# D'ou les choix : le GPU vise la precision (medium, 13/13, le GPU absorbe le
# cout) ; le CPU vise la latence (small, 11/13 — soit AUTANT que le MLX utilise
# aujourd'hui — pour 1,4 s ici). `medium` sur CPU serait plus juste, mais 4,4 s
# ont ete mesures sur un CPU Apple Silicon, rapide : sur un CPU Intel ou Windows,
# le meme modele peut demander trois a quatre fois plus. La commande de
# diagnostic (etape 4) mesurera la machine reelle et pourra promouvoir `medium`.
MODELES_PAR_MOTEUR = {
"mlx": "mlx-community/whisper-medium",
"faster-gpu": "Systran/faster-whisper-medium",
"faster-cpu": "Systran/faster-whisper-small",
# Pas la variante « -tts » : celle-ci ne fait que produire de la voix a
# partir de texte, elle ne transcrit pas. Il faut un modele qui ACCEPTE
# l'audio en entree.
"gemini": "gemini-3.1-flash-lite",
# Mesure du 2026-09-25 (phrase de 12 s / dictee de 94 s, 2 a 3 essais) :
# 3.5-flash-lite 1,4 s / 3-5 s AJOUTE une phrase non dite -> ecarte
# 3.1-flash-lite 1,5 s / 3,7 s « Gitea » -> « GitHub » sur le long
# 3.8-flash 4-9 s / 9 s tout juste, a chaque essai -> retenu
# « 3.8-flash-lite-tts » est refuse par OpenRouter (HTTP 400) : c'est un
# modele texte -> voix, il n'accepte pas d'audio en entree.
"openrouter": "google/gemini-3.8-flash",
}
def modele_retenu(nom_moteur, forcage=None):
"""Modele finalement utilise : le forcage manuel s'il est renseigne, sinon la
table. Une chaine vide est traitee comme une absence de forcage."""
return forcage or modele_pour_moteur(nom_moteur)
def modele_pour_moteur(nom):
"""Rend le depot de modele adapte a ce moteur. Leve KeyError si inconnu."""
if nom not in MODELES_PAR_MOTEUR:
raise KeyError(f"Aucun modele connu pour le moteur : {nom}")
return MODELES_PAR_MOTEUR[nom]
def validate_config(config):
"""Valide la config ; lève ValueError si une valeur est aberrante."""
if config["min_duration_s"] < 0:
+312
View File
@@ -0,0 +1,312 @@
"""`zonza --diagnostic` : éprouve toute la chaîne sur la machine cible et dit où
elle casse.
Aucune machine Windows n'étant disponible pour ce portage, cette commande est
le SEUL moyen pour l'utilisateur de valider `systeme/windows.py` en conditions
réelles. D'où deux règles strictes suivies ici :
1. Chaque épreuve attrape SES PROPRES erreurs : une panne (micro absent,
moteur introuvable, `say` manquant...) ne doit jamais faire tomber les
épreuves suivantes ni le rapport entier.
2. Une épreuve qui n'a pas pu être menée est marquée `reussite=None`
(« non concluante ») — JAMAIS `True`. Mentir sur une réussite serait pire
que ne rien dire : mieux vaut « je ne sais pas » qu'un faux vert.
Deux fonctions séparées, pour rester testable : `diagnostiquer()` rend une
liste de `ResultatEpreuve` (aucun `print`) ; `mettre_en_forme()` la traduit en
texte affichable. Toutes les dépendances matérielles/plateforme sont
injectables, ce qui permet d'éprouver l'ENCHAÎNEMENT des épreuves depuis ce
Mac sans jamais toucher un vrai micro ni un vrai moteur de transcription.
"""
import os
import platform
import shutil
import subprocess
import sys
import tempfile
import time
import uuid
import wave
from dataclasses import dataclass
import numpy as np
import core
import moteur
from moteur import _cuda_disponible
from systeme import choisir_systeme
@dataclass(frozen=True)
class ResultatEpreuve:
"""Le compte-rendu d'une seule épreuve.
`reussite` : True (menée, et tout va bien), False (menée, et quelque chose
est cassé), ou None (n'a PAS pu être menée — non concluante, jamais
présentée comme une réussite).
"""
nom: str
reussite: bool | None
detail: str
# --- épreuve 1 : plateforme -------------------------------------------------
def epreuve_plateforme(plateforme=None, machine=None, cuda_disponible=None):
"""Décrit la machine : plateforme Python, architecture, présence de CUDA."""
plateforme = plateforme if plateforme is not None else sys.platform
machine = machine if machine is not None else platform.machine()
sonder_cuda = cuda_disponible if cuda_disponible is not None else _cuda_disponible
try:
cuda = sonder_cuda()
except Exception as e:
return ResultatEpreuve("Plateforme", None, f"sonde CUDA en échec : {e}")
cuda_txt = "présent" if cuda else "absent"
return ResultatEpreuve("Plateforme", True, f"{plateforme} / {machine} / CUDA {cuda_txt}")
# --- épreuve 2 : moteur retenu ----------------------------------------------
def epreuve_moteur(detecter_moteur=None, resoudre_modele=None, creer_transcripteur=None):
"""Décide du moteur, résout son modèle, instancie le transcripteur.
Rend `(resultat, transcripteur_ou_None)` : le transcripteur sert à
l'épreuve Transcription sans être reconstruit — mais seulement si cette
épreuve a réussi ; sinon la Transcription sera honnêtement sautée.
"""
detecter_moteur = detecter_moteur or moteur.detecter_moteur
resoudre_modele = resoudre_modele or core.modele_retenu
creer_transcripteur = creer_transcripteur or moteur.choisir_transcripteur
try:
nom = detecter_moteur()
nom_modele = resoudre_modele(nom, core.CONFIG["model"])
transcripteur = creer_transcripteur(
nom=nom, modele=nom_modele, langue=core.CONFIG["language"],
amorce=core.CONFIG["initial_prompt"],
)
except Exception as e:
return ResultatEpreuve("Moteur retenu", None, f"détection impossible : {e}"), None
return ResultatEpreuve("Moteur retenu", True, f"{nom} + {nom_modele}"), transcripteur
# --- épreuve 3 : micro -------------------------------------------------------
def epreuve_micro(ouvrir_flux=None, sample_rate=None, duree_s=1.0, dormir=None):
"""Ouvre VRAIMENT le micro pendant `duree_s` et compte les blocs reçus.
`ouvrir_flux` suit le même contrat que `Recorder._ouvrir_flux` dans
`capture.py` : injectable, pour que les tests ne touchent à aucun
périphérique réel.
Le débit vient de CONFIG et n'est pas codé en dur : le diagnostic doit
éprouver le micro AU DÉBIT QUE L'APPLICATION UTILISE VRAIMENT, sinon il
cesserait silencieusement d'être représentatif si ce réglage changeait.
"""
from core import CONFIG
sample_rate = sample_rate or CONFIG["sample_rate"]
dormir = dormir or time.sleep
if ouvrir_flux is None:
try:
import sounddevice as sd
except Exception as e:
return ResultatEpreuve("Micro", None, f"sounddevice indisponible : {e}")
ouvrir_flux = lambda **kw: sd.InputStream(**kw) # noqa: E731
compte = {"n": 0}
def on_block(indata, frames, time_info, status):
compte["n"] += 1
try:
flux = ouvrir_flux(samplerate=sample_rate, channels=1, dtype="float32",
callback=on_block)
flux.start()
except Exception as e:
return ResultatEpreuve("Micro", None, f"ouverture du micro impossible : {e}")
dormir(duree_s)
for action in ("stop", "close"):
try:
getattr(flux, action)()
except Exception:
pass # meilleur effort : le compte déjà reçu reste valable
n = compte["n"]
return ResultatEpreuve("Micro", n > 0, f"{n} blocs reçus en {duree_s:g} s")
# --- épreuve 4 : transcription -----------------------------------------------
def _generer_echantillon_parole(texte="Bonjour, ceci est un test de dictée Zonza.",
sample_rate=None):
"""Synthétise un court échantillon de parole via la commande `say` de macOS.
LÈVE si `say` est absent, échoue, ou si le WAV produit est illisible —
c'est volontaire : l'appelant retombe alors sur du silence plutôt que
d'inventer un résultat. Sur une machine sans `say` (Windows, Linux),
c'est TOUJOURS ce chemin qui est emprunté.
"""
from core import CONFIG
sample_rate = sample_rate or CONFIG["sample_rate"]
if shutil.which("say") is None:
raise RuntimeError("commande `say` absente (probablement pas macOS)")
chemin = tempfile.NamedTemporaryFile(suffix=".wav", delete=False).name
try:
subprocess.run(
["say", "-o", chemin, f"--data-format=LEI16@{sample_rate}", texte],
check=True, capture_output=True, timeout=15,
)
with wave.open(chemin, "rb") as w:
brut = w.readframes(w.getnframes())
return np.frombuffer(brut, dtype="<i2").astype("float32") / 32768.0
finally:
try:
os.unlink(chemin)
except OSError:
pass
def epreuve_transcription(transcripteur, generer_parole=None, horloge=None):
"""Transcrit VRAIMENT un échantillon — parole synthétisée si possible,
sinon une seconde de silence — et rapporte la durée réellement mesurée."""
horloge = horloge or time.monotonic
generer_parole = generer_parole or _generer_echantillon_parole
try:
audio = generer_parole()
source = "parole synthétisée"
except Exception as e:
from core import CONFIG
audio = np.zeros(CONFIG["sample_rate"], dtype="float32") # une seconde
source = f"silence (synthèse vocale indisponible : {e})"
debut = horloge()
try:
texte = transcripteur.transcrire(audio)
except Exception as e:
return ResultatEpreuve("Transcription", None,
f"échec de la transcription ({source}) : {e}")
duree = horloge() - debut
return ResultatEpreuve("Transcription", True, f"{duree:.2f} s — « {texte} » [{source}]")
# --- épreuve 5 : presse-papier ------------------------------------------------
def epreuve_presse_papier(copier=None, lire=None):
"""Copie un texte marqueur, le relit, vérifie qu'il n'a pas changé en route."""
if copier is None or lire is None:
try:
import pyperclip
except Exception as e:
return ResultatEpreuve("Presse-papier", None, f"pyperclip indisponible : {e}")
copier = copier or pyperclip.copy
lire = lire or pyperclip.paste
texte = f"zonza-diagnostic-{uuid.uuid4().hex[:8]}"
try:
copier(texte)
relu = lire()
except Exception as e:
return ResultatEpreuve("Presse-papier", None, f"copie/collage impossible : {e}")
if relu == texte:
return ResultatEpreuve("Presse-papier", True, "copie puis relecture identiques")
return ResultatEpreuve("Presse-papier", False, f"attendu {texte!r}, relu {relu!r}")
# --- épreuve 6 : raccourci ----------------------------------------------------
def epreuve_raccourci(systeme):
"""Rapporte le raccourci par défaut de cette plateforme (aucune action)."""
try:
raccourci = systeme.raccourci_defaut()
except Exception as e:
return ResultatEpreuve("Raccourci", None, f"indisponible : {e}")
return ResultatEpreuve("Raccourci", True, raccourci)
# --- épreuve 7 : verrou --------------------------------------------------------
def epreuve_verrou(systeme, chemin=None):
"""Vérifie qu'une seconde tentative de verrou sur le même chemin est refusée."""
chemin = chemin or os.path.join(
tempfile.mkdtemp(prefix="zonza-diagnostic-"), "diagnostic.lock"
)
try:
premier = systeme.acquerir_verrou_instance(chemin)
second = systeme.acquerir_verrou_instance(chemin)
except Exception as e:
return ResultatEpreuve("Verrou", None, f"indisponible : {e}")
if premier is not None:
try:
premier.close()
except Exception:
pass
if premier is None:
return ResultatEpreuve("Verrou", None, "impossible d'obtenir le premier verrou")
if second is None:
return ResultatEpreuve("Verrou", True, "seconde instance refusée")
return ResultatEpreuve("Verrou", False, "seconde instance NON refusée — verrou inefficace")
# --- orchestration -------------------------------------------------------------
def diagnostiquer(systeme=None, plateforme=None, machine=None, cuda_disponible=None,
detecter_moteur=None, resoudre_modele=None, creer_transcripteur=None,
ouvrir_flux=None, duree_micro_s=1.0, dormir=None,
generer_parole=None, horloge=None, copier=None, lire=None):
"""Éprouve toute la chaîne, dans l'ordre du rapport.
Ne lève JAMAIS : chaque épreuve attrape ses propres erreurs et se marque
non concluante plutôt que de faire tomber les suivantes ou le diagnostic
entier. Rend une liste de `ResultatEpreuve` — pas de `print` ici, pour que
ce soit testable indépendamment de l'affichage (cf `mettre_en_forme`).
"""
systeme = systeme or choisir_systeme()
resultats = [epreuve_plateforme(plateforme, machine, cuda_disponible)]
resultat_moteur, transcripteur = epreuve_moteur(
detecter_moteur, resoudre_modele, creer_transcripteur
)
resultats.append(resultat_moteur)
resultats.append(epreuve_micro(ouvrir_flux, duree_s=duree_micro_s, dormir=dormir))
if transcripteur is not None:
resultats.append(epreuve_transcription(transcripteur, generer_parole, horloge))
else:
resultats.append(ResultatEpreuve(
"Transcription", None, "sautée — aucun moteur n'a pu être instancié"
))
resultats.append(epreuve_presse_papier(copier, lire))
resultats.append(epreuve_raccourci(systeme))
resultats.append(epreuve_verrou(systeme))
return resultats
_SYMBOLES = {True: "✅", False: "❌", None: "➖"}
def formater_resultat(resultat):
"""Une ligne du rapport : symbole, nom aligné, détail."""
symbole = _SYMBOLES.get(resultat.reussite, "➖")
return f"{symbole} {resultat.nom.ljust(16)}: {resultat.detail}"
def mettre_en_forme(resultats):
"""Traduit une liste de `ResultatEpreuve` en texte affichable.
Séparée de `diagnostiquer()` à dessein : cette fonction ne fait aucun
travail, uniquement de la mise en forme — donc testable sur des résultats
fabriqués à la main, sans jamais exécuter une seule épreuve.
"""
lignes = [formater_resultat(r) for r in resultats]
reussies = sum(1 for r in resultats if r.reussite is True)
echouees = sum(1 for r in resultats if r.reussite is False)
non_conclu = sum(1 for r in resultats if r.reussite is None)
lignes.append("")
lignes.append(
f"{reussies}/{len(resultats)} réussies, {echouees} en échec, "
f"{non_conclu} non concluante(s)."
)
return "\n".join(lignes)
@@ -67,13 +67,28 @@ sinon → faster-whisper CPU
La **décision** est une fonction pure recevant `(plateforme, processeur, cuda_disponible)` ; l'**interrogation** de la machine est séparée. Les quatre configurations sont donc testables depuis un seul Mac.
| Configuration | Moteur | Modèle | Latence |
|---|---|---|---|
| Mac Apple Silicon | MLX | `whisper-medium` | ~1,3 s (mesuré) |
| Windows + NVIDIA | faster-whisper CUDA | `medium` float16 | 1–2 s (**à mesurer**) |
| Mac Intel · Windows CPU | faster-whisper CPU | `small` int8 | 3–8 s (**à mesurer**) |
**Mesuré le 2026-08-23** — trois échantillons, amorce de vocabulaire active, sur le CPU d'un Apple Silicon :
`medium` est écarté sur CPU : 15 à 30 s tuerait l'usage.
| Moteur et modèle | Latence | Termes techniques | Taille |
|---|---|---|---|
| faster-whisper `base` int8 CPU | 0,52 s | 10/13 | 142 Mo |
| faster-whisper `small` int8 CPU | 1,42 s | 11/13 | 464 Mo |
| faster-whisper `medium` int8 CPU | 4,40 s | **13/13** | 1460 Mo |
| MLX `medium` GPU *(moteur actuel)* | 0,75 s | 11/13 | 489 Mo |
🔴 **Ces chiffres démentent l'estimation initiale de cette spec**, qui annonçait 15 à 30 s pour `medium` sur CPU et l'écartait d'avance. C'est 4,4 s. Et `faster-whisper medium` sur CPU est **plus juste que le moteur GPU en service** (13/13 contre 11/13).
Choix retenus dans `core.MODELES_PAR_MOTEUR` :
| Configuration | Moteur | Modèle |
|---|---|---|
| Mac Apple Silicon | MLX | `mlx-community/whisper-medium` |
| Windows + NVIDIA | faster-whisper CUDA | `Systran/faster-whisper-medium` |
| Mac Intel · Windows CPU | faster-whisper CPU | `Systran/faster-whisper-small` |
⚠️ **Réserve sur le CPU** : les 4,4 s de `medium` ont été mesurées sur un CPU Apple Silicon, rapide. Un CPU Intel ou Windows peut demander trois à quatre fois plus. D'où `small` par défaut — même précision (11/13) que le MLX utilisé aujourd'hui, pour 1,4 s ici. **La commande de diagnostic mesurera la machine réelle et pourra promouvoir `medium`.**
`CONFIG["model"]`, désormais à `None`, force manuellement le modèle et prime sur la table.
**Première étape de l'implémentation : mesurer réellement les deux dernières lignes.** Le 2026-08-21, une mesure analogue a renversé l'intuition — `large-v3-turbo` s'est révélé PIRE que `medium`.
@@ -151,7 +166,7 @@ Le périmètre dépasse un seul plan. Quatre étapes, chacune livrable et vérif
**Étape 1 — Extraire les contrats.** Créer `moteur/`, `interface/`, `systeme/` ; y déplacer le code Cocoa **sans le modifier** ; faire passer `controller.py` par les contrats. Aucun changement de comportement : les 80 tests actuels doivent rester verts, et Zonza fonctionner à l'identique sur le Mac. C'est l'étape qui dé-risque toutes les suivantes.
**Étape 2 — Moteur interchangeable.** Ajouter `faster-whisper`, la détection pure, et **mesurer** les latences réelles CPU. Vérifiable immédiatement sur le Mac, en forçant le moteur CPU.
**Étape 2 — Moteur interchangeable.** ✅ **Fait le 2026-08-23** (`2ccb614`, 115 tests) : `moteur/faster.py`, table `MODELES_PAR_MOTEUR`, forçage manuel, et les mesures ci-dessus.
**Étape 3 — Interface Qt.** Rejouer la suite de contrat de l'overlay contre l'implémentation Qt. Vérifiable sur le Mac : Qt y tourne aussi, ce qui permet de comparer les deux overlays côte à côte.
+66
View File
@@ -0,0 +1,66 @@
# Installe Zonza sur Windows.
#
# powershell -ExecutionPolicy Bypass -File .\install.ps1
#
# Cree l'environnement, installe UNIQUEMENT ce qui sert a Windows (les marqueurs
# de requirements.txt ecartent MLX et pyobjc), pose le lancement au demarrage,
# puis lance le diagnostic.
#
# AVERTISSEMENT D'HONNETETE : ce script n'a JAMAIS ete execute sur une machine
# Windows. Il a ete ecrit depuis un Mac, d'apres la documentation. Le diagnostic
# de la derniere etape est justement la pour dire ce qui marche et ce qui casse.
$ErrorActionPreference = "Stop"
Set-Location -Path $PSScriptRoot
Write-Host "-> Verification de Python..."
$py = Get-Command python -ErrorAction SilentlyContinue
if (-not $py) { throw "Python introuvable. Installe-le depuis python.org, en cochant 'Add python.exe to PATH'." }
$version = & python -c "import sys; print('.'.join(map(str, sys.version_info[:3])))"
Write-Host " Python $version"
& python -c "import sys; sys.exit(0 if sys.version_info >= (3, 11) else 1)"
if ($LASTEXITCODE -ne 0) { throw "Python 3.11 minimum requis, trouve $version." }
if (-not (Test-Path ".venv")) {
Write-Host "-> Creation de l'environnement..."
& python -m venv .venv
} else {
Write-Host "-> Environnement deja present, on le reutilise."
}
Write-Host "-> Installation des dependances de CETTE machine..."
& .\.venv\Scripts\python.exe -m pip install --quiet --upgrade pip
& .\.venv\Scripts\python.exe -m pip install --quiet -r requirements.txt
if ($LASTEXITCODE -ne 0) { throw "L'installation des dependances a echoue." }
Write-Host "-> Lancement au demarrage..."
# pythonw.exe : sans lui, une console noire resterait ouverte en permanence.
$demarrage = [Environment]::GetFolderPath("Startup")
$raccourci = Join-Path $demarrage "Zonza.lnk"
$shell = New-Object -ComObject WScript.Shell
$lien = $shell.CreateShortcut($raccourci)
$lien.TargetPath = (Resolve-Path ".\.venv\Scripts\pythonw.exe").Path
$lien.Arguments = '"' + (Resolve-Path ".\lanceur_qt.py").Path + '"'
$lien.WorkingDirectory = $PSScriptRoot
$lien.Description = "Zonza - dictee vocale locale"
$lien.Save()
Write-Host " $raccourci"
Write-Host "-> Diagnostic..."
& .\.venv\Scripts\python.exe app.py --diagnostic
if ($LASTEXITCODE -ne 0) {
Write-Host ""
Write-Host "Le diagnostic signale un probleme. Rien n'est casse : les lignes"
Write-Host "ci-dessus disent laquelle des epreuves a echoue."
exit 1
}
Write-Host ""
Write-Host "Installe."
Write-Host ""
Write-Host "Usage : Ctrl+Alt+D pour demarrer et arreter la dictee."
Write-Host "Journal : %LOCALAPPDATA%\Zonza\Zonza.log"
Write-Host "Lancer maintenant, sans redemarrer : .\.venv\Scripts\pythonw.exe lanceur_qt.py"
Write-Host ""
Write-Host "Si le raccourci ne repond pas, relance le diagnostic :"
Write-Host " .\.venv\Scripts\python.exe app.py --diagnostic"
Executable
+101
View File
@@ -0,0 +1,101 @@
#!/usr/bin/env bash
#
# Installe Zonza sur macOS (et Linux, pour la partie portable).
#
# ./install.sh
#
# Cree l'environnement, installe UNIQUEMENT ce qui sert a cette machine (les
# marqueurs de requirements.txt s'en chargent), construit l'app macOS, puis lance
# le diagnostic. Il ne masque aucune erreur : si une etape echoue, il s'arrete.
set -euo pipefail
cd "$(dirname "$0")"
MINIMUM="3.11"
PY="${PYTHON:-python3}"
echo "→ Verification de Python…"
command -v "$PY" >/dev/null || { echo "✗ '$PY' introuvable. Installe Python $MINIMUM ou plus, ou renseigne PYTHON=/chemin/vers/python3." >&2; exit 1; }
"$PY" - <<PYCODE || { echo "✗ Python $MINIMUM minimum requis." >&2; exit 1; }
import sys
raise SystemExit(0 if sys.version_info >= tuple(int(n) for n in "$MINIMUM".split(".")) else 1)
PYCODE
echo " $("$PY" --version)"
if [[ ! -d .venv ]]; then
echo "→ Creation de l'environnement…"
"$PY" -m venv .venv
else
echo "→ Environnement deja present, on le reutilise."
fi
echo "→ Installation des dependances de CETTE machine…"
.venv/bin/python -m pip install --quiet --upgrade pip
.venv/bin/python -m pip install --quiet -r requirements.txt
echo " $(.venv/bin/python -m pip list 2>/dev/null | wc -l | tr -d ' ') paquets installes"
if [[ "$(uname)" == "Darwin" ]]; then
# Garde-fou : build_app.sh grave le chemin de CE dossier dans le binaire de
# /Applications/Zonza.app. Si une Zonza y est deja installee depuis un AUTRE
# dossier, reconstruire la detournerait vers ici — et changerait son cdhash,
# ce qui REVOQUE silencieusement les autorisations macOS deja accordees.
# Faute de ce garde-fou, un simple essai depuis une copie casse l'installation
# en service (arrive le 2026-08-23, rattrape de justesse).
CIBLE_ACTUELLE=""
if [[ -x /Applications/Zonza.app/Contents/MacOS/zonza ]]; then
CIBLE_ACTUELLE="$(strings /Applications/Zonza.app/Contents/MacOS/zonza 2>/dev/null | grep -m1 -E '/app\.py$' || true)"
fi
ICI="$(pwd)/app.py"
if [[ -n "$CIBLE_ACTUELLE" && "$CIBLE_ACTUELLE" != "$ICI" ]]; then
echo
echo "⚠ Une Zonza est deja installee, et elle pointe ailleurs :"
echo " installee ici : $CIBLE_ACTUELLE"
echo " ce dossier : $ICI"
echo
echo " Reconstruire la detournerait vers ce dossier et REVOQUERAIT ses"
echo " autorisations macOS (Accessibilite, Surveillance de l'entree)."
echo " L'application n'est donc PAS reconstruite."
echo
echo " Si c'est bien ce que tu veux, lance : ZONZA_REMPLACER=1 ./install.sh"
if [[ "${ZONZA_REMPLACER:-}" != "1" ]]; then
CONSTRUIRE=0
else
echo " ZONZA_REMPLACER=1 : reconstruction demandee explicitement."
CONSTRUIRE=1
fi
else
CONSTRUIRE=1
fi
if [[ "$CONSTRUIRE" == "1" ]]; then
echo "→ Construction de Zonza.app…"
./build_app.sh
fi
fi
echo "→ Diagnostic…"
.venv/bin/python app.py --diagnostic || {
echo
echo "✗ Le diagnostic signale un probleme. Rien n'est casse : lis les lignes"
echo " ci-dessus, elles disent laquelle des epreuves a echoue." >&2
exit 1
}
cat <<'FIN'
✓ Installe.
Il reste UNE chose que seul toi peux faire, et sans elle le raccourci ne
fonctionnera pas : accorder les autorisations macOS.
Reglages systeme > Confidentialite et securite
• Accessibilite → ajoute /Applications/Zonza.app, puis active
• Surveillance de l'entree → la meme chose
Puis quitte Zonza par son icone 🎙️ et relance-la : l'autorisation n'est lue
qu'au demarrage.
Usage : Cmd+Ctrl+D pour demarrer et arreter la dictee.
Journal : ~/Library/Logs/Zonza.log
FIN
+12 -2
View File
@@ -1,4 +1,11 @@
"""Choisit l'implémentation d'overlay selon la plateforme."""
"""Choisit l'implémentation d'overlay selon la plateforme.
Convention des trois frontières (`moteur/`, `interface/`, `systeme/`) : **un module
plat par plateforme** — `interface/cocoa.py`, `systeme/macos.py`, `moteur/mlx.py`.
On ne passe à un sous-paquet que le jour où une plateforme exige réellement
plusieurs fichiers. Sans cette règle écrite, la couche Windows aurait eu à deviner
lequel des deux motifs imiter.
"""
import sys
@@ -6,6 +13,9 @@ 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
from interface.cocoa import PulseWindow
return PulseWindow
if plateforme.startswith("win32"):
from interface.qt import OverlayQt
return OverlayQt
raise NotImplementedError(f"Aucune interface pour : {plateforme}")
View File
+343
View File
@@ -0,0 +1,343 @@
"""Overlay de dictée : barre sombre arrondie posée en bas de l'écran (Qt).
Copie conforme de `interface/cocoa.py`, pour les plateformes sans Cocoa
(Windows aujourd'hui). Mêmes trois zones, de gauche à droite : un bouton stop
rond rouge-orange, une forme d'onde en barres verticales, une icône de micro.
Au-dessus, une bulle sombre affiche l'état en cours.
Qt natif (PySide6). Tout le dessin a lieu sur le thread principal via un
QTimer ; les niveaux audio arrivent d'un thread de fond par set_level(), et
l'historique qui alimente la forme d'onde est protégé par un verrou — même
disposition que Cocoa.
Repère : Qt place l'origine en HAUT à gauche (y croît vers le bas), alors que
Cocoa la place en BAS à gauche (y croît vers le haut). C'est la source
d'erreur la plus probable d'un portage Cocoa → Qt : décaler un signe à un
seul endroit du dessin fait sortir une forme du mauvais côté sans qu'aucun
test ne le révèle (cf. les 42 tests verts du 21/08 pendant qu'une icône
sortait en tache blanche — seule une capture d'écran l'a montré). On bascule
donc le repère une seule fois, ici, avant de dessiner les formes qui
raisonnent comme Cocoa ; le texte de la bulle est dessiné à part, hors de ce
repère basculé, pour ne pas sortir à l'envers.
"""
import threading
from PySide6.QtCore import QPointF, QRectF, Qt, QTimer
from PySide6.QtGui import QColor, QFontMetrics, QGuiApplication, QPainter, QPainterPath, QPen
from PySide6.QtWidgets import QApplication, QWidget
from audio_level import bar_heights, push_level
from core import CONFIG, bar_metrics
_BOTTOM_MARGIN = 80.0 # au-dessus de la barre des tâches
_BUBBLE_GAP = 12.0 # espace entre la bulle et la barre
_BUBBLE_PAD = 12.0 # marge intérieure de la bulle
_TAIL = 7.0 # demi-largeur de la pointe de la bulle
_BOTTOM_INSET = 8.0 # évite que l'arrondi bas de la barre touche le bord
def _qcolor(triplet, alpha=1.0):
r, g, b = triplet[:3]
c = QColor()
c.setRgbF(r, g, b, alpha)
return c
class VueOverlay(QWidget):
"""Dessine la barre, la forme d'onde et la bulle d'état.
Widget Qt top-level SANS PARENT : Qt n'exige pas, contrairement à Cocoa,
une fenêtre (NSWindow) séparée de sa vue de contenu (NSView) — un QWidget
sans parent EST déjà une fenêtre. Cette classe joue donc les deux rôles :
« vue qui dessine » ET « fenêtre affichée », la façade `OverlayQt`
n'ayant qu'à la montrer/masquer.
"""
def __init__(self):
super().__init__()
self._bars = []
self._status = ""
# Fenêtre sans cadre, toujours au-dessus, à fond transparent, et
# transparente aux clics : la souris doit traverser l'overlay jusqu'à
# l'application en dessous (Qt.WindowTransparentForInput).
self.setWindowFlags(
Qt.WindowType.FramelessWindowHint
| Qt.WindowType.WindowStaysOnTopHint
| Qt.WindowType.Tool
| Qt.WindowType.WindowTransparentForInput
)
self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground)
self.setAttribute(Qt.WidgetAttribute.WA_ShowWithoutActivating, True)
def set_bars(self, bars):
self._bars = bars
def set_status(self, text):
self._status = text or ""
# --- morceaux de dessin (repère Cocoa : y croît vers le HAUT) ---------
def _draw_bar_background(self, painter, x, y, w, h):
painter.setBrush(_qcolor(CONFIG["bar_bg_rgba"], CONFIG["bar_bg_rgba"][3]))
painter.setPen(Qt.PenStyle.NoPen)
painter.drawRoundedRect(
QRectF(x, y, w, h), CONFIG["bar_radius"], CONFIG["bar_radius"]
)
def _draw_stop_button(self, painter, cx, cy, diametre):
painter.setBrush(_qcolor(CONFIG["stop_button_rgb"]))
painter.setPen(Qt.PenStyle.NoPen)
painter.drawEllipse(QPointF(cx, cy), diametre / 2, diametre / 2)
# carré arrondi blanc au centre = « arrêter »
side = diametre * 0.34
painter.setBrush(_qcolor(CONFIG["stop_glyph_rgb"]))
painter.drawRoundedRect(QRectF(cx - side / 2, cy - side / 2, side, side), 2.0, 2.0)
def _draw_waveform(self, painter, x0, cy, width):
bars = self._bars
if not bars:
return
n = len(bars)
step = width / float(n)
bw = max(2.0, step * 0.38)
half_max = CONFIG["bar_height"] * 0.30 # amplitude verticale max
painter.setBrush(_qcolor(CONFIG["waveform_rgb"]))
painter.setPen(Qt.PenStyle.NoPen)
for i, level in enumerate(bars):
cxi = x0 + step * (i + 0.5)
half = max(bw / 2.0, level * half_max) # jamais moins qu'un point rond
painter.drawRoundedRect(
QRectF(cxi - bw / 2, cy - half, bw, half * 2), bw / 2, bw / 2
)
def _draw_mic(self, painter, cx, cy, m):
painter.setBrush(_qcolor(CONFIG["mic_rgb"]))
largeur = m["mic_largeur"]
body_h = m["mic_corps_hauteur"]
body_y = cy - body_h / 2 + body_h * 0.23
# capsule (le micro lui-même)
painter.setPen(Qt.PenStyle.NoPen)
painter.drawRoundedRect(
QRectF(cx - largeur / 2, body_y, largeur, body_h), largeur / 2, largeur / 2
)
# arceau en U qui passe SOUS la capsule. PIÈGE vérifié par capture
# d'écran (cf tâche) : `drawArc` mesure ses angles dans le repère
# ÉCRAN natif de Qt (0° = 3 h, sens visuellement anti-horaire), et ne
# tient PAS compte du basculement `scale(1, -1)` appliqué plus haut —
# contrairement aux formes remplies (rect, ellipse), qui suivent la
# transformation courante. Un premier essai avec les bornes de Cocoa
# telles quelles (195°→345°) donnait un arc reflété verticalement
# (ouvert vers le BAS, au-dessus de la capsule au lieu de dessous).
# On compense en reflétant l'angle : depart' = -(depart+etendue),
# etendue' = etendue.
rayon = m["mic_arceau_rayon"]
arc_cx, arc_cy = cx, body_y + body_h * 0.23
depart_cocoa, etendue_cocoa = 195, 345 - 195
depart = -(depart_cocoa + etendue_cocoa) % 360
# QPen neuf : `painter.pen()` aurait hérité le style NoPen laissé par
# le dessin de la capsule juste au-dessus — un pen NoPen ignore
# silencieusement couleur et épaisseur, l'arc ne se serait jamais vu.
pen = QPen(_qcolor(CONFIG["mic_rgb"]))
pen.setWidthF(max(1.0, body_h * 0.12))
pen.setCapStyle(Qt.PenCapStyle.RoundCap)
painter.setPen(pen)
painter.setBrush(Qt.BrushStyle.NoBrush)
rect_arc = QRectF(arc_cx - rayon, arc_cy - rayon, rayon * 2, rayon * 2)
painter.drawArc(rect_arc, depart * 16, etendue_cocoa * 16)
# pied vertical sous l'arceau
pied = max(1.2, largeur * 0.18)
painter.setPen(Qt.PenStyle.NoPen)
painter.setBrush(_qcolor(CONFIG["mic_rgb"]))
painter.drawRoundedRect(
QRectF(cx - pied / 2, body_y - body_h * 0.62, pied, body_h * 0.31),
pied / 2, pied / 2,
)
def _bubble_box(self, bar_x, bar_top, bar_w):
"""Géométrie de la bulle (repère Cocoa), ou None si aucun texte."""
if not self._status:
return None
font = self.font()
font.setPointSizeF(CONFIG["bubble_font_size"])
metrics = QFontMetrics(font)
bounds = metrics.boundingRect(
QRectF(0, 0, CONFIG["bubble_max_width"], 400.0).toRect(),
int(Qt.TextFlag.TextWordWrap),
self._status,
)
tw, th = float(bounds.width()), float(bounds.height())
w, h = tw + 2 * _BUBBLE_PAD, th + 2 * _BUBBLE_PAD
x = bar_x + CONFIG["bar_height"] * 0.46 # alignée à gauche, comme la référence
y = bar_top + _BUBBLE_GAP
return {"x": x, "y": y, "w": w, "h": h, "tw": tw, "th": th, "font": font}
def _draw_bubble_shape(self, painter, box):
painter.setBrush(_qcolor(CONFIG["bar_bg_rgba"], CONFIG["bar_bg_rgba"][3]))
painter.setPen(Qt.PenStyle.NoPen)
x, y, w, h = box["x"], box["y"], box["w"], box["h"]
painter.drawRoundedRect(QRectF(x, y, w, h), 10.0, 10.0)
# petite pointe vers la barre, en bas à gauche
tail = QPainterPath()
tail.moveTo(x + 20.0 - _TAIL, y + 1.0)
tail.lineTo(x + 20.0 + _TAIL, y + 1.0)
tail.lineTo(x + 20.0, y - 7.0)
tail.closeSubpath()
painter.drawPath(tail)
def paintEvent(self, event):
# `with` ferme le QPainter (.end()) meme si une exception surgit en
# cours de dessin : sans ca, la fermeture ne tenait qu'au comptage de
# references de PySide6 — suffisant aujourd'hui, fragile demain.
with QPainter(self) as painter:
painter.setRenderHint(QPainter.RenderHint.Antialiasing, True)
w = CONFIG["bar_width"]
h = CONFIG["bar_height"]
x = (self.width() - w) / 2.0
y = _BOTTOM_INSET
cy = y + h / 2.0
m = bar_metrics(h)
pad, stop_d, mic_w, ecart = (
m["marge_laterale"], m["stop_diametre"], m["mic_largeur"], m["ecart_interne"]
)
bubble = self._bubble_box(x, y + h, w)
# Bascule de repère : à partir d'ici, y croît vers le HAUT comme dans
# Cocoa (voir docstring du module). On la retire (`restore`) avant de
# dessiner le texte de la bulle plus bas, pour qu'il ne sorte pas à
# l'envers.
painter.save()
painter.translate(0, self.height())
painter.scale(1, -1)
self._draw_bar_background(painter, x, y, w, h)
self._draw_stop_button(painter, x + pad + stop_d / 2, cy, stop_d)
self._draw_mic(painter, x + w - pad - mic_w / 2, cy, m)
wave_x = x + pad + stop_d + ecart
wave_w = (x + w - pad - mic_w - ecart) - wave_x
self._draw_waveform(painter, wave_x, cy, wave_w)
if bubble is not None:
self._draw_bubble_shape(painter, bubble)
painter.restore()
if bubble is not None:
# Repère Qt normal (y vers le bas) : on convertit le coin bas-gauche
# « repère Cocoa » de la bulle en un rectangle Qt haut-gauche.
text_top_qt = self.height() - (bubble["y"] + _BUBBLE_PAD + bubble["th"])
painter.setFont(bubble["font"])
painter.setPen(_qcolor(CONFIG["bubble_text_rgb"]))
painter.drawText(
QRectF(bubble["x"] + _BUBBLE_PAD, text_top_qt, bubble["tw"], bubble["th"]),
int(Qt.TextFlag.TextWordWrap),
self._status,
)
class OverlayQt:
"""Fenêtre flottante Qt : show(), hide(), set_level(), set_status(). Thread-safe.
Contrairement à `PulseWindow` (NSObject, exige `alloc().init()`), c'est une
classe Python ordinaire : `classe_overlay_pour("win32")()` suffit.
"""
def __init__(self):
self._history = []
self._status = ""
self._lock = threading.Lock()
self._view = None
self._timer = None
self._garde = None # ferme l'overlay quoi qu'il arrive (cf show)
def _window_size(self):
return (CONFIG["bar_width"] + 80.0,
CONFIG["bar_height"] + _BOTTOM_INSET + 140.0)
def _ensure_view(self):
if self._view is not None:
return
# PySide6 est déjà exigé dès l'IMPORT de ce module (QWidget, QTimer...
# ci-dessus) : le report réel n'est pas ici, mais dans la fabrique
# (interface/__init__.py::classe_overlay_pour), qui n'importe
# `interface.qt` que pour la branche win32 — jamais sur ce Mac, où
# Cocoa suffit.
if QApplication.instance() is None:
QApplication([])
ww, wh = self._window_size()
self._view = VueOverlay()
self._view.resize(int(ww), int(wh))
ecran = QGuiApplication.primaryScreen().geometry()
px = ecran.left() + (ecran.width() - ww) / 2.0
py = ecran.top() + ecran.height() - _BOTTOM_MARGIN - wh
self._view.move(int(px), int(py))
def set_level(self, level):
"""Pousse le dernier niveau audio (appelable depuis un thread de fond)."""
with self._lock:
self._history = push_level(self._history, level, CONFIG["waveform_bars"])
def set_status(self, text):
"""Texte affiché dans la bulle (thread de fond autorisé)."""
with self._lock:
self._status = text or ""
def _tick(self):
with self._lock:
history, status = list(self._history), self._status
if self._view is not None:
self._view.set_bars(
bar_heights(history, CONFIG["waveform_bars"], CONFIG["waveform_floor"])
)
self._view.set_status(status)
self._view.update()
def show(self):
"""Affiche la fenêtre et démarre l'animation. Thread principal."""
self._ensure_view()
with self._lock:
self._history = []
self._view.show()
self._timer = QTimer()
self._timer.timeout.connect(self._tick)
self._timer.start(int(1000.0 / float(CONFIG["fps"])))
# Garde-fou : la fenêtre porte sa propre durée de vie. Si le contrôleur
# échoue à la refermer — transcription bloquée, thread mort, bug en
# amont — elle se ferme seule plutôt que de rester figée à l'écran
# (même correctif que Cocoa, 2026-08-21).
self._garde = QTimer()
self._garde.setSingleShot(True)
self._garde.timeout.connect(self._expiration)
self._garde.start(int(float(CONFIG["duree_max_overlay_s"]) * 1000.0))
def rearmer_garde(self, duree_s):
"""Remplace la garde en cours par une plus courte. Thread principal.
Appelée au passage en transcription : la garde longue couvre un
enregistrement de 2 min, mais une fois l'enregistrement fini il ne
reste que quelques secondes de calcul — inutile de laisser un halo
trois minutes.
"""
if self._garde is not None:
self._garde.stop()
self._garde = QTimer()
self._garde.setSingleShot(True)
self._garde.timeout.connect(self._expiration)
self._garde.start(int(float(duree_s) * 1000.0))
def _expiration(self):
print("[garde] overlay ouvert trop longtemps — fermeture forcée")
self.hide()
def hide(self):
"""Arrête l'animation et masque la fenêtre. Thread principal."""
if self._timer is not None:
self._timer.stop()
self._timer = None
if self._garde is not None:
self._garde.stop()
self._garde = None
if self._view is not None:
self._view.hide()
with self._lock:
self._status = ""
+227
View File
@@ -0,0 +1,227 @@
"""Point d'entrée : application Qt de barre de tâches (PySide6, Windows).
Pendant Qt de `app.py` (Cocoa/macOS) : même icône de barre (🎙️/🔴/⏳), même
menu à deux entrées (Sons, Quitter), même hotkey globale via pynput. Ce qui
diffère de Cocoa est ISOLÉ ici — boucle d'événements Qt, `QSystemTrayIcon`,
marshalage par signal Qt — le reste (choix du système, verrou d'instance,
transcripteur, overlay, contrôleur) vient d'`assemblage.py`, partagé avec
`app.py`. Dupliquer cet assemblage créerait deux copies vouées à diverger en
silence (cf docstring d'`assemblage.py`).
Marshalage thread de fond → thread Qt : `AppHelper.callAfter` (Cocoa) n'a pas
d'équivalent Qt. La voie sûre est un signal Qt émis depuis le thread de fond
et connecté en `QueuedConnection` — Qt place alors l'exécution du slot dans
la boucle d'événements du thread PROPRIÉTAIRE de l'objet récepteur (le thread
principal, puisque `_Pont` y est instancié), quel que soit le thread émetteur.
`QTimer.singleShot` depuis un thread non-Qt n'offre pas cette garantie : un
QTimer démarré hors du thread principal s'exécute (ou pas) sur CE thread, pas
sur le thread principal — comportement non documenté comme sûr, à l'inverse
des connexions en file, documentées explicitement thread-safe.
"""
import sys
# --diagnostic doit s'évaluer AVANT tout import PySide6/pynput : sur une
# machine qui ne les a pas, c'est le seul moyen d'éprouver la chaîne (même
# raison que dans app.py — voir sa docstring).
if "--diagnostic" in sys.argv:
from diagnostic import diagnostiquer, mettre_en_forme
print(mettre_en_forme(diagnostiquer()))
sys.exit(0)
import os
import threading
from PySide6.QtCore import QObject, Qt, Signal
from PySide6.QtGui import QAction, QIcon, QPixmap
from PySide6.QtWidgets import QApplication, QMenu, QSystemTrayIcon
from pynput import keyboard
from assemblage import acquerir_verrou, construire
from core import CONFIG
from systeme import choisir_systeme
_STATE_TITLES = {
"idle": "🎙️",
"recording": "🔴",
"transcribing": "⏳",
}
def _icone_pour(emoji):
"""Rend un QIcon dessiné à partir d'un emoji — pas de fichier .ico à livrer."""
pixmap = QPixmap(64, 64)
pixmap.fill(Qt.GlobalColor.transparent)
from PySide6.QtGui import QFont, QPainter
with QPainter(pixmap) as painter:
painter.setRenderHint(QPainter.RenderHint.Antialiasing, True)
font = QFont()
font.setPointSize(40)
painter.setFont(font)
painter.drawText(pixmap.rect(), int(Qt.AlignmentFlag.AlignCenter), emoji)
return QIcon(pixmap)
class _Pont(QObject):
"""Fait passer un appel du thread de fond (hotkey, contrôleur) au thread
Qt principal. `etat_change` porte le nom d'état ("idle"/"recording"/
"transcribing") ; `appel_demande` porte une fonction à exécuter sans
argument (utilisée pour `controller.toggle`).
Émettre un signal Qt depuis un thread quelconque est thread-safe par
construction ; c'est la connexion en Qt.QueuedConnection (posée à la
construction de `ZonzaTray`) qui garantit que le SLOT, lui, s'exécute
sur le thread principal.
"""
etat_change = Signal(str)
appel_demande = Signal(object)
class ZonzaTray:
"""Icône de barre de tâches + menu + hotkey + contrôleur, côté Qt."""
def __init__(self, controller, systeme):
self.controller = controller
self.systeme = systeme
self._listener = None
self._pont = _Pont()
self._pont.etat_change.connect(
self._apply_title, Qt.ConnectionType.QueuedConnection
)
self._pont.appel_demande.connect(
lambda fn: fn(), Qt.ConnectionType.QueuedConnection
)
self._current_state = "idle"
self.tray = QSystemTrayIcon(_icone_pour(_STATE_TITLES["idle"]))
self.tray.setToolTip("Zonza")
self.menu = QMenu()
self.sounds_action = QAction("Sons", self.menu)
self.sounds_action.setCheckable(True)
self.sounds_action.setChecked(bool(CONFIG["sounds"]))
self.sounds_action.triggered.connect(self.toggle_sounds)
self.menu.addAction(self.sounds_action)
self.menu.addSeparator()
self.quit_action = QAction("Quitter", self.menu)
self.quit_action.triggered.connect(self.quit_app)
self.menu.addAction(self.quit_action)
self.tray.setContextMenu(self.menu)
def run_on_main(self, fn):
"""Passé au contrôleur comme `run_on_main` : marshale `fn` sur le thread Qt."""
self._pont.appel_demande.emit(fn)
def on_state(self, state):
"""Passé au contrôleur comme `on_state` : peut arriver d'un thread de fond."""
self._pont.etat_change.emit(state)
def _apply_title(self, state):
# `_current_state` n'est pas utilisé pour dessiner (l'icône vient de
# `_STATE_TITLES` juste en dessous) : il existe pour que les tests
# puissent vérifier la correspondance état → icône sans avoir à
# comparer deux QIcon pixel à pixel.
self._current_state = state
self.tray.setIcon(_icone_pour(_STATE_TITLES.get(state, "🎙️")))
def show(self):
self.tray.show()
self._start_hotkey()
self._warmup_async()
# --- actions du menu ------------------------------------------------
def toggle_sounds(self, checked):
CONFIG["sounds"] = bool(checked)
self.sounds_action.setChecked(CONFIG["sounds"])
def quit_app(self):
if self._listener is not None:
self._listener.stop()
QApplication.instance().quit()
# --- hotkey globale ---------------------------------------------------
def _start_hotkey(self):
# pynput exécute ce callback sur SON thread d'écoute, pas le thread
# Qt principal. toggle() touche l'overlay (Qt) → marshalage requis,
# même raison que app.py avec AppHelper.callAfter.
def on_activate():
self.run_on_main(self.controller.toggle)
self._listener = keyboard.GlobalHotKeys(
{self.systeme.raccourci_defaut(): on_activate}
)
self._listener.start()
# --- warmup modèle ----------------------------------------------------
def _warmup_async(self):
def run():
try:
self.controller.transcripteur.warmup()
except Exception as e:
print(f"[erreur] warmup modèle : {e}")
threading.Thread(target=run, daemon=True).start()
def _rediriger_sortie_vers_le_journal(chemin):
"""Renvoie stdout/stderr vers `chemin`.
Sur macOS, c'est `launcher/stub.c` (freopen) qui s'en charge AVANT même
de lancer Python, parce que le bundle n'a pas de terminal. Ce lanceur n'a
pas de stub équivalent sur Windows (lancé directement, pas via un
exécutable C) : c'est donc lui qui doit faire cette redirection, le plus
tôt possible dans main(), avant le premier print.
"""
parent = os.path.dirname(chemin)
if parent:
os.makedirs(parent, exist_ok=True)
journal = open(chemin, "a", buffering=1) # ligne par ligne, comme setvbuf(_IOLBF) côté stub
sys.stdout = journal
sys.stderr = journal
# Variable d'environnement permettant de faire tourner ce lanceur SANS toucher
# aux chemins réels de l'utilisateur (verrou d'instance, journal) : c'est ce
# qui rend possible de le lancer manuellement à côté d'une instance Zonza
# déjà installée, pour une vérification, sans lui disputer son verrou ni
# mélanger les journaux. Absente (cas normal) : les chemins réels du système
# sont utilisés, comme app.py.
_ENV_DOSSIER_DONNEES = "ZONZA_DOSSIER_DONNEES_TEST"
def main():
app = QApplication(sys.argv)
app.setQuitOnLastWindowClosed(False) # la fenêtre, c'est l'icône de la barre
systeme = choisir_systeme()
dossier_test = os.environ.get(_ENV_DOSSIER_DONNEES)
if dossier_test:
chemin_verrou = os.path.join(dossier_test, "instance.lock")
chemin_log = os.path.join(dossier_test, "Zonza.log")
else:
chemin_verrou = None # None -> systeme.dossier_donnees(), le chemin réel
chemin_log = systeme.chemin_log()
_rediriger_sortie_vers_le_journal(chemin_log)
lock = acquerir_verrou(systeme, chemin_verrou)
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
# Dépendance circulaire : le contrôleur a besoin de run_on_main/on_state
# du tray, mais le tray a besoin du contrôleur pour être construit. On la
# casse avec une case remplie APRÈS coup : les deux fermetures ci-dessous
# ne lisent `case["tray"]` qu'au moment de l'appel, jamais à la
# construction — ce qui laisse le temps de la remplir.
case = {}
assemblage = construire(
run_on_main=lambda fn: case["tray"].run_on_main(fn),
on_state=lambda s: case["tray"].on_state(s),
)
tray = ZonzaTray(assemblage["controller"], assemblage["systeme"])
case["tray"] = tray
tray.show()
main._tray = tray # référence gardée pour toute la vie du processus
sys.exit(app.exec())
if __name__ == "__main__":
main()
+32 -2
View File
@@ -12,8 +12,17 @@ def nom_moteur_pour(plateforme, machine, cuda_disponible):
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`."""
def detecter_moteur(forcage=None):
"""Le moteur forcé par `CONFIG["moteur"]` s'il est renseigné ; sinon on
interroge la machine et on délègue la décision à `nom_moteur_pour`."""
if forcage is None:
from core import CONFIG
forcage = CONFIG.get("moteur")
return forcage or _moteur_local()
def _moteur_local():
"""Le moteur qui tourne sur CETTE machine, sans réseau."""
return nom_moteur_pour(sys.platform, platform.machine(), _cuda_disponible())
@@ -59,4 +68,25 @@ def choisir_transcripteur(nom, modele, langue, amorce):
if nom == "mlx":
from moteur.mlx import TranscripteurMLX
return TranscripteurMLX(modele, langue, amorce)
if nom in ("faster-cpu", "faster-gpu"):
# C'est ICI, et nulle part dans TranscripteurFasterWhisper, que le
# peripherique et le type de calcul sont decides : la classe reste
# testable sans jamais sonder de vrai materiel.
from moteur.faster import TranscripteurFasterWhisper
peripherique = "cuda" if nom == "faster-gpu" else "cpu"
type_calcul = "float16" if nom == "faster-gpu" else "int8"
return TranscripteurFasterWhisper(modele, langue, amorce, peripherique, type_calcul)
if nom in ("gemini", "openrouter"):
# Toujours enveloppe d'un repli : la dictee ne doit jamais etre perdue
# parce que le reseau ou l'API tombe (cf moteur/gemini.py).
from core import modele_pour_moteur
from moteur.gemini import TranscripteurAvecRepli, TranscripteurGemini
from moteur.openrouter import TranscripteurOpenRouter
classe = TranscripteurGemini if nom == "gemini" else TranscripteurOpenRouter
local = _moteur_local()
return TranscripteurAvecRepli(
classe(modele, langue, amorce),
fabriquer_secours=lambda: choisir_transcripteur(
local, modele_pour_moteur(local), langue, amorce),
)
raise NotImplementedError(f"Moteur non pris en charge à cette étape : {nom}")
+48
View File
@@ -0,0 +1,48 @@
"""Transcription par faster-whisper (CTranslate2) — tourne partout : Mac Intel,
Windows avec ou sans GPU. Complement de moteur/mlx.py, qui lui est reserve a
l'Apple Silicon.
"""
import numpy as np
from core import clean_transcript
class TranscripteurFasterWhisper:
"""Implementation du contrat Transcripteur pour faster-whisper.
Le peripherique (`cpu`/`cuda`) et le type de calcul (`int8`/`float16`...)
sont recus en parametres plutot que devines ici : cette classe ne sonde
jamais le materiel elle-meme. C'est `moteur/__init__.py::choisir_transcripteur`
qui tranche, d'apres `_cuda_disponible` — garder la decision hors de cette
classe la rend testable sans machine CUDA ni GPU.
"""
def __init__(self, modele, langue, amorce, peripherique, type_calcul):
self.modele = modele
self.langue = langue
self.amorce = amorce
self.peripherique = peripherique
self.type_calcul = type_calcul
self._instance = None # modele CTranslate2 charge paresseusement
def warmup(self):
"""Charge le modele une fois, sur une seconde de silence."""
self.transcrire(np.zeros(16000, dtype="float32"))
def transcrire(self, audio):
# Import differe dans la methode, jamais au niveau module : sinon toute
# machine sans faster-whisper installe casserait a l'import de ce
# fichier, y compris pour les tests qui n'exercent que la fabrique
# (meme raison que moteur/mlx.py pour mlx_whisper).
from faster_whisper import WhisperModel
if self._instance is None:
self._instance = WhisperModel(
self.modele, device=self.peripherique, compute_type=self.type_calcul
)
segments, _info = self._instance.transcribe(
audio, language=self.langue, initial_prompt=self.amorce
)
# faster-whisper rend un generateur de segments, pas un texte unique :
# il faut le consommer entierement avant de pouvoir le joindre.
return clean_transcript("".join(segment.text for segment in segments))
+197
View File
@@ -0,0 +1,197 @@
"""Transcription par l'API Gemini — l'audio part chez Google.
Contrairement à `mlx` et `faster`, ce moteur N'EST PAS local : chaque dictée
est envoyée à `generativelanguage.googleapis.com`. Il est donc toujours
enveloppé d'un `TranscripteurAvecRepli` : réseau coupé, clé refusée, quota
épuisé ou réponse vide, la dictée repasse par le moteur local au lieu d'être
perdue.
Aucune dépendance nouvelle : l'appel HTTP passe par `urllib` de la bibliothèque
standard, et la clé se lit dans le trousseau macOS (`security`) ou, à défaut,
dans la variable d'environnement GEMINI_API_KEY.
"""
import base64
import io
import json
import os
import subprocess
import sys
import urllib.error
import urllib.request
import wave
import numpy as np
from core import clean_transcript
URL_API = "https://generativelanguage.googleapis.com/v1beta/models/{modele}:generateContent"
SERVICE_TROUSSEAU = "zonza-gemini"
# Borné sous `duree_max_transcription_s` (20 s) : si l'API ne répond pas à
# temps, il doit rester de quoi basculer sur le moteur local avant que la
# garde du halo ne tombe. 15 s couvrent les 9 s mesurées de gemini-3.8-flash
# sur une dictée de 94 s.
DELAI_RESEAU_S = 15.0
CONSIGNE = (
"Tu es un moteur de dictée. Transcris mot pour mot ce que dit la personne, "
"dans la langue parlée ({langue}), avec la ponctuation et les majuscules "
"d'un texte écrit. Ne réponds pas, ne résume pas, ne commente pas, "
"n'ajoute rien : rends UNIQUEMENT le texte dicté. Si l'enregistrement ne "
"contient aucune parole, rends une chaîne vide. Termes techniques à "
"orthographier exactement : {amorce}"
)
class ErreurGemini(RuntimeError):
"""Tout échec côté Gemini : clé, réseau, quota, réponse inexploitable."""
def audio_en_wav(audio, sample_rate):
"""Encode un signal float32 mono dans [-1, 1] en WAV PCM 16 bits."""
pcm = (np.clip(audio, -1.0, 1.0) * 32767).astype("<i2")
tampon = io.BytesIO()
with wave.open(tampon, "wb") as w:
w.setnchannels(1)
w.setsampwidth(2)
w.setframerate(sample_rate)
w.writeframes(pcm.tobytes())
return tampon.getvalue()
def _lire_trousseau(service=SERVICE_TROUSSEAU):
"""Clé rangée par `security add-generic-password -s <service> ...`.
Le trousseau, et non `~/.zshrc` : Zonza est lancée par le Finder, qui ne
lit jamais la configuration du shell — une clé exportée dans `.zshrc`
serait invisible pour l'application.
"""
if sys.platform != "darwin":
return None
try:
sortie = subprocess.run(
["security", "find-generic-password", "-s", service, "-w"],
capture_output=True, text=True, timeout=5,
)
except Exception:
return None
if sortie.returncode != 0:
return None
return sortie.stdout.strip() or None
def lire_cle(trousseau=None, environ=None, service=SERVICE_TROUSSEAU,
variable="GEMINI_API_KEY"):
"""La clé du trousseau, sinon la variable d'environnement, sinon None."""
trousseau = trousseau or (lambda: _lire_trousseau(service))
environ = os.environ if environ is None else environ
return trousseau() or environ.get(variable) or None
def _envoyer(url, cle, corps, delai_s):
"""La clé part en en-tête, jamais dans l'URL (qui finit en journal)."""
return poster_json(url, {"x-goog-api-key": cle}, corps, delai_s)
def poster_json(url, entetes, corps, delai_s):
"""POST JSON ; toute erreur HTTP ou réseau devient une ErreurGemini."""
requete = urllib.request.Request(
url,
data=json.dumps(corps).encode("utf-8"),
headers={"Content-Type": "application/json", **entetes},
method="POST",
)
try:
with urllib.request.urlopen(requete, timeout=delai_s) as reponse:
return json.loads(reponse.read())
except urllib.error.HTTPError as e:
brut = e.read().decode("utf-8", "replace")
try:
detail = json.loads(brut)["error"]["message"]
except Exception:
detail = " ".join(brut.split())[:300]
raise ErreurGemini(f"HTTP {e.code} : {detail}") from e
except (urllib.error.URLError, TimeoutError, OSError) as e:
raise ErreurGemini(f"réseau : {e}") from e
class TranscripteurGemini:
"""Implémentation du contrat Transcripteur via l'API Gemini."""
def __init__(self, modele, langue, amorce, sample_rate=16000,
lire_cle=lire_cle, envoyer=_envoyer):
self.modele = modele
self.langue = langue
self.amorce = amorce or ""
self.sample_rate = sample_rate
self._lire_cle = lire_cle
self._envoyer = envoyer
def _cle(self):
cle = self._lire_cle()
if not cle:
raise ErreurGemini(
"aucune clé Gemini : lancer `security add-generic-password "
f"-s {SERVICE_TROUSSEAU} -a zonza -w` puis relancer Zonza"
)
return cle
def warmup(self):
"""Rien à charger : on vérifie seulement la clé, sans appel payant."""
self._cle()
def transcrire(self, audio):
corps = {
"system_instruction": {"parts": [{"text": CONSIGNE.format(
langue=self.langue, amorce=self.amorce)}]},
"contents": [{"role": "user", "parts": [
{"inline_data": {
"mime_type": "audio/wav",
"data": base64.b64encode(audio_en_wav(audio, self.sample_rate)).decode("ascii"),
}},
{"text": "Transcris cet enregistrement."},
]}],
}
reponse = self._envoyer(URL_API.format(modele=self.modele), self._cle(), corps, DELAI_RESEAU_S)
candidats = reponse.get("candidates") or []
if not candidats:
raise ErreurGemini(f"réponse sans transcription : {json.dumps(reponse)[:300]}")
parts = (candidats[0].get("content") or {}).get("parts") or []
return clean_transcript("".join(p.get("text", "") for p in parts))
class TranscripteurAvecRepli:
"""Essaie le principal ; au moindre échec, bascule sur le moteur local.
Le moteur local n'est fabriqué qu'au premier besoin : tant que Gemini
répond, on n'immobilise pas 500 Mo de modèle en mémoire pour rien.
"""
def __init__(self, principal, fabriquer_secours):
self.principal = principal
self._fabriquer_secours = fabriquer_secours
self._secours = None
@property
def secours_charge(self):
return self._secours is not None
def _secours_pret(self):
if self._secours is None:
self._secours = self._fabriquer_secours()
self._secours.warmup()
return self._secours
def warmup(self):
try:
self.principal.warmup()
except Exception as e:
print(f"[api] indisponible ({e}) — moteur local chargé en secours")
self._secours_pret()
def transcrire(self, audio):
try:
return self.principal.transcrire(audio)
except Exception as e:
print(f"[api] échec ({e}) — transcription locale")
return self._secours_pret().transcrire(audio)
+79
View File
@@ -0,0 +1,79 @@
"""Transcription par OpenRouter — même famille Gemini, autre guichet.
Comme `moteur/gemini.py`, ce moteur N'EST PAS local : l'audio part chez
OpenRouter, qui le relaie à Google. L'encodage, la consigne, le repli local et
la lecture de la clé sont partagés avec le moteur Gemini direct ; seuls
l'adresse, l'authentification et le format de requête (compatible OpenAI,
partie `input_audio`) diffèrent.
"""
import base64
import json
from core import clean_transcript
from moteur.gemini import (
CONSIGNE, DELAI_RESEAU_S, ErreurGemini, audio_en_wav, lire_cle, poster_json,
)
URL_API = "https://openrouter.ai/api/v1/chat/completions"
SERVICE_TROUSSEAU = "zonza-openrouter"
def lire_cle_openrouter(trousseau=None, environ=None):
"""La clé du trousseau `zonza-openrouter`, sinon OPENROUTER_API_KEY."""
return lire_cle(trousseau, environ, service=SERVICE_TROUSSEAU,
variable="OPENROUTER_API_KEY")
def _envoyer(url, entetes, corps, delai_s):
return poster_json(url, entetes, corps, delai_s)
class TranscripteurOpenRouter:
"""Implémentation du contrat Transcripteur via OpenRouter."""
def __init__(self, modele, langue, amorce, sample_rate=16000,
lire_cle=lire_cle_openrouter, envoyer=_envoyer):
self.modele = modele
self.langue = langue
self.amorce = amorce or ""
self.sample_rate = sample_rate
self._lire_cle = lire_cle
self._envoyer = envoyer
def _cle(self):
cle = self._lire_cle()
if not cle:
raise ErreurGemini(
"aucune clé OpenRouter : lancer `security add-generic-password "
f"-s {SERVICE_TROUSSEAU} -a zonza -w` puis relancer Zonza"
)
return cle
def warmup(self):
"""Rien à charger : on vérifie seulement la clé, sans appel payant."""
self._cle()
def transcrire(self, audio):
wav = base64.b64encode(audio_en_wav(audio, self.sample_rate)).decode("ascii")
corps = {
"model": self.modele,
"messages": [
{"role": "system", "content": CONSIGNE.format(
langue=self.langue, amorce=self.amorce)},
{"role": "user", "content": [
{"type": "input_audio", "input_audio": {"data": wav, "format": "wav"}},
{"type": "text", "text": "Transcris cet enregistrement."},
]},
],
}
entetes = {"Authorization": f"Bearer {self._cle()}", "X-Title": "Zonza"}
reponse = self._envoyer(URL_API, entetes, corps, DELAI_RESEAU_S)
# OpenRouter peut rendre un 200 portant un objet `error` quand le
# fournisseur en aval tombe : ce n'est PAS une transcription vide.
if "error" in reponse:
erreur = reponse["error"]
raise ErreurGemini(f"OpenRouter : {erreur.get('message', erreur)}")
choix = reponse.get("choices") or []
if not choix:
raise ErreurGemini(f"réponse sans transcription : {json.dumps(reponse)[:300]}")
return clean_transcript(choix[0].get("message", {}).get("content") or "")
+26 -1
View File
@@ -1,7 +1,32 @@
mlx-whisper>=0.4.3
# Zonza — dependances, avec marqueurs de plateforme.
#
# `pip install -r requirements.txt` n'installe QUE ce qui sert a la machine :
# MLX et pyobjc sont propres a macOS, PySide6 sert l'overlay Qt. Sans ces
# marqueurs, l'installation echouerait sur Windows (mlx-whisper n'y existe pas).
# --- transcription -------------------------------------------------------
# Deux moteurs, car aucun ne couvre a lui seul les plateformes visees
# (cf moteur/__init__.py::nom_moteur_pour).
mlx-whisper>=0.4.3; sys_platform == "darwin" and platform_machine == "arm64"
faster-whisper>=1.0
# --- interface -----------------------------------------------------------
# Cocoa sert l'overlay macOS. Ces paquets n'etaient PAS declares ici : une
# installation neuve sur un autre Mac aurait produit une Zonza qui ne demarre pas.
pyobjc-core>=12.2; sys_platform == "darwin"
pyobjc-framework-Cocoa>=12.2; sys_platform == "darwin"
pyobjc-framework-Quartz>=12.2; sys_platform == "darwin"
pyobjc-framework-ApplicationServices>=12.2; sys_platform == "darwin"
# Qt sert l'overlay partout ailleurs. Installe aussi sur macOS : la suite de
# tests y eprouve les DEUX interfaces, c'est ce qui protege le portage.
PySide6>=6.11
# --- capture, collage, calcul --------------------------------------------
sounddevice>=0.5.5
numpy>=2.4
pynput>=1.8.2
pyperclip>=1.11
pillow>=11
# --- developpement -------------------------------------------------------
pytest>=8.0
+3
View File
@@ -12,4 +12,7 @@ def choisir_systeme(plateforme=None):
if plateforme.startswith("darwin"):
from systeme.macos import SystemeMacOS
return SystemeMacOS()
if plateforme.startswith("win32"):
from systeme.windows import SystemeWindows
return SystemeWindows()
raise NotImplementedError(f"Plateforme non prise en charge : {plateforme}")
+31 -4
View File
@@ -9,8 +9,24 @@ _SONS = {
"error": "/System/Library/Sounds/Basso.aiff",
}
# Masque des modificateurs (ctrl, shift, alt, cmd) dans les drapeaux Quartz.
_MASQUE_MODIFICATEURS = 0x000F0000
def _masque_modificateurs():
"""Masque des modificateurs, COMPOSE depuis les constantes nommees de Quartz.
Il etait ecrit 0x000F0000 a la main : ce masque couvrait Shift, Control et Alt
mais PAS Command (0x00100000), pourtant la touche du raccourci de Zonza. Un
masque calcule de tete se trompe en silence ; compose depuis les constantes,
il ne peut plus. Repli sur la valeur complete si Quartz est indisponible.
"""
try:
from Quartz import (kCGEventFlagMaskAlternate, kCGEventFlagMaskCommand,
kCGEventFlagMaskControl, kCGEventFlagMaskShift)
return (kCGEventFlagMaskShift | kCGEventFlagMaskControl
| kCGEventFlagMaskAlternate | kCGEventFlagMaskCommand)
except Exception:
return 0x001E0000 # sans le verrouillage majuscule : il resterait actif en permanence
_MASQUE_MODIFICATEURS = _masque_modificateurs()
class SystemeMacOS:
@@ -25,10 +41,21 @@ class SystemeMacOS:
except Exception:
pass # le son ne doit jamais casser la dictée
def __init__(self, sonde_modificateurs=None):
"""`sonde_modificateurs` est injectable : sans cela, tout test de cette
methode lirait le CLAVIER PHYSIQUE, et son resultat dependrait des mains
de l'utilisateur au moment ou la suite tourne. Meme forme que
SystemeWindows, qui recoit deja son clavier."""
self._sonde_modificateurs = sonde_modificateurs or self._drapeaux_reels
@staticmethod
def _drapeaux_reels():
from Quartz import CGEventSourceFlagsState
return CGEventSourceFlagsState(1)
def modificateurs_enfonces(self):
try:
from Quartz import CGEventSourceFlagsState
return bool(CGEventSourceFlagsState(1) & _MASQUE_MODIFICATEURS)
return bool(self._sonde_modificateurs() & _MASQUE_MODIFICATEURS)
except Exception:
return False # dans le doute, ne jamais bloquer le collage
+120
View File
@@ -0,0 +1,120 @@
"""Appels propres à Windows. Rien ici ne doit être importé directement par le noyau.
Miroir de `systeme/macos.py`, mais avec une contrainte que macOS n'a pas :
`winsound`, `msvcrt` et `ctypes.windll` n'EXISTENT PAS sur macOS. Un import en
tête de fichier rendrait ce module impossible à charger — donc impossible à
tester — depuis la machine de développement (aucune machine Windows n'est
disponible pour ce portage). Les trois dépendances sont donc INJECTABLES :
`jouer`, `clavier`, `verrouiller`. Par défaut (None), elles se résolvent
paresseusement vers l'API Windows réelle au moment de l'appel, jamais à
l'import. Les tests injectent des doublures et n'exécutent donc jamais une
ligne de code Windows — ils éprouvent la LOGIQUE de cette classe (mapping des
sons, masque des touches, construction des chemins, séquence d'acquisition du
verrou), pas le comportement réel de winsound/msvcrt/user32, qu'aucune
machine ici ne peut vérifier.
"""
import os
# Alias système Windows joués par winsound.PlaySound(alias, SND_ALIAS | SND_ASYNC).
_SONS = {
"start": "SystemAsterisk",
"done": "SystemExclamation",
"error": "SystemHand",
}
# Virtual-key codes Windows des cinq touches modificatrices à surveiller.
# ctrl, shift, alt (menu), windows gauche, windows droite.
_TOUCHES_MODIFICATRICES = (0x11, 0x10, 0x12, 0x5B, 0x5C)
# Dans le retour de GetAsyncKeyState, le bit de poids fort indique que la
# touche est enfoncée EN CE MOMENT (les bits bas concernent un état passé
# qui ne nous intéresse pas ici).
_BIT_ENFONCEE = 0x8000
class SystemeWindows:
"""Implémentation du contrat Systeme pour Windows.
jouer(alias) : joue un son. clavier(vk) : rend l'état d'une touche.
verrouiller(fichier) : pose un verrou exclusif non bloquant sur `fichier`,
lève OSError s'il est déjà pris. Les trois par défaut à None : résolus
paresseusement vers l'API Windows réelle au premier usage.
"""
def __init__(self, jouer=None, clavier=None, verrouiller=None):
self._jouer = jouer
self._clavier = clavier
self._verrouiller = verrouiller
# --- son ------------------------------------------------------------
def jouer_son(self, nom):
alias = _SONS.get(nom)
if not alias:
return
try:
(self._jouer or self._jouer_reel)(alias)
except Exception:
pass # le son ne doit jamais casser la dictée (même raison qu'en macOS)
@staticmethod
def _jouer_reel(alias):
import winsound
winsound.PlaySound(alias, winsound.SND_ALIAS | winsound.SND_ASYNC)
# --- clavier ----------------------------------------------------------
def modificateurs_enfonces(self):
sonder = self._clavier or self._clavier_reel
try:
return any(sonder(vk) & _BIT_ENFONCEE for vk in _TOUCHES_MODIFICATRICES)
except Exception:
return False # dans le doute, ne jamais bloquer le collage (idem macOS)
@staticmethod
def _clavier_reel(vk):
import ctypes
return ctypes.windll.user32.GetAsyncKeyState(vk)
# --- raccourci --------------------------------------------------------
def raccourci_defaut(self):
# <cmd> n'existe pas sur Windows : pas d'équivalent à la touche Commande.
return "<ctrl>+<alt>+d"
# --- chemins ------------------------------------------------------------
def chemin_log(self):
return os.path.join(self._dossier_local_appdata(), "Zonza", "Zonza.log")
def dossier_donnees(self):
return os.path.join(self._dossier_local_appdata(), "Zonza")
@staticmethod
def _dossier_local_appdata():
# os.environ est standard : lisible (et falsifiable par les tests via
# monkeypatch.setenv) depuis n'importe quelle plateforme, contrairement
# à winsound/ctypes.windll/msvcrt.
return os.environ.get("LOCALAPPDATA", os.path.expanduser("~"))
# --- verrou d'instance --------------------------------------------------
def acquerir_verrou_instance(self, chemin):
parent = os.path.dirname(chemin)
if parent:
os.makedirs(parent, exist_ok=True)
fichier = open(chemin, "a+")
fichier.seek(0)
# msvcrt.locking verrouille un OCTET à partir de la position courante ;
# il exige que cet octet existe déjà dans le fichier, d'où cette garde.
if os.fstat(fichier.fileno()).st_size == 0:
fichier.write("0")
fichier.flush()
fichier.seek(0)
try:
(self._verrouiller or self._verrouiller_reel)(fichier)
except OSError:
fichier.close()
return None
return fichier
@staticmethod
def _verrouiller_reel(fichier):
import msvcrt
fichier.seek(0)
msvcrt.locking(fichier.fileno(), msvcrt.LK_NBLCK, 1)
+122
View File
@@ -0,0 +1,122 @@
"""L'assemblage commun aux deux lanceurs — sans jamais ouvrir de fenêtre ni de micro.
Le transcripteur réel charge un modèle (opération lourde, parfois absente sur
la machine de test) : on injecte donc toujours un `choisir_transcripteur` et
un `detecter_moteur` doublés. L'overlay réel (Cocoa ou Qt), lui, est fait pour
être construit sans effet visible tant que `show()` n'est pas appelé — les
tests le laissent donc réel, et vérifient justement qu'aucune fenêtre ne
s'affiche.
"""
import os
import tempfile
from assemblage import acquerir_verrou, construire
from controller import ZonzaController
from systeme import choisir_systeme
class _TranscripteurFactice:
def __init__(self, modele, langue, amorce):
self.modele = modele
self.langue = langue
self.amorce = amorce
def warmup(self):
pass
def transcrire(self, audio):
return "texte factice"
def _choisir_transcripteur_factice(nom, modele, langue, amorce):
return _TranscripteurFactice(modele, langue, amorce)
def _tmp_verrou():
return os.path.join(tempfile.mkdtemp(), "instance.lock")
# --- construire() ------------------------------------------------------------
def test_construire_rend_un_controleur_cable_sur_les_bons_objets():
resultat = construire(
plateforme="darwin",
detecter_moteur=lambda: "mlx",
choisir_transcripteur=_choisir_transcripteur_factice,
)
assert isinstance(resultat["controller"], ZonzaController)
# le contrôleur pointe vers CE système, CE transcripteur, CET overlay —
# pas une copie recréée à côté.
assert resultat["controller"].systeme is resultat["systeme"]
assert resultat["controller"].transcripteur is resultat["transcripteur"]
assert resultat["controller"].pulse is resultat["overlay"]
assert isinstance(resultat["transcripteur"], _TranscripteurFactice)
def test_construire_fonctionne_aussi_pour_win32():
resultat = construire(
plateforme="win32",
detecter_moteur=lambda: "faster-cpu",
choisir_transcripteur=_choisir_transcripteur_factice,
)
from interface.qt import OverlayQt
assert isinstance(resultat["overlay"], OverlayQt)
def test_construire_transmet_run_on_main_et_on_state():
appels = []
resultat = construire(
plateforme="darwin",
run_on_main=lambda fn: appels.append(fn) or fn(),
on_state=lambda s: appels.append(s),
detecter_moteur=lambda: "mlx",
choisir_transcripteur=_choisir_transcripteur_factice,
)
resultat["controller"].on_state("recording")
assert "recording" in appels
def test_construire_ne_leve_aucune_erreur_de_config():
# validate_config(CONFIG) est appelée à l'intérieur : si elle levait, le
# test échouerait ici plutôt qu'au premier lancement réel.
construire(
plateforme="darwin",
detecter_moteur=lambda: "mlx",
choisir_transcripteur=_choisir_transcripteur_factice,
)
# --- acquerir_verrou() --------------------------------------------------------
def test_acquerir_verrou_rend_un_handle_pour_la_premiere_instance():
systeme = choisir_systeme()
handle = acquerir_verrou(systeme, _tmp_verrou())
assert handle is not None
def test_acquerir_verrou_refuse_une_seconde_instance():
systeme = choisir_systeme()
chemin = _tmp_verrou()
premier = acquerir_verrou(systeme, chemin)
assert premier is not None
assert acquerir_verrou(systeme, chemin) is None
def test_acquerir_verrou_sans_chemin_utilise_le_dossier_de_donnees_du_systeme():
# Pas de deuxième prise ici (on toucherait le verrou réel de l'utilisateur) :
# on vérifie seulement que le chemin par défaut est bien dérivé du système.
class _SystemeSonde:
def __init__(self):
self.chemin_demande = None
def dossier_donnees(self):
return "/tmp/zonza-sonde"
def acquerir_verrou_instance(self, chemin):
self.chemin_demande = chemin
return "handle-factice"
sonde = _SystemeSonde()
resultat = acquerir_verrou(sonde)
assert resultat == "handle-factice"
assert sonde.chemin_demande == "/tmp/zonza-sonde/instance.lock"
+326
View File
@@ -0,0 +1,326 @@
"""`zonza --diagnostic` : enchaînement des épreuves et mise en forme.
Chaque épreuve qui toucherait normalement du matériel réel (micro, moteur de
transcription, presse-papier, verrou) est ici exercée avec une doublure
injectée — ces tests éprouvent l'ORCHESTRATION (ordre, propagation d'un échec,
non-invention d'un résultat), pas le matériel lui-même.
"""
import numpy as np
import pytest
from diagnostic import (
ResultatEpreuve,
diagnostiquer,
epreuve_micro,
epreuve_moteur,
epreuve_plateforme,
epreuve_presse_papier,
epreuve_raccourci,
epreuve_transcription,
epreuve_verrou,
formater_resultat,
mettre_en_forme,
)
# --- epreuve_plateforme ------------------------------------------------------
def test_epreuve_plateforme_rapporte_cuda_present():
r = epreuve_plateforme(plateforme="win32", machine="AMD64", cuda_disponible=lambda: True)
assert r.reussite is True
assert r.detail == "win32 / AMD64 / CUDA présent"
def test_epreuve_plateforme_rapporte_cuda_absent():
r = epreuve_plateforme(plateforme="darwin", machine="arm64", cuda_disponible=lambda: False)
assert "CUDA absent" in r.detail
def test_epreuve_plateforme_non_concluante_si_la_sonde_cuda_leve():
def leve():
raise RuntimeError("ctypes en échec")
r = epreuve_plateforme(plateforme="win32", machine="AMD64", cuda_disponible=leve)
assert r.reussite is None
# --- epreuve_moteur -----------------------------------------------------------
def test_epreuve_moteur_rapporte_le_nom_et_le_modele():
r, transcripteur = epreuve_moteur(
detecter_moteur=lambda: "faster-cpu",
resoudre_modele=lambda nom, forcage: "Systran/faster-whisper-small",
creer_transcripteur=lambda **kw: "un-faux-transcripteur",
)
assert r.reussite is True
assert r.detail == "faster-cpu + Systran/faster-whisper-small"
assert transcripteur == "un-faux-transcripteur"
def test_epreuve_moteur_non_concluante_si_la_detection_echoue():
def leve():
raise RuntimeError("aucun moteur compatible")
r, transcripteur = epreuve_moteur(detecter_moteur=leve)
assert r.reussite is None
assert transcripteur is None
# --- epreuve_micro -------------------------------------------------------------
class _FauxFlux:
def __init__(self, callback, n_blocs):
self._callback = callback
self._n_blocs = n_blocs
def start(self):
for _ in range(self._n_blocs):
self._callback(np.zeros(10, dtype="float32"), 10, None, None)
def stop(self):
pass
def close(self):
pass
def test_epreuve_micro_compte_les_blocs_recus():
def ouvrir_flux(**kw):
return _FauxFlux(kw["callback"], n_blocs=7)
r = epreuve_micro(ouvrir_flux=ouvrir_flux, dormir=lambda s: None)
assert r.reussite is True
assert r.detail == "7 blocs reçus en 1 s"
def test_epreuve_micro_echoue_si_aucun_bloc_recu():
def ouvrir_flux(**kw):
return _FauxFlux(kw["callback"], n_blocs=0)
r = epreuve_micro(ouvrir_flux=ouvrir_flux, dormir=lambda s: None)
assert r.reussite is False
def test_epreuve_micro_non_concluante_si_louverture_echoue():
def ouvrir_flux(**kw):
raise OSError("aucun périphérique")
r = epreuve_micro(ouvrir_flux=ouvrir_flux, dormir=lambda s: None)
assert r.reussite is None
def test_epreuve_micro_utilise_bien_la_duree_demandee():
appels = []
def ouvrir_flux(**kw):
return _FauxFlux(kw["callback"], n_blocs=1)
def dormir(s):
appels.append(s)
epreuve_micro(ouvrir_flux=ouvrir_flux, duree_s=2.5, dormir=dormir)
assert appels == [2.5]
# --- epreuve_transcription -----------------------------------------------------
class _FauxTranscripteur:
def __init__(self, texte="bonjour", leve=False):
self._texte = texte
self._leve = leve
def transcrire(self, audio):
if self._leve:
raise RuntimeError("modèle indisponible")
return self._texte
def test_epreuve_transcription_rapporte_duree_et_texte():
horloges = iter([10.0, 10.8])
r = epreuve_transcription(
_FauxTranscripteur(texte="bonjour le monde"),
generer_parole=lambda: np.zeros(16000, dtype="float32"),
horloge=lambda: next(horloges),
)
assert r.reussite is True
assert "0.80 s" in r.detail
assert "bonjour le monde" in r.detail
def test_epreuve_transcription_retombe_sur_le_silence_si_la_synthese_echoue():
def leve():
raise RuntimeError("`say` absent")
r = epreuve_transcription(
_FauxTranscripteur(texte=""),
generer_parole=leve,
horloge=lambda: 0.0,
)
assert r.reussite is True
assert "silence" in r.detail
def test_epreuve_transcription_non_concluante_si_la_transcription_echoue():
r = epreuve_transcription(
_FauxTranscripteur(leve=True),
generer_parole=lambda: np.zeros(16000, dtype="float32"),
horloge=lambda: 0.0,
)
assert r.reussite is None
# --- epreuve_presse_papier -------------------------------------------------------
def test_epreuve_presse_papier_reussit_si_la_relecture_est_identique():
stockage = {}
r = epreuve_presse_papier(
copier=lambda t: stockage.__setitem__("v", t),
lire=lambda: stockage["v"],
)
assert r.reussite is True
def test_epreuve_presse_papier_echoue_si_la_relecture_differe():
r = epreuve_presse_papier(copier=lambda t: None, lire=lambda: "autre chose")
assert r.reussite is False
def test_epreuve_presse_papier_non_concluante_si_la_copie_leve():
def leve(t):
raise RuntimeError("presse-papier indisponible")
r = epreuve_presse_papier(copier=leve, lire=lambda: "")
assert r.reussite is None
# --- epreuve_raccourci -----------------------------------------------------------
class _FauxSysteme:
def __init__(self, raccourci="<ctrl>+<alt>+d", verrou_sequence=None):
self._raccourci = raccourci
self._verrou_sequence = iter(verrou_sequence or [])
def raccourci_defaut(self):
return self._raccourci
def acquerir_verrou_instance(self, chemin):
return next(self._verrou_sequence)
def test_epreuve_raccourci_rapporte_la_valeur_de_la_plateforme():
r = epreuve_raccourci(_FauxSysteme(raccourci="<ctrl>+<alt>+d"))
assert r.reussite is True
assert r.detail == "<ctrl>+<alt>+d"
# --- epreuve_verrou ---------------------------------------------------------------
def test_epreuve_verrou_reussit_si_la_seconde_prise_est_refusee():
r = epreuve_verrou(_FauxSysteme(verrou_sequence=["handle", None]), chemin="/tmp/x.lock")
assert r.reussite is True
def test_epreuve_verrou_echoue_si_la_seconde_prise_nest_pas_refusee():
r = epreuve_verrou(_FauxSysteme(verrou_sequence=["h1", "h2"]), chemin="/tmp/x.lock")
assert r.reussite is False
def test_epreuve_verrou_non_concluante_si_le_premier_verrou_echoue_deja():
r = epreuve_verrou(_FauxSysteme(verrou_sequence=[None, None]), chemin="/tmp/x.lock")
assert r.reussite is None
# --- diagnostiquer : enchaînement complet -----------------------------------------
def test_diagnostiquer_rend_les_sept_epreuves_dans_lordre():
resultats = diagnostiquer(
systeme=_FauxSysteme(verrou_sequence=["h", None]),
plateforme="win32", machine="AMD64", cuda_disponible=lambda: False,
detecter_moteur=lambda: "faster-cpu",
resoudre_modele=lambda nom, forcage: "modele-test",
creer_transcripteur=lambda **kw: _FauxTranscripteur(texte="ok"),
ouvrir_flux=lambda **kw: _FauxFlux(kw["callback"], n_blocs=3),
dormir=lambda s: None,
generer_parole=lambda: np.zeros(16000, dtype="float32"),
horloge=lambda: 0.0,
copier=lambda t: None,
lire=lambda: "peu importe",
)
assert [r.nom for r in resultats] == [
"Plateforme", "Moteur retenu", "Micro", "Transcription",
"Presse-papier", "Raccourci", "Verrou",
]
def test_diagnostiquer_saute_honnetement_la_transcription_si_le_moteur_echoue():
def leve():
raise RuntimeError("aucun moteur")
resultats = diagnostiquer(
systeme=_FauxSysteme(verrou_sequence=["h", None]),
detecter_moteur=leve,
ouvrir_flux=lambda **kw: _FauxFlux(kw["callback"], n_blocs=1),
dormir=lambda s: None,
copier=lambda t: None,
lire=lambda: "peu importe",
)
transcription = [r for r in resultats if r.nom == "Transcription"][0]
# jamais une réussite inventée quand rien n'a pu être mené.
assert transcription.reussite is None
assert "sautée" in transcription.detail
class _SystemeTombeToujours:
"""Doublure dont chaque méthode lève — pour prouver que rien n'est jamais
inventé même quand la couche système est totalement en panne."""
def raccourci_defaut(self):
raise RuntimeError("tout casse")
def acquerir_verrou_instance(self, chemin):
raise RuntimeError("tout casse")
def test_diagnostiquer_ne_leve_jamais_meme_si_toutes_les_epreuves_echouent():
def leve(*a, **kw):
raise RuntimeError("tout casse")
resultats = diagnostiquer(
systeme=_SystemeTombeToujours(),
plateforme="win32", machine="AMD64", cuda_disponible=leve,
detecter_moteur=leve,
ouvrir_flux=leve,
dormir=lambda s: None,
copier=leve,
lire=leve,
)
assert len(resultats) == 7
assert all(r.reussite is None for r in resultats)
# --- mise en forme -----------------------------------------------------------------
def test_formater_resultat_aligne_le_nom_et_prefixe_un_symbole():
r = ResultatEpreuve("Plateforme", True, "darwin / arm64 / CUDA absent")
ligne = formater_resultat(r)
assert ligne == "✅ Plateforme : darwin / arm64 / CUDA absent"
def test_formater_resultat_marque_lechec():
r = ResultatEpreuve("Micro", False, "0 blocs reçus en 1 s")
assert formater_resultat(r).startswith("❌ ")
def test_formater_resultat_marque_le_non_concluant():
r = ResultatEpreuve("Transcription", None, "sautée")
assert formater_resultat(r).startswith("➖ ")
def test_mettre_en_forme_ne_prend_aucun_travail_a_sa_charge():
# fabriqué à la main : aucune épreuve n'est exécutée ici.
resultats = [
ResultatEpreuve("Plateforme", True, "darwin / arm64 / CUDA absent"),
ResultatEpreuve("Moteur retenu", True, "mlx + mlx-community/whisper-medium"),
ResultatEpreuve("Micro", False, "0 blocs reçus en 1 s"),
ResultatEpreuve("Transcription", None, "sautée"),
]
rapport = mettre_en_forme(resultats)
for r in resultats:
assert formater_resultat(r) in rapport
assert "1 en échec" in rapport
assert "1 non concluante" in rapport
def test_mettre_en_forme_sur_zero_resultat_ne_leve_pas():
assert "0/0" in mettre_en_forme([])
+170
View File
@@ -0,0 +1,170 @@
"""Moteur Gemini : la transcription part chez Google, avec repli local.
Aucun test ne touche le réseau : l'envoi HTTP et la lecture de la clé sont
injectés. Ce qu'on éprouve ici, c'est la forme de la requête, la lecture de
la réponse, et surtout le repli — une dictée ne doit JAMAIS être perdue parce
que le réseau ou l'API tombe.
"""
import base64
import io
import wave
import numpy as np
import pytest
from moteur import choisir_transcripteur, detecter_moteur
from moteur.gemini import (
ErreurGemini, TranscripteurAvecRepli, TranscripteurGemini, audio_en_wav,
lire_cle,
)
def _reponse(texte):
return {"candidates": [{"content": {"parts": [{"text": texte}]}}]}
def _transcripteur(envoyer, cle="cle-test"):
return TranscripteurGemini(
"gemini-x", "fr", "Vocabulaire : Gitea, iCloud.",
lire_cle=lambda: cle, envoyer=envoyer,
)
# --- encodage audio ---------------------------------------------------------
def test_l_audio_devient_un_wav_16_bits_mono_a_16_khz():
audio = np.array([0.0, 0.5, -0.5, 1.0], dtype="float32")
with wave.open(io.BytesIO(audio_en_wav(audio, 16000))) as w:
assert (w.getnchannels(), w.getsampwidth(), w.getframerate()) == (1, 2, 16000)
echantillons = np.frombuffer(w.readframes(4), dtype="<i2")
assert list(echantillons) == [0, 16383, -16383, 32767]
def test_un_audio_sature_est_borne_au_lieu_de_deborder():
with wave.open(io.BytesIO(audio_en_wav(np.array([3.0, -3.0], dtype="float32"), 16000))) as w:
assert list(np.frombuffer(w.readframes(2), dtype="<i2")) == [32767, -32767]
# --- requête et réponse -----------------------------------------------------
def test_la_requete_porte_l_audio_le_modele_et_le_vocabulaire():
vu = {}
def envoyer(url, cle, corps, delai_s):
vu.update(url=url, cle=cle, corps=corps)
return _reponse("Bonjour.")
_transcripteur(envoyer).transcrire(np.zeros(1600, dtype="float32"))
assert vu["url"].endswith("/models/gemini-x:generateContent")
assert vu["cle"] == "cle-test"
parts = vu["corps"]["contents"][0]["parts"]
audio = next(p["inline_data"] for p in parts if "inline_data" in p)
assert audio["mime_type"] == "audio/wav"
assert base64.b64decode(audio["data"])[:4] == b"RIFF"
consigne = vu["corps"]["system_instruction"]["parts"][0]["text"]
assert "Gitea" in consigne and "iCloud" in consigne
def test_la_cle_ne_part_jamais_dans_l_url():
vu = {}
_transcripteur(lambda url, cle, corps, delai_s: vu.setdefault("url", url) and _reponse("x")) \
.transcrire(np.zeros(1600, dtype="float32"))
assert "cle-test" not in vu["url"]
def test_le_texte_rendu_est_nettoye_et_les_morceaux_joints():
reponse = {"candidates": [{"content": {"parts": [{"text": "Je pousse "}, {"text": "sur Gitea. "}]}}]}
t = _transcripteur(lambda *a: reponse)
assert t.transcrire(np.zeros(1600, dtype="float32")) == "Je pousse sur Gitea."
def test_une_reponse_sans_candidat_est_une_erreur_explicite():
t = _transcripteur(lambda *a: {"promptFeedback": {"blockReason": "OTHER"}})
with pytest.raises(ErreurGemini) as e:
t.transcrire(np.zeros(1600, dtype="float32"))
assert "OTHER" in str(e.value)
def test_une_cle_absente_est_signalee_des_le_warmup():
t = _transcripteur(lambda *a: _reponse("x"), cle=None)
with pytest.raises(ErreurGemini) as e:
t.warmup()
assert "zonza-gemini" in str(e.value)
def test_le_warmup_ne_consomme_aucun_appel_payant():
appels = []
_transcripteur(lambda *a: appels.append(a) or _reponse("x")).warmup()
assert appels == []
# --- lecture de la clé ------------------------------------------------------
def test_la_cle_du_trousseau_prime_sur_l_environnement():
assert lire_cle(trousseau=lambda: "du-trousseau", environ={"GEMINI_API_KEY": "env"}) == "du-trousseau"
def test_sans_trousseau_on_se_rabat_sur_l_environnement():
assert lire_cle(trousseau=lambda: None, environ={"GEMINI_API_KEY": "env"}) == "env"
def test_sans_aucune_cle_on_rend_none():
assert lire_cle(trousseau=lambda: None, environ={}) is None
# --- repli local ------------------------------------------------------------
class _Local:
def __init__(self):
self.appels = 0
def warmup(self):
pass
def transcrire(self, audio):
self.appels += 1
return "texte local"
def test_si_gemini_tombe_la_dictee_passe_par_le_moteur_local():
def en_panne(audio):
raise ErreurGemini("réseau coupé")
principal = type("P", (), {"warmup": lambda s: None, "transcrire": lambda s, a: en_panne(a)})()
local = _Local()
t = TranscripteurAvecRepli(principal, fabriquer_secours=lambda: local)
assert t.transcrire(np.zeros(10, dtype="float32")) == "texte local"
def test_le_moteur_local_n_est_charge_qu_au_premier_repli():
fabrications = []
principal = type("P", (), {"warmup": lambda s: None, "transcrire": lambda s, a: "de gemini"})()
t = TranscripteurAvecRepli(principal, fabriquer_secours=lambda: fabrications.append(1) or _Local())
t.warmup()
assert t.transcrire(np.zeros(10, dtype="float32")) == "de gemini"
assert fabrications == []
def test_une_cle_absente_au_warmup_ne_bloque_pas_le_demarrage():
def sans_cle(self):
raise ErreurGemini("aucune clé")
principal = type("P", (), {"warmup": sans_cle, "transcrire": lambda s, a: "x"})()
local = _Local()
t = TranscripteurAvecRepli(principal, fabriquer_secours=lambda: local)
t.warmup() # ne lève pas : le moteur local prendra le relais
assert t.secours_charge
# --- sélection du moteur ----------------------------------------------------
def test_le_forcage_de_config_prime_sur_la_detection_materielle():
assert detecter_moteur(forcage="gemini") == "gemini"
def test_la_fabrique_rend_gemini_enveloppe_d_un_repli():
t = choisir_transcripteur(nom="gemini", modele="gemini-x", langue="fr", amorce="")
assert isinstance(t, TranscripteurAvecRepli)
assert isinstance(t.principal, TranscripteurGemini)
assert callable(t.warmup) and callable(t.transcrire)
+83
View File
@@ -14,3 +14,86 @@ 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)
def test_la_fabrique_rend_overlayqt_pour_win32():
classe = classe_overlay_pour("win32")
from interface.qt import OverlayQt
assert classe is OverlayQt
def test_l_overlay_qt_declare_tout_le_contrat():
classe = classe_overlay_pour("win32")
for methode in ("show", "hide", "set_level", "set_status", "rearmer_garde"):
assert callable(getattr(classe, methode, None)), f"{methode} manque"
# --- garde de duree de vie : annulation reelle a hide() ---------------------
#
# Exception assumee au principe "aucune fenetre" du haut de ce fichier : la
# seule facon honnete de prouver qu'une garde est REELLEMENT annulee (et pas
# seulement oubliee dans le champ `_garde`) est d'interroger l'etat vrai de la
# minuterie native — NSTimer.isValid() cote Cocoa, QTimer.isActive() cote Qt.
# Une doublure (fake) ne fournirait aucune garantie sur ce point : elle ne
# ferait que rejouer ce que le code de production affirme deja faire.
#
# Piege du 21/08 (halo fige) : une garde dereferencee (`_garde = None`) mais
# jamais invalidee continue de courir en coulisse, et peut refermer un overlay
# rouvert entre-temps, en pleine dictee. D'ou : on capture l'objet minuterie
# AVANT hide()/rearmer_garde(), pour verifier SON etat a lui, pas seulement le
# champ qui le reference.
from interface.cocoa import PulseWindow
from interface.qt import OverlayQt
def _garde_active_cocoa(garde):
return bool(garde.isValid())
def _garde_active_qt(garde):
return garde.isActive()
_IMPLEMENTATIONS_REELLES = [
(lambda: PulseWindow.alloc().init(), _garde_active_cocoa),
(OverlayQt, _garde_active_qt),
]
_IDS_IMPLEMENTATIONS = ["cocoa", "qt"]
@pytest.fixture(params=_IMPLEMENTATIONS_REELLES, ids=_IDS_IMPLEMENTATIONS)
def overlay_et_garde_active(request):
"""Une vraie instance d'overlay, plus la fonction qui lit l'etat reel de sa
minuterie de garde. Ferme toujours la fenetre a la fin, meme si le test echoue."""
construire, garde_active = request.param
ov = construire()
try:
yield ov, garde_active
finally:
ov.hide()
def test_show_arme_une_garde_valide(overlay_et_garde_active):
ov, garde_active = overlay_et_garde_active
ov.show()
assert ov._garde is not None
assert garde_active(ov._garde)
def test_hide_annule_reellement_la_garde(overlay_et_garde_active):
ov, garde_active = overlay_et_garde_active
ov.show()
garde = ov._garde # reference gardee AVANT hide()
ov.hide()
assert ov._garde is None
assert not garde_active(garde), "la minuterie doit etre invalidee, pas seulement oubliee"
def test_rearmer_garde_ne_laisse_qu_une_seule_garde_active(overlay_et_garde_active):
ov, garde_active = overlay_et_garde_active
ov.show()
ancienne_garde = ov._garde
ov.rearmer_garde(5.0)
assert ov._garde is not ancienne_garde
assert not garde_active(ancienne_garde), "l'ancienne garde doit etre invalidee au remplacement"
assert garde_active(ov._garde)
+124
View File
@@ -0,0 +1,124 @@
"""Ce qui est testable dans le lanceur Qt SANS boucle d'événements.
`ZonzaTray.__init__` construit le menu et l'icône de façon synchrone (aucun
`.show()`, aucune boucle) : c'est déjà éprouvable. Le marshalage par signal
Qt.QueuedConnection, lui, N'EST PAS testé ici — sa livraison dépend d'une
boucle d'événements qui tourne, ce qu'aucun test de ce fichier ne fait
tourner. On teste donc directement les méthodes qu'un slot appellerait
(`_apply_title`, `toggle_sounds`), pas le trajet signal → file → slot.
"""
import pytest
PySide6 = pytest.importorskip("PySide6")
from PySide6.QtWidgets import QApplication # noqa: E402
from core import CONFIG # noqa: E402
from lanceur_qt import _STATE_TITLES, ZonzaTray # noqa: E402
@pytest.fixture(scope="module", autouse=True)
def _app_qt():
# QIcon/QPixmap/QSystemTrayIcon exigent une QApplication déjà construite —
# sans .exec() : on ne fait jamais tourner la boucle d'événements.
app = QApplication.instance() or QApplication([])
yield app
class _ControleurFactice:
"""Ni toggle() ni transcripteur ne sont appelés tant que .show() ne l'est
pas — ce que ces tests évitent justement."""
def toggle(self):
raise AssertionError("ne doit pas être appelé sans .show()")
class _SystemeFactice:
def raccourci_defaut(self):
return "<ctrl>+<alt>+d"
@pytest.fixture
def tray():
t = ZonzaTray(_ControleurFactice(), _SystemeFactice())
yield t
# évite qu'un menu/tray fantôme traîne d'un test à l'autre (pas de .show()
# appelé ici, donc rien n'a jamais été rendu visible).
t.menu.deleteLater()
# --- construction du menu -----------------------------------------------
def test_le_menu_contient_exactement_sons_et_quitter():
t = ZonzaTray(_ControleurFactice(), _SystemeFactice())
titres = [a.text() for a in t.menu.actions() if not a.isSeparator()]
assert titres == ["Sons", "Quitter"]
def test_sons_est_une_case_a_cocher():
t = ZonzaTray(_ControleurFactice(), _SystemeFactice())
assert t.sounds_action.isCheckable()
def test_l_etat_initial_de_la_case_suit_la_config():
CONFIG["sounds"] = True
t = ZonzaTray(_ControleurFactice(), _SystemeFactice())
assert t.sounds_action.isChecked() is True
def test_le_menu_est_bien_celui_du_tray():
t = ZonzaTray(_ControleurFactice(), _SystemeFactice())
assert t.tray.contextMenu() is t.menu
# --- bascule des sons -----------------------------------------------------
def test_toggle_sounds_coupe_le_son(tray):
CONFIG["sounds"] = True
tray.toggle_sounds(False)
assert CONFIG["sounds"] is False
assert tray.sounds_action.isChecked() is False
def test_toggle_sounds_reactive_le_son(tray):
CONFIG["sounds"] = False
tray.toggle_sounds(True)
assert CONFIG["sounds"] is True
assert tray.sounds_action.isChecked() is True
def test_toggle_sounds_ne_touche_a_rien_d_autre_dans_la_config(tray):
avant = dict(CONFIG)
tray.toggle_sounds(True)
apres = dict(CONFIG)
del avant["sounds"]
del apres["sounds"]
assert avant == apres
# --- correspondance état → icône ------------------------------------------
def test_trois_etats_sont_couverts():
assert set(_STATE_TITLES) == {"idle", "recording", "transcribing"}
def test_etat_idle_par_defaut(tray):
assert tray._current_state == "idle"
@pytest.mark.parametrize("etat", ["idle", "recording", "transcribing"])
def test_apply_title_memorise_l_etat_recu(tray, etat):
tray._apply_title(etat)
assert tray._current_state == etat
def test_apply_title_pose_une_icone_non_nulle_pour_chaque_etat(tray):
for etat in _STATE_TITLES:
tray._apply_title(etat)
assert not tray.tray.icon().isNull()
def test_un_etat_inconnu_retombe_sur_l_icone_de_repos(tray):
# _apply_title ne doit jamais planter sur un état imprévu — repli sur
# "idle" (même garde que app.py : `_STATE_TITLES.get(state, "🎙️")`).
tray._apply_title("etat-invente")
assert not tray.tray.icon().isNull()
+65
View File
@@ -0,0 +1,65 @@
"""À chaque moteur son modèle.
`app.py` passait `CONFIG["model"]` — un dépôt MLX — quel que soit le moteur détecté.
Sur une machine sans Apple Silicon, `faster-whisper` aurait reçu un dépôt qu'il ne
sait pas charger. Le défaut était invisible ici, puisque la détection rend toujours
`mlx` sur ce Mac.
"""
import pytest
from core import MODELES_PAR_MOTEUR, modele_pour_moteur
def test_chaque_moteur_connu_a_son_modele():
for nom in ("mlx", "faster-cpu", "faster-gpu"):
assert modele_pour_moteur(nom)
def test_mlx_garde_le_modele_d_aujourd_hui():
assert modele_pour_moteur("mlx") == "mlx-community/whisper-medium"
def test_les_moteurs_faster_ne_recoivent_jamais_un_depot_mlx():
for nom in ("faster-cpu", "faster-gpu"):
assert "mlx" not in modele_pour_moteur(nom).lower()
def test_le_gpu_vise_la_precision_le_cpu_la_latence():
# mesuré le 2026-08-23 : medium = 13/13 termes mais 4,4 s sur CPU ; small = 11/13 en 1,4 s
assert "medium" in modele_pour_moteur("faster-gpu")
assert "small" in modele_pour_moteur("faster-cpu")
def test_un_moteur_inconnu_est_refuse_clairement():
with pytest.raises(KeyError) as e:
modele_pour_moteur("vosk")
assert "vosk" in str(e.value)
def test_la_table_ne_contient_que_les_moteurs_reconnus():
assert set(MODELES_PAR_MOTEUR) == {"mlx", "faster-cpu", "faster-gpu", "gemini", "openrouter"}
def test_le_reglage_manuel_prime_sur_la_table():
from core import modele_retenu
assert modele_retenu("faster-cpu", forcage=None) == modele_pour_moteur("faster-cpu")
assert modele_retenu("faster-cpu", forcage="Systran/faster-whisper-medium") \
== "Systran/faster-whisper-medium"
def test_un_forcage_vide_est_ignore():
from core import modele_retenu
assert modele_retenu("mlx", forcage="") == modele_pour_moteur("mlx")
def test_les_moteurs_distants_ne_recoivent_jamais_un_modele_tts():
# Un modele « -tts » produit de la voix a partir de texte : il refuse
# l'audio en entree (HTTP 400 chez OpenRouter, verifie le 2026-09-25).
for nom in ("gemini", "openrouter"):
assert "tts" not in modele_pour_moteur(nom).lower()
def test_le_delai_reseau_laisse_la_place_au_repli_sous_la_garde():
from core import CONFIG
from moteur.gemini import DELAI_RESEAU_S
assert DELAI_RESEAU_S < CONFIG["duree_max_transcription_s"]
+20
View File
@@ -64,3 +64,23 @@ 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
# --- faster-whisper : meme garantie que MLX ci-dessus — la fabrique rend un
# transcripteur complet sans jamais importer faster_whisper au niveau module,
# pour rester executable sur une machine qui ne l'a pas installe.
def test_la_fabrique_rend_un_transcripteur_faster_cpu_complet():
t = choisir_transcripteur(nom="faster-cpu", modele="Systran/faster-whisper-small",
langue="fr", amorce="")
assert callable(t.warmup) and callable(t.transcrire)
assert t.peripherique == "cpu"
assert t.type_calcul == "int8"
def test_la_fabrique_rend_un_transcripteur_faster_gpu_complet():
t = choisir_transcripteur(nom="faster-gpu", modele="Systran/faster-whisper-small",
langue="fr", amorce="")
assert callable(t.warmup) and callable(t.transcrire)
assert t.peripherique == "cuda"
assert t.type_calcul == "float16"
+90
View File
@@ -0,0 +1,90 @@
"""Moteur OpenRouter : même modèle Gemini, facturé et routé par OpenRouter.
Aucun test ne touche le réseau : l'envoi HTTP et la lecture de la clé sont
injectés. Le repli local, lui, est déjà éprouvé dans test_gemini.py.
"""
import base64
import numpy as np
import pytest
from moteur import choisir_transcripteur
from moteur.gemini import ErreurGemini, TranscripteurAvecRepli
from moteur.openrouter import TranscripteurOpenRouter, lire_cle_openrouter
def _reponse(texte):
return {"choices": [{"message": {"role": "assistant", "content": texte}}]}
def _transcripteur(envoyer, cle="cle-test"):
return TranscripteurOpenRouter(
"google/gemini-x", "fr", "Vocabulaire : Gitea, iCloud.",
lire_cle=lambda: cle, envoyer=envoyer,
)
def test_la_requete_porte_l_audio_le_modele_et_le_vocabulaire():
vu = {}
def envoyer(url, entetes, corps, delai_s):
vu.update(url=url, entetes=entetes, corps=corps)
return _reponse("Bonjour.")
_transcripteur(envoyer).transcrire(np.zeros(1600, dtype="float32"))
assert vu["url"] == "https://openrouter.ai/api/v1/chat/completions"
assert vu["entetes"]["Authorization"] == "Bearer cle-test"
assert vu["corps"]["model"] == "google/gemini-x"
systeme, utilisateur = vu["corps"]["messages"]
assert "Gitea" in systeme["content"] and "iCloud" in systeme["content"]
audio = next(p["input_audio"] for p in utilisateur["content"] if p["type"] == "input_audio")
assert audio["format"] == "wav"
assert base64.b64decode(audio["data"])[:4] == b"RIFF"
def test_la_cle_ne_part_jamais_dans_l_url():
vu = {}
_transcripteur(lambda url, e, c, d: vu.setdefault("url", url) and _reponse("x")) \
.transcrire(np.zeros(1600, dtype="float32"))
assert "cle-test" not in vu["url"]
def test_le_texte_rendu_est_nettoye():
t = _transcripteur(lambda *a: _reponse(" Je pousse sur Gitea. "))
assert t.transcrire(np.zeros(1600, dtype="float32")) == "Je pousse sur Gitea."
def test_une_erreur_rendue_dans_un_200_est_une_erreur_explicite():
# OpenRouter peut répondre 200 avec un objet `error` (fournisseur en panne).
t = _transcripteur(lambda *a: {"error": {"code": 502, "message": "fournisseur indisponible"}})
with pytest.raises(ErreurGemini) as e:
t.transcrire(np.zeros(1600, dtype="float32"))
assert "fournisseur indisponible" in str(e.value)
def test_une_reponse_sans_choix_est_une_erreur_explicite():
t = _transcripteur(lambda *a: {"choices": []})
with pytest.raises(ErreurGemini):
t.transcrire(np.zeros(1600, dtype="float32"))
def test_une_cle_absente_est_signalee_des_le_warmup_sans_appel():
appels = []
t = _transcripteur(lambda *a: appels.append(a), cle=None)
with pytest.raises(ErreurGemini) as e:
t.warmup()
assert "zonza-openrouter" in str(e.value)
assert appels == []
def test_la_cle_se_lit_sous_son_propre_nom():
assert lire_cle_openrouter(trousseau=lambda: None,
environ={"OPENROUTER_API_KEY": "or", "GEMINI_API_KEY": "g"}) == "or"
assert lire_cle_openrouter(trousseau=lambda: None, environ={"GEMINI_API_KEY": "g"}) is None
def test_la_fabrique_rend_openrouter_enveloppe_d_un_repli():
t = choisir_transcripteur(nom="openrouter", modele="google/gemini-x", langue="fr", amorce="")
assert isinstance(t, TranscripteurAvecRepli)
assert isinstance(t.principal, TranscripteurOpenRouter)
+34 -2
View File
@@ -35,5 +35,37 @@ def test_le_verrou_refuse_une_seconde_prise():
assert s.acquerir_verrou_instance(chemin) is None
def test_aucun_modificateur_enfonce_pendant_les_tests():
assert choisir_systeme().modificateurs_enfonces() is False
def test_les_modificateurs_sont_lus_par_le_masque():
"""La sonde est injectee : sans cela ce test lirait le CLAVIER PHYSIQUE et
echouerait au hasard, selon qu'une touche est enfoncee ou non pendant la
suite. C'est arrive le 2026-08-23."""
from systeme.macos import SystemeMacOS
assert SystemeMacOS(sonde_modificateurs=lambda: 0x00000000).modificateurs_enfonces() is False
# chaque modificateur, un par un, depuis les constantes nommees
from Quartz import (kCGEventFlagMaskAlternate, kCGEventFlagMaskCommand,
kCGEventFlagMaskControl, kCGEventFlagMaskShift)
for drapeau in (kCGEventFlagMaskShift, kCGEventFlagMaskControl,
kCGEventFlagMaskAlternate, kCGEventFlagMaskCommand):
assert SystemeMacOS(sonde_modificateurs=lambda d=drapeau: d).modificateurs_enfonces() is True
def test_une_sonde_qui_leve_ne_bloque_jamais_le_collage():
from systeme.macos import SystemeMacOS
def sonde_cassee():
raise RuntimeError("Quartz indisponible")
assert SystemeMacOS(sonde_modificateurs=sonde_cassee).modificateurs_enfonces() is False
def test_le_masque_couvre_command():
"""Command etait ABSENT du masque ecrit a la main : Zonza ne voyait jamais
cette touche, pourtant celle de son propre raccourci."""
from Quartz import kCGEventFlagMaskCommand
from systeme.macos import _MASQUE_MODIFICATEURS
assert kCGEventFlagMaskCommand & _MASQUE_MODIFICATEURS
def test_la_sonde_reelle_rend_un_booleen_sans_lever():
"""On n'affirme PAS sa valeur : elle depend du clavier de l'utilisateur."""
assert isinstance(choisir_systeme().modificateurs_enfonces(), bool)
+154
View File
@@ -0,0 +1,154 @@
"""Couche Windows du contrat Systeme — éprouvée depuis macOS par injection.
`winsound`, `msvcrt` et `ctypes.windll` n'existent pas sur macOS : c'est
pourquoi `SystemeWindows` les reçoit en paramètres injectables (jouer, clavier,
verrouiller) plutôt que de les importer en tête de module. Ces tests éprouvent
donc la LOGIQUE française de la classe — quel alias pour quel son, quel bit
pour quelle touche, comment les chemins et le verrou sont construits — jamais
le comportement réel de winsound.PlaySound, GetAsyncKeyState ou msvcrt.locking :
aucune machine ici ne peut le vérifier. Seule une vraie machine Windows le peut.
"""
import os
import pytest
from systeme.windows import SystemeWindows
# --- jouer_son ---------------------------------------------------------
def test_jouer_son_start_produit_lalias_systemasterisk():
appels = []
s = SystemeWindows(jouer=appels.append)
s.jouer_son("start")
assert appels == ["SystemAsterisk"]
def test_jouer_son_done_produit_lalias_systemexclamation():
appels = []
s = SystemeWindows(jouer=appels.append)
s.jouer_son("done")
assert appels == ["SystemExclamation"]
def test_jouer_son_error_produit_lalias_systemhand():
appels = []
s = SystemeWindows(jouer=appels.append)
s.jouer_son("error")
assert appels == ["SystemHand"]
def test_jouer_son_nom_inconnu_ne_produit_aucun_appel():
appels = []
s = SystemeWindows(jouer=appels.append)
s.jouer_son("inconnu")
assert appels == []
def test_jouer_son_qui_echoue_ne_leve_pas():
def echoue(alias):
raise OSError("winsound indisponible")
s = SystemeWindows(jouer=echoue)
s.jouer_son("start") # ne doit jamais lever : le son ne casse pas la dictée
# --- modificateurs_enfonces --------------------------------------------
@pytest.mark.parametrize("vk", [0x11, 0x10, 0x12, 0x5B, 0x5C])
def test_modificateurs_enfonces_vrai_pour_chacune_des_cinq_touches(vk):
# bit de poids fort posé UNIQUEMENT pour la touche ciblée par ce cas.
s = SystemeWindows(clavier=lambda sonde, cible=vk: 0x8000 if sonde == cible else 0)
assert s.modificateurs_enfonces() is True
def test_modificateurs_enfonces_faux_si_aucun_bit_de_poids_fort_pose():
# une valeur non nulle mais sans le bit 0x8000 ne doit pas compter.
s = SystemeWindows(clavier=lambda vk: 0x0001)
assert s.modificateurs_enfonces() is False
def test_modificateurs_enfonces_faux_si_toutes_les_touches_sont_relachees():
s = SystemeWindows(clavier=lambda vk: 0)
assert s.modificateurs_enfonces() is False
def test_modificateurs_enfonces_faux_si_la_sonde_leve():
def leve(vk):
raise OSError("user32 indisponible")
s = SystemeWindows(clavier=leve)
assert s.modificateurs_enfonces() is False
# --- raccourci_defaut ----------------------------------------------------
def test_raccourci_defaut_est_ctrl_alt_d():
# <cmd> n'existe pas sur Windows, contrairement à macOS.
assert SystemeWindows().raccourci_defaut() == "<ctrl>+<alt>+d"
# --- chemins ---------------------------------------------------------------
def test_chemin_log_derive_de_localappdata(monkeypatch, tmp_path):
monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
assert SystemeWindows().chemin_log() == os.path.join(str(tmp_path), "Zonza", "Zonza.log")
def test_dossier_donnees_derive_de_localappdata(monkeypatch, tmp_path):
monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
assert SystemeWindows().dossier_donnees() == os.path.join(str(tmp_path), "Zonza")
def test_chemin_log_retombe_sur_le_dossier_utilisateur_si_localappdata_absent(monkeypatch):
monkeypatch.delenv("LOCALAPPDATA", raising=False)
attendu = os.path.join(os.path.expanduser("~"), "Zonza", "Zonza.log")
assert SystemeWindows().chemin_log() == attendu
# --- verrou d'instance -------------------------------------------------------
def test_le_verrou_rend_un_objet_la_premiere_fois(tmp_path):
chemin = str(tmp_path / "sous-dossier" / "z.lock")
s = SystemeWindows(verrouiller=lambda f: None)
verrou = s.acquerir_verrou_instance(chemin)
assert verrou is not None
verrou.close()
def test_le_verrou_cree_le_dossier_parent_manquant(tmp_path):
chemin = str(tmp_path / "sous-dossier" / "z.lock")
s = SystemeWindows(verrouiller=lambda f: None)
assert not os.path.isdir(os.path.dirname(chemin))
verrou = s.acquerir_verrou_instance(chemin)
assert os.path.isdir(os.path.dirname(chemin))
verrou.close()
def test_le_verrou_refuse_une_seconde_prise(tmp_path):
chemin = str(tmp_path / "z.lock")
pris = {"valeur": False}
def verrouiller(f):
if pris["valeur"]:
raise OSError("verrou déjà pris")
pris["valeur"] = True
s = SystemeWindows(verrouiller=verrouiller)
premier = s.acquerir_verrou_instance(chemin)
assert premier is not None
second = s.acquerir_verrou_instance(chemin)
assert second is None
premier.close()
def test_un_verrou_dont_la_prise_echoue_referme_le_fichier(tmp_path):
chemin = str(tmp_path / "z.lock")
captures = {}
def echoue(f):
captures["fichier"] = f
raise OSError("verrou pris")
s = SystemeWindows(verrouiller=echoue)
resultat = s.acquerir_verrou_instance(chemin)
assert resultat is None
assert captures["fichier"].closed