Spaces:
Running
Running
File size: 5,184 Bytes
1053c8b 6f061b9 1053c8b e5c4db0 6f061b9 f9b4788 1053c8b 6f061b9 1053c8b 6f061b9 1053c8b 6f061b9 1053c8b 6f061b9 1053c8b 6f061b9 1053c8b f9b4788 1053c8b f9b4788 1053c8b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 |
import gradio as gr
import librosa
import numpy as np
import whisperx
from transformers import pipeline
from pydub import AudioSegment
import os
import scipy.signal as signal
import torch
from pydub.silence import detect_nonsilent # Correct import
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)
device = "cuda" if torch.cuda.is_available() else "cpu"
diarize_model = whisperx.DiarizationPipeline(use_auth_token=hf_token, device=device)
print("WhisperX Model Loaded Successfully for Diarization.")
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 audio format to WAV using pydub
sound = AudioSegment.from_file(audio_path)
wav_path = "converted_audio.wav"
sound.export(wav_path, format="wav")
return wav_path
import gradio as gr
import librosa
import numpy as np
import whisperx
from transformers import pipeline
from pydub import AudioSegment
import os
import scipy.signal as signal
import torch
import pandas as pd
from pydub.silence import detect_nonsilent
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):
print(f"Received audio file: {audio_path}")
try:
# Load the audio file using librosa
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)
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)
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
transcriptions = []
for chunk in non_silent_chunks:
chunk.export("chunk.wav", format="wav")
chunk_audio, chunk_sr = librosa.load("chunk.wav", sr=None)
transcription = stt_pipeline(chunk_audio) # Transcribe using Wav2Vec2
# Match transcription segment with diarization result
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
speaker_label = label
break
transcriptions.append(f"Speaker {speaker_label}: {transcription['text']}")
# Clean up temporary files
os.remove("chunk.wav")
return "\n".join(transcriptions)
except Exception as e:
print(f"Error: {str(e)}")
return f"Error: {str(e)}"
# Create Gradio interface
iface = gr.Interface(
fn=process_audio,
inputs=gr.Audio(type="filepath"),
outputs="text",
title="Speaker Diarization & Transcription",
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.")
|