Transcription audio en texte en local ou on-premise : Whisper, Parakeet, Voxtral et Qwen3-ASR testés en français

"""Prépare l'échantillon de test à partir de FLEURS (français, partie test, CC BY 4.0).

Pour 120 phrases lues par de vrais locuteurs :
  propre/      -> l'enregistrement d'origine, 16 kHz
  telephone/   -> la même phrase dégradée comme un appel : bande 300-3 400 Hz, 8 kHz,
                  compression μ-law, bruit de fond à 15 dB de rapport signal sur bruit
  references.json -> la transcription humaine de chaque phrase

Usage : python preparer.py donnees/fleurs-fr-test.parquet
"""

import io
import json
import sys
from pathlib import Path

import numpy as np
import pyarrow.parquet as pq
import soundfile as sf
from scipy.signal import butter, resample_poly, sosfilt

ICI = Path(__file__).parent
N = 120
rng = np.random.default_rng(2026)


def mu_law(x: np.ndarray, mu: float = 255.0) -> np.ndarray:
    """Compression puis décompression μ-law sur 8 bits, comme sur une ligne téléphonique."""
    y = np.sign(x) * np.log1p(mu * np.abs(x)) / np.log1p(mu)
    y = np.round((y + 1) / 2 * 255) / 255 * 2 - 1
    return np.sign(y) * np.expm1(np.abs(y) * np.log1p(mu)) / mu


