Story 8
This commit is contained in:
Ryan
@@ -0,0 +1,204 @@
|
||||
"""
|
||||
Candidate model validation for active learning.
|
||||
|
||||
Evaluates a candidate checkpoint on the held-out validation set,
|
||||
computes F1, ECE, accuracy, precision, and recall, then checks
|
||||
quality gates before promotion.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Dict, Optional, Tuple
|
||||
|
||||
import numpy as np
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Validator:
|
||||
"""Evaluate a candidate model and enforce quality gates."""
|
||||
|
||||
def __init__(self, config, current_f1: float = 0.0):
|
||||
al = config.get_section("active_learning")
|
||||
val_cfg = al.get("validation", {})
|
||||
self._min_f1_delta: float = float(val_cfg.get("min_f1_improvement", 0.02))
|
||||
self._max_ece: float = float(val_cfg.get("max_ece", 0.08))
|
||||
self._current_f1 = current_f1
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public API
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def validate(self, checkpoint_path: str, dataset_path: str) -> Dict:
|
||||
"""
|
||||
Run validation and return a results dict including gate pass/fail.
|
||||
|
||||
Keys: f1, ece, accuracy, precision, recall, gates_passed, gate_details
|
||||
"""
|
||||
try:
|
||||
import torch
|
||||
probs, labels = self._run_inference(checkpoint_path, dataset_path, torch)
|
||||
return self._compute_metrics(probs, labels)
|
||||
except ImportError as exc:
|
||||
logger.error("PyTorch not available: %s", exc)
|
||||
return self._empty_result()
|
||||
except Exception as exc:
|
||||
logger.error("Validation failed: %s", exc, exc_info=True)
|
||||
return self._empty_result()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Inference
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _run_inference(self, checkpoint_path: str, dataset_path: str, torch) -> Tuple[np.ndarray, np.ndarray]:
|
||||
import torch.nn as nn
|
||||
import torchvision.models as models
|
||||
import torchvision.transforms as T
|
||||
from torch.utils.data import DataLoader
|
||||
|
||||
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
|
||||
|
||||
model = self._load_checkpoint(checkpoint_path, nn, models, torch, device)
|
||||
model.eval()
|
||||
|
||||
transform = T.Compose([
|
||||
T.Resize((224, 224)),
|
||||
T.ToTensor(),
|
||||
T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
|
||||
])
|
||||
|
||||
from .trainer import _CropDataset, _split_samples
|
||||
crops_root = Path(dataset_path) / "crops"
|
||||
all_samples = _CropDataset._discover(None, crops_root) # type: ignore[arg-type]
|
||||
_, val_samples = _split_samples(all_samples, 0.80, seed=42)
|
||||
|
||||
if not val_samples:
|
||||
raise ValueError("Validation set is empty")
|
||||
|
||||
val_dataset = _ValDataset(val_samples, transform)
|
||||
loader = DataLoader(val_dataset, batch_size=32, shuffle=False, num_workers=2, pin_memory=True)
|
||||
|
||||
all_probs, all_labels = [], []
|
||||
with torch.no_grad():
|
||||
for images, batch_labels in loader:
|
||||
images = images.to(device)
|
||||
outputs = model(images)
|
||||
probs_batch = torch.softmax(outputs, dim=1)[:, 1].cpu().numpy()
|
||||
all_probs.extend(probs_batch)
|
||||
all_labels.extend(batch_labels.numpy())
|
||||
|
||||
return np.array(all_probs, dtype=np.float32), np.array(all_labels, dtype=np.int32)
|
||||
|
||||
@staticmethod
|
||||
def _load_checkpoint(checkpoint_path: str, nn, models, torch, device):
|
||||
model = models.mobilenet_v3_small(weights=None)
|
||||
model.classifier = nn.Sequential(
|
||||
nn.Linear(576, 256),
|
||||
nn.ReLU(),
|
||||
nn.Dropout(0.3),
|
||||
nn.Linear(256, 2),
|
||||
)
|
||||
state = torch.load(checkpoint_path, map_location=device)
|
||||
model.load_state_dict(state)
|
||||
return model.to(device)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Metrics
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _compute_metrics(self, probs: np.ndarray, labels: np.ndarray) -> Dict:
|
||||
from sklearn.metrics import (
|
||||
accuracy_score, f1_score, precision_score, recall_score,
|
||||
)
|
||||
|
||||
preds = (probs >= 0.5).astype(np.int32)
|
||||
f1 = float(f1_score(labels, preds, average="macro", zero_division=0))
|
||||
accuracy = float(accuracy_score(labels, preds))
|
||||
precision = float(precision_score(labels, preds, average="macro", zero_division=0))
|
||||
recall = float(recall_score(labels, preds, average="macro", zero_division=0))
|
||||
ece = compute_ece(probs, labels)
|
||||
|
||||
delta_f1 = f1 - self._current_f1
|
||||
gate_f1 = delta_f1 > self._min_f1_delta
|
||||
gate_ece = ece < self._max_ece
|
||||
gates_passed = gate_f1 and gate_ece
|
||||
|
||||
gate_details = {
|
||||
"delta_f1": round(delta_f1, 4),
|
||||
"min_f1_improvement": self._min_f1_delta,
|
||||
"gate_f1_passed": gate_f1,
|
||||
"ece": round(ece, 4),
|
||||
"max_ece": self._max_ece,
|
||||
"gate_ece_passed": gate_ece,
|
||||
}
|
||||
|
||||
logger.info(
|
||||
"Validation results: f1=%.4f delta_f1=%.4f ece=%.4f gates_passed=%s",
|
||||
f1, delta_f1, ece, gates_passed,
|
||||
)
|
||||
|
||||
return {
|
||||
"f1": round(f1, 4),
|
||||
"ece": round(ece, 4),
|
||||
"accuracy": round(accuracy, 4),
|
||||
"precision": round(precision, 4),
|
||||
"recall": round(recall, 4),
|
||||
"gates_passed": gates_passed,
|
||||
"gate_details": gate_details,
|
||||
}
|
||||
|
||||
def _empty_result(self) -> Dict:
|
||||
return {
|
||||
"f1": 0.0, "ece": 1.0, "accuracy": 0.0,
|
||||
"precision": 0.0, "recall": 0.0,
|
||||
"gates_passed": False,
|
||||
"gate_details": {},
|
||||
}
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# ECE calculation (module-level for reuse in tests)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def compute_ece(predictions: np.ndarray, labels: np.ndarray, n_bins: int = 15) -> float:
|
||||
"""
|
||||
Expected Calibration Error.
|
||||
|
||||
Bins predictions by confidence and measures average |confidence - accuracy|
|
||||
weighted by bin population.
|
||||
"""
|
||||
bin_boundaries = np.linspace(0, 1, n_bins + 1)
|
||||
ece = 0.0
|
||||
n = len(predictions)
|
||||
if n == 0:
|
||||
return 0.0
|
||||
|
||||
for i in range(n_bins):
|
||||
mask = (predictions >= bin_boundaries[i]) & (predictions < bin_boundaries[i + 1])
|
||||
if mask.sum() > 0:
|
||||
bin_confidence = float(predictions[mask].mean())
|
||||
bin_accuracy = float(labels[mask].mean())
|
||||
ece += (mask.sum() / n) * abs(bin_confidence - bin_accuracy)
|
||||
|
||||
return float(ece)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Minimal dataset wrapper for validation
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
class _ValDataset:
|
||||
def __init__(self, samples, transform):
|
||||
self._samples = samples
|
||||
self._transform = transform
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._samples)
|
||||
|
||||
def __getitem__(self, idx: int):
|
||||
from PIL import Image
|
||||
path, label = self._samples[idx]
|
||||
image = Image.open(path).convert("RGB")
|
||||
if self._transform:
|
||||
image = self._transform(image)
|
||||
return image, label
|
||||
Reference in New Issue
Block a user