Ajoute la couche Windows et la commande de diagnostic
This commit is contained in:
+303
@@ -0,0 +1,303 @@
|
||||
"""`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=16000, 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.
|
||||
"""
|
||||
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=16000):
|
||||
"""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é.
|
||||
"""
|
||||
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:
|
||||
audio = np.zeros(16000, dtype="float32")
|
||||
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)
|
||||
Reference in New Issue
Block a user