Spaces:
Sleeping
Sleeping
File size: 2,946 Bytes
f234112 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """Evaluation helpers for validation and test splits."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
import torch
from torch.utils.data import DataLoader
from .dataset import AccidentParquetDataset, collate_detection
from .metrics import evaluate_coco, threshold_analysis
from .model import build_detector
from .train import gather_predictions
from .utils import FIGURE_DIR, MODEL_DIR, REPORT_DIR, get_device
def evaluate_split_from_checkpoint(
split: str,
checkpoint_path: Path,
config: dict[str, Any],
dataset_dir: Path | None = None,
score_threshold: float = 0.05,
) -> dict[str, Any]:
"""Evaluate a split using a saved checkpoint.
Returns ground-truth records, prediction records, COCO metrics, and
threshold analysis.
"""
device = get_device()
image_size = tuple(config["image_size"])
ds = AccidentParquetDataset(
split=split, target_size=image_size, dataset_dir=dataset_dir
)
loader = DataLoader(
ds,
batch_size=config["batch_size"],
shuffle=False,
num_workers=config.get("num_workers", 0),
collate_fn=collate_detection,
)
model = build_detector(num_classes=config["num_classes"], pretrained=False)
model.load_state_dict(torch.load(checkpoint_path, map_location=device, weights_only=True))
model.to(device)
gt_records, pred_records = gather_predictions(model, loader, device, score_threshold)
metrics = evaluate_coco(gt_records, pred_records, image_size)
thresholds = threshold_analysis(gt_records, pred_records, image_size=image_size)
return {
"split": split,
"metrics": metrics,
"threshold_analysis": thresholds,
"num_images": len(ds),
"num_ground_truth_objects": sum(len(rec["boxes"]) for rec in gt_records),
"num_predictions": sum(len(rec["boxes"]) for rec in pred_records),
}
def run_test_evaluation(
config: dict[str, Any],
checkpoint_path: Path = MODEL_DIR / "best_detector.pth",
dataset_dir: Path | None = None,
report_dir: Path = REPORT_DIR,
) -> dict[str, Any]:
"""Run the ONE final test evaluation after validation-based selection."""
print("Running final test evaluation with best validation checkpoint...")
result = evaluate_split_from_checkpoint(
split="test",
checkpoint_path=checkpoint_path,
config=config,
dataset_dir=dataset_dir,
)
report_dir.mkdir(parents=True, exist_ok=True)
test_path = report_dir / "test_results.json"
with open(test_path, "w", encoding="utf-8") as f:
json.dump(result, f, indent=2)
print(
f"Test mAP@50={result['metrics']['overall']['mAP_50']:.4f} — "
f"accident AP={result['metrics']['per_class']['accident']['AP_50']:.4f} — "
f"accident recall={result['metrics']['per_class']['accident']['recall']:.4f}"
)
return result
|