Synthèse vocale en local avec Ollama : Phonon testé en français

brew install ollama            # version 0.40.9 ou plus récente
ollama serve &
ollama pull phonon:fr          # 122 Mo

# En ligne de commande : le fichier WAV sort sur la sortie standard
ollama run phonon:fr "Bonjour, je suis une voix de synthèse locale." > bonjour.wav

# Par l'API compatible OpenAI
curl -s http://localhost:11434/v1/audio/speech \
  -H "Content-Type: application/json" \
  -d '{"model": "phonon:fr", "input": "Votre rendez-vous est confirmé."}' \
  --output rendez-vous.wav

{
  "nombres": [
    "Le montant de la facture s'élève à 1 245,80 euros toutes taxes comprises.",
    "Notre équipe compte 37 personnes réparties sur 4 sites.",
    "Le taux de disponibilité du service a atteint 99,95 % sur l'année.",
    "Vous avez 3 messages non lus et 12 tâches en attente.",
    "Le serveur dispose de 256 gigaoctets de mémoire et de 64 cœurs."
  ],
  "sigles": [
    "La DSI a validé la conformité au RGPD avant la mise en production.",
    "Le PDG présentera les résultats à l'AG des actionnaires.",
    "Le flux passe par un VPN puis par l'API REST de l'ERP.",
    "La SNCF et la RATP ont annoncé une grève pour jeudi.",
    "Votre numéro de TVA intracommunautaire figure sur le bon de commande."
  ]
}

"""Synthèse vocale en français : vitesse et fidélité de Phonon (Ollama) face à la voix de macOS.

Pour chaque phrase du corpus et chaque moteur :
  vitesse   -> délai avant le premier échantillon audio, durée de génération, facteur de temps réel
  fidélité  -> l'audio est retranscrit par un modèle de reconnaissance vocale, puis comparé au texte
               (taux d'erreur sur les mots, WER, et sur les caractères, CER), par catégorie

Usage :
  export AOAI_ENDPOINT=… AOAI_KEY=… STT_DEPLOYMENT=gpt-4o-transcribe
  python bench_tts.py
"""

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

import httpx
import jiwer
from num2words import num2words
from openai import AzureOpenAI

ICI = Path(__file__).parent
AUDIO = ICI / "audio"
AUDIO.mkdir(exist_ok=True)
CORPUS = json.loads((ICI / "corpus.json").read_text())
OLLAMA = "http://localhost:11434/v1/audio/speech"
MOTEURS = ["phonon:fr", "phonon:40m-fr", "phonon:multilingual", "phonon:latest", "macos:Thomas",
           "azure:fr-FR-DeniseNeural", "azure:fr-FR-VivienneMultilingualNeural", "azure:fr-FR-Soleil:MAI-Voice-2.1-Flash"]
# Sous-ensemble à mesurer : MOTEURS=azure:fr-FR-DeniseNeural,… (les résultats existants sont conservés).
if os.environ.get("MOTEURS"):
    MOTEURS = os.environ["MOTEURS"].split(",")
stt = AzureOpenAI(azure_endpoint=os.environ["AOAI_ENDPOINT"], api_key=os.environ["AOAI_KEY"], api_version="2025-03-01-preview")
STT = os.environ.get("STT_DEPLOYMENT", "gpt-4o-transcribe")


def duree_wav(chemin: Path) -> float:
    # En flux, l'en-tête WAV ne connaît pas encore la longueur : il porte une valeur
    # provisoire. On calcule donc la durée d'après la taille des données (PCM 16 bits mono).
    with wave.open(str(chemin)) as w:
        octets_par_seconde = w.getframerate() * w.getsampwidth() * w.getnchannels()
    return (chemin.stat().st_size - 44) / octets_par_seconde


