File size: 7,027 Bytes
9ad60eb | 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 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | #!/usr/bin/env python3
"""
Audio Processor - Handles audio loading, preprocessing and caching
===================================================================
"""
import os
import hashlib
import logging
import numpy as np
import soundfile as sf
from pathlib import Path
from typing import Tuple, Optional
import subprocess
import tempfile
logger = logging.getLogger(__name__)
class AudioProcessor:
"""Handles audio file processing and caching"""
def __init__(self, cache_dir: str = "/tmp/audio_cache", target_sr: int = 16000):
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.target_sr = target_sr
def load_audio(self, audio_path: str, use_cache: bool = True) -> Tuple[np.ndarray, int]:
"""Load audio file and convert to proper format"""
audio_path = Path(audio_path)
if not audio_path.exists():
raise FileNotFoundError(f"Audio file not found: {audio_path}")
# Check cache
if use_cache:
cached_audio = self._get_cached_audio(audio_path)
if cached_audio is not None:
logger.info(f"Loaded audio from cache: {audio_path.name}")
return cached_audio, self.target_sr
# Load audio based on extension
audio_data = None
sample_rate = None
if audio_path.suffix.lower() in ['.wav', '.flac']:
# Direct load for supported formats
audio_data, sample_rate = sf.read(str(audio_path))
elif audio_path.suffix.lower() in ['.mp3', '.m4a', '.ogg', '.webm']:
# Convert using ffmpeg for other formats
audio_data, sample_rate = self._convert_with_ffmpeg(audio_path)
else:
raise ValueError(f"Unsupported audio format: {audio_path.suffix}")
# Convert to mono if stereo
if len(audio_data.shape) > 1:
audio_data = np.mean(audio_data, axis=1)
# Resample if needed
if sample_rate != self.target_sr:
audio_data = self._resample_audio(audio_data, sample_rate, self.target_sr)
sample_rate = self.target_sr
# Normalize audio
audio_data = self._normalize_audio(audio_data)
# Cache the processed audio
if use_cache:
self._cache_audio(audio_path, audio_data)
return audio_data, sample_rate
def _convert_with_ffmpeg(self, audio_path: Path) -> Tuple[np.ndarray, int]:
"""Convert audio using ffmpeg"""
logger.info(f"Converting {audio_path.suffix} file with ffmpeg...")
with tempfile.NamedTemporaryFile(suffix='.wav') as tmp_wav:
# Convert to WAV using ffmpeg
cmd = [
'ffmpeg', '-i', str(audio_path),
'-ar', str(self.target_sr),
'-ac', '1', # Mono
'-f', 'wav',
'-y', # Overwrite
tmp_wav.name
]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=30
)
if result.returncode != 0:
logger.error(f"FFmpeg error: {result.stderr}")
raise RuntimeError(f"FFmpeg conversion failed: {result.stderr}")
# Load the converted WAV
audio_data, sample_rate = sf.read(tmp_wav.name)
return audio_data, sample_rate
except subprocess.TimeoutExpired:
logger.error("FFmpeg conversion timed out")
raise
except FileNotFoundError:
logger.error("FFmpeg not found. Please install ffmpeg")
raise RuntimeError("FFmpeg not found. Install with: apt-get install ffmpeg")
def _resample_audio(self, audio: np.ndarray, orig_sr: int, target_sr: int) -> np.ndarray:
"""Resample audio to target sample rate"""
if orig_sr == target_sr:
return audio
# Simple resampling using numpy
duration = len(audio) / orig_sr
n_samples = int(duration * target_sr)
# Use linear interpolation for resampling
x_old = np.linspace(0, duration, len(audio))
x_new = np.linspace(0, duration, n_samples)
resampled = np.interp(x_new, x_old, audio)
return resampled
def _normalize_audio(self, audio: np.ndarray) -> np.ndarray:
"""Normalize audio to [-1, 1] range"""
# Avoid division by zero
max_val = np.max(np.abs(audio))
if max_val > 0:
audio = audio / max_val * 0.95 # Leave some headroom
return audio.astype(np.float32)
def _get_cache_key(self, audio_path: Path) -> str:
"""Generate cache key for audio file"""
# Use file path and modification time for cache key
stat = audio_path.stat()
key_str = f"{audio_path}_{stat.st_mtime}_{stat.st_size}"
return hashlib.md5(key_str.encode()).hexdigest()
def _get_cached_audio(self, audio_path: Path) -> Optional[np.ndarray]:
"""Retrieve cached audio if available"""
cache_key = self._get_cache_key(audio_path)
cache_file = self.cache_dir / f"{cache_key}.npy"
if cache_file.exists():
try:
return np.load(cache_file)
except Exception as e:
logger.warning(f"Failed to load cached audio: {e}")
cache_file.unlink() # Remove corrupted cache
return None
def _cache_audio(self, audio_path: Path, audio_data: np.ndarray):
"""Save processed audio to cache"""
cache_key = self._get_cache_key(audio_path)
cache_file = self.cache_dir / f"{cache_key}.npy"
try:
np.save(cache_file, audio_data)
logger.debug(f"Cached audio: {cache_key}")
except Exception as e:
logger.warning(f"Failed to cache audio: {e}")
def clear_cache(self):
"""Clear all cached audio files"""
cache_files = list(self.cache_dir.glob("*.npy"))
for cache_file in cache_files:
cache_file.unlink()
logger.info(f"Cleared {len(cache_files)} cached audio files")
def preprocess_for_whisper(self, audio_data: np.ndarray) -> np.ndarray:
"""Preprocess audio specifically for Whisper model"""
# Whisper expects float32 audio in [-1, 1] range
if audio_data.dtype != np.float32:
audio_data = audio_data.astype(np.float32)
# Ensure proper range
audio_data = np.clip(audio_data, -1.0, 1.0)
return audio_data |