"""
Moteur de transcription vocale FastConformer Nemotron 3.5 8-bit via MLX Audio.
"""
import mlx.core as mx
import numpy as np
from mlx_audio.stt import load
from ...core.logger import getMirzaLogger
from ...core.textCleaner import TextCleaner
from ..interfaces.baseStt import BaseStt
logger = getMirzaLogger("mirza.stt.nemotron")
[docs]
class MlxNemotronStt(BaseStt):
"""
FastConformer-RNNT Nemotron 3.5 ASR local sur Apple Silicon (MLX).
Modele streaming 600M quantifie en 8 bits, rapide et optimisé pour Apple Silicon.
"""
[docs]
def __init__(self, modelRepo: str = "mlx-community/nemotron-3.5-asr-streaming-0.6b-8bit"):
"""
Initialise le moteur Nemotron ASR.
Parameters
----------
modelRepo : str, optional
Depot HuggingFace du modele FastConformer quantifie. Defaut :
"mlx-community/nemotron-3.5-asr-streaming-0.6b-8bit".
"""
self.modelRepo = modelRepo
self.model = None
[docs]
def warmUp(self) -> None:
"""
Pre-charge integralement les poids Nemotron 8 bits et calibre le
pipeline Metal.
"""
try:
if self.model is None:
self.model = load(self.modelRepo)
# Audio silencieux : il pre-charge le pipeline Metal
dummyAudio = mx.zeros((16000,), dtype=mx.float32)
self.model.generate(dummyAudio, language="fr", chunk_duration=None)
mx.clear_cache()
print(f"[STT (MLX_NEMOTRON)] Nemotron 3.5 ASR 8-bit pré-chargé et calibré en VRAM ({self.modelRepo}).")
except Exception as e:
print(f"[STT (MLX_NEMOTRON)] Avertissement warmUp : {e}")
[docs]
def transcribe(
self,
audioArray: np.ndarray,
language: str = "fr",
initialPrompt: str = "",
) -> str:
"""
Transcrit le flux audio de maniere optimisee avec Nemotron FastConformer-RNNT.
Parameters
----------
audioArray : np.ndarray
Echantillons audio 16 kHz mono au format float32.
language : str, optional
Code langue ISO (ex: 'fr'). Defaut : "fr".
initialPrompt : str, optional
Invite textuelle initiale, non utilisee par FastConformer.
Defaut : "".
Returns
-------
str
Chaine transcrite filtree du bruit ou vide si aucun signal intelligible.
"""
if self.model is None:
self.warmUp()
try:
if isinstance(audioArray, np.ndarray):
if len(audioArray) == 0:
return ""
# Centrage DC et attenuation du clipping seulement : les signaux
# faibles sont amplifies avec un gain plafonne a 4x, pour
# compenser les micros eloignes sans saturer le mel-spectrogramme.
mxAudio = mx.array(audioArray, dtype=mx.float32)
else:
mxAudio = audioArray
# chunk_duration=None : un seul bloc pour les segments de moins de
# 30 s (commandes vocales typiques : 1 a 8 s). Segmenter degraderait
# la coherence acoustique sur les courts segments.
result = self.model.generate(mxAudio, language=language, chunk_duration=None)
mx.clear_cache()
rawText = getattr(result, "text", str(result)).strip()
cleanedText = TextCleaner.cleanSttText(rawText)
if TextCleaner.isNoiseTranscription(cleanedText):
return ""
return cleanedText
except Exception as e:
print(f"[STT (MLX_NEMOTRON)] Erreur transcription : {e}")
return ""
[docs]
def unload(self) -> None:
"""
Libere le reseau de neurones et les tenseurs Metal associes.
Relaisable : ``transcribe`` recharge paresseusement au prochain appel.
"""
self.model = None
logger.info("Modele Nemotron ASR decharge.")
@property
def isLoaded(self) -> bool:
"""
Indique si les poids resident en memoire.
Returns
-------
bool
True une fois le modele charge.
"""
return self.model is not None