BBPlease / src /audio_processor.py
hamza-ksr's picture
Upload src/audio_processor.py with huggingface_hub
c3545d8 verified
Raw
History Blame Contribute Delete
28 kB
"""
Baby Cry AI - Audio Processing Module
Handles audio feature extraction, preprocessing, validation, and diagnostics
"""
import librosa
import numpy as np
from scipy import signal
from sklearn.preprocessing import StandardScaler
import warnings
warnings.filterwarnings('ignore')
class AudioProcessor:
# Validation thresholds
MIN_DURATION = 3.0 # seconds
MAX_DURATION = 30.0 # seconds
MIN_AUDIO_LEVEL_DB = -40 # dB
MAX_SILENCE_RATIO = 0.80 # 80%
def __init__(self, sample_rate=22050, duration=5.0, fast_mode=False):
"""
Initialize AudioProcessor
Args:
sample_rate: Target sample rate (default 22050 Hz)
duration: Max duration to load (default 5 seconds, optimized for speed)
fast_mode: If True, skip expensive features (CQT, enhanced chroma) for faster processing
"""
self.sample_rate = sample_rate
self.duration = duration
self.fast_mode = fast_mode # Skip expensive features for faster analysis
self.scaler = StandardScaler()
# ==================== DIAGNOSTICS ====================
def calculate_snr(self, y, sr):
"""
Calculate Signal-to-Noise Ratio (SNR) in dB
Uses spectral subtraction approach:
- Signal: energy in typical cry frequency range (300-3000 Hz)
- Noise: energy in low frequencies (< 100 Hz) and high frequencies (> 5000 Hz)
"""
try:
# Compute power spectral density
freqs = np.fft.rfftfreq(len(y), 1/sr)
fft = np.fft.rfft(y)
psd = np.abs(fft) ** 2
# Signal: 300-3000 Hz (typical cry range)
signal_mask = (freqs >= 300) & (freqs <= 3000)
signal_power = np.sum(psd[signal_mask])
# Noise: < 100 Hz and > 5000 Hz
noise_mask = (freqs < 100) | (freqs > 5000)
noise_power = np.sum(psd[noise_mask])
# Avoid division by zero
if noise_power < 1e-10:
noise_power = 1e-10
snr_linear = signal_power / noise_power
snr_db = 10 * np.log10(snr_linear)
return float(snr_db)
except Exception:
return -100.0 # Invalid SNR
def get_audio_stats(self, file_path):
"""
Get comprehensive audio statistics for diagnostics
Returns dict with:
- duration_seconds: actual duration of audio
- sample_rate: detected sample rate
- audio_level_db: average audio level in dB
- peak_level_db: peak audio level in dB
- silence_ratio: ratio of silent frames
- rms_energy: root mean square energy
- snr_db: signal-to-noise ratio in dB
- has_audio_content: boolean if meaningful audio detected
"""
try:
# Load full audio without duration limit for stats
y, sr = librosa.load(file_path, sr=None, duration=None)
if y is None or len(y) == 0:
return {'error': 'Could not load audio', 'valid': False}
duration = len(y) / sr
# Calculate RMS energy
rms = librosa.feature.rms(y=y)[0]
rms_mean = float(np.mean(rms))
# Calculate audio levels in dB
# Avoid log(0) by adding small epsilon
eps = 1e-10
audio_level_db = float(20 * np.log10(rms_mean + eps))
peak_level_db = float(20 * np.log10(np.max(np.abs(y)) + eps))
# Calculate silence ratio (frames below threshold)
silence_threshold = 0.01 # amplitude threshold
silent_frames = np.sum(np.abs(y) < silence_threshold)
silence_ratio = float(silent_frames / len(y))
# Spectral centroid (indicator of "brightness" - cries are typically high)
spectral_centroid = librosa.feature.spectral_centroid(y=y, sr=sr)
centroid_mean = float(np.mean(spectral_centroid))
# Zero crossing rate (high for noisy/cry sounds)
zcr = librosa.feature.zero_crossing_rate(y)
zcr_mean = float(np.mean(zcr))
# Calculate SNR
snr_db = self.calculate_snr(y, sr)
# Basic cry detection heuristic
# Cries typically have: high spectral centroid, moderate-high energy, not silence
has_audio_content = (
audio_level_db > self.MIN_AUDIO_LEVEL_DB and
silence_ratio < self.MAX_SILENCE_RATIO
)
# Cry likelihood based on audio characteristics
# Baby cries typically: 300-600 Hz fundamental, high harmonics
cry_likelihood = 0.0
if has_audio_content:
# Higher centroid = more likely cry (cries are high-pitched)
if centroid_mean > 1000:
cry_likelihood += 0.3
if centroid_mean > 2000:
cry_likelihood += 0.2
# High ZCR suggests cry/voice
if zcr_mean > 0.05:
cry_likelihood += 0.2
# Good energy level
if audio_level_db > -30:
cry_likelihood += 0.3
return {
'valid': True,
'duration_seconds': round(duration, 2),
'sample_rate': int(sr),
'audio_level_db': round(audio_level_db, 1),
'peak_level_db': round(peak_level_db, 1),
'silence_ratio': round(silence_ratio, 3),
'rms_energy': round(rms_mean, 6),
'spectral_centroid_mean': round(centroid_mean, 1),
'zcr_mean': round(zcr_mean, 4),
'snr_db': round(snr_db, 1),
'has_audio_content': has_audio_content,
'cry_likelihood': round(min(cry_likelihood, 1.0), 2)
}
except Exception as e:
return {'error': str(e), 'valid': False}
def get_feature_summary(self, features):
"""
Get a summary of key features for diagnostics
Args:
features: dict of extracted features
Returns:
dict with key feature values
"""
if not features:
return {}
summary = {}
# Key MFCC features (first few are most important)
for i in range(3):
key = f'mfcc_{i}_mean'
if key in features:
summary[key] = round(features[key], 2)
# Spectral features
if 'spectral_centroid_mean' in features:
summary['spectral_centroid'] = round(features['spectral_centroid_mean'], 1)
if 'spectral_rolloff_mean' in features:
summary['spectral_rolloff'] = round(features['spectral_rolloff_mean'], 1)
# Energy
if 'rms_mean' in features:
summary['rms_energy'] = round(features['rms_mean'], 4)
# Tempo
if 'tempo' in features:
summary['tempo'] = round(features['tempo'], 1)
return summary
# ==================== VALIDATION ====================
def validate_audio(self, file_path):
"""
Validate audio file before analysis
Returns:
tuple: (is_valid: bool, message: str, stats: dict)
"""
stats = self.get_audio_stats(file_path)
if not stats.get('valid', False):
return False, f"Could not load audio: {stats.get('error', 'Unknown error')}", stats
# Check duration
duration = stats.get('duration_seconds', 0)
if duration < self.MIN_DURATION:
return False, f"Recording too short ({duration:.1f}s). Minimum {self.MIN_DURATION}s required.", stats
if duration > self.MAX_DURATION:
return False, f"Recording too long ({duration:.1f}s). Maximum {self.MAX_DURATION}s allowed.", stats
# Check audio level
audio_level = stats.get('audio_level_db', -100)
if audio_level < self.MIN_AUDIO_LEVEL_DB:
return False, f"Recording too quiet ({audio_level:.1f} dB). Please record closer to the baby.", stats
# Check silence ratio
silence_ratio = stats.get('silence_ratio', 1.0)
if silence_ratio > self.MAX_SILENCE_RATIO:
return False, f"Recording is mostly silence ({silence_ratio*100:.0f}%). Please record actual crying.", stats
# Check for audio content
if not stats.get('has_audio_content', False):
return False, "No meaningful audio detected. Please check your microphone.", stats
# All checks passed
return True, "Audio validated successfully", stats
def validate_audio_array(self, y, sr):
"""
Validate audio array (for already-loaded audio)
Returns:
tuple: (is_valid: bool, message: str, stats: dict)
"""
if y is None or len(y) == 0:
return False, "Empty audio data", {}
duration = len(y) / sr
# Calculate stats
rms = librosa.feature.rms(y=y)[0]
rms_mean = float(np.mean(rms))
eps = 1e-10
audio_level_db = float(20 * np.log10(rms_mean + eps))
silent_frames = np.sum(np.abs(y) < 0.01)
silence_ratio = float(silent_frames / len(y))
stats = {
'duration_seconds': round(duration, 2),
'audio_level_db': round(audio_level_db, 1),
'silence_ratio': round(silence_ratio, 3)
}
# Validate
if duration < self.MIN_DURATION:
return False, f"Recording too short ({duration:.1f}s)", stats
if audio_level_db < self.MIN_AUDIO_LEVEL_DB:
return False, f"Recording too quiet ({audio_level_db:.1f} dB)", stats
if silence_ratio > self.MAX_SILENCE_RATIO:
return False, f"Recording is mostly silence ({silence_ratio*100:.0f}%)", stats
return True, "Audio validated", stats
# ==================== LOADING ====================
def load_audio(self, file_path, sr=None, duration=None):
"""Load audio file with librosa"""
if sr is None:
sr = self.sample_rate
if duration is None:
duration = self.duration
try:
y, sr = librosa.load(file_path, sr=sr, duration=duration)
return y, sr
except Exception as e:
print(f"Error loading audio file {file_path}: {e}")
return None, None
def load_audio_full(self, file_path):
"""Load full audio file without duration limit"""
try:
y, sr = librosa.load(file_path, sr=self.sample_rate, duration=None)
return y, sr
except Exception as e:
print(f"Error loading audio file {file_path}: {e}")
return None, None
# ==================== FEATURE EXTRACTION ====================
def extract_mfcc_features(self, y, sr, n_mfcc=13):
"""Extract MFCC features"""
mfccs = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=n_mfcc)
return np.mean(mfccs, axis=1), np.std(mfccs, axis=1)
def extract_spectral_features(self, y, sr):
"""Extract spectral features"""
spectral_centroids = librosa.feature.spectral_centroid(y=y, sr=sr)
spectral_rolloff = librosa.feature.spectral_rolloff(y=y, sr=sr)
spectral_bandwidth = librosa.feature.spectral_bandwidth(y=y, sr=sr)
zcr = librosa.feature.zero_crossing_rate(y)
return {
'spectral_centroid_mean': float(np.mean(spectral_centroids)),
'spectral_centroid_std': float(np.std(spectral_centroids)),
'spectral_rolloff_mean': float(np.mean(spectral_rolloff)),
'spectral_rolloff_std': float(np.std(spectral_rolloff)),
'spectral_bandwidth_mean': float(np.mean(spectral_bandwidth)),
'spectral_bandwidth_std': float(np.std(spectral_bandwidth)),
'zcr_mean': float(np.mean(zcr)),
'zcr_std': float(np.std(zcr))
}
def extract_rhythm_features(self, y, sr):
"""Extract rhythm and tempo features"""
tempo, beats = librosa.beat.beat_track(y=y, sr=sr)
rms = librosa.feature.rms(y=y)[0]
onset_frames = librosa.onset.onset_detect(y=y, sr=sr)
onset_strength = librosa.onset.onset_strength(y=y, sr=sr)
return {
'tempo': float(tempo),
'rms_mean': float(np.mean(rms)),
'rms_std': float(np.std(rms)),
'onset_rate': float(len(onset_frames) / (len(y) / sr)),
'onset_strength_mean': float(np.mean(onset_strength)),
'onset_strength_std': float(np.std(onset_strength))
}
def extract_tonal_features(self, y, sr):
"""Extract tonal and harmonic features"""
chroma = librosa.feature.chroma_stft(y=y, sr=sr)
tonnetz = librosa.feature.tonnetz(y=y, sr=sr)
y_harmonic, y_percussive = librosa.effects.hpss(y)
harmonic_energy = np.mean(y_harmonic**2)
percussive_energy = np.mean(y_percussive**2)
total_energy = harmonic_energy + percussive_energy
return {
'chroma_mean': np.mean(chroma, axis=1).tolist(),
'chroma_std': np.std(chroma, axis=1).tolist(),
'tonnetz_mean': np.mean(tonnetz, axis=1).tolist(),
'harmonic_ratio': harmonic_energy / total_energy if total_energy > 0 else 0.5
}
def extract_mel_spectrogram_features(self, y, sr, n_mels=128):
"""Extract Mel-spectrogram features (better for cry classification)"""
mel_spec = librosa.feature.melspectrogram(y=y, sr=sr, n_mels=n_mels)
mel_spec_db = librosa.power_to_db(mel_spec, ref=np.max)
# Statistical features from mel-spectrogram
return {
'mel_spec_mean': float(np.mean(mel_spec_db)),
'mel_spec_std': float(np.std(mel_spec_db)),
'mel_spec_max': float(np.max(mel_spec_db)),
'mel_spec_min': float(np.min(mel_spec_db)),
'mel_spec_median': float(np.median(mel_spec_db)),
# Energy in different frequency bands (important for cry classification)
'mel_low_band_energy': float(np.mean(mel_spec_db[:n_mels//4])), # Low frequencies
'mel_mid_band_energy': float(np.mean(mel_spec_db[n_mels//4:3*n_mels//4])), # Mid frequencies
'mel_high_band_energy': float(np.mean(mel_spec_db[3*n_mels//4:])), # High frequencies
}
def extract_spectral_contrast(self, y, sr):
"""Extract spectral contrast features (useful for distinguishing cry types)"""
spectral_contrast = librosa.feature.spectral_contrast(y=y, sr=sr)
return {
'spectral_contrast_mean': np.mean(spectral_contrast, axis=1).tolist(),
'spectral_contrast_std': np.std(spectral_contrast, axis=1).tolist(),
'spectral_contrast_max': np.max(spectral_contrast, axis=1).tolist(),
'spectral_contrast_min': np.min(spectral_contrast, axis=1).tolist(),
}
def extract_polyphonic_features(self, y, sr):
"""Extract polyphonic features (for complex audio analysis)
Note: CQT is computationally expensive. Consider using fast_mode=True
to skip this for faster processing.
"""
# CQT (Constant-Q Transform) for better frequency resolution
# This is expensive - use hop_length=512 to speed up
cqt = np.abs(librosa.cqt(y, sr=sr, hop_length=512))
# Chroma from CQT (more accurate than STFT-based chroma)
chroma_cqt = librosa.feature.chroma_cqt(y=y, sr=sr, hop_length=512)
return {
'cqt_mean': float(np.mean(cqt)),
'cqt_std': float(np.std(cqt)),
'cqt_max': float(np.max(cqt)),
'chroma_cqt_mean': np.mean(chroma_cqt, axis=1).tolist(),
'chroma_cqt_std': np.std(chroma_cqt, axis=1).tolist(),
}
def extract_time_domain_features(self, y, sr):
"""Extract time-domain features"""
# Zero-crossing rate (already in spectral, but add more detail)
zcr = librosa.feature.zero_crossing_rate(y)[0]
# Energy envelope
rms = librosa.feature.rms(y=y)[0]
# Autocorrelation (for periodicity detection)
autocorr = np.correlate(y, y, mode='full')
autocorr = autocorr[len(autocorr)//2:]
autocorr = autocorr / autocorr[0] if autocorr[0] != 0 else autocorr
# Find dominant period (for fundamental frequency estimation)
# Look for peaks in autocorrelation (excluding first few samples)
if len(autocorr) > 100:
peak_idx = np.argmax(autocorr[10:100]) + 10
dominant_period = peak_idx / sr if peak_idx > 0 else 0
fundamental_freq = sr / peak_idx if peak_idx > 0 else 0
else:
dominant_period = 0
fundamental_freq = 0
return {
'zcr_mean': float(np.mean(zcr)),
'zcr_std': float(np.std(zcr)),
'zcr_max': float(np.max(zcr)),
'rms_mean': float(np.mean(rms)),
'rms_std': float(np.std(rms)),
'rms_max': float(np.max(rms)),
'dominant_period': float(dominant_period),
'fundamental_freq': float(fundamental_freq),
'energy_variance': float(np.var(rms)),
}
def extract_enhanced_chroma(self, y, sr):
"""Extract enhanced chroma features with multiple variants
Note: CQT-based chroma is expensive. Consider using fast_mode=True
to skip this for faster processing.
"""
# Standard chroma (fast)
chroma_stft = librosa.feature.chroma_stft(y=y, sr=sr)
# Chroma from CQT (more accurate but expensive - use hop_length=512)
chroma_cqt = librosa.feature.chroma_cqt(y=y, sr=sr, hop_length=512)
# Chroma from CENS (Chroma Energy Normalized Statistics)
chroma_cens = librosa.feature.chroma_cens(y=y, sr=sr)
return {
'chroma_stft_mean': np.mean(chroma_stft, axis=1).tolist(),
'chroma_cqt_mean': np.mean(chroma_cqt, axis=1).tolist(),
'chroma_cens_mean': np.mean(chroma_cens, axis=1).tolist(),
'chroma_stft_std': np.std(chroma_stft, axis=1).tolist(),
'chroma_cqt_std': np.std(chroma_cqt, axis=1).tolist(),
}
def extract_all_features(self, y, sr):
"""Extract all audio features (including advanced features)"""
features = {}
# MFCC features
mfcc_mean, mfcc_std = self.extract_mfcc_features(y, sr)
features.update({
f'mfcc_{i}_mean': float(mfcc_mean[i]) for i in range(len(mfcc_mean))
})
features.update({
f'mfcc_{i}_std': float(mfcc_std[i]) for i in range(len(mfcc_std))
})
# Spectral features
spectral_features = self.extract_spectral_features(y, sr)
features.update(spectral_features)
# Rhythm features
rhythm_features = self.extract_rhythm_features(y, sr)
features.update(rhythm_features)
# Tonal features
tonal_features = self.extract_tonal_features(y, sr)
for key, value in tonal_features.items():
if isinstance(value, list):
for i, v in enumerate(value):
features[f'{key}_{i}'] = float(v)
else:
features[key] = float(value)
# NEW: Mel-spectrogram features (better for cry classification)
mel_features = self.extract_mel_spectrogram_features(y, sr)
features.update(mel_features)
# NEW: Spectral contrast features
contrast_features = self.extract_spectral_contrast(y, sr)
for key, value in contrast_features.items():
if isinstance(value, list):
for i, v in enumerate(value):
features[f'{key}_{i}'] = float(v)
else:
features[key] = float(value)
# NEW: Polyphonic features (expensive - skip in fast mode)
if not self.fast_mode:
poly_features = self.extract_polyphonic_features(y, sr)
for key, value in poly_features.items():
if isinstance(value, list):
for i, v in enumerate(value):
features[f'{key}_{i}'] = float(v)
else:
features[key] = float(value)
# NEW: Time-domain features
time_features = self.extract_time_domain_features(y, sr)
features.update(time_features)
# NEW: Enhanced chroma features (expensive - skip in fast mode)
if not self.fast_mode:
chroma_features = self.extract_enhanced_chroma(y, sr)
for key, value in chroma_features.items():
if isinstance(value, list):
for i, v in enumerate(value):
features[f'{key}_{i}'] = float(v)
else:
features[key] = float(value)
return features
# ==================== PREPROCESSING ====================
def preprocess_audio(self, y, sr):
"""
Preprocess audio signal with consistent pipeline:
1. Resample to target sample rate
2. Apply high-pass filter (remove low frequency noise)
3. Normalize amplitude
4. Trim silence
5. Pad/truncate to consistent length
"""
# Resample if needed
if sr != self.sample_rate:
y = librosa.resample(y, orig_sr=sr, target_sr=self.sample_rate)
sr = self.sample_rate
# Apply high-pass filter at 100Hz to remove low-frequency noise
# Baby cries are typically 300-600Hz fundamental
y = self._apply_highpass_filter(y, sr, cutoff=100)
# Normalize audio amplitude
y = librosa.util.normalize(y)
# Trim silence from beginning and end
y, _ = librosa.effects.trim(y, top_db=30)
# Ensure minimum length (pad with zeros if too short)
min_samples = int(self.MIN_DURATION * sr)
if len(y) < min_samples:
y = np.pad(y, (0, min_samples - len(y)), mode='constant')
# Truncate to max duration if too long
max_samples = int(self.duration * sr)
if len(y) > max_samples:
y = y[:max_samples]
return y, sr
def _apply_highpass_filter(self, y, sr, cutoff=100):
"""Apply high-pass filter to remove low-frequency noise"""
try:
# Design Butterworth high-pass filter
nyquist = sr / 2
normalized_cutoff = cutoff / nyquist
# Ensure cutoff is valid
if normalized_cutoff >= 1:
return y
b, a = signal.butter(4, normalized_cutoff, btype='high')
y_filtered = signal.filtfilt(b, a, y)
return y_filtered
except Exception:
# If filter fails, return original
return y
# ==================== MAIN EXTRACTION METHODS ====================
def extract_features_from_file(self, file_path, validate=False):
"""
Extract features from audio file
Args:
file_path: Path to audio file
validate: If True, validate audio before extraction
Returns:
features dict, or None if failed
If validate=True, also returns validation info
"""
if validate:
is_valid, message, stats = self.validate_audio(file_path)
if not is_valid:
return None, {'valid': False, 'message': message, 'stats': stats}
y, sr = self.load_audio(file_path)
if y is None:
if validate:
return None, {'valid': False, 'message': 'Could not load audio'}
return None
# Preprocess
y, sr = self.preprocess_audio(y, sr)
# Extract features
features = self.extract_all_features(y, sr)
if validate:
return features, {'valid': True, 'message': 'Success'}
return features
def extract_features_with_diagnostics(self, file_path):
"""
Extract features with full diagnostics
Returns:
tuple: (features, diagnostics)
- features: dict of extracted features or None
- diagnostics: dict with validation, stats, and feature summary
"""
diagnostics = {
'valid': False,
'message': '',
'stats': {},
'feature_summary': {}
}
# Get audio stats
stats = self.get_audio_stats(file_path)
diagnostics['stats'] = stats
if not stats.get('valid', False):
diagnostics['message'] = f"Could not analyze audio: {stats.get('error', 'Unknown')}"
return None, diagnostics
# Validate
is_valid, message, _ = self.validate_audio(file_path)
diagnostics['valid'] = is_valid
diagnostics['message'] = message
if not is_valid:
return None, diagnostics
# Load and extract features
y, sr = self.load_audio(file_path)
if y is None:
diagnostics['message'] = 'Failed to load audio for feature extraction'
return None, diagnostics
# Preprocess
y, sr = self.preprocess_audio(y, sr)
# Extract features
features = self.extract_all_features(y, sr)
# Add feature summary to diagnostics
diagnostics['feature_summary'] = self.get_feature_summary(features)
return features, diagnostics
def extract_features_from_array(self, y, sr=None):
"""Extract features from audio array"""
if sr is None:
sr = self.sample_rate
# Preprocess
y, sr = self.preprocess_audio(y, sr)
# Extract features
features = self.extract_all_features(y, sr)
return features
def get_feature_names(self):
"""Get list of feature names"""
y = np.random.randn(22050) # 1 second of random audio
sr = self.sample_rate
features = self.extract_all_features(y, sr)
return list(features.keys())
# Example usage and testing
if __name__ == "__main__":
processor = AudioProcessor()
print("🎵 Audio Processor Test")
print("=" * 50)
# Show configuration
print(f"\n⚙️ Configuration:")
print(f" Sample rate: {processor.sample_rate} Hz")
print(f" Max duration: {processor.duration} seconds")
print(f" Min duration: {processor.MIN_DURATION} seconds")
print(f" Min audio level: {processor.MIN_AUDIO_LEVEL_DB} dB")
print(f" Max silence ratio: {processor.MAX_SILENCE_RATIO * 100}%")
feature_names = processor.get_feature_names()
print(f"\n📊 Total features: {len(feature_names)}")
print("🔍 Feature categories:")
print(" • MFCC features (26)")
print(" • Spectral features (8)")
print(" • Rhythm features (6)")
print(" • Tonal features (25)")
print("\n✅ Audio processor ready!")