Lundi

app.py

@@ -2,44 +2,61 @@ import gradio as gr
 import librosa
 import numpy as np
 import soundfile as sf
-from transformers import pipeline
 import os
+from transformers import pipeline
+import torchaudio
+from pyannote.audio import Pipeline
 
-print("Chargement du modèle Wav2Vec2...")  
+# Charger le modèle de reconnaissance vocale
+print("Chargement du modèle Wav2Vec2...")
 stt_pipeline = pipeline("automatic-speech-recognition", model="boumehdi/wav2vec2-large-xlsr-moroccan-darija")
 print("Modèle chargé avec succès !")
 
+# Charger le pipeline de diarisation (détection des speakers)
+print("Chargement du modèle de diarisation...")
+diarization_pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization", use_auth_token="YOUR_HF_TOKEN")
+print("Modèle de diarisation chargé !")
+
 def reduce_noise(audio, sr):
-    """Réduction du bruit avec un filtre passe-haut et suppression des faibles amplitudes"""
+    """Réduction du bruit pour améliorer la transcription"""
     audio = librosa.effects.preemphasis(audio)
     noise_threshold = np.percentile(np.abs(audio), 10)
     audio = np.where(np.abs(audio) > noise_threshold, audio, 0)
     return audio
 
-def segment_audio(audio, sr, segment_length=2.0, silence_threshold=0.01):
+def diarize_audio(audio_path):
+    """
+    Diarisation de l'audio : détecte qui parle et à quel moment.
+    Retourne une liste de (speaker, début, fin).
+    """
+    diarization = diarization_pipeline(audio_path)
+    speaker_segments = {}
+
+    for turn, _, speaker in diarization.itertracks(yield_label=True):
+        start, end = turn.start, turn.end
+        if speaker not in speaker_segments:
+            speaker_segments[speaker] = []
+        speaker_segments[speaker].append((start, end))
+
+    return speaker_segments
+
+def merge_speaker_segments(audio, sr, speaker_segments):
     """
-    Découpe l'audio en segments courts de ~0.5s à 3s, en détectant les silences.
+    Fusionne les segments d’un même speaker pour améliorer la précision.
+    Retourne un dictionnaire {speaker: signal_audio_fusionné}.
     """
-    intervals = librosa.effects.split(audio, top_db=20)  # Détecte les zones non silencieuses
-    segments = []
-
-    for start, end in intervals:
-        segment = audio[start:end]
-        duration = (end - start) / sr  # Convertir en secondes
-
-        # S'assurer que le segment est dans l'intervalle 0.5s - 3s
-        if duration < 0.5:
-            continue  # Trop court, on l'ignore
-        elif duration > 3.0:
-            num_subsegments = int(duration / segment_length)
-            subsegment_length = len(segment) // num_subsegments
-            for i in range(num_subsegments):
-                subseg = segment[i * subsegment_length: (i + 1) * subsegment_length]
-                segments.append(subseg)
-        else:
-            segments.append(segment)
-
-    return segments
+    merged_audio = {}
+
+    for speaker, segments in speaker_segments.items():
+        combined_audio = np.array([])
+        for start, end in segments:
+            start_sample = int(start * sr)
+            end_sample = int(end * sr)
+            combined_audio = np.concatenate((combined_audio, audio[start_sample:end_sample]))
+
+        merged_audio[speaker] = combined_audio
+
+    return merged_audio
 
 def process_audio(audio_path):
     print(f"Fichier reçu : {audio_path}")
@@ -52,23 +69,25 @@ def process_audio(audio_path):
         # Réduction du bruit
         audio = reduce_noise(audio, sr)
 
-        # Découpage de l’audio en segments courts
-        segments = segment_audio(audio, sr, segment_length=2.0)
-        print(f"Nombre de segments générés : {len(segments)}")
+        # Étape de diarisation : détection des speakers
+        speaker_segments = diarize_audio(audio_path)
+        print(f"Speakers détectés : {list(speaker_segments.keys())}")
+
+        # Fusionner les segments de chaque speaker
+        merged_audio = merge_speaker_segments(audio, sr, speaker_segments)
 
-        # Transcrire chaque segment
+        # Transcrire chaque speaker
         result = []
-        for i, segment in enumerate(segments):
-            temp_filename = f"temp_segment_{i}.wav"
-            sf.write(temp_filename, np.array(segment), sr)
+        for speaker, audio_data in merged_audio.items():
+            temp_filename = f"temp_{speaker}.wav"
+            sf.write(temp_filename, np.array(audio_data), sr)
 
-            # Transcription du segment
+            # Transcription du segment fusionné
             transcription = stt_pipeline(temp_filename)
             text = transcription["text"].strip()
 
-            # Vérification du contenu
             if text:
-                result.append(f"Segment {i+1}: {text}")
+                result.append(f"{speaker}: {text}")
 
             # Supprimer le fichier temporaire
             os.remove(temp_filename)
@@ -88,8 +107,8 @@ iface = gr.Interface(
     fn=process_audio,
     inputs=gr.Audio(type="filepath"),
     outputs="text",
-    title="Transcription optimisée",
-    description="Upload un fichier audio pour transcription par segments courts."
+    title="Transcription avec Diarisation",
+    description="Upload un fichier audio pour une transcription avec détection des speakers."
 )
 
 iface.launch()