Spaces:
Running on Zero
Running on Zero
| """Comprehensive Evaluation, Metrics Computation, Confusion Matrix, and Error Analysis.""" | |
| import os | |
| import json | |
| import time | |
| from pathlib import Path | |
| from typing import Dict, Any, List, Optional | |
| import numpy as np | |
| import pandas as pd | |
| import matplotlib | |
| matplotlib.use('Agg') | |
| import matplotlib.pyplot as plt | |
| import seaborn as sns | |
| from sklearn.metrics import ( | |
| accuracy_score, | |
| precision_recall_fscore_support, | |
| confusion_matrix, | |
| classification_report | |
| ) | |
| from src.utils.config import ( | |
| EXPERIMENTS_DIR, | |
| EMOTION_CLASSES, | |
| ID_TO_EMOTION, | |
| EMOTION_TO_ID | |
| ) | |
| class ExperimentLogger: | |
| """Standardized Experiment Logger saving configs, metrics, plots, and error analysis.""" | |
| def __init__(self, experiment_id: str, description: str = ""): | |
| self.experiment_id = experiment_id | |
| self.exp_dir = EXPERIMENTS_DIR / experiment_id | |
| self.exp_dir.mkdir(parents=True, exist_ok=True) | |
| self.model_dir = self.exp_dir / "model" | |
| self.model_dir.mkdir(parents=True, exist_ok=True) | |
| self.description = description | |
| self.start_time = time.time() | |
| def log_config(self, config: Dict[str, Any]): | |
| config_path = self.exp_dir / "config.json" | |
| with open(config_path, "w", encoding="utf-8") as f: | |
| json.dump(config, f, indent=2) | |
| def evaluate_and_log( | |
| self, | |
| y_true: List[str], | |
| y_pred: List[str], | |
| y_proba: Optional[np.ndarray] = None, | |
| audio_ids: Optional[List[str]] = None, | |
| split_name: str = "test", | |
| extra_metrics: Optional[Dict[str, Any]] = None | |
| ) -> Dict[str, Any]: | |
| """Compute full metrics suite, plot confusion matrix, and generate error_analysis.csv.""" | |
| training_time = round(time.time() - self.start_time, 2) | |
| # 1. Macro and weighted metrics | |
| acc = float(accuracy_score(y_true, y_pred)) | |
| p_macro, r_macro, f1_macro, _ = precision_recall_fscore_support( | |
| y_true, y_pred, average="macro", zero_division=0 | |
| ) | |
| p_weighted, r_weighted, f1_weighted, _ = precision_recall_fscore_support( | |
| y_true, y_pred, average="weighted", zero_division=0 | |
| ) | |
| # 2. Per-class metrics | |
| p_per_class, r_per_class, f1_per_class, support = precision_recall_fscore_support( | |
| y_true, y_pred, labels=EMOTION_CLASSES, zero_division=0 | |
| ) | |
| per_class_metrics = {} | |
| for idx, emo in enumerate(EMOTION_CLASSES): | |
| per_class_metrics[emo] = { | |
| "precision": float(round(p_per_class[idx], 4)), | |
| "recall": float(round(r_per_class[idx], 4)), | |
| "f1_score": float(round(f1_per_class[idx], 4)), | |
| "support": int(support[idx]) | |
| } | |
| # 3. Build metrics payload | |
| metrics = { | |
| "experiment_id": self.experiment_id, | |
| "split_evaluated": split_name, | |
| "training_time_sec": training_time, | |
| "accuracy": float(round(acc, 4)), | |
| "accuracy_pct": float(round(acc * 100.0, 2)), | |
| "macro_f1": float(round(f1_macro, 4)), | |
| "macro_precision": float(round(p_macro, 4)), | |
| "macro_recall": float(round(r_macro, 4)), | |
| "weighted_f1": float(round(f1_weighted, 4)), | |
| "per_class": per_class_metrics | |
| } | |
| if extra_metrics: | |
| metrics.update(extra_metrics) | |
| # Save metrics.json | |
| metrics_path = self.exp_dir / "metrics.json" | |
| with open(metrics_path, "w", encoding="utf-8") as f: | |
| json.dump(metrics, f, indent=2) | |
| # Save classification_report.json | |
| clf_report_dict = classification_report(y_true, y_pred, labels=EMOTION_CLASSES, output_dict=True, zero_division=0) | |
| with open(self.exp_dir / "classification_report.json", "w", encoding="utf-8") as f: | |
| json.dump(clf_report_dict, f, indent=2) | |
| # 4. Confusion Matrix Plot | |
| cm = confusion_matrix(y_true, y_pred, labels=EMOTION_CLASSES) | |
| cm_norm = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis] | |
| fig, ax = plt.subplots(figsize=(8, 6)) | |
| sns.heatmap( | |
| cm, | |
| annot=True, | |
| fmt="d", | |
| cmap="Blues", | |
| xticklabels=EMOTION_CLASSES, | |
| yticklabels=EMOTION_CLASSES, | |
| ax=ax, | |
| cbar=True | |
| ) | |
| ax.set_title(f"{self.experiment_id}: Confusion Matrix (Acc: {acc*100:.1f}%)", fontsize=14, fontweight="bold", pad=12) | |
| ax.set_ylabel("Actual Emotion", fontsize=12) | |
| ax.set_xlabel("Predicted Emotion", fontsize=12) | |
| plt.tight_layout() | |
| cm_path = self.exp_dir / "confusion_matrix.png" | |
| fig.savefig(cm_path, dpi=300) | |
| plt.close(fig) | |
| # 5. Error Analysis CSV | |
| if audio_ids is not None: | |
| error_records = [] | |
| for i, (actual, pred) in enumerate(zip(y_true, y_pred)): | |
| conf = 0.0 | |
| if y_proba is not None and i < len(y_proba): | |
| pred_idx = EMOTION_TO_ID.get(pred, 0) | |
| conf = float(round(y_proba[i][pred_idx], 4)) | |
| is_error = (actual != pred) | |
| error_records.append({ | |
| "audio_id": audio_ids[i], | |
| "actual_emotion": actual, | |
| "predicted_emotion": pred, | |
| "confidence": conf, | |
| "is_misclassified": is_error, | |
| "error_pair": f"{actual} -> {pred}" if is_error else "CORRECT" | |
| }) | |
| err_df = pd.DataFrame(error_records) | |
| err_csv_path = self.exp_dir / "error_analysis.csv" | |
| err_df.to_csv(err_csv_path, index=False) | |
| print(f"\n[EVALUATION] {self.experiment_id} Results:") | |
| print(f" Accuracy: {acc * 100:.2f}%") | |
| print(f" Macro F1: {f1_macro * 100:.2f}%") | |
| print(f" Weighted F1: {f1_weighted * 100:.2f}%") | |
| print(f" Saved artifacts to: {self.exp_dir}") | |
| return metrics | |