Ralph Mayola
3e269dfb5e
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>
313 lines
12 KiB
Python
313 lines
12 KiB
Python
"""`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)
|