Buckets:
| """ | |
| evaluate.py | |
| Evaluates the trained CNN-LSTM baseline model on held-out test partition. | |
| Uses validation partition ONLY for decision threshold tuning. | |
| Generates test_metrics.json, confusion_matrix.png, training_curves.png, and model_summary.txt. | |
| """ | |
| import os | |
| import sys | |
| import json | |
| import platform | |
| import argparse | |
| import numpy as np | |
| import pandas as pd | |
| import torch | |
| import matplotlib | |
| matplotlib.use("Agg") | |
| import matplotlib.pyplot as plt | |
| import seaborn as sns | |
| from sklearn.metrics import ( | |
| precision_score, recall_score, f1_score, accuracy_score, | |
| precision_recall_curve, auc, roc_auc_score, roc_curve, | |
| confusion_matrix | |
| ) | |
| from model import build_model_from_config | |
| from dataset import get_data_loaders | |
| def find_optimal_threshold(val_y_true, val_y_probs): | |
| """ | |
| Find optimal classification threshold on VALIDATION data ONLY (maximizing F1). | |
| """ | |
| precisions, recalls, thresholds = precision_recall_curve(val_y_true, val_y_probs) | |
| f1_scores = [] | |
| # thresholds array has length len(precisions) - 1 | |
| for p, r in zip(precisions[:-1], recalls[:-1]): | |
| if (p + r) > 0: | |
| f1 = 2 * (p * r) / (p + r) | |
| else: | |
| f1 = 0.0 | |
| f1_scores.append(f1) | |
| if len(f1_scores) > 0 and max(f1_scores) > 0: | |
| best_idx = int(np.argmax(f1_scores)) | |
| best_threshold = float(thresholds[best_idx]) | |
| best_f1 = float(f1_scores[best_idx]) | |
| print(f"Validation optimal threshold found: {best_threshold:.4f} (Val F1: {best_f1:.4f})") | |
| return best_threshold | |
| else: | |
| print("Fallback to default threshold 0.5") | |
| return 0.5 | |
| def run_inference(model, data_loader, device): | |
| model.eval() | |
| all_y_true = [] | |
| all_y_probs = [] | |
| all_ids = [] | |
| with torch.no_grad(): | |
| for x, y, window_ids in data_loader: | |
| x = x.to(device, non_blocking=True) | |
| logits = model(x) | |
| probs = torch.softmax(logits, dim=1)[:, 1].cpu().numpy() | |
| all_y_true.extend(y.numpy()) | |
| all_y_probs.extend(probs) | |
| all_ids.extend(window_ids) | |
| return np.array(all_y_true), np.array(all_y_probs), all_ids | |
| def plot_confusion_matrix(cm, save_path, title="Test Set Confusion Matrix"): | |
| plt.figure(figsize=(6, 5)) | |
| sns.heatmap( | |
| cm, | |
| annot=True, | |
| fmt="d", | |
| cmap="Blues", | |
| xticklabels=["Non-Seizure (0)", "Seizure (1)"], | |
| yticklabels=["Non-Seizure (0)", "Seizure (1)"] | |
| ) | |
| plt.xlabel("Predicted Label") | |
| plt.ylabel("True Label") | |
| plt.title(title) | |
| plt.tight_layout() | |
| plt.savefig(save_path, dpi=300) | |
| plt.close() | |
| print(f"Confusion matrix saved to {save_path}") | |
| def plot_training_curves(metrics_csv_path, save_path): | |
| if not os.path.exists(metrics_csv_path): | |
| print(f"Metrics CSV not found at {metrics_csv_path}. Skipping curve plotting.") | |
| return | |
| df = pd.read_csv(metrics_csv_path) | |
| if len(df) == 0: | |
| return | |
| fig, axes = plt.subplots(2, 2, figsize=(12, 10)) | |
| # 1. Loss Curve | |
| axes[0, 0].plot(df["epoch"], df["train_loss"], label="Train Loss", color="blue", marker="o") | |
| axes[0, 0].plot(df["epoch"], df["val_loss"], label="Val Loss", color="orange", marker="s") | |
| axes[0, 0].set_title("Training & Validation Loss") | |
| axes[0, 0].set_xlabel("Epoch") | |
| axes[0, 0].set_ylabel("Loss") | |
| axes[0, 0].grid(True) | |
| axes[0, 0].legend() | |
| # 2. PR-AUC Curve | |
| axes[0, 1].plot(df["epoch"], df["val_pr_auc"], label="Val PR-AUC", color="green", marker="^") | |
| axes[0, 1].set_title("Validation PR-AUC") | |
| axes[0, 1].set_xlabel("Epoch") | |
| axes[0, 1].set_ylabel("PR-AUC") | |
| axes[0, 1].grid(True) | |
| axes[0, 1].legend() | |
| # 3. F1-Score Curve | |
| axes[1, 0].plot(df["epoch"], df["val_f1"], label="Val F1-Score", color="purple", marker="d") | |
| axes[1, 0].set_title("Validation F1-Score") | |
| axes[1, 0].set_xlabel("Epoch") | |
| axes[1, 0].set_ylabel("F1-Score") | |
| axes[1, 0].grid(True) | |
| axes[1, 0].legend() | |
| # 4. Sensitivity & Specificity | |
| axes[1, 1].plot(df["epoch"], df["val_sensitivity"], label="Val Sensitivity", color="red", marker="v") | |
| axes[1, 1].plot(df["epoch"], df["val_specificity"], label="Val Specificity", color="teal", marker="x") | |
| axes[1, 1].set_title("Validation Sensitivity & Specificity") | |
| axes[1, 1].set_xlabel("Epoch") | |
| axes[1, 1].set_ylabel("Score") | |
| axes[1, 1].grid(True) | |
| axes[1, 1].legend() | |
| plt.tight_layout() | |
| plt.savefig(save_path, dpi=300) | |
| plt.close() | |
| print(f"Training curves saved to {save_path}") | |
| def evaluate_pipeline(config_path="configs/baseline_config.json", checkpoint_path="checkpoints/best_model.pt"): | |
| with open(config_path, "r") as f: | |
| config = json.load(f) | |
| reports_dir = config["training"].get("reports_dir", "reports") | |
| log_dir = config["training"].get("log_dir", "logs") | |
| os.makedirs(reports_dir, exist_ok=True) | |
| # Device | |
| device = torch.device("cuda" if torch.cuda.is_available() and config.get("hardware", {}).get("device", "cuda") == "cuda" else "cpu") | |
| print(f"Evaluation running on device: {device}") | |
| # Load Model | |
| print(f"Loading checkpoint from {checkpoint_path}...") | |
| if not os.path.exists(checkpoint_path): | |
| raise FileNotFoundError(f"Checkpoint file not found: {checkpoint_path}") | |
| checkpoint = torch.load(checkpoint_path, map_location=device) | |
| model = build_model_from_config(config).to(device) | |
| model.load_state_dict(checkpoint["model_state_dict"]) | |
| model.eval() | |
| # Load DataLoaders | |
| data_dict = get_data_loaders(config) | |
| val_loader = data_dict["val_loader"] | |
| test_loader = data_dict["test_loader"] | |
| # 1. Validation Inference for Threshold Selection | |
| print("Running validation inference for threshold calibration...") | |
| val_y_true, val_y_probs, _ = run_inference(model, val_loader, device) | |
| optimal_threshold = find_optimal_threshold(val_y_true, val_y_probs) | |
| # 2. Test Inference | |
| print("Running held-out test inference...") | |
| test_y_true, test_y_probs, test_ids = run_inference(model, test_loader, device) | |
| test_y_preds = (test_y_probs >= optimal_threshold).astype(int) | |
| # Compute Metrics on Held-out Test Set | |
| cm = confusion_matrix(test_y_true, test_y_preds, labels=[0, 1]) | |
| tn, fp, fn, tp = cm.ravel() if cm.shape == (2, 2) else (0, 0, 0, 0) | |
| precision = float(precision_score(test_y_true, test_y_preds, pos_label=1, zero_division=0)) | |
| recall = float(recall_score(test_y_true, test_y_preds, pos_label=1, zero_division=0)) | |
| specificity = float(tn / (tn + fp)) if (tn + fp) > 0 else 0.0 | |
| f1 = float(f1_score(test_y_true, test_y_preds, pos_label=1, zero_division=0)) | |
| accuracy = float(accuracy_score(test_y_true, test_y_preds)) | |
| # Test PR-AUC & ROC-AUC | |
| try: | |
| p_c, r_c, _ = precision_recall_curve(test_y_true, test_y_probs) | |
| pr_auc = float(auc(r_c, p_c)) | |
| except Exception: | |
| pr_auc = 0.0 | |
| try: | |
| roc_auc = float(roc_auc_score(test_y_true, test_y_probs)) if len(np.unique(test_y_true)) > 1 else 0.5 | |
| except Exception: | |
| roc_auc = 0.5 | |
| # Count Support | |
| test_total = len(test_y_true) | |
| test_seizures = int(np.sum(test_y_true == 1)) | |
| test_non_seizures = int(np.sum(test_y_true == 0)) | |
| total_params = sum(p.numel() for p in model.parameters()) | |
| results = { | |
| "evaluation_timestamp": pd.Timestamp.utcnow().isoformat(), | |
| "checkpoint_used": checkpoint_path, | |
| "checkpoint_epoch": checkpoint.get("epoch", "unknown"), | |
| "best_validation_metric": checkpoint.get("best_metric_val", "unknown"), | |
| "classification_threshold": round(optimal_threshold, 4), | |
| "test_support": { | |
| "total_windows": test_total, | |
| "non_seizure_windows (0)": test_non_seizures, | |
| "seizure_windows (1)": test_seizures, | |
| "seizure_percentage": round(test_seizures / test_total * 100, 4) | |
| }, | |
| "confusion_matrix": { | |
| "true_negatives": int(tn), | |
| "false_positives": int(fp), | |
| "false_negatives": int(fn), | |
| "true_positives": int(tp) | |
| }, | |
| "metrics": { | |
| "accuracy": round(accuracy, 4), | |
| "precision": round(precision, 4), | |
| "recall_sensitivity": round(recall, 4), | |
| "specificity": round(specificity, 4), | |
| "f1_score": round(f1, 4), | |
| "pr_auc": round(pr_auc, 4), | |
| "roc_auc": round(roc_auc, 4) | |
| }, | |
| "model_architecture": { | |
| "total_parameters": total_params, | |
| "input_shape": [18, 2560], | |
| "num_classes": 2 | |
| }, | |
| "software_environment": { | |
| "os": platform.platform(), | |
| "python": sys.version.split()[0], | |
| "pytorch": torch.__version__, | |
| "cuda_available": torch.cuda.is_available(), | |
| "device": str(device) | |
| }, | |
| "clinical_disclaimer": ( | |
| "Window-level classification metrics reflect performance on isolated 10-second temporal slices. " | |
| "They do NOT establish clinical seizure-event detection performance or regulatory validity. " | |
| "Real-world clinical utility requires patient-level event-based post-processing, latency analysis, and alarm rates." | |
| ) | |
| } | |
| # Print summary | |
| print("\n" + "=" * 60) | |
| print("HELD-OUT TEST SET EVALUATION RESULTS") | |
| print("=" * 60) | |
| print(f"Threshold: {results['classification_threshold']}") | |
| print(f"Accuracy: {results['metrics']['accuracy']:.4f}") | |
| print(f"Precision: {results['metrics']['precision']:.4f}") | |
| print(f"Recall/Sensitivity: {results['metrics']['recall_sensitivity']:.4f}") | |
| print(f"Specificity: {results['metrics']['specificity']:.4f}") | |
| print(f"F1-Score: {results['metrics']['f1_score']:.4f}") | |
| print(f"PR-AUC: {results['metrics']['pr_auc']:.4f}") | |
| print(f"ROC-AUC: {results['metrics']['roc_auc']:.4f}") | |
| print(f"Confusion Matrix: TP={tp}, FP={fp}, TN={tn}, FN={fn}") | |
| print("=" * 60) | |
| # Save reports | |
| test_metrics_json_path = os.path.join(reports_dir, "test_metrics.json") | |
| with open(test_metrics_json_path, "w") as f: | |
| json.dump(results, f, indent=2) | |
| print(f"Test metrics saved to {test_metrics_json_path}") | |
| cm_img_path = os.path.join(reports_dir, "confusion_matrix.png") | |
| plot_confusion_matrix(cm, cm_img_path) | |
| metrics_csv_path = os.path.join(log_dir, "metrics.csv") | |
| curves_img_path = os.path.join(reports_dir, "training_curves.png") | |
| plot_training_curves(metrics_csv_path, curves_img_path) | |
| # Model Summary text | |
| model_summary_path = os.path.join(reports_dir, "model_summary.txt") | |
| with open(model_summary_path, "w") as f: | |
| f.write("=" * 70 + "\n") | |
| f.write("CNN-LSTM BASELINE MODEL SUMMARY\n") | |
| f.write("=" * 70 + "\n\n") | |
| f.write(str(model) + "\n\n") | |
| f.write(f"Total Parameters: {total_params:,}\n") | |
| f.write(f"Trainable Parameters: {sum(p.numel() for p in model.parameters() if p.requires_grad):,}\n") | |
| f.write(f"Input Shape: (batch_size, 18, 2560)\n") | |
| f.write(f"Output Shape: (batch_size, 2)\n") | |
| f.write(f"Evaluation Threshold: {optimal_threshold:.4f}\n") | |
| f.write(f"Test F1: {f1:.4f} | PR-AUC: {pr_auc:.4f} | Recall: {recall:.4f} | Precision: {precision:.4f}\n") | |
| print(f"Model summary written to {model_summary_path}") | |
| return results | |
| if __name__ == "__main__": | |
| parser = argparse.ArgumentParser(description="Evaluate CNN-LSTM on test set") | |
| parser.add_argument("--config", default="configs/baseline_config.json") | |
| parser.add_argument("--checkpoint", default="checkpoints/best_model.pt") | |
| args = parser.parse_args() | |
| evaluate_pipeline(config_path=args.config, checkpoint_path=args.checkpoint) | |
Xet Storage Details
- Size:
- 11.8 kB
- Xet hash:
- 0554d619bcb361413cd2197a9cb8422b6b37508ead425a9d5b5756e302f8ac07
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.