def synthese(moteur: str, texte: str, sortie: Path) -> dict:
    debut = time.perf_counter()
    if moteur.startswith("azure:"):
        # Azure AI Speech, voix neuronales : SSML en entrée, WAV 24 kHz en flux en sortie.
        voix = moteur.split(":", 1)[1]
        ssml = (f"<speak version='1.0' xml:lang='fr-FR'><voice name='{voix}'>"
                f"{texte.replace('&', '&amp;').replace('<', '&lt;')}</voice></speak>")
        premier, morceaux = None, []
        with httpx.stream("POST", f"https://{os.environ['TTS_REGION']}.tts.speech.microsoft.com/cognitiveservices/v1",
                          content=ssml.encode(), timeout=120, headers={
                              "Ocp-Apim-Subscription-Key": os.environ["SPEECH_KEY"],
                              "Content-Type": "application/ssml+xml",
                              "X-Microsoft-OutputFormat": "riff-24khz-16bit-mono-pcm"}) as r:
            r.raise_for_status()
            for morceau in r.iter_bytes():
                morceaux.append(morceau)
                if premier is None and sum(map(len, morceaux)) > 44:
                    premier = time.perf_counter() - debut
        total = time.perf_counter() - debut
        sortie.write_bytes(b"".join(morceaux))
    elif moteur.startswith("macos:"):
        aiff = sortie.with_suffix(".aiff")
        subprocess.run(["say", "-v", moteur.split(":")[1], "-o", str(aiff), texte], check=True)
        subprocess.run(["afconvert", "-f", "WAVE", "-d", "LEI16@24000", str(aiff), str(sortie)], check=True)
        aiff.unlink()
        total = time.perf_counter() - debut
        premier = total  # say n'écrit le fichier qu'à la fin
    else:
        premier, morceaux = None, []
        with httpx.stream("POST", OLLAMA, json={"model": moteur, "input": texte}, timeout=300) as r:
            r.raise_for_status()
            for morceau in r.iter_bytes():
                morceaux.append(morceau)
                # L'en-tête WAV fait 44 octets : le premier son arrive avec les octets suivants.
                if premier is None and sum(map(len, morceaux)) > 44:
                    premier = time.perf_counter() - debut
        total = time.perf_counter() - debut
        sortie.write_bytes(b"".join(morceaux))
    audio = duree_wav(sortie)
    return {"premier_son_s": round(premier, 3), "generation_s": round(total, 3), "audio_s": round(audio, 2),
            "temps_reel_x": round(audio / total, 1)}


def normaliser(texte: str) -> str:
    """Forme parlée comparable des deux côtés : nombres en lettres, symboles développés, sans ponctuation."""
    t = texte.lower().replace(" ", " ").replace(" ", " ")
    t = re.sub(r"(\d) (\d{3})\b", r"\1\2", t)                      # 1 245 -> 1245
    t = re.sub(r"(\d+) ?h ?(\d+)", r"\1 heures \2", t)            # 9 h 30
    t = re.sub(r"(\d+) ?h\b", r"\1 heures", 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)                                # ponctuation, traits d'union, apostrophes
    return " ".join(t.split())


def transcrire(chemin: Path) -> str:
    with chemin.open("rb") as f:
        return stt.audio.transcriptions.create(model=STT, file=f, language="fr").text


def main() -> None:
    lignes = []
    for moteur in MOTEURS:
        if not moteur.startswith("macos:"):
            synthese(moteur, "Échauffement.", AUDIO / "echauffement.wav")  # chargement du modèle, connexion
        for categorie, phrases in CORPUS.items():
            for i, texte in enumerate(phrases):
                fichier = AUDIO / f"{re.sub(r'[^\w.-]', '_', moteur)}-{categorie}-{i}.wav"
                mesure = synthese(moteur, texte, fichier)
                transcription = transcrire(fichier)
                ref, hyp = normaliser(texte), normaliser(transcription)
                lignes.append({"moteur": moteur, "categorie": categorie, "texte": texte, "transcription": transcription,
                               "wer": round(jiwer.wer(ref, hyp), 3), "cer": round(jiwer.cer(ref, hyp), 3), **mesure})
                print(f"{moteur:20} {categorie:12} WER {lignes[-1]['wer']:.2f}  x{mesure['temps_reel_x']}")
    fichier = ICI / "resultats.json"
    anciens = [l for l in json.loads(fichier.read_text()) if l["moteur"] not in MOTEURS] if fichier.exists() else []
    lignes = anciens + lignes
    fichier.write_text(json.dumps(lignes, ensure_ascii=False, indent=1))
    MOTEURS[:] = list(dict.fromkeys(l["moteur"] for l in lignes))

    print("\n| Moteur | WER moyen | CER moyen | Premier son (médiane) | Temps réel (médiane) |")
    print("|---|---|---|---|---|")
    for moteur in MOTEURS:
        m = [l for l in lignes if l["moteur"] == moteur]
        print(f"| {moteur} | {statistics.mean(l['wer'] for l in m):.1%} | {statistics.mean(l['cer'] for l in m):.1%} | "
              f"{statistics.median(l['premier_son_s'] for l in m):.2f} s | x{statistics.median(l['temps_reel_x'] for l in m)} |")
    print("\nWER par catégorie :\n")
    print("| Catégorie | " + " | ".join(MOTEURS) + " |")
    print("|---|" + "---|" * len(MOTEURS))
    for categorie in CORPUS:
        print(f"| {categorie} | " + " | ".join(
            f"{statistics.mean(l['wer'] for l in lignes if l['moteur'] == m and l['categorie'] == categorie):.0%}"
            for m in MOTEURS) + " |")


