NoiseReductionInASR.py

#!/usr/bin/env python
# coding: utf-8

# In[ ]:


get_ipython().system('pip install webrtcvad')


# In[ ]:


# import librosa
# import numpy as np
# import scipy.signal
# import webrtcvad
# import soundfile as sf  # New library for saving audio
# import matplotlib.pyplot as plt

# # Function to apply a high-pass filter
# def high_pass_filter(audio, sr, cutoff=100):
#     # Design a high-pass Butterworth filter
#     sos = scipy.signal.butter(10, cutoff, btype='highpass', fs=sr, output='sos')
#     filtered_audio = scipy.signal.sosfilt(sos, audio)
#     return filtered_audio

# # Function to apply Wiener filter for noise reduction
# def wiener_filter(audio):
#     return scipy.signal.wiener(audio)

# # Voice Activity Detection using WebRTC VAD
# def apply_vad(audio, sr, frame_duration=30, aggressiveness=3):
#     vad = webrtcvad.Vad(aggressiveness)  # aggressiveness: 0 (least aggressive) to 3 (most aggressive)
    
#     # Convert audio to 16-bit PCM (required by webrtcvad)
#     audio_int16 = np.int16(audio * 32767)  # assuming `audio` is in range [-1, 1]
    
#     frame_size = int(sr * frame_duration / 1000)  # frame size in samples
#     frames = [audio_int16[i:i+frame_size] for i in range(0, len(audio_int16), frame_size)]
    
#     voiced_audio = np.concatenate([frame for frame in frames if vad.is_speech(frame.tobytes(), sample_rate=sr)])
    
#     # Convert back to float32
#     voiced_audio = np.float32(voiced_audio) / 32767
#     return voiced_audio

# # Load the audio file
# def load_audio(filepath):
#     # Load with librosa
#     audio, sr = librosa.load(filepath, sr=None)
#     return audio, sr

# # Save the audio file using soundfile
# def save_audio(filepath, audio, sr):
#     # Use soundfile.write to save the audio
#     sf.write(filepath, audio, sr)

# # Full noise reduction pipeline
# def noise_reduction_pipeline(filepath):
#     # Step 1: Load audio
#     audio, sr = load_audio(filepath)
#     print(f"Loaded audio with sample rate: {sr}, duration: {len(audio) / sr:.2f} seconds")
    
#     # Step 2: Apply high-pass filter
#     audio_hp = high_pass_filter(audio, sr, cutoff=100)  # Remove low-frequency noise below 100 Hz
    
#     # Step 3: Apply Wiener filter (for noise reduction)
#     audio_wiener = wiener_filter(audio_hp)
    
#     # Step 4: Apply Voice Activity Detection (VAD)
#     audio_vad = apply_vad(audio_wiener, sr)
    
#     # Step 5: Save processed audio
#     output_filepath = "processed_output.wav"
#     save_audio(output_filepath, audio_vad, sr)
    
#     print(f"Processed audio saved to {output_filepath}")
#     return output_filepath

# # Optional: Plot the original and processed audio signals
# def plot_signals(original, processed, sr):
#     plt.figure(figsize=(14, 6))
#     plt.subplot(2, 1, 1)
#     librosa.display.waveshow(original, sr=sr)
#     plt.title("Original Audio")
    
#     plt.subplot(2, 1, 2)
#     librosa.display.waveshow(processed, sr=sr)
#     plt.title("Processed Audio")
    
#     plt.tight_layout()
#     plt.show()

# # Example usage:
# if __name__ == "__main__":
#     # Replace 'input.wav' with your audio file path
#     input_filepath = 'C:/Users/WCHL/Desktop/hindi_dataset/train/hp_sounds/crm/hi/1728268478957.wav'  # input file to process
#     processed_filepath = noise_reduction_pipeline(input_filepath)
#     # processed_filepath=
#     # Load original and processed audio for visualization
#     original_audio, sr = load_audio(input_filepath)
#     processed_audio, _ = load_audio(processed_filepath)
    
#     # Plot the original and processed signals
#     plot_signals(original_audio, processed_audio, sr)


# In[ ]:


# !pip install speechbrain


# ##########################
# 

# In[1]:


# Load the Hugging Face ASR pipeline
from transformers import pipeline
hindi_pipe = pipeline("automatic-speech-recognition", model="cdactvm/w2v-bert-2.0-hindi_new")
whisper_pipe = pipeline("automatic-speech-recognition", model="openai/whisper-large-v3")
eng_pipe = pipeline(task="automatic-speech-recognition", model="C:/Users/WCHL/Desktop/huggingface_english/hf_eng")


