""" 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!")