if __name__ == "__main__":
    main()

"""Écrit les nombres en lettres avant la synthèse : Phonon lit mal les décimales et les milliers.

  « 1 245,80 euros » -> « mille deux cent quarante-cinq virgule quatre-vingts euros »
  « 99,95 % »        -> « quatre-vingt-dix-neuf virgule quatre-vingt-quinze pour cent »
"""

import re

from num2words import num2words


def nombres_en_lettres(texte: str) -> str:
    t = texte.replace(" ", " ").replace(" ", " ")
    t = re.sub(r"(\d) (\d{3})\b", r"\1\2", t)                       # séparateur de milliers
    t = re.sub(r"(\d+) ?h ?(\d{2})", r"\1 heures \2", t)             # 9 h 30
    t = re.sub(r"(\d+) ?%", r"\1 pour cent", t)
    t = re.sub(r"(\d+),(\d+)", lambda m: f"{m.group(1)} virgule {m.group(2)}", t)
    return re.sub(r"\d+", lambda m: num2words(int(m.group()), lang="fr"), t)


if __name__ == "__main__":
    print(nombres_en_lettres("Le montant s'élève à 1 245,80 euros, soit 99,95 % du budget, à 9 h 30."))

"""Assistant vocal 100 % local : un LLM répond en flux, Phonon lit chaque phrase dès qu'elle est complète.

On compare le délai avant le premier son avec la méthode naïve (attendre toute la réponse).

Usage : python assistant_vocal.py "Comment sécuriser l'accès à distance de nos collaborateurs ?"
"""

import re
import sys
import time
import wave
from pathlib import Path

import httpx
from openai import OpenAI

from nombres_en_lettres import nombres_en_lettres

LLM = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
MODELE, VOIX = "qwen3.5:9b", "phonon:fr"
FIN_DE_PHRASE = re.compile(r"(?<=[.!?…])\s+")


def dire(phrase: str) -> bytes:
    """Audio PCM d'une phrase (sans l'en-tête WAV de 44 octets)."""
    r = httpx.post("http://localhost:11434/v1/audio/speech", timeout=120,
                   json={"model": VOIX, "input": nombres_en_lettres(phrase)})
    r.raise_for_status()
    return r.content[44:]


def phrases_en_flux(question: str):
    """Rend chaque phrase de la réponse dès qu'elle est terminée."""
    flux = LLM.chat.completions.create(
        model=MODELE, stream=True, extra_body={"reasoning_effort": "none"},
        messages=[{"role": "system", "content": "Tu réponds à l'oral, en français, en quatre ou cinq phrases, sans liste ni mise en forme."},
                  {"role": "user", "content": question}])
    tampon = ""
    for e in flux:
        if e.choices and e.choices[0].delta.content:
            tampon += e.choices[0].delta.content
            *finies, tampon = FIN_DE_PHRASE.split(tampon)
            yield from (p for p in finies if p.strip())
    if tampon.strip():
        yield tampon


def main(question: str) -> None:
    debut = time.perf_counter()
    audio, premier_son, texte = [], None, []
    for phrase in phrases_en_flux(question):
        texte.append(phrase)
        audio.append(dire(phrase))
        if premier_son is None:
            premier_son = time.perf_counter() - debut  # la lecture peut commencer ici
    total = time.perf_counter() - debut
    with wave.open("audio/assistant.wav", "wb") as w:
        w.setnchannels(1), w.setsampwidth(2), w.setframerate(24000)
        w.writeframes(b"".join(audio))
    duree = sum(map(len, audio)) / 48000
    print(" ".join(texte))
    print(f"\nPremier son après {premier_son:.2f} s ; réponse complète (texte + voix) après {total:.2f} s ; "
          f"{duree:.1f} s d'audio.")
    # Méthode naïve, pour comparaison : toute la réponse d'abord, puis une seule synthèse.
    debut = time.perf_counter()
    reponse = " ".join(phrases_en_flux(question))
    dire(reponse)
    print(f"Méthode naïve : premier son après {time.perf_counter() - debut:.2f} s.")


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "Comment sécuriser l'accès à distance de nos collaborateurs ?")