Compare commits

..
9 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
17 changed files with 1427 additions and 42 deletions
+30 -26
View File
@@ -25,7 +25,6 @@ if "--diagnostic" in sys.argv:
print(mettre_en_forme(diagnostiquer()))
sys.exit(0)
import os
import threading
from objc import super
@@ -41,11 +40,8 @@ from Foundation import NSObject, NSTimer
from PyObjCTools import AppHelper
from pynput import keyboard
from controller import ZonzaController
from core import CONFIG, modele_retenu, 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": "🎙️",
@@ -55,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
@@ -144,23 +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)
moteur = detecter_moteur()
transcripteur = choisir_transcripteur(
nom=moteur,
modele=modele_retenu(moteur, 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.
@@ -170,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,
}
+18
View File
@@ -6,6 +6,11 @@ Tout ce qui est ici est testable sans micro, sans clavier, sans modèle.
import re
CONFIG = {
# 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,
@@ -65,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",
@@ -186,6 +193,17 @@ 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",
}
+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
+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()
+24 -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())
@@ -67,4 +76,17 @@ def choisir_transcripteur(nom, modele, langue, amorce):
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}")
+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 "")
+25 -7
View File
@@ -1,14 +1,32 @@
# mlx-whisper : GPU Apple Silicon uniquement (moteur "mlx", machine actuelle).
mlx-whisper>=0.4.3
# faster-whisper : coeur CTranslate2, tourne partout (Mac Intel, Windows avec ou
# sans GPU). Les deux coexistent car aucun des deux ne couvre a lui seul toutes
# les plateformes visees par le portage — cf moteur/__init__.py::nom_moteur_pour.
# 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
# PySide6 : uniquement hors macOS (Cocoa/pyobjc y sert deja l'overlay) — cf interface/qt.py.
PySide6>=6.11
+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
+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"
+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)
+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()
+14 -1
View File
@@ -37,7 +37,7 @@ def test_un_moteur_inconnu_est_refuse_clairement():
def test_la_table_ne_contient_que_les_moteurs_reconnus():
assert set(MODELES_PAR_MOTEUR) == {"mlx", "faster-cpu", "faster-gpu"}
assert set(MODELES_PAR_MOTEUR) == {"mlx", "faster-cpu", "faster-gpu", "gemini", "openrouter"}
def test_le_reglage_manuel_prime_sur_la_table():
@@ -50,3 +50,16 @@ def test_le_reglage_manuel_prime_sur_la_table():
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"]
+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)