Update app.py

app.py

@@ -1,20 +1,23 @@
 # -*- coding: utf-8 -*-
-"""RobotsMali_ASR_Demo.ipynb
-Automatically generated by Colab.
-Original file is located at
-    https://colab.research.google.com/drive/1fCpSvqwoSbpEBC62cZrQuqQGr4U1BNsh
+"""RobotsMali_ASR_Demo.ipynb - Script Final pour Démo Fluide et Stable
+Version optimisée pour la RAM, la vitesse et l'affichage 'Lyrics'.
 """
 import gradio as gr
-from transformers import pipeline
 import time
 import os
+import librosa
+import soundfile as sf
+import numpy as np
+
+# --- IMPORTS NEMO ---
+import nemo.collections.asr as nemo_asr
+# --------------------
 
 # ----------------------------------------------------------------------
-# 1. CONFIGURATION DES MODÈLES
+# CONSTANTES DE CONFIGURATION
 # ----------------------------------------------------------------------
-# Liste des identifiants exacts des modèles RobotsMali
 ROBOTSMALI_MODELS = [
-    "RobotsMali/soloba-ctc-0.6b-v0",  # Le modèle qui posait problème
+    "RobotsMali/soloba-ctc-0.6b-v0",  
     "RobotsMali/soloni-114m-tdt-ctc-v1",
     "RobotsMali/soloni-114m-tdt-ctc-V0",
     "RobotsMali/stt-bm-quartznet5x5-V0",
@@ -22,109 +25,200 @@ ROBOTSMALI_MODELS = [
     "RobotsMali/soloba-ctc-0.6b-v1"
 ]
 
-# Cache pour stocker les pipelines ASR déjà chargés.
+CHUNK_DURATION_SEC = 25  # Durée par segment (secondes) pour économiser la RAM
+SR_TARGET = 16000        # Taux d'échantillonnage cible pour NeMo ASR (16kHz)
+
+# Cache pour stocker les modèles NeMo chargés.
 asr_pipelines = {}
 
+# ----------------------------------------------------------------------
+# 1. FONCTIONS DE GESTION DES MODÈLES (CHARGEMENT + WARM-UP)
+# ----------------------------------------------------------------------
 def load_pipeline(model_name):
     """
-    Charge le pipeline ASR (modèle + processeur) pour un modèle donné et le met en cache.
+    Charge le modèle NeMo, le met en cache et effectue un warm-up.
     """
     if model_name not in asr_pipelines:
-        print(f"-> Tentative de chargement du modèle: {model_name}...")
+        print(f"-> Tentative de chargement du modèle NeMo: {model_name}...")
+        temp_warmup_file = "dummy_warmup.wav"
+
         try:
-            # Cette ligne charge le modèle directement depuis le Hub Hugging Face
-            asr_pipelines[model_name] = pipeline(
-                "automatic-speech-recognition",
-                model=model_name,
-                # device=0  # Décommenter si un GPU est disponible et souhaité
-            )
-            print(f"-> Modèle {model_name} chargé avec succès.")
+            # 🚀 CHARGEMENT NEMO
+            model_instance = nemo_asr.models.ASRModel.from_pretrained(model_name=model_name)
+            model_instance.eval() 
+
+            asr_pipelines[model_name] = model_instance
+            print(f"-> Modèle NeMo {model_name} chargé avec succès.")
+
+            # ----------------------------------------------------
+            # WARM-UP (Inférence à blanc)
+            # ----------------------------------------------------
+            print(f"   [Warmup] Exécution d'une inférence à blanc...")
+            
+            dummy_audio = np.random.randn(SR_TARGET).astype(np.float32) # 1s d'audio
+            sf.write(temp_warmup_file, dummy_audio, SR_TARGET)
+
+            model_instance.transcribe([temp_warmup_file], batch_size=1)
+            
+            print(f"   [Warmup] Terminé.")
+            # ----------------------------------------------------
+
         except Exception as e:
-            # Meilleure gestion d'erreur : affiche la trace complète dans les logs du Space
-            print(f"!!! Erreur de chargement pour {model_name}: {e}")
-            # Si vous avez besoin de la trace complète dans les logs du Space
-            # import traceback; traceback.print_exc() 
-            raise RuntimeError(f"Impossible de charger le modèle {model_name}. Vérifiez sa configuration.")
-    return asr_pipelines[model_name]
+            if model_name in asr_pipelines:
+                 del asr_pipelines[model_name]
+            print(f"!!! Erreur de chargement NeMo pour {model_name}: {e}")
+            raise RuntimeError(f"Impossible de charger le modèle {model_name}. Détail: {e}")
+            
+        finally:
+            if os.path.exists(temp_warmup_file):
+                os.remove(temp_warmup_file)
+            
+    return asr_pipelines.get(model_name)
 
 # ----------------------------------------------------------------------
-# 2. FONCTION PRINCIPALE D'INFÉRENCE
+# 2. FONCTION PRINCIPALE D'INFÉRENCE AVEC STREAMING ET DÉCOUPAGE
 # ----------------------------------------------------------------------
-def transcribe_audio(model_name: str, audio_path: str) -> str:
+def transcribe_audio(model_name: str, audio_path: str):
     """
