14-02-24 10 25

app.py

@@ -11,17 +11,18 @@ 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...")
 stt_pipeline = pipeline("automatic-speech-recognition", model="boumehdi/wav2vec2-large-xlsr-moroccan-darija")
 print("Model Loaded Successfully.")
 
-# Initialize WhisperX with diarization (not transcription)
+# 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 for Diarization.")
+print("WhisperX Model Loaded Successfully.")
 
 def remove_phone_tonalities(audio, sr):
     nyquist = 0.5 * sr
@@ -31,18 +32,15 @@ def remove_phone_tonalities(audio, sr):
     filtered_audio = signal.filtfilt(b, a, audio)
     return filtered_audio
 
-
-
 def convert_audio_to_wav(audio_path):
-    # Check the audio file format before conversion
+    """ 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']:  # Add other valid formats if necessary
+
+    if audio_info['format_name'] not in ['wav', 'mp3', 'flac', 'ogg']:  
         raise ValueError(f"Unsupported audio format: {audio_info['format_name']}")
-    
+
     try:
-        # Convert any audio format to WAV using pydub
         sound = AudioSegment.from_file(audio_path)
         wav_path = "converted_audio.wav"
         sound.export(wav_path, format="wav")
@@ -51,88 +49,57 @@ def convert_audio_to_wav(audio_path):
         print(f"Error converting audio: {e}")
         raise
 
-
-
-
-hf_token = os.getenv('diarizationToken')
-
-print("Initializing Speech-to-Text Model...")
-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 process_audio(audio_path):
+    """ Process the audio: remove noise, split, diarize, and transcribe. """
     print(f"Received audio file: {audio_path}")
-    
+
     try:
-        # Load the audio file using librosa
+        # Load the audio file
         audio, sr = librosa.load(audio_path, sr=None, duration=30)
         print(f"Audio loaded: {len(audio)} samples at {sr} Hz")
         
-        # Remove phone tonalities (if any)
+        # 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)
-        
-        # Silence detection: split based on silence
-        min_silence_len = 1000  # minimum silence length in ms
-        silence_thresh = sound.dBFS - 14  # threshold for silence (adjust as needed)
-        
-        # Correct usage of detect_nonsilent from pydub.silence
-        nonsilent_chunks = detect_nonsilent(
-            sound,
-            min_silence_len=min_silence_len,
-            silence_thresh=silence_thresh
-        )
-        
-        non_silent_chunks = [
-            sound[start:end] for start, end in nonsilent_chunks
-        ]
-        
-        # Apply diarization (WhisperX)
+
+        # Silence detection
+        min_silence_len = 1000  # Minimum silence length in ms
+        silence_thresh = sound.dBFS - 14  # Threshold for silence detection
+
+        nonsilent_chunks = detect_nonsilent(sound, min_silence_len=min_silence_len, silence_thresh=silence_thresh)
+
+        # Apply diarization
         diarization = diarize_model(audio_path)
-        
-        # Check if diarization is a DataFrame and process accordingly
+
         if isinstance(diarization, pd.DataFrame):
-            print("Diarization is a DataFrame")
-            diarization = diarization.to_dict(orient="records")  # Convert DataFrame to a list of dicts
-        
+            diarization = diarization.to_dict(orient="records")  
+
         transcriptions = []
-        for chunk in non_silent_chunks:
+
+        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
+
             chunk_audio, chunk_sr = librosa.load("chunk.wav", sr=None)
-            transcription = stt_pipeline(chunk_audio)  # Transcribe using Wav2Vec2
-            
-            # Match transcription segment with diarization result
+            transcription = stt_pipeline(chunk_audio)  
+
+            # Match transcription segment with diarization
             speaker_label = "Unknown"
             for speaker in diarization:
                 spk_start, spk_end, label = speaker['start'], speaker['end'], speaker['label']
-                # Adjust timestamp matching
-                if spk_start <= (chunk.start_time / 1000) <= spk_end:  # Convert ms to seconds
+                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']}")
-        
-        # Clean up temporary files
-        os.remove("chunk.wav")
+
+        os.remove("chunk.wav")  # Clean up temporary file
         
         return "\n".join(transcriptions)