diarizationNative

app.py

@@ -1,113 +1,59 @@
 import gradio as gr
 import librosa
 import numpy as np
-import whisperx
+from sklearn.preprocessing import StandardScaler
+from sklearn.cluster import KMeans
 from transformers import pipeline
-from pydub import AudioSegment
-import os
-import scipy.signal as signal
-import torch
-from pydub.utils import mediainfo
-from pydub.silence import detect_nonsilent  # Correct import
-import pandas as pd
 
-# Load Hugging Face token
-hf_token = os.getenv('diarizationToken')
-
-print("Initializing Speech-to-Text Model...")
+print("Chargement du modèle Wav2Vec2...")  
 stt_pipeline = pipeline("automatic-speech-recognition", model="boumehdi/wav2vec2-large-xlsr-moroccan-darija")
-print("Model Loaded Successfully.")
-
-# Initialize WhisperX with diarization
-device = "cuda" if torch.cuda.is_available() else "cpu"
-whisper_model = whisperx.load_model("large-v2", device)
-diarize_model = whisperx.DiarizationPipeline(use_auth_token=hf_token, device=device)
-print("WhisperX Model Loaded Successfully.")
-
-def remove_phone_tonalities(audio, sr):
-    nyquist = 0.5 * sr
-    low_cut = 300 / nyquist
-    high_cut = 3400 / nyquist
-    b, a = signal.butter(1, [low_cut, high_cut], btype='band')
-    filtered_audio = signal.filtfilt(b, a, audio)
-    return filtered_audio
-
-def convert_audio_to_wav(audio_path):
-    """ Convert any supported audio format to WAV. """
-    audio_info = mediainfo(audio_path)
-    print(f"Audio file info: {audio_info}")
-
-    if audio_info['format_name'] not in ['wav', 'mp3', 'flac', 'ogg']:  
-        raise ValueError(f"Unsupported audio format: {audio_info['format_name']}")
-
-    try:
-        sound = AudioSegment.from_file(audio_path)
-        wav_path = "converted_audio.wav"
-        sound.export(wav_path, format="wav")
-        return wav_path
-    except Exception as e:
-        print(f"Error converting audio: {e}")
-        raise
+print("Modèle chargé avec succès !")
 
 def process_audio(audio_path):
-    """ Process the audio: remove noise, split, diarize, and transcribe. """
-    print(f"Received audio file: {audio_path}")
+    print(f"Fichier reçu : {audio_path}")
 
     try:
-        # Load the audio file
+        # Charger uniquement les 30 premières secondes
         audio, sr = librosa.load(audio_path, sr=None, duration=30)
-        print(f"Audio loaded: {len(audio)} samples at {sr} Hz")
-        
-        # Remove phone tonalities
-        audio = remove_phone_tonalities(audio, sr)
-        print("Phone tonalities removed")
-
-        # Convert to AudioSegment for silence detection
-        sound = AudioSegment.from_wav(audio_path)
+        print(f"Audio chargé : {len(audio)} échantillons à {sr} Hz")
 
-        # Silence detection
-        min_silence_len = 1000  # Minimum silence length in ms
-        silence_thresh = sound.dBFS - 14  # Threshold for silence detection
+        # Extraction des MFCC
+        mfccs = librosa.feature.mfcc(y=audio, sr=sr, n_mfcc=13)
+        print(f"MFCC extrait, shape: {mfccs.shape}")
 
-        nonsilent_chunks = detect_nonsilent(sound, min_silence_len=min_silence_len, silence_thresh=silence_thresh)
+        # Normalisation
+        scaler = StandardScaler()
+        mfccs_scaled = scaler.fit_transform(mfccs.T)
+        print("MFCC normalisé.")
 
-        # Apply diarization
-        diarization = diarize_model(audio_path)
-
-        if isinstance(diarization, pd.DataFrame):
-            diarization = diarization.to_dict(orient="records")  
+        # Clustering avec KMeans
+        kmeans = KMeans(n_clusters=2, random_state=42, n_init=10)
+        speaker_labels = kmeans.fit_predict(mfccs_scaled)
+        print(f"Clustering terminé, {len(set(speaker_labels))} locuteurs détectés.")
 
+        # Segmentation et transcription
         transcriptions = []
+        segment_duration = len(audio) // len(speaker_labels)
 
-        for start, end in nonsilent_chunks:
-            chunk = sound[start:end]
-            chunk.export("chunk.wav", format="wav")
-
-            # Track start time manually
-            chunk_start_time = start / 1000.0  # Convert ms to seconds
+        print("Début de la transcription...")
+        for i in range(0, len(audio), sr * 5):
+            segment = audio[i : i + sr * 5]
+            if len(segment) < sr:
+                continue
 
-            chunk_audio, chunk_sr = librosa.load("chunk.wav", sr=None)
-            transcription = stt_pipeline(chunk_audio)  
+            transcription = stt_pipeline(segment)  # Transcription
+            transcriptions.append(f"Speaker {speaker_labels[i // segment_duration]}: {transcription['text']}")
+            print(f"Segment {i // sr}-{(i + sr * 5) // sr}s transcrit.")
 
-            # Match transcription segment with diarization
-            speaker_label = "Unknown"
-            for speaker in diarization:
-                spk_start, spk_end, label = speaker['start'], speaker['end'], speaker['label']
-                if spk_start <= chunk_start_time <= spk_end:  # Use manually tracked start time
-                    speaker_label = label
-                    break
-
-            transcriptions.append(f"Speaker {speaker_label}: {transcription['text']}")
-
-        os.remove("chunk.wav")  # Clean up temporary file
-        
+        print("Transcription terminée !")
         return "\n".join(transcriptions)
-    
+
     except Exception as e:
-        print(f"Error: {str(e)}")
-        return f"Error: {str(e)}"
+        print(f"Erreur : {e}")
+        return "Une erreur s'est produite."
 
-# Create Gradio interface
+# Interface Gradio
+print("Démarrage de Gradio...")
 iface = gr.Interface(
     fn=process_audio,
     inputs=gr.Audio(type="filepath"),
@@ -116,6 +62,5 @@ iface = gr.Interface(
     description="Upload an audio file to detect speakers and transcribe speech for each segment."
 )
 
-print("Launching Gradio Interface...")
 iface.launch()
-print("Gradio Interface Launched Successfully.")
+print("Interface lancée avec succès !")