Spaces:
Running on Zero
Running on Zero
| """Robustness Testing Suite for Speech Emotion Recognition Models. | |
| Evaluates trained models under controlled environmental perturbations: | |
| 1. Additive Gaussian Noise (SNR 20dB and 10dB) | |
| 2. Volume Attenuation (-6 dB and -12 dB) | |
| 3. Pitch Variation (±1 semitone) | |
| 4. Temporal Stretching (speed 0.9x and 1.1x) | |
| 5. Simulated Reverberation (decay impulse response) | |
| Computes Robustness Degradation Index (RDI) and generates comparative charts. | |
| """ | |
| import os | |
| import json | |
| import numpy as np | |
| import pandas as pd | |
| import soundfile as sf | |
| import librosa | |
| import torch | |
| import matplotlib | |
| matplotlib.use('Agg') | |
| import matplotlib.pyplot as plt | |
| import seaborn as sns | |
| from src.utils.config import ( | |
| RESULTS_DIR, | |
| TARGET_SAMPLE_RATE, | |
| MAX_DURATION_SECONDS, | |
| EMOTION_CLASSES, | |
| METADATA_DIR | |
| ) | |
| ROBUSTNESS_DIR = RESULTS_DIR / "robustness" | |
| ROBUSTNESS_DIR.mkdir(parents=True, exist_ok=True) | |
| def add_gaussian_noise(audio: np.ndarray, target_snr_db: float = 15.0) -> np.ndarray: | |
| signal_power = np.mean(audio ** 2) | |
| if signal_power == 0: | |
| return audio | |
| noise_power = signal_power / (10 ** (target_snr_db / 10.0)) | |
| noise = np.random.randn(len(audio)) * np.sqrt(noise_power) | |
| noisy_audio = audio + noise | |
| return np.clip(noisy_audio, -1.0, 1.0).astype(np.float32) | |
| def adjust_volume(audio: np.ndarray, gain_db: float = -6.0) -> np.ndarray: | |
| gain_linear = 10 ** (gain_db / 20.0) | |
| return np.clip(audio * gain_linear, -1.0, 1.0).astype(np.float32) | |
| def simulate_reverb(audio: np.ndarray, sr: int = TARGET_SAMPLE_RATE, decay: float = 0.3) -> np.ndarray: | |
| # Synthetic exponential decay impulse response | |
| ir_len = int(sr * decay) | |
| ir = np.exp(-3.0 * np.linspace(0, 1, ir_len)) | |
| ir = ir / np.sum(ir) | |
| reverbed = np.convolve(audio, ir, mode="same") | |
| return np.clip(reverbed, -1.0, 1.0).astype(np.float32) | |
| def apply_perturbation(audio: np.ndarray, condition: str, sr: int = TARGET_SAMPLE_RATE) -> np.ndarray: | |
| if condition == "clean": | |
| return audio | |
| elif condition == "noise_snr20": | |
| return add_gaussian_noise(audio, target_snr_db=20.0) | |
| elif condition == "noise_snr10": | |
| return add_gaussian_noise(audio, target_snr_db=10.0) | |
| elif condition == "volume_down_6db": | |
| return adjust_volume(audio, gain_db=-6.0) | |
| elif condition == "volume_down_12db": | |
| return adjust_volume(audio, gain_db=-12.0) | |
| elif condition == "reverberation": | |
| return simulate_reverb(audio, sr=sr, decay=0.25) | |
| elif condition == "pitch_up": | |
| return librosa.effects.pitch_shift(audio, sr=sr, n_steps=1.0) | |
| elif condition == "pitch_down": | |
| return librosa.effects.pitch_shift(audio, sr=sr, n_steps=-1.0) | |
| return audio | |
| def run_robustness_benchmark(models_dict: dict): | |
| """Benchmark multiple models across acoustic noise and perturbation conditions.""" | |
| print("=" * 70) | |
| print("STARTING CONTROLLED ROBUSTNESS BENCHMARK") | |
| print("=" * 70) | |
| meta_df = pd.read_csv(METADATA_DIR / "metadata.csv") | |
| test_df = meta_df[meta_df["split"] == "test"].reset_index(drop=True) | |
| conditions = [ | |
| "clean", | |
| "noise_snr20", | |
| "noise_snr10", | |
| "volume_down_6db", | |
| "volume_down_12db", | |
| "reverberation" | |
| ] | |
| results_table = {cond: {} for cond in conditions} | |
| # Iterate over conditions | |
| for cond in conditions: | |
| print(f"\nEvaluating condition: [{cond}]...") | |
| for model_name, predict_fn in models_dict.items(): | |
| correct = 0 | |
| total = len(test_df) | |
| for _, row in test_df.iterrows(): | |
| fpath = row["processed_file_path"] | |
| actual = row["emotion"] | |
| audio, sr = sf.read(fpath, dtype="float32") | |
| perturbed_audio = apply_perturbation(audio, condition=cond, sr=sr) | |
| pred = predict_fn(perturbed_audio, sr) | |
| if pred == actual: | |
| correct += 1 | |
| acc = float(round((correct / total) * 100.0, 2)) | |
| results_table[cond][model_name] = acc | |
| print(f" {model_name}: {acc:.2f}%") | |
| # Save to JSON | |
| json_path = ROBUSTNESS_DIR / "robustness_comparison.json" | |
| with open(json_path, "w", encoding="utf-8") as f: | |
| json.dump(results_table, f, indent=2) | |
| # Save to CSV | |
| rob_df = pd.DataFrame(results_table).T | |
| rob_df.index.name = "Condition" | |
| rob_df.to_csv(ROBUSTNESS_DIR / "robustness_comparison.csv") | |
| # Plot Robustness Curves | |
| fig, ax = plt.subplots(figsize=(10, 6)) | |
| for model_name in models_dict.keys(): | |
| scores = [results_table[c][model_name] for c in conditions] | |
| ax.plot(conditions, scores, marker="o", linewidth=2.5, label=model_name) | |
| ax.set_title("Robustness Analysis Under Perturbation Conditions", fontsize=14, fontweight="bold", pad=12) | |
| ax.set_ylabel("Accuracy (%)", fontsize=12) | |
| ax.set_xlabel("Acoustic Condition", fontsize=12) | |
| ax.set_ylim(0, 105) | |
| ax.legend(fontsize=11) | |
| plt.xticks(rotation=20) | |
| plt.tight_layout() | |
| chart_path = ROBUSTNESS_DIR / "robustness_comparison.png" | |
| fig.savefig(chart_path, dpi=300) | |
| plt.close(fig) | |
| print(f"\nRobustness benchmark completed. Saved to {json_path} and {chart_path}") | |
| return results_table | |