-    Effectue la transcription ASR sur un fichier audio en utilisant le modèle sélectionné.
+    Effectue la transcription ASR avec découpage (chunking) et streaming d'état.
     """
     if audio_path is None:
-        return "⚠️ Veuillez d'abord télécharger ou enregistrer un fichier audio."
+        yield "⚠️ Veuillez d'abord télécharger ou enregistrer un fichier audio."
+        return
     if not ROBOTSMALI_MODELS:
-        return "Liste de modèles ASR indisponible."
+        yield "Liste de modèles ASR indisponible."
+        return
     
     start_time = time.time()
+    model_short_name = model_name.split('/')[-1]
+    temp_chunk_paths = [] # Pour le nettoyage final
+
     try:
-        if model_name not in ROBOTSMALI_MODELS:
-            return f"Modèle sélectionné ({model_name}) non reconnu."
+        # ----------------------------------------------------------------
+        # ÉTAPE 1 : PRÉPARATION ET CHARGEMENT AUDIO
+        # ----------------------------------------------------------------
+        yield f"**[1/4] CHARGEMENT AUDIO...** Préparation du fichier original (Mono @ 16kHz). ⚙️"
+        
+        full_audio_data, sr = librosa.load(audio_path, sr=SR_TARGET, mono=True)
         
-        # Le modèle est chargé ou récupéré depuis le cache ici
-        asr_pipe = load_pipeline(model_name)
+        total_duration = len(full_audio_data) / SR_TARGET
+        samples_per_chunk = int(CHUNK_DURATION_SEC * SR_TARGET)
         
-        # S'assure que le fichier audio existe avant de l'envoyer au pipeline
-        if not os.path.exists(audio_path):
-             return "❌ Erreur : Fichier audio introuvable ou chemin invalide."
-             
-        result = asr_pipe(audio_path)
+        # ----------------------------------------------------------------
+        # ÉTAPE 2 : CHARGEMENT/VÉRIFICATION DU MODÈLE ET DÉCOUPAGE
+        # ----------------------------------------------------------------
+        yield f"**[2/4] PRÉ-CALCUL...** Chargement du modèle et découpage ({total_duration:.1f}s en segments de {CHUNK_DURATION_SEC}s). 🧠"
+        
+        asr_model = load_pipeline(model_name)
+        
+        # Logique de DÉCOUPAGE
+        audio_segments = []
+        for i in range(0, len(full_audio_data), samples_per_chunk):
+            audio_segments.append(full_audio_data[i:i + samples_per_chunk])
+            
+        num_chunks = len(audio_segments)
+        full_transcription_text = ""
+        
+        # ----------------------------------------------------------------
+        # ÉTAPE 3 : TRANSCRIPTION PAR SEGMENT
+        # ----------------------------------------------------------------
+        
+        for idx, segment_data in enumerate(audio_segments):
+            
+            # Message d'état clé pour l'utilisateur
+            yield f"**[3/4] TRANSCRIPTION EN COURS...** Analyse du segment {idx + 1}/{num_chunks}. ⏳" 
+            
+            # Écriture du chunk temporaire
+            chunk_path = f"{os.path.splitext(os.path.basename(audio_path))[0]}_chunk_{idx}.wav"
+            sf.write(chunk_path, segment_data, SR_TARGET)
+            temp_chunk_paths.append(chunk_path)
+            
+            # 🚀 INFÉRENCE NEMO
+            transcriptions = asr_model.transcribe([chunk_path], batch_size=1)
+            
+            # --- GESTION DE L'OBJET HYPOTHESIS (CORRIGÉE) ---
+            segment_text = ""
+            if transcriptions and transcriptions[0]:
+                hyp_object = transcriptions[0] 
+                
+                # Accède à l'attribut .text de l'objet Hypothesis
+                if hasattr(hyp_object, 'text'):
+                    segment_text = hyp_object.text.strip()
+                elif isinstance(hyp_object, str):
+                    segment_text = hyp_object.strip()
+                # Gère le cas où transcribe retourne une liste de listes
+                elif isinstance(hyp_object, list) and hasattr(hyp_object[0], 'text'):
+                    segment_text = hyp_object[0].text.strip()
+                
+            if not segment_text:
+                segment_text = "[Transcription vide]"
+                
+            # Ajout d'un double saut de ligne pour le format "Lyrics" (paragraphe par segment)
+            full_transcription_text += segment_text + "\n\n" 
+            
+        # ----------------------------------------------------
+        # ÉTAPE 4 : RÉSULTAT FINAL
+        # ----------------------------------------------------
         end_time = time.time()
         duration = end_time - start_time
         
-        transcription_text = result.get("text", "Transcription non disponible.")
-        model_short_name = model_name.split('/')[-1]
+        transcription_text_final = full_transcription_text.strip()
         
-        output = f"**Modèle Utilisé :** `{model_short_name}`\n\n"
+        # 1. EN-TÊTE D'INFORMATION
+        output = f"**Modèle Utilisé :** `{model_short_name}` (NeMo)\n"
+        output += f"**Durée de l'Audio :** {total_duration:.1f} secondes\n"
+        output += f"**Temps de Traitement Total :** {duration:.2f} secondes\n"
+        output += f"**DÉCOUPAGE :** {CHUNK_DURATION_SEC} secondes ({num_chunks} segments)\n"
         output += f"***\n"
-        output += f"**Temps d'inférence (hors chargement) :** {duration:.2f} secondes\n"
-        output += f"**RÉSULTAT DE LA TRANSCRIPTION :**\n"
-        output += f"**{transcription_text.strip()}**"
         
-        return output
+        # 2. PRÉSENTATION LYRICS PROPRE
+        output += "**RÉSULTAT DE LA TRANSCRIPTION (Lyrics) :**\n"
+        # Utilisation du bloc de citation Markdown pour la structure
+        output += ">>> " + transcription_text_final.replace('\n\n', '\n>>> ')
+        
+        # 3. NOTE FINALE
+        output += "\n\n*Note : Audio converti en **Mono @ 16kHz** pour la transcription.*"
+        
+        # Le dernier 'yield' envoie le résultat final
+        yield output
     
     except RuntimeError as e:
-        # Ceci capture l'erreur de chargement (si elle n'a pas été capturée au pré-chargement)
-        return f"❌ Erreur critique : {str(e)}"
+        yield f"❌ Erreur critique lors du chargement : {str(e)}"
     except Exception as e:
-        return f"❌ Erreur lors de la transcription avec {model_name}: {e}"
+        # Affiche le texte partiel en cas d'erreur
+        if 'full_transcription_text' in locals() and full_transcription_text:
+             yield f"❌ Erreur lors de la transcription, le traitement s'est arrêté. Texte partiel:\n>>> {full_transcription_text.strip().replace('\n\n', '\n>>> ')}"
+        yield f"❌ Erreur générale : {e}"
+    finally:
+        # Nettoyage
+        for chunk_path in temp_chunk_paths:
+            if os.path.exists(chunk_path):
+                os.remove(chunk_path)
+        print(f"-> {len(temp_chunk_paths)} fichiers temporaires de segments supprimés.")
+
 
 # ----------------------------------------------------------------------
-# 2.5. PRÉ-CHARGEMENT DU MODÈLE PAR DÉFAUT (NOUVEAU)
+# 4. PRÉ-CHARGEMENT ET INTERFACE GRADIO
 # ----------------------------------------------------------------------
 
-INITIAL_DESCRIPTION = "Sélectionnez un modèle ASR de RobotsMali, puis enregistrez ou téléchargez un fichier audio (MP3 recommandé) pour obtenir la transcription."
+INITIAL_DESCRIPTION = "Sélectionnez un modèle ASR de RobotsMali, puis enregistrez ou téléchargez un fichier audio pour obtenir la transcription."
 
 if ROBOTSMALI_MODELS:
     default_model = ROBOTSMALI_MODELS[0]
     try:
-        # Tente de charger le modèle par défaut au démarrage du script
-        load_pipeline(default_model)
+        load_pipeline(default_model) 
         default_model_short_name = default_model.split('/')[-1]
         INITIAL_DESCRIPTION = (
-            f"✅ Le modèle par défaut `{default_model_short_name}` a été préchargé avec succès. "
+            f"✅ Le modèle par défaut `{default_model_short_name}` (NeMo) a été **préchargé et réchauffé** avec succès. "
             f"Téléchargez ou enregistrez votre audio pour transcrire."
         )
     except RuntimeError as e:
-        # Si le chargement échoue, informe l'utilisateur que le modèle par défaut est cassé
         default_model_short_name = default_model.split('/')[-1]
         INITIAL_DESCRIPTION = (
-            f"❌ ERREUR CRITIQUE AU DÉMARRAGE : Impossible de charger le modèle "
-            f"`{default_model_short_name}`. "
-            f"**Veuillez sélectionner un autre modèle dans la liste** (ex: le second). "
+            f"❌ ERREUR CRITIQUE AU DÉMARRAGE : Impossible de charger le modèle `{default_model_short_name}`. "
+            f"**Veuillez sélectionner un autre modèle dans la liste**. "
             f"Détails de l'erreur : {str(e)}"
         )
-    except Exception:
-        # Capture toute autre erreur non gérée par RuntimeError
-        INITIAL_DESCRIPTION = "❌ ERREUR CRITIQUE AU DÉMARRAGE : Problème inconnu lors du pré-chargement du modèle."
-
-
-# ----------------------------------------------------------------------
-# 3. INTERFACE GRADIO ET LANCEMENT
-# ----------------------------------------------------------------------
 
 model_dropdown = gr.Dropdown(
     label="1. Sélectionner un Modèle RobotsMali",
@@ -146,11 +240,9 @@ interface = gr.Interface(
     fn=transcribe_audio,
     inputs=[model_dropdown, audio_input],
     outputs=text_output,
-    title="🤖 RobotsMali ASR Multi-Modèles (Test Colab)",
-    # Utilise la description générée par la phase de pré-chargement
+    title="🤖 RobotsMali ASR Multi-Modèles (Démo NeMo Fluide)",
     description=INITIAL_DESCRIPTION, 
     allow_flagging="never")
 
-# Lancement de l'Interface Gradio sur Colab
 print("Lancement de l'interface Gradio...")
 interface.launch(share=True)