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>

core.py

@@ -6,6 +6,10 @@ 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" = l'audio part chez Google, avec le
+    # moteur local en repli si l'API ne repond pas (cf moteur/gemini.py).
+    "moteur": "gemini",
     # 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,
@@ -186,6 +190,10 @@ 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",
 }
 
 

moteur/__init__.py

@@ -12,8 +12,17 @@ def nom_moteur_pour(plateforme, machine, cuda_disponible):
     return "faster-gpu" if cuda_disponible else "faster-cpu"
 
 
-def detecter_moteur():
+def detecter_moteur(forcage=None):
-    """Interroge la machine, puis délègue la décision à `nom_moteur_pour`."""
+    """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,15 @@ 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 == "gemini":
+        # 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
+        local = _moteur_local()
+        return TranscripteurAvecRepli(
+            TranscripteurGemini(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}")

moteur/gemini.py

@@ -0,0 +1,190 @@
+"""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 Gemini ne répond pas à
+# temps, il doit rester de quoi basculer sur le moteur local avant que la
+# garde du halo ne tombe.
+DELAI_RESEAU_S = 12.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():
+    """Clé rangée par `security add-generic-password -s zonza-gemini ...`.
+
+    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_TROUSSEAU, "-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):
+    """La clé du trousseau, sinon GEMINI_API_KEY, sinon None."""
+    trousseau = trousseau or _lire_trousseau
+    environ = os.environ if environ is None else environ
+    return trousseau() or environ.get("GEMINI_API_KEY") or None
+
+
+def _envoyer(url, cle, corps, delai_s):
+    """POST JSON. La clé part en en-tête, jamais dans l'URL (qui finit en journal)."""
+    requete = urllib.request.Request(
+        url,
+        data=json.dumps(corps).encode("utf-8"),
+        headers={"Content-Type": "application/json", "x-goog-api-key": cle},
+        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"[gemini] 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"[gemini] échec ({e}) — transcription locale")
+            return self._secours_pret().transcrire(audio)

tests/test_gemini.py

@@ -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)

tests/test_modele_par_moteur.py

@@ -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"}
 
 
 def test_le_reglage_manuel_prime_sur_la_table():