Whisper Speech Recognition Code Examples
1. Basic Whisper Setup and Usage
import whisper
import torch
import numpy as np
import soundfile as sf
import librosa
# Load Whisper model
def load_whisper_model(model_size="base"):
"""
Load Whisper model with appropriate device configuration
Args:
model_size (str): Size of Whisper model ('tiny', 'base', 'small', 'medium', 'large')
Returns:
whisper.model: Loaded Whisper model
"""
# Check if CUDA is available
device = "cuda" if torch.cuda.is_available() else "cpu"
print(f"Loading Whisper model ({model_size}) on {device}...")
model = whisper.load_model(model_size, device=device)
return model
# Example usage
model = load_whisper_model("base")
# Transcribe an audio file
def transcribe_audio_file(model, audio_path):
"""
Transcribe audio from a file using Whisper
Args:
model: Loaded Whisper model
audio_path (str): Path to audio file
Returns:
str: Transcribed text
"""
# Load audio file
result = model.transcribe(audio_path)
return result["text"]
# Example usage
# transcription = transcribe_audio_file(model, "speech_sample.wav")
# print(f"Transcription: {transcription}")
2. Real-time Audio Processing with PyAudio
import pyaudio
import numpy as np
import threading
import queue
from collections import deque
import time
class RealTimeWhisperProcessor:
def __init__(self, model, chunk_duration=2.0, sample_rate=16000):
"""
Real-time Whisper processor for continuous audio input
Args:
model: Loaded Whisper model
chunk_duration (float): Duration of audio chunks in seconds
sample_rate (int): Audio sample rate
"""
self.model = model
self.chunk_duration = chunk_duration
self.sample_rate = sample_rate
self.chunk_size = int(self.sample_rate * self.chunk_duration)
# Audio stream parameters
self.format = pyaudio.paFloat32
self.channels = 1
self.rate = self.sample_rate
# Audio buffer for real-time processing
self.audio_buffer = deque(maxlen=self.chunk_size * 5) # 5 seconds of audio
# Thread control
self.recording = False
self.processing_thread = None
self.audio_queue = queue.Queue()
# Initialize PyAudio
self.pyaudio = pyaudio.PyAudio()
# Voice activity detection threshold
self.vad_threshold = 0.01 # Adjust based on your environment
def start_recording(self):
"""Start real-time audio recording"""
self.recording = True
# Open audio stream
self.stream = self.pyaudio.open(
format=self.format,
channels=self.channels,
rate=self.rate,
input=True,
frames_per_buffer=self.chunk_size
)
# Start processing thread
self.processing_thread = threading.Thread(target=self._processing_loop)
self.processing_thread.daemon = True
self.processing_thread.start()
print("Real-time audio recording started")
def stop_recording(self):
"""Stop real-time audio recording"""
self.recording = False
if hasattr(self, 'stream'):
self.stream.stop_stream()
self.stream.close()
if self.processing_thread:
self.processing_thread.join()
# Terminate PyAudio
self.pyaudio.terminate()
print("Real-time audio recording stopped")
def _processing_loop(self):
"""Main processing loop for real-time audio"""
while self.recording:
try:
# Read audio chunk
data = self.stream.read(self.chunk_size, exception_on_overflow=False)
audio_chunk = np.frombuffer(data, dtype=np.float32)
# Add to buffer
self.audio_buffer.extend(audio_chunk)
# Check for voice activity
if self._is_voice_active(audio_chunk):
# Process the accumulated audio if it's long enough
if len(self.audio_buffer) > self.sample_rate * 0.5: # At least 0.5 seconds
audio_segment = np.array(self.audio_buffer)
self._process_audio_segment(audio_segment)
except Exception as e:
print(f"Error in processing loop: {e}")
time.sleep(0.1) # Brief pause to prevent busy waiting
def _is_voice_active(self, audio_chunk):
"""Simple voice activity detection based on energy"""
energy = np.mean(np.abs(audio_chunk))
return energy > self.vad_threshold
def _process_audio_segment(self, audio_segment):
"""Process an audio segment with Whisper"""
try:
# Convert to float32 if needed
if audio_segment.dtype != np.float32:
audio_segment = audio_segment.astype(np.float32)
# Pad or trim to required length for Whisper
audio = whisper.pad_or_trim(audio_segment)
# Convert to log-Mel spectrogram
mel = whisper.log_mel_spectrogram(audio).to(self.model.device)
# Decode audio
options = whisper.DecodingOptions()
result = whisper.decode(self.model, mel, options)
if result.text.strip(): # Only process non-empty transcriptions
print(f"Transcription: {result.text}")
# In a real application, you would send this to your command processor
self._handle_transcription(result.text)
except Exception as e:
print(f"Error processing audio segment: {e}")
def _handle_transcription(self, transcription):
"""Handle the transcribed text"""
# This would typically trigger command processing in a VLA system
print(f"Processing command: {transcription}")
# In practice, this would send the transcription to your command interpretation system
# Example usage
model = load_whisper_model("base")
real_time_processor = RealTimeWhisperProcessor(model)
# Start real-time processing (uncomment to run)
# real_time_processor.start_recording()
# Stop after some time (uncomment to run)
# time.sleep(10)
# real_time_processor.stop_recording()
3. Whisper with Audio Preprocessing
import librosa
import soundfile as sf
from scipy import signal
class WhisperAudioPreprocessor:
def __init__(self):
"""Initialize audio preprocessing utilities for Whisper"""
pass
def preprocess_audio(self, audio_path, target_sr=16000):
"""
Preprocess audio for optimal Whisper performance
Args:
audio_path (str): Path to audio file
target_sr (int): Target sample rate
Returns:
np.ndarray: Preprocessed audio array
"""
# Load audio with librosa (handles various formats better than soundfile)
y, sr = librosa.load(audio_path, sr=target_sr)
# Normalize audio
y = librosa.util.normalize(y)
# Apply pre-emphasis filter to boost high frequencies
y = self._preemphasis_filter(y)
# Remove silence at beginning and end
y = self._remove_silence(y, sr)
return y
def _preemphasis_filter(self, audio, coef=0.97):
"""Apply pre-emphasis filter to boost high frequencies"""
return signal.lfilter([1, -coef], [1], audio)
def _remove_silence(self, audio, sr, threshold=0.01, frame_length=2048):
"""Remove silence from beginning and end of audio"""
# Calculate energy for each frame
frames = librosa.util.frame(audio, frame_length=frame_length, hop_length=frame_length//2)
frame_energy = np.mean(frames**2, axis=0)
# Find non-silent frames
non_silent_frames = np.where(frame_energy > threshold)[0]
if len(non_silent_frames) > 0:
start_frame = non_silent_frames[0]
end_frame = non_silent_frames[-1]
start_sample = start_frame * frame_length // 2
end_sample = (end_frame + 1) * frame_length // 2
return audio[start_sample:end_sample]
return audio # Return original if no non-silent frames found
def transcribe_with_preprocessing(model, audio_path):
"""
Transcribe audio with preprocessing for better results
Args:
model: Loaded Whisper model
audio_path (str): Path to audio file
Returns:
str: Transcribed text
"""
preprocessor = WhisperAudioPreprocessor()
# Preprocess audio
processed_audio = preprocessor.preprocess_audio(audio_path)
# Convert to the format expected by Whisper
# Whisper expects 16kHz mono audio
if len(processed_audio.shape) > 1:
# If stereo, convert to mono
processed_audio = processed_audio.mean(axis=1)
# Pad or trim to required length
audio = whisper.pad_or_trim(processed_audio)
# Convert to log-Mel spectrogram
mel = whisper.log_mel_spectrogram(audio).to(model.device)
# Decode audio
options = whisper.DecodingOptions()
result = whisper.decode(model, mel, options)
return result.text
# Example usage
# model = load_whisper_model("base")
# transcription = transcribe_with_preprocessing(model, "noisy_speech.wav")
# print(f"Preprocessed transcription: {transcription}")
4. Whisper with Language Detection
def transcribe_with_language_detection(model, audio_path):
"""
Transcribe audio with automatic language detection
Args:
model: Loaded Whisper model
audio_path (str): Path to audio file
Returns:
dict: Transcription result with language information
"""
# Load audio
audio = whisper.load_audio(audio_path)
audio = whisper.pad_or_trim(audio)
# Convert to log-Mel spectrogram
mel = whisper.log_mel_spectrogram(audio).to(model.device)
# Detect language
_, probs = model.detect_language(mel)
detected_language = max(probs, key=probs.get)
# Decode audio with detected language
options = whisper.DecodingOptions(
language=detected_language,
without_timestamps=True
)
result = whisper.decode(model, mel, options)
return {
"text": result.text,
"language": detected_language,
"language_probability": probs[detected_language]
}
# Example usage
# model = load_whisper_model("base")
# result = transcribe_with_language_detection(model, "multilingual_speech.wav")
# print(f"Text: {result['text']}")
# print(f"Language: {result['language']} (confidence: {result['language_probability']:.2f})")
5. Batch Processing Multiple Audio Files
import os
from pathlib import Path
class WhisperBatchProcessor:
def __init__(self, model):
"""
Batch processor for multiple audio files
Args:
model: Loaded Whisper model
"""
self.model = model
def process_directory(self, input_dir, output_dir, file_extensions=None):
"""
Process all audio files in a directory
Args:
input_dir (str): Input directory containing audio files
output_dir (str): Output directory for transcriptions
file_extensions (list): List of file extensions to process
Returns:
list: List of processing results
"""
if file_extensions is None:
file_extensions = ['.wav', '.mp3', '.m4a', '.flac', '.aac']
input_path = Path(input_dir)
output_path = Path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
results = []
for file_path in input_path.rglob('*'):
if file_path.suffix.lower() in file_extensions:
try:
# Transcribe the file
transcription = self.model.transcribe(str(file_path))["text"]
# Save transcription to output file
output_file = output_path / f"{file_path.stem}.txt"
with open(output_file, 'w', encoding='utf-8') as f:
f.write(transcription)
results.append({
"file": str(file_path),
"output": str(output_file),
"success": True,
"transcription_length": len(transcription)
})
print(f"Processed: {file_path.name}")
except Exception as e:
results.append({
"file": str(file_path),
"success": False,
"error": str(e)
})
print(f"Error processing {file_path.name}: {e}")
return results
# Example usage
# model = load_whisper_model("base")
# batch_processor = WhisperBatchProcessor(model)
# results = batch_processor.process_directory("input_audio/", "output_transcriptions/")
# print(f"Processed {len(results)} files")
6. Whisper with Confidence Scoring
def transcribe_with_confidence(model, audio_path):
"""
Transcribe audio with confidence scoring
Args:
model: Loaded Whisper model
audio_path (str): Path to audio file
Returns:
dict: Transcription with confidence scores
"""
# Load and process audio
audio = whisper.load_audio(audio_path)
audio = whisper.pad_or_trim(audio)
mel = whisper.log_mel_spectrogram(audio).to(model.device)
# Get the full transcription result with tokens
options = whisper.DecodingOptions(
return_segments=True,
temperature=0.0 # For consistent results
)
result = model.transcribe(
audio_path, # Use transcribe method which provides more detailed output
return_segments=True
)
# Calculate overall confidence based on segment confidence
if 'segments' in result and result['segments']:
avg_confidence = np.mean([seg.get('avg_logprob', 0) for seg in result['segments']])
# Convert log probability to a more interpretable confidence score
confidence_score = 1 / (1 + np.exp(-avg_confidence)) # Sigmoid transformation
else:
confidence_score = 0.0
return {
"text": result["text"],
"confidence": confidence_score,
"segments": result.get("segments", []),
"language": result.get("language", "unknown")
}
# Example usage
# model = load_whisper_model("base")
# result = transcribe_with_confidence(model, "speech_sample.wav")
# print(f"Transcription: {result['text']}")
# print(f"Confidence: {result['confidence']:.2f}")
7. Whisper Integration with VLA System
class VLAWhisperInterface:
def __init__(self, model_size="base"):
"""
Whisper interface for Vision-Language-Action system
Args:
model_size (str): Size of Whisper model to use
"""
self.model = load_whisper_model(model_size)
self.preprocessor = WhisperAudioPreprocessor()
def process_voice_command(self, audio_data):
"""
Process voice command for VLA system
Args:
audio_data: Audio data (file path or numpy array)
Returns:
dict: Processing result for VLA system
"""
try:
if isinstance(audio_data, str): # File path
# Load and preprocess audio file
processed_audio = self.preprocessor.preprocess_audio(audio_data)
audio = whisper.pad_or_trim(processed_audio)
else: # Audio array
audio = whisper.pad_or_trim(audio_data.astype(np.float32))
# Convert to log-Mel spectrogram
mel = whisper.log_mel_spectrogram(audio).to(self.model.device)
# Decode audio
options = whisper.DecodingOptions()
result = whisper.decode(self.model, mel, options)
# Detect language
_, probs = self.model.detect_language(mel)
detected_language = max(probs, key=probs.get)
# Return structured result for VLA system
return {
"success": True,
"transcription": result.text,
"language": detected_language,
"language_confidence": probs[detected_language],
"timestamp": time.time(),
"processing_time": time.time() # Placeholder
}
except Exception as e:
return {
"success": False,
"error": str(e),
"timestamp": time.time()
}
def process_real_time_audio(self, callback_function):
"""
Process real-time audio with callback for VLA system
Args:
callback_function: Function to call with each transcription
"""
real_time_processor = RealTimeWhisperProcessor(self.model)
# Override the _handle_transcription method to use our callback
def custom_handler(transcription):
result = {
"transcription": transcription,
"timestamp": time.time()
}
callback_function(result)
# Replace the default handler
real_time_processor._handle_transcription = custom_handler
return real_time_processor
# Example usage in VLA context
def vla_command_callback(result):
"""Callback function for processing VLA commands"""
command = result["transcription"]
print(f"VLA received command: {command}")
# In a real VLA system, this would trigger:
# 1. Command interpretation
# 2. Plan generation
# 3. Action execution
# 4. Feedback to user
# Initialize VLA Whisper interface
vla_whisper = VLAWhisperInterface("base")
# Process a single command
# result = vla_whisper.process_voice_command("command.wav")
# print(f"Command processed: {result}")
# For real-time processing
# real_time_processor = vla_whisper.process_real_time_audio(vla_command_callback)
# real_time_processor.start_recording()
8. Whisper Performance Optimization
class OptimizedWhisperProcessor:
def __init__(self, model_size="base", device=None, use_fp16=True):
"""
Optimized Whisper processor for better performance
Args:
model_size (str): Size of Whisper model
device (str): Device to run model on ('cuda', 'cpu', or None for auto)
use_fp16 (bool): Whether to use float16 precision for faster inference
"""
if device is None:
device = "cuda" if torch.cuda.is_available() else "cpu"
self.device = device
# Load model with optimizations
self.model = whisper.load_model(
model_size,
device=device,
in_memory=True
)
# Apply further optimizations if using CUDA
if device == "cuda" and use_fp16:
self.model = self.model.half() # Use float16 for faster inference
# Warm up the model
self._warm_up()
print(f"Optimized Whisper model loaded on {device}")
def _warm_up(self):
"""Warm up the model with dummy input"""
# Create a short dummy audio segment
dummy_audio = torch.zeros(16000, dtype=torch.float32) # 1 second of silence
dummy_audio = whisper.pad_or_trim(dummy_audio)
mel = whisper.log_mel_spectrogram(dummy_audio).to(self.model.device)
# Run a dummy inference
options = whisper.DecodingOptions()
_ = whisper.decode(self.model, mel, options)
def transcribe_optimized(self, audio_path, temperature=0.0):
"""
Optimized transcription with performance considerations
Args:
audio_path (str): Path to audio file
temperature (float): Temperature for decoding (lower for more consistent results)
Returns:
str: Transcribed text
"""
start_time = time.time()
# Use the optimized transcribe method
result = self.model.transcribe(
audio_path,
temperature=temperature,
compression_ratio_threshold=None, # Disable compression ratio threshold
logprob_threshold=None, # Disable logprob threshold
no_speech_threshold=None # Disable no speech threshold
)
end_time = time.time()
return {
"text": result["text"],
"processing_time": end_time - start_time,
"language": result.get("language", "unknown")
}
# Example usage
# optimized_processor = OptimizedWhisperProcessor("base", use_fp16=True)
# result = optimized_processor.transcribe_optimized("speech.wav")
# print(f"Transcription: {result['text']}")
# print(f"Processing time: {result['processing_time']:.2f}s")
These examples demonstrate various aspects of Whisper speech recognition implementation:
- Basic Setup: Loading and using Whisper models with different sizes
- Real-time Processing: Continuous audio capture and processing using PyAudio
- Audio Preprocessing: Techniques to improve transcription quality
- Language Detection: Automatic language identification
- Batch Processing: Processing multiple files efficiently
- Confidence Scoring: Measuring transcription reliability
- VLA Integration: Connecting Whisper to Vision-Language-Action systems
- Performance Optimization: Techniques for faster inference
The code follows best practices for Whisper integration and provides a solid foundation for building speech recognition capabilities in robotics applications.