"""
Moteur de synthese vocale locale hors-ligne Kokoro ONNX.
"""
import os
import urllib.request
import numpy as np
from ...config import TtsConfig
from ...core.logger import getMirzaLogger
from ..interfaces.baseTts import BaseTts
logger = getMirzaLogger("mirza.engine.tts.kokoro")
[docs]
class KokoroTts(BaseTts):
"""
Synthese vocale locale Kokoro ONNX (~80 Mo, 30 ms de latence).
Parameters
----------
ttsConfig : TtsConfig
Configuration de la synthese vocale.
modelPath : str, optional
Chemin du fichier modele ONNX local. Defaut : "mirza/data/models/kokoro-v0_19.onnx".
voicesPath : str, optional
Chemin du fichier de voix. Defaut : "mirza/data/models/voices-v1.0.bin".
Attributes
----------
config : TtsConfig
Configuration locale.
modelPath : str
Emplacement du modele ONNX.
voicesPath : str
Emplacement du fichier de voix.
kokoro : kokoro_onnx.Kokoro | None
Instance du moteur Kokoro chargee en memoire.
"""
[docs]
def __init__(
self,
ttsConfig: TtsConfig,
modelPath: str = "mirza/data/models/kokoro-v1.0.onnx",
voicesPath: str = "mirza/data/models/voices-v1.0.bin",
):
"""
Initialise le moteur Kokoro ONNX standard, non quantifie.
Parameters
----------
ttsConfig : TtsConfig
Configuration acoustique (voix, vitesse).
modelPath : str, optional
Chemin local du modele ONNX non quantifie. Defaut :
"mirza/data/models/kokoro-v1.0.onnx".
voicesPath : str, optional
Chemin local du fichier de voix binaire. Defaut :
"mirza/data/models/voices-v1.0.bin".
"""
self.config = ttsConfig
self.modelPath = modelPath
self.voicesPath = voicesPath
self.kokoro = None
self._styleCache: dict[str, np.ndarray] = {}
def _resolveVoice(self, voiceName: str):
"""
Resout le nom de voix ou le mixage de styles vocaux.
Prend en charge :
- Nom de voix direct : "ff_siwis", "bm_george", "bm_lewis", etc.
- Presets :
- "fr_male_soft" : voix masculine douce et feutree (45% ff_siwis + 55% bm_lewis)
- "fr_male_warm" : voix masculine chaleureuse (45% ff_siwis + 55% am_fenrir)
- "fr_male_calm" : voix masculine posee (40% ff_siwis + 60% bm_lewis)
- Mixage pondere personnalise : "voix1:poids1+voix2:poids2"
Exemple : "ff_siwis:0.45+bm_lewis:0.55"
"""
PRESETS = {
"fr_siwis": "ff_siwis",
"fr_female": "ff_siwis",
"fr_default": "ff_siwis",
"fr_male_soft": "ff_siwis",
"fr_male_warm": "ff_siwis",
"fr_male_calm": "ff_siwis",
}
voiceSpec = PRESETS.get(voiceName.strip().lower(), voiceName.strip())
if "+" in voiceSpec:
if hasattr(self, "_styleCache") and voiceSpec in self._styleCache:
return self._styleCache[voiceSpec]
parts = voiceSpec.split("+")
items = []
totalWeight = 0.0
for part in parts:
part = part.strip()
if not part:
continue
if ":" in part:
vName, wStr = part.split(":", 1)
try:
w = float(wStr)
except ValueError:
w = 1.0
else:
vName = part
w = 1.0
items.append((vName.strip(), w))
totalWeight += w
if not items or totalWeight <= 0:
return voiceName
blended = None
for vName, w in items:
normW = w / totalWeight
try:
vStyle = self.kokoro.get_voice_style(vName)
if blended is None:
blended = normW * vStyle
else:
blended = blended + normW * vStyle
except Exception as exc:
logger.warning("Erreur chargement style de voix %s : %s", vName, exc)
if blended is not None:
if not hasattr(self, "_styleCache"):
self._styleCache = {}
self._styleCache[voiceSpec] = blended
return blended
if voiceName.lower() in ("george", "bm_george"):
return "bm_george"
return voiceName
def _ensureEngine(self) -> None:
"""
Assure le telechargement et l'initialisation a la demande du moteur Kokoro.
"""
if self.kokoro is not None:
return
import kokoro_onnx
if not os.path.exists(self.modelPath):
# Modele officiel v1.0 multilingue (~310 Mo)
url = "https://github.com/thewh1teagle/kokoro-onnx/releases/download/model-files-v1.0/kokoro-v1.0.onnx"
urllib.request.urlretrieve(url, self.modelPath)
if not os.path.exists(self.voicesPath):
# Pack de voix binaire, compatible kokoro-onnx
url = "https://github.com/thewh1teagle/kokoro-onnx/releases/download/model-files-v1.0/voices-v1.0.bin"
urllib.request.urlretrieve(url, self.voicesPath)
self.kokoro = kokoro_onnx.Kokoro(self.modelPath, self.voicesPath)
[docs]
def warmUp(self) -> None:
"""
Prechauffe le runtime ONNX par une courte synthese initiale, pour
supprimer la latence du premier tour.
"""
try:
self._ensureEngine()
if self.kokoro is not None:
self.synthesize("Bonjour.")
except Exception as exc:
logger.warning("Prechauffage Kokoro impossible, la synthese sera plus lente au premier tour : %s", exc)
[docs]
async def synthesizeAsync(self, text: str) -> tuple[np.ndarray | None, int | None]:
"""
Synthetise un texte en audio de maniere asynchrone (delegue a synthesize).
Parameters
----------
text : str
Texte a synthetiser.
Returns
-------
tuple[np.ndarray | None, int | None]
Echantillons audio float32 et taux d'echantillonnage, ou (None, None).
"""
return self.synthesize(text)
[docs]
def synthesize(self, text: str) -> tuple[np.ndarray | None, int | None]:
"""
Genere l'audio phonetique via le runtime local Kokoro ONNX.
Parameters
----------
text : str
Texte a synthetiser.
Returns
-------
tuple[np.ndarray | None, int | None]
Echantillons audio float32 et frequence d'echantillonnage, ou (None, None).
"""
if not text or not text.strip():
return None, None
try:
self._ensureEngine()
if self.kokoro is None:
return None, None
cleanText = text.strip()
rawVoice = getattr(self.config, "voice", "ff_siwis") or "ff_siwis"
voiceTarget = self._resolveVoice(rawVoice)
# Vitesse de diction : 1.30 par defaut, surcharge par la config
speedFactor = 1.30
voiceSpeed = getattr(self.config, "voiceSpeedRate", None)
if voiceSpeed:
rateStr = voiceSpeed.strip().replace("%", "")
try:
speedFactor = max(0.5, min(2.0, 1.0 + (float(rateStr) / 100.0)))
except ValueError:
speedFactor = 1.15
# Langue : Kokoro attend un code a deux lettres et un pays (fr-fr par defaut)
rawLang = getattr(self.config, "language", "fr-fr") or "fr-fr"
langCode = rawLang.lower()
if langCode in ("fr", "fr-fr"):
langCode = "fr-fr"
elif langCode in ("en", "us", "en-us"):
langCode = "en-us"
elif langCode in ("gb", "en-gb", "uk"):
langCode = "en-gb"
else:
langCode = "fr-fr"
samples, sampleRate = self.kokoro.create(
cleanText,
voice=voiceTarget,
speed=speedFactor,
lang=langCode,
)
return samples.astype(np.float32), sampleRate
except Exception as exc:
logger.error("Erreur synthese Kokoro : %s", exc)
return None, None
[docs]
def unload(self) -> None:
"""
Libere la session ONNX Kokoro.
Relaisable : ``synthesize`` reconstruit la session au prochain appel.
"""
self.kokoro = None
self._styleCache.clear()
logger.info("Moteur Kokoro ONNX decharge.")
@property
def isLoaded(self) -> bool:
"""
Indique si les poids resident en memoire.
Returns
-------
bool
True une fois la session ONNX ouverte.
"""
return self.kokoro is not None