def telephone(x: np.ndarray, sr: int) -> np.ndarray:
    x = sosfilt(butter(4, [300, 3400], btype="bandpass", fs=sr, output="sos"), x)
    x8 = resample_poly(x, 1, sr // 8000)
    bruit = rng.normal(0, 1, len(x8))
    bruit *= np.sqrt(np.mean(x8 ** 2) / 10 ** (15 / 10)) / np.sqrt(np.mean(bruit ** 2))
    x8 = mu_law(np.clip(x8 + bruit, -1, 1))
    return resample_poly(x8, sr // 8000, 1)  # les modèles attendent du 16 kHz


def main(parquet: str) -> None:
    lignes = pq.read_table(parquet).to_pylist()
    # Une phrase par identifiant (FLEURS fait lire certaines phrases à plusieurs locuteurs).
    vues, choix = set(), []
    for l in lignes:
        if l["id"] not in vues:
            vues.add(l["id"])
            choix.append(l)
    choix = [choix[i] for i in sorted(rng.choice(len(choix), N, replace=False))]
    for d in ("propre", "telephone"):
        (ICI / d).mkdir(exist_ok=True)
    references, duree = {}, 0.0
    for l in choix:
        x, sr = sf.read(io.BytesIO(l["audio"]["bytes"]), dtype="float32")
        nom = f"{l['id']}.wav"
        sf.write(ICI / "propre" / nom, x, sr)
        sf.write(ICI / "telephone" / nom, telephone(x, sr).astype(np.float32), sr)
        references[nom] = l["raw_transcription"]
        duree += len(x) / sr
    (ICI / "references.json").write_text(json.dumps(references, ensure_ascii=False, indent=1))
    print(f"{len(references)} phrases, {duree / 60:.1f} minutes d'audio")


if __name__ == "__main__":
    main(sys.argv[1])

python3.12 -m venv .venv && source .venv/bin/activate
pip install mlx-audio==0.6.4 mlx-whisper "mistral-common" pyarrow scipy soundfile jiwer num2words openai httpx

# Jeu de test FLEURS, partie française (environ 450 Mo)
curl -L -o donnees/fleurs-fr-test.parquet \
  https://huggingface.co/api/datasets/google/fleurs/parquet/fr_fr/test/0.parquet
python preparer.py donnees/fleurs-fr-test.parquet

"""Banc d'essai de transcription (speech to text) en français : modèles ouverts sur Mac et cloud.

Pour chaque phrase de l'échantillon FLEURS (voir preparer.py), en audio propre et en audio
dégradé façon téléphone :
  fidélité -> taux d'erreur sur les mots (WER) et les caractères (CER) face à la transcription humaine
  vitesse  -> durée de l'audio rapportée au temps de transcription (× temps réel)
  mémoire  -> pic de mémoire des modèles MLX

Usage :
  export AOAI_ENDPOINT=… AOAI_KEY=… SPEECH_ENDPOINT=… SPEECH_KEY=…
  python bench.py                          # tous les moteurs
  MOTEURS=whisper,parakeet python bench.py # un sous-ensemble ; les autres résultats sont conservés
"""

import json
import os
import re
import statistics
import time
from pathlib import Path

import httpx
import jiwer
import mlx.core as mx
import soundfile as sf
from num2words import num2words
from openai import AzureOpenAI

ICI = Path(__file__).parent
REFERENCES = json.loads((ICI / "references.json").read_text())
CONDITIONS = ["propre", "telephone"]
MOTEURS = {
    "whisper-turbo": "mlx-community/whisper-large-v3-turbo",
    "parakeet-v3": "mlx-community/parakeet-tdt-0.6b-v3",
    "qwen3-asr": "mlx-community/Qwen3-ASR-1.7B-8bit",
    "voxtral-realtime": "mlx-community/Voxtral-Mini-4B-Realtime-2602-4bit",
    "azure-gpt-4o-transcribe": "gpt-4o-transcribe",
    "azure-mai-transcribe-2": "MAI-Transcribe-2",
}
CHOIX = os.environ["MOTEURS"].split(",") if os.environ.get("MOTEURS") else list(MOTEURS)


class Moteur:
    def __init__(self, nom: str):
        self.nom, self.modele = nom, MOTEURS[nom]
        if nom == "whisper-turbo":
            import mlx_whisper
            self.whisper = mlx_whisper
        elif nom.startswith("azure-gpt"):
            self.client = AzureOpenAI(azure_endpoint=os.environ["AOAI_ENDPOINT"], api_key=os.environ["AOAI_KEY"],
                                      api_version="2025-03-01-preview")
        elif not nom.startswith("azure"):
            from mlx_audio.stt.utils import load_model
            self.m = load_model(self.modele)

    def transcrire(self, chemin: Path) -> str:
        if self.nom == "whisper-turbo":
            return self.whisper.transcribe(str(chemin), path_or_hf_repo=self.modele, language="fr")["text"]
        if self.nom.startswith("azure-gpt"):
            with chemin.open("rb") as f:
                return self.client.audio.transcriptions.create(model=self.modele, file=f, language="fr").text
        if self.nom.startswith("azure-mai"):
            return transcription_rapide(chemin)
        options = {"language": "French"} if self.nom == "qwen3-asr" else {}
        return self.m.generate(str(chemin), **options).text


def transcription_rapide(chemin: Path) -> str:
    """API Fast Transcription d'Azure AI Speech, avec le modèle MAI-Transcribe-2."""
    definition = {"enhancedMode": {"enabled": True, "model": "MAI-Transcribe-2"}}
    url = f"{os.environ['SPEECH_ENDPOINT']}/speechtotext/transcriptions:transcribe?api-version=2025-10-15"
    for essai in range(6):
        with chemin.open("rb") as audio:
            r = httpx.post(url, timeout=300, headers={"Ocp-Apim-Subscription-Key": os.environ["SPEECH_KEY"]},
                           files={"audio": (chemin.name, audio, "audio/wav"),
                                  "definition": (None, json.dumps(definition), "application/json")})
        if r.status_code != 429:
            break
        time.sleep(float(r.headers.get("Retry-After", 2 ** (essai + 1))))  # quota par minute
    r.raise_for_status()
    return " ".join(p["text"] for p in r.json()["combinedPhrases"])


def normaliser(texte: str) -> str:
    """Forme comparable : minuscules, nombres en lettres, sans ponctuation ni apostrophes typographiques."""
    t = texte.lower().replace("’", "'").replace(" ", " ").replace(" ", " ")
    t = re.sub(r"(\d) (\d{3})\b", r"\1\2", t)
    t = t.replace("%", " pour cent ").replace("€", " euros ")
    t = re.sub(r"(\d+),(\d+)", lambda m: f"{m.group(1)} virgule {m.group(2)}", t)
    t = re.sub(r"\d+", lambda m: " " + num2words(int(m.group()), lang="fr") + " ", t)
    t = re.sub(r"[^\w\s]|_", " ", t)
    return " ".join(t.split())


def main() -> None:
    lignes = []
    for nom in CHOIX:
        debut = time.perf_counter()
        moteur = Moteur(nom)
        moteur.transcrire(ICI / "propre" / next(iter(REFERENCES)))  # échauffement
        chargement = time.perf_counter() - debut
        mx.reset_peak_memory()
        for condition in CONDITIONS:
            for fichier, reference in REFERENCES.items():
                chemin = ICI / condition / fichier
                t0 = time.perf_counter()
                texte = moteur.transcrire(chemin)
                duree = time.perf_counter() - t0
                audio = sf.info(chemin).duration
                ref, hyp = normaliser(reference), normaliser(texte)
                lignes.append({"moteur": nom, "condition": condition, "fichier": fichier, "reference": reference,
                               "transcription": texte, "wer": round(jiwer.wer(ref, hyp), 4),
                               "cer": round(jiwer.cer(ref, hyp), 4), "audio_s": round(audio, 2),
                               "traitement_s": round(duree, 3)})
            m = [l for l in lignes if l["moteur"] == nom and l["condition"] == condition]
            print(f"{nom:24} {condition:10} WER {wer_global(m):.1%}")
        if not nom.startswith("azure"):
            lignes[-1]["pic_memoire_go"] = round(mx.get_peak_memory() / 2**30, 1)
        lignes[-1]["chargement_s"] = round(chargement, 1)
        del moteur
        mx.clear_cache()
    fichier = ICI / "resultats.json"
    anciens = [l for l in json.loads(fichier.read_text()) if l["moteur"] not in CHOIX] if fichier.exists() else []
    lignes = anciens + lignes
    fichier.write_text(json.dumps(lignes, ensure_ascii=False, indent=1))
    rapport(lignes)


def wer_global(lignes: list[dict]) -> float:
    """WER sur l'ensemble du corpus (erreurs cumulées / mots de référence), pas la moyenne des phrases."""
    refs = [normaliser(l["reference"]) for l in lignes]
    hyps = [normaliser(l["transcription"]) for l in lignes]
    return jiwer.wer(refs, hyps)


def rapport(lignes: list[dict]) -> None:
    noms = list(dict.fromkeys(l["moteur"] for l in lignes))
    print("\n| Moteur | WER propre | WER téléphone | CER propre | × temps réel | Pic mémoire |")
    print("|---|---|---|---|---|---|")
    for n in noms:
        p = [l for l in lignes if l["moteur"] == n and l["condition"] == "propre"]
        t = [l for l in lignes if l["moteur"] == n and l["condition"] == "telephone"]
        cer = jiwer.cer([normaliser(l["reference"]) for l in p], [normaliser(l["transcription"]) for l in p])
        vitesse = sum(l["audio_s"] for l in p + t) / sum(l["traitement_s"] for l in p + t)
        pic = next((l["pic_memoire_go"] for l in lignes if l["moteur"] == n and "pic_memoire_go" in l), None)
        print(f"| {n} | {wer_global(p):.1%} | {wer_global(t):.1%} | {cer:.1%} | {vitesse:.0f} × | {f'{pic} Go' if pic else '–'} |")


if __name__ == "__main__":
    main()