Views
No views yet
1from transformers import WhisperProcessor, WhisperForConditionalGeneration
2import torchaudio
3import torch
4
5model_name = "TransferRapid/whisper-large-v3-turbo_ro"
6
7# Load processor and model
8processor = WhisperProcessor.from_pretrained(model_name)
9model = WhisperForConditionalGeneration.from_pretrained(model_name)
10
11# Move model to GPU if available
12device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
13model.to(device)
14model.eval()
15
16def preprocess_audio(audio_path, processor):
17 """Preprocess audio: load, resample if needed, and convert to model input format."""
18 waveform, sample_rate = torchaudio.load(audio_path)
19
20 # Resample to 16kHz if needed
21 if sample_rate != 16000:
22 resampler = torchaudio.transforms.Resample(orig_freq=sample_rate, new_freq=16000)
23 waveform = resampler(waveform)
24
25 # Process audio into model input format
26 inputs = processor(waveform.squeeze().numpy(), sampling_rate=16000, return_tensors="pt")
27
28 # Move inputs to device
29 inputs = {key: val.to(device) for key, val in inputs.items()}
30
31 return inputs
32
33def transcribe(audio_path, model, processor, language="romanian", task="transcribe"):
34 """Generate transcription for an audio file."""
35 inputs = preprocess_audio(audio_path, processor)
36
37 forced_decoder_ids = processor.tokenizer.get_decoder_prompt_ids(language=language, task=task)
38
39 with torch.no_grad():
40 generated_ids = model.generate(inputs["input_features"], forced_decoder_ids=forced_decoder_ids)
41
42 transcription = processor.tokenizer.batch_decode(generated_ids, skip_special_tokens=True)
43
44 return transcription[0]
45
46# Define audio path
47audio_file = "audio.wav"
48
49# Run transcription
50transcription = transcribe(audio_file, model, processor)
51print("Transcription:", transcription)1import os
2import torchaudio
3import numpy as np
4import librosa
5import webrtcvad
6import soundfile as sf
7from pydub import AudioSegment
8from transformers import WhisperProcessor, WhisperForConditionalGeneration
9import torch
10
11# Load model from Hugging Face
12model_name = "TransferRapid/whisper-large-v3-turbo_ro"
13processor = WhisperProcessor.from_pretrained(model_name)
14model = WhisperForConditionalGeneration.from_pretrained(model_name)
15
16# Move model to GPU if available
17device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
18model.to(device)
19model.eval()
20
21def convert_mp3_to_wav(mp3_file_path):
22 """Convert MP3 to WAV (16kHz)."""
23 audio = AudioSegment.from_mp3(mp3_file_path)
24 wav_16k_file_path = mp3_file_path.replace(".mp3", "_16k.wav")
25 audio.set_frame_rate(16000).export(wav_16k_file_path, format="wav")
26 return wav_16k_file_path
27
28def extract_audio_channels(wav_file_path):
29 """Extract left and right channels from stereo WAV."""
30 y, sr = librosa.load(wav_file_path, sr=None, mono=False)
31
32 if len(y.shape) == 1:
33 mono_file = wav_file_path.replace(".wav", "_mono.wav")
34 sf.write(mono_file, y, sr)
35 return y, None, sr, mono_file, None
36
37 left_channel, right_channel = y[0], y[1]
38 left_file = wav_file_path.replace(".wav", "_left.wav")
39 right_file = wav_file_path.replace(".wav", "_right.wav")
40
41 sf.write(left_file, left_channel, sr)
42 sf.write(right_file, right_channel, sr)
43
44 return left_channel, right_channel, sr, left_file, right_file
45
46def detect_speech_intervals(channel_data, sr, vad_level=3):
47 """Detect speech activity using VAD (30ms frames)."""
48 vad = webrtcvad.Vad(vad_level)
49 frame_duration = 30
50 frame_length = int(sr * frame_duration / 1000)
51
52 frames = librosa.util.frame(channel_data, frame_length=frame_length, hop_length=frame_length)
53 speech_intervals = []
54
55 for i, frame in enumerate(frames.T):
56 pcm_data = (frame * np.iinfo(np.int16).max).astype(np.int16).tobytes()
57 if vad.is_speech(pcm_data, sr):
58 start_time, end_time = (i * frame_duration) / 1000, ((i + 1) * frame_duration) / 1000
59 speech_intervals.append((start_time, end_time))
60
61 return speech_intervals
62
63def merge_intervals(intervals, merge_threshold=1):
64 """Merge speech intervals with a gap smaller than merge_threshold."""
65 if not intervals:
66 return []
67
68 merged = [list(intervals[0])]
69 for start, end in intervals[1:]:
70 if (start - merged[-1][1]) <= merge_threshold:
71 merged[-1][1] = end
72 else:
73 merged.append([start, end])
74
75 return merged
76
77def save_segments(channel_data, sr, intervals, output_dir="segments", prefix="segment"):
78 """Save detected speech segments."""
79 os.makedirs(output_dir, exist_ok=True)
80 segment_paths = []
81
82 for idx, (start, end) in enumerate(intervals):
83 start_sample = int(start * sr)
84 end_sample = int(end * sr)
85 segment = channel_data[start_sample:end_sample]
86 segment_path = os.path.join(output_dir, f"{prefix}_{idx+1}.wav")
87 sf.write(segment_path, segment, sr)
88 segment_paths.append((start, end, segment_path, prefix))
89
90 return segment_paths
91
92def preprocess_audio(audio_path, processor, device):
93 """Preprocess audio: load, resample if needed, and convert to model input format."""
94 waveform, sample_rate = torchaudio.load(audio_path)
95
96 if sample_rate != 16000:
97 resampler = torchaudio.transforms.Resample(orig_freq=sample_rate, new_freq=16000)
98 waveform = resampler(waveform)
99
100 inputs = processor(waveform.squeeze().numpy(), sampling_rate=16000, return_tensors="pt")
101 inputs = {key: val.to(device) for key, val in inputs.items()}
102
103 return inputs
104
105def transcribe(audio_path, model, processor, device, language="romanian", task="transcribe"):
106 """Generate transcription for an audio file."""
107 inputs = preprocess_audio(audio_path, processor, device)
108 forced_decoder_ids = processor.tokenizer.get_decoder_prompt_ids(language=language, task=task)
109
110 with torch.no_grad():
111 generated_ids = model.generate(inputs["input_features"], forced_decoder_ids=forced_decoder_ids)
112
113 transcription = processor.tokenizer.batch_decode(generated_ids, skip_special_tokens=True)
114
115 return transcription[0]
116
117# Load audio file (MP3 or WAV)
118audio_file = "audio.mp3"
119
120# Convert MP3 to WAV if needed
121if audio_file.endswith(".mp3"):
122 wav_file = convert_mp3_to_wav(audio_file)
123else:
124 wav_file = audio_file
125
126# Process stereo or mono file
127left_channel, right_channel, sr, left_file, right_file = extract_audio_channels(wav_file)
128
129# Process left channel (or mono)
130if left_channel is not None:
131 left_intervals = detect_speech_intervals(left_channel, sr)
132 merged_left_intervals = merge_intervals(left_intervals)
133 left_segments = save_segments(left_channel, sr, merged_left_intervals, output_dir="left_segments", prefix="Left")
134else:
135 left_segments = []
136
137# Process right channel (if stereo)
138if right_channel is not None:
139 right_intervals = detect_speech_intervals(right_channel, sr)
140 merged_right_intervals = merge_intervals(right_intervals)
141 right_segments = save_segments(right_channel, sr, merged_right_intervals, output_dir="right_segments", prefix="Right")
142else:
143 right_segments = []
144
145# Combine all segments and sort by start time
146all_segments = left_segments + right_segments
147all_segments.sort(key=lambda x: x[0])
148
149# Transcribe each segment
150for idx, (start, end, segment_path, channel) in enumerate(all_segments, start=1):
151 transcription = transcribe(segment_path, model, processor, device)
152 print(f"{idx}. {start:.2f}s → {end:.2f}s | {channel}: {transcription}")