Upload app (1).py

app (1).py

@@ -0,0 +1,391 @@
+#!/usr/bin/env python
+# coding: utf-8
+
+# In[1]:
+
+
+#!/usr/bin/env python
+# coding: utf-8
+
+# In[3]:
+
+
+#!pip install torchaudio
+
+
+# In[2]:
+
+
+from IPython.display import Audio
+import IPython.display as ipd
+from scipy.io import wavfile
+import numpy as np
+import warnings
+import re
+warnings.filterwarnings("ignore")
+import soundfile as sf
+import librosa
+import torch
+import os
+import soundfile as sf
+import librosa
+import noisereduce as nr
+import numpy as np
+import gradio as gr
+import pyloudnorm as pyln
+# import torchaudio
+from transformers import Wav2Vec2ForCTC, Wav2Vec2Tokenizer
+from transformers import AutoModelForCTC, AutoProcessor, AutoTokenizer, AutoModelForCausalLM
+from transformers import pipeline, AutoProcessor, AutoModelForSpeechSeq2Seq
+import pandas as pd
+from transformers import pipeline, AutoModelForAudioClassification, AutoProcessor
+
+
+# In[3]:
+
+
+# In[3]:
+
+
+# Set device and dtype
+device = "cuda:0" if torch.cuda.is_available() else "cpu"
+# device= "cpu"
+torch_dtype = torch.float16 if torch.cuda.is_available() else torch.float32
+lid_model_id = "facebook/mms-lid-126"
+lid_pipeline = pipeline("audio-classification", model=lid_model_id,device=device)
+language_mapping = {
+"hin": "hindi",
+"ben": "bengali",
+"eng": "english",
+"guj": "gujarati"
+}
+
+
+# In[4]:
+
+
+def detect_language_for_audio_file(audio_file_path, lid_pipeline, target_sampling_rate=16000):
+    """
+    Detects the language of a given audio file and returns a DataFrame.
+
+    Parameters:
+    - audio_file_path (str): The path to the audio file.
+    - lid_pipeline: The language identification pipeline.
+    - target_sampling_rate (int): The target sampling rate for the audio file. Default is 16000.
+
+    Returns:
+    - df (pd.DataFrame): A DataFrame containing the detected language and filename.
+    """
+    detected_languages = []
+    audio_filenames = []
+
+    filename = os.path.basename(audio_file_path)
+    waveform, original_sampling_rate = librosa.load(audio_file_path, sr=None)
+
+    if len(waveform.shape) > 1:
+        waveform = librosa.to_mono(waveform)
+
+    if original_sampling_rate != target_sampling_rate:
+        waveform = librosa.resample(waveform, orig_sr=original_sampling_rate, target_sr=target_sampling_rate)
+
+    # Perform language identification
+    lid_result = lid_pipeline(waveform, sampling_rate=target_sampling_rate)
+    detected_language = lid_result[0]['label'].split('_')[0]
+    print(f"Detected language for {filename}: {detected_language}")
+
+    detected_languages.append(detected_language)
+    audio_filenames.append(filename)
+
+    df = pd.DataFrame({
+        "Detected_Language": detected_languages,
+        "Audio_Filename": audio_filenames
+    })
+
+    # removing nondetected languages
+
+    df['Detected_Language'] = df['Detected_Language'].map(language_mapping)
+    
+    df.dropna(inplace=True, axis= 0)
+
+    # adding model names based on language
+    model_names = []
+
+    for index, row in df.iterrows():
+        detected_language = row['Detected_Language']
+    
+        model_name = "ai4bharat/indicwav2vec_v1_" + detected_language
+    
+        model_names.append(model_name)
+    
+    df['Model_Name'] = model_names
+
+    return df
+# Example usage:
+# audio_file_path = 'processed_audio.wav'
+# df = detect_language_for_audio_file(audio_file_path, lid_pipeline)
+# print(df)
+
+
+# In[11]:
+
+
+loaded_models = {}
+current_loaded_model = None
+
+def load_model_and_tokenizer(standardized_language):
+    global current_loaded_model
+    
+    # If the requested model is already loaded, return it
+    if standardized_language in loaded_models:
+        return loaded_models[standardized_language]
+    
+    # Check if the current loaded model is the same as the new one
+    if current_loaded_model == standardized_language:
+        return loaded_models[standardized_language]
+    
+    # Clear the specific model currently loaded on the GPU, if any
+    elif current_loaded_model is not None:
+        del loaded_models[current_loaded_model]
+        torch.cuda.empty_cache()
+        current_loaded_model = None
+    
+    # Determine the model name based on the standardized language
+    if standardized_language == 'hindi':
+        model_name = "ai4bharat/indicwav2vec-hindi"
+    elif standardized_language == 'odia':
+        model_name = "ai4bharat/indicwav2vec-odia"
+    elif standardized_language == 'english':
+        model_name = "facebook/wav2vec2-large-960h-lv60-self"
+    else:
+        model_name = "ai4bharat/indicwav2vec_v1_" + standardized_language
+    
+    # Load the model and tokenizer
+    model = Wav2Vec2ForCTC.from_pretrained(model_name)
+    tokenizer = Wav2Vec2Tokenizer.from_pretrained(model_name)
+    
+    # Update the loaded models and current loaded model
+    loaded_models[standardized_language] = (model, tokenizer)
+    current_loaded_model = standardized_language
+    
+    return model, tokenizer
+
+
+# In[6]:
+
+
+# In[5]:
+
+
+def perform_transcription(df):
+
+    transcriptions = []
+
+    for index, row in df.iterrows():
+        audio_file_path = row['Audio_Filename']
+        detected_language = row['Detected_Language']
+
+        standardized_language = language_mapping.get(detected_language, detected_language)
+        model, tokenizer = load_model_and_tokenizer(standardized_language)
+
+        input_audio, _ = librosa.load(audio_file_path, sr=16000)
+        input_values = tokenizer(input_audio, return_tensors="pt").input_values
+
+        with torch.no_grad():
+            logits = model(input_values).logits
+
+        predicted_ids = torch.argmax(logits, dim=-1)
+        text = tokenizer.batch_decode(predicted_ids)[0]
+
+        transcriptions.append(text)
+
+    df['Transcription'] = transcriptions
+    
+    return df
+
+
+# In[8]:
+
+
+# In[7]:
+
+
+# Loading the tokenizer and model from Hugging Face's model hub.
+tokenizer = AutoTokenizer.from_pretrained("soketlabs/pragna-1b", token=os.environ.get('HF_TOKEN'))
+model = AutoModelForCausalLM.from_pretrained(
+    "soketlabs/pragna-1b",
+    token=os.environ.get('HF_TOKEN'),
+    revision='3c5b8b1309f7d89710331ba2f164570608af0de7'
+)
+model.load_adapter('soketlabs/pragna-1b-it-v0.1', token=os.environ.get('HF_TOKEN'))
+model = model.to(device)
+
+
+# Function to generate response
+def generate_response(transcription):
+    try:
+        messages = [
+            {"role": "system", "content": " you are a friendly bot to help the user"},
+            {"role": "user", "content": transcription},
+        ]
+        tokenized_chat = tokenizer.apply_chat_template(messages, tokenize=True, add_generation_prompt=True, return_tensors="pt")
+        input_ids = tokenized_chat[0].to(device)
+        if len(input_ids.shape) == 1:
+            input_ids = input_ids.unsqueeze(0)
+        with torch.no_grad():
+            output = model.generate(
+                input_ids,
+                max_new_tokens=300,
+                do_sample=True,
+                top_k=5,
+                num_beams=1,
+                use_cache=False,
+                temperature=0.2,
+                repetition_penalty=1.1,
+            )
+        generated_text = tokenizer.decode(output[0], skip_special_tokens=True)
+        return find_last_sentence(generated_text)
+    except Exception as e:
+        print("Error during response generation:", e)
+        return "Response generation error: " + str(e)
+
+# Function to find last sentence in generated text
+def find_last_sentence(text):
+    sentence_endings = re.finditer(r'[।?!]', text)
+    end_positions = [ending.end() for ending in sentence_endings]
+    if end_positions:
+        return text[:end_positions[-1]]
+    return text
+
+
+# In[9]:
+
+
+# In[15]:
+
+
+def generate_text_and_display_audio(row, model, tokenizer):
+    audio_file = row['Audio_Filename']
+    transcription = row['Transcription']
+    
+    # Generate text
+    generated_text = generate_response(transcription)
+    
+    generated_text = find_last_sentence(generated_text)
+    # Display audio
+    # display(ipd.Audio(audio_path))
+    return transcription, generated_text
+    # Display prompt and generated text
+    # print("Transcribed Text:", transcription)
+    # print("Generated Text:", generated_text)
+
+
+# In[12]:
+
+
+# In[16]:
+
+def spectral_subtraction(audio_data, sample_rate):
+    # Compute short-time Fourier transform (STFT)
+    stft = librosa.stft(audio_data)
+
+    # Compute power spectrogram
+    power_spec = np.abs(stft)**2
+
+    # Estimate noise power spectrum
+    noise_power = np.median(power_spec, axis=1)
+
+    # Apply spectral subtraction
+    alpha = 2.0  # Adjustment factor, typically between 1.0 and 2.0
+    denoised_spec = np.maximum(power_spec - alpha * noise_power[:, np.newaxis], 0)
+
+    # Inverse STFT to obtain denoised audio
+    denoised_audio = librosa.istft(np.sqrt(denoised_spec) * np.exp(1j * np.angle(stft)))
+
+    return denoised_audio
+
+def apply_compression(audio_data, sample_rate):
+    # Apply dynamic range compression
+    meter = pyln.Meter(sample_rate)  # create BS.1770 meter
+    loudness = meter.integrated_loudness(audio_data)
+    
+    # Normalize audio to target loudness of -24 LUFS
+    loud_norm = pyln.normalize.loudness(audio_data, loudness, -24.0)
+    
+    return loud_norm
+
+def process_audio(audio_file_path):
+    try:
+        # Read audio data
+        audio_data, sample_rate = librosa.load(audio_file_path)
+        print(f"Read audio data: {audio_file_path}, Sample Rate: {sample_rate}")
+
+        # Apply noise reduction using noisereduce
+        reduced_noise = nr.reduce_noise(y=audio_data, sr=sample_rate)
+        print("Noise reduction applied")
+
+        # Apply spectral subtraction for additional noise reduction
+        denoised_audio = spectral_subtraction(reduced_noise, sample_rate)
+        print("Spectral subtraction applied")
+
+        # Apply dynamic range compression to make foreground louder
+        compressed_audio = apply_compression(denoised_audio, sample_rate)
+        print("Dynamic range compression applied")
+
+        # Remove silent spaces
+        final_audio = librosa.effects.trim(compressed_audio)[0]
+        print("Silences trimmed")
+
+        # Save the final processed audio to a file with a fixed name
+        processed_file_path = 'processed_audio.wav'
+        sf.write(processed_file_path, final_audio, sample_rate)
+        print(f"Processed audio saved to: {processed_file_path}")
+
+        # Check if file exists to confirm it was saved
+        if not os.path.isfile(processed_file_path):
+            raise FileNotFoundError(f"Processed file not found: {processed_file_path}")
+
+        # Load the processed audio for transcription
+        processed_audio_data, _ = librosa.load(processed_file_path)
+        print(f"Processed audio reloaded for transcription: {processed_file_path}")
+        
+        df = detect_language_for_audio_file(processed_file_path, lid_pipeline)
+        print(df)
+        df_transcription= perform_transcription(df)
+        print(df_transcription)
+        for index, row in df_transcription.iterrows():
+            print(index, row)
+            transcription, response = generate_text_and_display_audio(row, model, tokenizer)
+
+        
+        # Transcribe audio
+        # transcription = transcribe_audio(processed_audio_data)
+        # print("Transcription completed")
+
+        # # Generate response
+        # response = generate_response(transcription)
+        # print("Response generated")
+
+        return processed_file_path, transcription, response
+    except Exception as e:
+        print("Error during audio processing:", e)
+        return "Error during audio processing:", str(e)
+
+
+# Create Gradio interface
+iface = gr.Interface(
+    fn=process_audio,
+    inputs=gr.Audio(label="Record Audio", type="filepath"),
+    outputs=[gr.Audio(label="Processed Audio"), gr.Textbox(label="Transcription"), gr.Textbox(label="Response")]
+)
+
+iface.launch(share=True)
+
+
+# In[ ]:
+
+
+# In[ ]:
+
+
+
+