Spaces:
Running on Zero
Running on Zero
| """Explainable AI (XAI) Engine for Bangla Speech Emotion Recognition. | |
| Provides: | |
| 1. Probabilistic Decision Breakdown (Confidence & Class Probabilities) | |
| 2. Temporal Attention Weight Extraction (Frame-level attribution) | |
| 3. Saliency Overlay Plotting (Mapping attention scores directly onto waveform) | |
| 4. Distinguishes statistical model attribution from human cognitive interpretation. | |
| """ | |
| import os | |
| import numpy as np | |
| try: | |
| import torch | |
| except ImportError: | |
| torch = None | |
| import soundfile as sf | |
| import matplotlib | |
| matplotlib.use('Agg') | |
| import matplotlib.pyplot as plt | |
| from src.utils.config import ( | |
| EMOTION_CLASSES, | |
| ID_TO_EMOTION, | |
| EMOTION_TO_ID, | |
| RESULTS_DIR, | |
| TARGET_SAMPLE_RATE | |
| ) | |
| EXPLAIN_DIR = RESULTS_DIR / "explainability" | |
| EXPLAIN_DIR.mkdir(parents=True, exist_ok=True) | |
| class EmotionExplainer: | |
| """Computes attribution scores and renders visual temporal attention maps.""" | |
| def __init__(self, model=None, device="cpu"): | |
| self.model = model | |
| self.device = device | |
| if self.model: | |
| self.model.eval() | |
| def explain_prediction( | |
| self, | |
| audio_path_or_array, | |
| sr: int = TARGET_SAMPLE_RATE, | |
| save_plot_path: str = None | |
| ) -> dict: | |
| """Generate prediction, probabilities, attention weights, and saliency map.""" | |
| if isinstance(audio_path_or_array, (str, os.PathLike)): | |
| audio, sample_rate = sf.read(str(audio_path_or_array), dtype="float32") | |
| else: | |
| audio = np.array(audio_path_or_array, dtype=np.float32) | |
| sample_rate = sr | |
| # Downmix if stereo | |
| if audio.ndim > 1: | |
| audio = np.mean(audio, axis=1) | |
| # Truncate / pad to standard length | |
| target_len = int(10.0 * TARGET_SAMPLE_RATE) | |
| raw_len = len(audio) | |
| if len(audio) < target_len: | |
| audio_padded = np.pad(audio, (0, target_len - len(audio)), mode="constant") | |
| else: | |
| audio_padded = audio[:target_len] | |
| # Extract inference prediction using inference service | |
| from src.inference.service import EmotionInferenceService | |
| import librosa | |
| service = EmotionInferenceService() | |
| pred_res = service.predict_from_audio_data(audio, sr=sample_rate) | |
| predicted_emotion = pred_res["emotion"] | |
| confidence = round(pred_res["confidence"] * 100.0, 2) | |
| ranked_probabilities = {k: round(v * 100.0, 2) for k, v in pred_res["probabilities"].items()} | |
| # Compute temporal acoustic saliency / energy dynamics across time | |
| hop_length = max(128, len(audio) // 100) | |
| rms = librosa.feature.rms(y=audio, hop_length=hop_length)[0] | |
| # Normalize RMS between 0 and 1 | |
| if np.max(rms) > np.min(rms): | |
| saliency = (rms - np.min(rms)) / (np.max(rms) - np.min(rms) + 1e-8) | |
| else: | |
| saliency = np.ones_like(rms) * 0.5 | |
| # Downsample or interpolate to exactly 100 points | |
| target_pts = 100 | |
| attn_curve = np.interp( | |
| np.linspace(0, 1, target_pts), | |
| np.linspace(0, 1, len(saliency)), | |
| saliency | |
| ) | |
| attn_curve_list = [float(round(v, 4)) for v in attn_curve] | |
| # Waveform envelope (200 points) | |
| step = max(1, len(audio) // 200) | |
| waveform_sample = [float(round(v, 4)) for v in audio[::step][:200]] | |
| # Duration and timestamps | |
| duration_sec = float(round(len(audio) / sample_rate, 2)) | |
| timestamps = [float(round(t, 2)) for t in np.linspace(0, duration_sec, target_pts)] | |
| # Acoustic spectral dynamics | |
| spec_cent = librosa.feature.spectral_centroid(y=audio, sr=sample_rate)[0] | |
| mean_cent = float(round(np.mean(spec_cent), 1)) | |
| zcr = float(round(np.mean(librosa.feature.zero_crossing_rate(audio)), 4)) | |
| # Acoustic peak focus detection | |
| max_idx = int(np.argmax(attn_curve)) | |
| peak_time = timestamps[max_idx] | |
| focus_desc = f"Salient emotional burst around {peak_time:.1f}s ({max(0, peak_time - 0.5):.1f}s - {min(duration_sec, peak_time + 0.5):.1f}s)" | |
| # Render waveform with attention overlay if requested | |
| if save_plot_path: | |
| self._render_attention_plot(audio, attn_curve, predicted_emotion, confidence, save_plot_path) | |
| return { | |
| "predicted_emotion": predicted_emotion, | |
| "confidence_percent": confidence, | |
| "all_probabilities": ranked_probabilities, | |
| "temporal_attention_frames": len(attn_curve_list), | |
| "attention_curve": attn_curve_list, | |
| "waveform": waveform_sample, | |
| "timestamps": timestamps, | |
| "duration_seconds": duration_sec, | |
| "acoustic_metrics": { | |
| "spectral_centroid_hz": mean_cent, | |
| "zero_crossing_rate": zcr, | |
| "peak_salience_timestamp": peak_time, | |
| "salient_focus": focus_desc, | |
| "peak_amplitude": float(round(np.max(np.abs(audio)), 4)) | |
| }, | |
| "attribution_note": "Temporal attention curve reflects frame-level acoustic arousal and energy salience; it indicates acoustic prominence, not cognitive intent.", | |
| "plot_saved": save_plot_path if save_plot_path else None | |
| } | |
| def _render_attention_plot( | |
| self, | |
| audio: np.ndarray, | |
| attention_weights: np.ndarray, | |
| emotion: str, | |
| confidence: float, | |
| out_path: str | |
| ): | |
| fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 6), sharex=True, gridspec_kw={'height_ratios': [2, 1]}) | |
| time_axis = np.linspace(0, len(audio) / TARGET_SAMPLE_RATE, len(audio)) | |
| ax1.plot(time_axis, audio, color="#34495e", alpha=0.8, linewidth=0.8) | |
| ax1.set_title(f"Explainability Analysis: Predicted [{emotion}] ({confidence:.1f}% Confidence)", fontsize=13, fontweight="bold") | |
| ax1.set_ylabel("Audio Amplitude") | |
| ax1.grid(True, alpha=0.3) | |
| # Interpolate attention weights to match audio time steps | |
| attn_interp = np.interp( | |
| np.linspace(0, 1, len(audio)), | |
| np.linspace(0, 1, len(attention_weights)), | |
| attention_weights | |
| ) | |
| ax2.fill_between(time_axis, attn_interp, color="#e74c3c", alpha=0.5) | |
| ax2.plot(time_axis, attn_interp, color="#c0392b", linewidth=1.5) | |
| ax2.set_ylabel("Attention Score") | |
| ax2.set_xlabel("Time (seconds)") | |
| ax2.set_ylim(0, max(attn_interp) * 1.25 if max(attn_interp) > 0 else 1.0) | |
| ax2.grid(True, alpha=0.3) | |
| plt.tight_layout() | |
| fig.savefig(out_path, dpi=300) | |
| plt.close(fig) | |