"""
Moteur de synthese vocale asynchrone Edge-TTS Neural (Microsoft Cloud).
"""
import asyncio
import io
import threading
import aiohttp
import edge_tts
import numpy as np
import soundfile as sf
from ...config import TtsConfig
from ..interfaces.baseTts import BaseTts
[docs]
class PersistentTCPConnector(aiohttp.TCPConnector):
"""
Connecteur TCP persistant, qui empeche la fermeture prematuree du socket.
"""
[docs]
async def close(self):
"""
Empeche la fermeture du socket TCP par session unitaire.
"""
[docs]
def force_close(self):
"""
Ferme definitivement la socket TCP sous-jacente.
Returns
-------
object
Ce que rend la fermeture de la classe parente.
"""
return super().close()
[docs]
class EdgeTts(BaseTts):
"""
Service neural Edge-TTS, a tres faible latence.
Utilise une boucle evenementielle persistante et rogne les silences de bord.
Parameters
----------
ttsConfig : TtsConfig
Configuration contenant la voix, le debit et la hauteur de pitch.
Attributes
----------
config : TtsConfig
Instance de configuration du moteur TTS.
loop : asyncio.AbstractEventLoop
Boucle evenementielle asynchrone dediee dans un thread d'arriere-plan.
connector : PersistentTCPConnector | None
Connecteur TCP persistant, reutilise d'un appel a l'autre.
workerThread : threading.Thread
Thread executant la boucle evenementielle asynchrone.
"""
[docs]
def __init__(self, ttsConfig: TtsConfig):
"""
Initialise le moteur et demarre la boucle d'evenements en arriere-plan.
Parameters
----------
ttsConfig : TtsConfig
Configuration acoustique (voix, debit, pitch).
"""
self.config = ttsConfig
self.loop = asyncio.new_event_loop()
self.connector = None
self.workerThread = threading.Thread(target=self._runEventLoop, daemon=True)
self.workerThread.start()
def _runEventLoop(self) -> None:
"""
Execute la boucle evenementielle asyncio en arriere-plan.
"""
asyncio.set_event_loop(self.loop)
self.loop.run_forever()
[docs]
async def synthesizeAsync(self, text: str) -> tuple[np.ndarray | None, int | None]:
"""
Synthetise un flux audio via Edge-TTS de maniere asynchrone.
Parameters
----------
text : str
Texte a synthetiser vocalement.
Returns
-------
tuple[np.ndarray | None, int | None]
Echantillons audio float32 et frequence d'echantillonnage en Hz,
ou (None, None) en cas d'erreur.
"""
if self.connector is None or self.connector.closed:
self.connector = PersistentTCPConnector(keepalive_timeout=30)
comm = edge_tts.Communicate(
text,
self.config.voice,
rate=self.config.voiceSpeedRate,
pitch=self.config.voicePitch,
connector=self.connector,
)
mp3Buffer = bytearray()
async for chunk in comm.stream():
if chunk["type"] == "audio":
mp3Buffer.extend(chunk["data"])
if not mp3Buffer:
return None, None
audioArray, sampleRate = sf.read(io.BytesIO(mp3Buffer), dtype="float32")
# Suppression des silences de debut et de fin, pour un enchainement fluide
if len(audioArray) > 0 and sampleRate is not None:
threshold = 0.006
padSamples = int(sampleRate * 0.015) # 15 ms de fondu
nonZero = np.where(np.abs(audioArray) > threshold)[0]
if len(nonZero) > 0:
startIdx = max(0, nonZero[0] - padSamples)
endIdx = min(len(audioArray), nonZero[-1] + padSamples)
audioArray = audioArray[startIdx:endIdx]
return audioArray, sampleRate
[docs]
def synthesize(self, text: str) -> tuple[np.ndarray | None, int | None]:
"""
Synthetise un flux audio de maniere synchrone en interrogeant la boucle dediee.
Parameters
----------
text : str
Texte a synthetiser vocalement.
Returns
-------
tuple[np.ndarray | None, int | None]
Echantillons audio float32 et frequence d'echantillonnage en Hz,
ou (None, None) en cas d'erreur.
"""
try:
future = asyncio.run_coroutine_threadsafe(self.synthesizeAsync(text), self.loop)
return future.result(timeout=8.0)
except Exception as e:
print(f"[TTS (EDGE_TTS)] Erreur synthese: {e}")
if self.connector is not None:
self.connector = None
return None, None