""" Service ASR principal avec gestion MQTT et reconnaissance vocale Vosk """ import asyncio import json import logging import time from pathlib import Path from typing import Dict, Any, Optional import paho.mqtt.client as mqtt from .audio_capture import AudioCapture from .vosk_engine import VoskEngine from .config import ASRConfig class ASRService: """Service principal de reconnaissance vocale automatique.""" def __init__(self) -> None: self.logger = logging.getLogger(__name__) self.mqtt_client = mqtt.Client() self.audio_capture = AudioCapture() self.vosk_engine: Optional[VoskEngine] = None self.config = ASRConfig() # État de session ASR self.session_active = False self.vad_active = False self.audio_buffer: list[bytes] = [] self.last_audio_time = 0.0 # Configuration MQTT self.mqtt_client.on_connect = self._on_mqtt_connect self.mqtt_client.on_message = self._on_mqtt_message async def start(self) -> None: """Démarre le service ASR.""" self.logger.info("Initialisation du service ASR") # Initialiser le moteur Vosk self.vosk_engine = VoskEngine(self.config.lang, self.config.sample_rate) await self.vosk_engine.initialize() # Connecter MQTT self.mqtt_client.connect("127.0.0.1", 1883, 60) self.mqtt_client.loop_start() # Publier les capacités (retained) await self._publish_capabilities() # Démarrer la capture audio await self.audio_capture.start() # Démarrer la boucle de traitement await self._run_processing_loop() async def stop(self) -> None: """Arrête le service ASR.""" self.logger.info("Arrêt du service ASR") if self.audio_capture: await self.audio_capture.stop() if self.vosk_engine: await self.vosk_engine.cleanup() self.mqtt_client.loop_stop() self.mqtt_client.disconnect() def _on_mqtt_connect(self, client: mqtt.Client, userdata, flags, rc: int) -> None: """Callback de connexion MQTT.""" if rc == 0: self.logger.info("Connexion MQTT établie") # S'abonner aux topics nécessaires client.subscribe("audio/vad") client.subscribe("audio/rms") client.subscribe("asr/config") else: self.logger.error(f"Échec connexion MQTT: {rc}") def _on_mqtt_message( self, client: mqtt.Client, userdata, msg: mqtt.MQTTMessage ) -> None: """Traite les messages MQTT reçus.""" try: topic = msg.topic payload = json.loads(msg.payload.decode()) if topic == "audio/vad": self._handle_vad_message(payload) elif topic == "asr/config": self._handle_config_message(payload) except json.JSONDecodeError: self.logger.error(f"JSON invalide sur {msg.topic}: {msg.payload}") except Exception as e: self.logger.error(f"Erreur traitement message {msg.topic}: {e}") def _handle_vad_message(self, payload: Dict[str, Any]) -> None: """Traite les messages VAD.""" active = payload.get("active", False) end_of_speech = payload.get("end_of_speech", False) if active and not self.vad_active: # Début d'activité vocale self.logger.debug("VAD actif - début de session ASR") self.vad_active = True self._start_asr_session() elif not active and self.vad_active: # Fin d'activité vocale self.vad_active = False if end_of_speech and self.session_active: # Fin de phrase détectée self.logger.debug("EOS détecté - finalisation transcription") asyncio.create_task(self._finalize_transcription()) def _handle_config_message(self, payload: Dict[str, Any]) -> None: """Traite les messages de configuration.""" self.logger.info(f"Mise à jour configuration: {payload}") if "lang" in payload: self.config.lang = payload["lang"] if "eos_ms" in payload: self.config.eos_ms = payload["eos_ms"] if "vad" in payload: vad_config = payload["vad"] if "threshold" in vad_config: self.config.vad_threshold = vad_config["threshold"] # Redémarrer le moteur si nécessaire if "lang" in payload and self.vosk_engine: asyncio.create_task(self._reload_engine()) def _start_asr_session(self) -> None: """Démarre une nouvelle session ASR.""" if not self.session_active: self.session_active = True self.audio_buffer.clear() self.last_audio_time = time.time() if self.vosk_engine: self.vosk_engine.reset() async def _finalize_transcription(self) -> None: """Finalise la transcription et publie le résultat.""" if not self.session_active or not self.vosk_engine: return try: # Traiter l'audio bufferisé start_time = time.time() for audio_chunk in self.audio_buffer: self.vosk_engine.process_audio(audio_chunk) # Obtenir le résultat final result = self.vosk_engine.get_final_result() processing_time = time.time() - start_time self.logger.info(f"Transcription terminée en {processing_time:.3f}s") if result and result.strip(): # Publier le texte transcrit await self._publish_text(result) except Exception as e: self.logger.error(f"Erreur lors de la transcription: {e}") finally: self.session_active = False self.audio_buffer.clear() async def _publish_text(self, text: str) -> None: """Publie le texte transcrit sur MQTT.""" message = {"ts_ms": int(time.time() * 1000), "text": text.lower().strip()} self.mqtt_client.publish("asr/text", json.dumps(message)) self.logger.info(f"Texte publié: '{text}'") async def _publish_capabilities(self) -> None: """Publie les capacités du service (retained).""" capabilities = { "engine": "vosk", "lang": self.config.lang, "sample_rate": self.config.sample_rate, } self.mqtt_client.publish( "asr/capabilities", json.dumps(capabilities), retain=True ) self.logger.info("Capacités publiées") async def _reload_engine(self) -> None: """Recharge le moteur Vosk avec la nouvelle configuration.""" if self.vosk_engine: await self.vosk_engine.cleanup() self.vosk_engine = VoskEngine(self.config.lang, self.config.sample_rate) await self.vosk_engine.initialize() await self._publish_capabilities() async def _run_processing_loop(self) -> None: """Boucle principale de traitement audio.""" self.logger.info("Démarrage de la boucle de traitement") try: while True: # Vérifier timeout EOS if self.session_active and self.vad_active: elapsed = (time.time() - self.last_audio_time) * 1000 if elapsed > self.config.eos_ms: self.logger.debug(f"Timeout EOS atteint ({elapsed:.0f}ms)") await self._finalize_transcription() # Traiter l'audio en attente if self.session_active: audio_data = await self.audio_capture.get_audio_chunk() if audio_data: self.audio_buffer.append(audio_data) self.last_audio_time = time.time() await asyncio.sleep(0.01) # 10ms except asyncio.CancelledError: self.logger.info("Boucle de traitement interrompue") except Exception as e: self.logger.error( f"Erreur dans la boucle de traitement: {e}", exc_info=True )