# In[12]:


import os
import re
import librosa
import nbimporter
import torchaudio
import numpy as np
import scipy.signal
import webrtcvad
import soundfile as sf
import warnings
warnings.filterwarnings("ignore")

from transformers import pipeline
from text2int import text_to_int
from isNumber import is_number
from Text2List import text_to_list
from convert2list import convert_to_list
from processDoubles import process_doubles
from replaceWords import replace_words
from applyVad import apply_vad
from wienerFilter import wiener_filter
from highPassFilter import high_pass_filter

def noise_reduction_pipeline(filepath):
    audio, sr = librosa.load(filepath, sr=None)
    print(sr)
    audio_hp = high_pass_filter(audio, sr, cutoff=100, order=5)
    audio_wiener = wiener_filter(audio_hp)
    audio_vad = apply_vad(audio_wiener, sr)
    output_filepath = "processed_output.wav"
    sf.write(output_filepath, audio_vad, sr)
    return output_filepath

# Hugging Face ASR pipeline integration
def transcribe_with_huggingface(filepath):
    asr_pipeline = pipeline("automatic-speech-recognition", model="cdactvm/w2v-bert-2.0-hindi_new")
    result = asr_pipeline(filepath)
    text_value=result['text']
    cleaned_text=text_value.replace("<s>", "")
    converted_to_list=convert_to_list(cleaned_text,text_to_list())
    processd_doubles=process_doubles(converted_to_list)
    replaced_words = replace_words(processd_doubles)
    converted_text=text_to_int(replaced_words)
    print("Transcription: ", converted_text)
    return converted_text

if __name__ == "__main__":
    # Step 1: Input file path
    input_filepath = 'C:/Users/WCHL/Desktop/hp_sounds/101003/crm/hi/1728685442307.wav'
    # input_file="enhanced.wav"
    
    # Step 2: Preprocess (Noise Reduction)
    processed_filepath = noise_reduction_pipeline(input_filepath)
    
    # Step 3: ASR (Automatic Speech Recognition) with Hugging Face pipeline
    transcription = transcribe_with_huggingface(processed_filepath)


# In[ ]:


# result = eng_pipe(filepath)
result = hindi_pipe("C:/Users/WCHL/Desktop/hp_sounds/101003/crm/hi/1728685502007.wav")
# result = hindi_pipe("./enhanced/1728268841215.wav")
# result = whisper_pipe(filepath)
text_value=result['text']
cleaned_text=text_value.replace("<s>", "")
converted_to_list=convert_to_list(cleaned_text,text_to_list())
processd_doubles=process_doubles(converted_to_list)
replaced_words = replace_words(processd_doubles)
converted_text=text_to_int(replaced_words)
# Output the transcription
print("Transcription: ", converted_text)

नमस्का जी 1 मन 2 पुलिस हेलप्लेन से बात कर रहे बताइए आपकी ाएमर्जेंसी है
नमिश्का जी 1 मन 2 पुलिस हेलप्लेन से बात कर रह बताइए आपकी क्या एमर्जेंसी है
नमस्का जी 1 मन 2 पुलिस हेलप्लेन से बात कर रह बताइए आपके क्या एमर्जेंसी हैवेल्कम 2 एमर्जनसी
वेल्कम 2 एमर्जनसी
वेलकम 2 एमर्जेंसी
और 9 र मलीख वेल्कम 2 एमर्जंसीनमस्कार जी 1 ्स 2 बारा पुलस हल्प्लाइन में आपका स्वागत ह बताइए आपकी के एमर्जेंसी है
नमस्कार जी 1 ्स दौबारा पुलिस हेल्प्लाइ में आपका स्वागत है बताइए आपकी के एमर्जेंसी है
नमस्कार जी 1 2 बारा पुलिस हल्प्लाइन में आपका स्वागत है बताइए आपकी क् एमर्जेंसी हैमस्कार जी 1 ्स 2 12 पुलस हल्प्लाइन में आपका स्वागत ह बताइए आपकी के एमर्जेंसी है
नमस्कार जी 1 ्स दौबारा पुलिस हेल्प्लाइ में आपका स्वागत है बताइए आपकी के एमर्जेंसी है
नमस्कार जी 1 2 12 पुलिस हल्प्लाइन में आपका स्वागत है बताइए आपकी क् एमर्जेंसी हैनमस्कार जी इक्सुबारा में आपका स्वागत हैइनम
नमस्कार जी इक्सुबारा में आपका स्वागत है कि इनमें
नमस्कार जी 1 ्सुबारा में आपका स्वागत हैइन
# In[ ]:


import os
import numpy as np
import scipy.signal
import webrtcvad
import soundfile as sf
import librosa
import logging
from transformers import pipeline
from text2int import text_to_int
from isNumber import is_number
from Text2List import text_to_list
from convert2list import convert_to_list
from processDoubles import process_doubles
from replaceWords import replace_words

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Noise reduction functions
def high_pass_filter(audio, sr, cutoff=100, order=5):
    try:
        sos = scipy.signal.butter(order, cutoff, btype='highpass', fs=sr, output='sos')
        filtered_audio = scipy.signal.sosfilt(sos, audio)
        return filtered_audio
    except Exception as e:
        logging.error(f"High-pass filter failed: {e}")
        return audio

def wiener_filter(audio):
    try:
        return scipy.signal.wiener(audio)
    except Exception as e:
        logging.error(f"Wiener filter failed: {e}")
        return audio

def apply_vad(audio, sr, frame_duration=30, aggressiveness=3):
    try:
        vad = webrtcvad.Vad(aggressiveness)
        audio_int16 = np.int16(audio * 32767)
        frame_size = int(sr * frame_duration / 1000)
        frames = [audio_int16[i:i + frame_size] for i in range(0, len(audio_int16), frame_size)]
        voiced_audio = np.concatenate([frame for frame in frames if vad.is_speech(frame.tobytes(), sample_rate=sr)])
        voiced_audio = np.float32(voiced_audio) / 32767
        return voiced_audio
    except Exception as e:
        logging.error(f"VAD processing failed: {e}")
        return audio

def load_audio(filepath):
    try:
        audio, sr = librosa.load(filepath, sr=None)
        return audio, sr
    except Exception as e:
        logging.error(f"Failed to load audio: {e}")
        return None, None

def save_audio(filepath, audio, sr):
    try:
        sf.write(filepath, audio, sr)
        logging.info(f"Audio saved at {filepath}")
    except Exception as e:
        logging.error(f"Failed to save audio: {e}")

def noise_reduction_pipeline(filepath):
    # Step 1: Load audio
    audio, sr = load_audio(filepath)
    if audio is None:
        return None
    
    # Step 2: Apply high-pass filter
    audio_hp = high_pass_filter(audio, sr)
    
    # Step 3: Apply Wiener filter
    audio_wiener = wiener_filter(audio_hp)
    
    # Step 4: Apply VAD
    audio_vad = apply_vad(audio_wiener, sr)
    
    # Step 5: Save cleaned audio
    output_filepath = "processed_output.wav"
    save_audio(output_filepath, audio_vad, sr)
    
    return output_filepath

# Hugging Face ASR pipeline integration
def transcribe_with_huggingface(filepath, model_name="cdactvm/w2v-bert-2.0-hindi_new"):
    try:
        # Load ASR model
        logging.info("Loading ASR model...")
        asr_pipeline = pipeline("automatic-speech-recognition", model=model_name)

        # Run the ASR pipeline on the processed audio
        result = asr_pipeline(filepath)
        text_value = result.get('text', '')
        
        # Clean and process transcription
        cleaned_text = text_value.replace("<s>", "")
        converted_to_list = convert_to_list(cleaned_text, text_to_list())
        processed_doubles = process_doubles(converted_to_list)
        replaced_words = replace_words(processed_doubles)
        converted_text = text_to_int(replaced_words)
        
        logging.info("Transcription completed.")
        return converted_text
    
    except Exception as e:
        logging.error(f"ASR transcription failed: {e}")
        return ""

if __name__ == "__main__":
    # Input file path
    input_filepath = 'C:/Users/WCHL/Desktop/hp_sounds/101005/crm/hi/1728268817091.wav'
    
    # Step 1: Preprocess (Noise Reduction)
    processed_filepath = noise_reduction_pipeline(input_filepath)
    
    # Step 2: Check if noise reduction succeeded
    if processed_filepath:
        # Step 3: ASR (Automatic Speech Recognition) with Hugging Face pipeline
        transcription = transcribe_with_huggingface(processed_filepath)
        if transcription:
            print("Transcription:", transcription)
        else:
            logging.warning("No transcription could be generated.")
    else:
        logging.warning("Noise reduction failed; skipping ASR transcription.")


# In[ ]: