diff --git a/docs/checkpoint_schema.md b/docs/checkpoint_schema.md index 529aa64c..29d67bb0 100644 --- a/docs/checkpoint_schema.md +++ b/docs/checkpoint_schema.md @@ -391,6 +391,15 @@ distributed as training checkpoints. For release compatibility, readers accept legacy best-metric aliases such as `best_mAP50_95`, `best_mAP50`, `best_metric`, and `best_metric_name`. +Custom-fitness training writes optional `fitness_source: "callback"` and +`best_metric_key: "fitness/custom"` in its training and averaged checkpoints. +`best_metric_value` and the legacy `best_mAP50_95` / `best_metric` aliases then +contain the custom score, while `best_mAP50` remains the validation mAP50 at +the selected epoch. The marker stores no callback code or state. Such files +can be loaded for inference or as weights for a new run, but cannot resume +training; neither can a default checkpoint resume with a custom scorer. +Absent `fitness_source` preserves the existing checkpoint/resume contract. + ## External Snapshot Exception The schema above governs LibreYOLO-authored `.pt` checkpoints. It does not diff --git a/docs/training_loggers.md b/docs/training_loggers.md index 5eedb723..31f2a56d 100644 --- a/docs/training_loggers.md +++ b/docs/training_loggers.md @@ -49,6 +49,54 @@ reference but cannot be transported by value (for example, one that captures a write-only file handle) uses the guarded-script compatibility launcher and emits a warning; keep that call under the `__main__` guard. +### Custom fitness + +For families using the shared `BaseTrainer`, including YOLO9 and RF-DETR, +one callback object may define `fitness(metrics)`. Its finite real scalar +return value selects `best.pt` and drives the existing `patience` setting; +higher is better. Plain functions remain epoch observers. No CLI code loader +is provided: this hook is a Python callable. + +```python +class ValidationLossFitness: + def fitness(self, metrics): + return -metrics["metrics/loss"] + +model.train( + data="coco8.yaml", val_loss=True, patience=10, + callbacks=ValidationLossFitness(), +) +``` + +The scorer receives a read-only copy of the scalar validation metrics, with +the same names as `TrainEpochEvent.val_metrics`. Use the metrics your task +actually reports; missing keys raise normally. It runs on rank zero before +checkpoint selection and epoch observers, once for each validation result +used for selection. Skipped validation does not call it. Precise-BN +revalidation and final averaged-weight validation also use it. Keep the +scorer deterministic and independent of invocation count. + +Events and checkpoints name the selected score `fitness/custom`; validation +metrics keep their original values. `average_best` ranks snapshots by the +same score. Ties keep the earlier best, and patience still counts epochs +since improvement, including epochs without validation. Multiple scorers, +non-numeric results, booleans, non-scalar tensors, NaN and infinity raise +before updating best state or writing that epoch's checkpoint. + +The returned training result's legacy `best_mAP50_95` field also contains the +selected score, as it already does for non-detection task metrics. Read raw +accuracy values from `val_metrics`; `epoch_metrics` pairs `current_metric` +and `best_metric` with their metric names. + +Custom-fitness training does not support `resume`. Callback code and state +are not serialized, and the library cannot establish whether a new scorer +is comparable with historical scores. Resuming with a scorer or from a +custom-fitness checkpoint raises. To continue from its weights, load that +checkpoint and start a new run with `resume=False` and a new output directory; +this starts fresh optimizer, best-score and patience tracking. Ordinary +training and ordinary resume are unchanged. VLM and VLA trainers do not +support this scorer. + ## Built-in loggers Built-in loggers are callback objects layered on the same universal hooks. @@ -145,7 +193,8 @@ monitor overlays it like every other family. This option is off by default because target assignment adds work and memory to validation. It runs under `torch.no_grad()` with the evaluation/EMA model, and distributed training computes it locally on rank 0 without collectives. -Best-checkpoint selection remains based on the configured accuracy metric. +Best-checkpoint selection uses the task's default metric unless a custom +fitness callback is supplied. Augmented validation, a task a family has not implemented it for, and inference-only (`g3`/`g4`) families all raise a clear configuration error. diff --git a/libreyolo/models/fomo/trainer.py b/libreyolo/models/fomo/trainer.py index b5c8a130..f662a641 100644 --- a/libreyolo/models/fomo/trainer.py +++ b/libreyolo/models/fomo/trainer.py @@ -317,7 +317,6 @@ def _run_validation( raise def _checkpoint_extra_metadata(self) -> Dict[str, Any]: - return { - "task": "point", - "best_metric_key": "metrics/grid_F1", - } + # Direct trainer users may omit the model wrapper, whose task would + # otherwise supply this value to the shared checkpoint writer. + return {"task": "point"} diff --git a/libreyolo/models/rfdetr/model.py b/libreyolo/models/rfdetr/model.py index 1f94349e..fb7f07e0 100644 --- a/libreyolo/models/rfdetr/model.py +++ b/libreyolo/models/rfdetr/model.py @@ -1236,7 +1236,9 @@ def train( ``/`` when omitted; ``project=`` / ``name=`` kwargs take precedence over this split. resume: Checkpoint path, or True to resume the loaded checkpoint. - callbacks: Optional training callback or iterable of callbacks. + callbacks: Optional callback or iterable. One object may define + fitness(metrics) to select best.pt and drive patience; custom + fitness requires a new run (resume=False). loggers: Optional built-in experiment loggers: a registered name, a configured logger instance, or an iterable mixing both. """ diff --git a/libreyolo/models/vla/training/trainer.py b/libreyolo/models/vla/training/trainer.py index 9a5b97db..ece50efe 100644 --- a/libreyolo/models/vla/training/trainer.py +++ b/libreyolo/models/vla/training/trainer.py @@ -216,6 +216,8 @@ def __init__( self.callbacks = TrainCallbackList(callbacks) for logger_cb in resolve_loggers(loggers): self.callbacks.append(logger_cb) + if self.callbacks.fitness is not None: + raise NotImplementedError("VLA training does not support custom fitness callbacks") self.save_dir = self._resolve_save_dir() def _resolve_save_dir(self) -> Path: diff --git a/libreyolo/models/vlm/training/trainer.py b/libreyolo/models/vlm/training/trainer.py index c5cfd52f..13e7eb88 100644 --- a/libreyolo/models/vlm/training/trainer.py +++ b/libreyolo/models/vlm/training/trainer.py @@ -111,6 +111,8 @@ def __init__( self.callbacks = TrainCallbackList(callbacks) for logger_cb in resolve_loggers(loggers): self.callbacks.append(logger_cb) + if self.callbacks.fitness is not None: + raise NotImplementedError("VLM training does not support custom fitness callbacks") if self.config.resume is True: # resume=True reads weights/last from this exact save_dir (see # _resolve_resume_dir); never let _resolve_save_dir() increment diff --git a/libreyolo/models/yolo9/model.py b/libreyolo/models/yolo9/model.py index ff511b79..57050b84 100644 --- a/libreyolo/models/yolo9/model.py +++ b/libreyolo/models/yolo9/model.py @@ -652,7 +652,9 @@ def train( pretrained: Optional training initialization weights. Use True to load the matching LibreYOLO9 detect checkpoint for transfer learning, or pass a checkpoint path/name. - callbacks: Optional training callback or iterable of callbacks. + callbacks: Optional callback or iterable. One object may define + fitness(metrics) to select best.pt and drive patience; custom + fitness requires a new run (resume=False). loggers: Optional built-in experiment loggers: a registered name, a configured logger instance, or an iterable mixing both. diff --git a/libreyolo/training/__init__.py b/libreyolo/training/__init__.py index bf779446..d81cc426 100644 --- a/libreyolo/training/__init__.py +++ b/libreyolo/training/__init__.py @@ -11,6 +11,7 @@ TrainEndEvent as TrainEndEvent, TrainEpochEvent as TrainEpochEvent, TrainExceptionEvent as TrainExceptionEvent, + TrainFitnessCallback as TrainFitnessCallback, TrainStartEvent as TrainStartEvent, ) from .config import ( diff --git a/libreyolo/training/callbacks.py b/libreyolo/training/callbacks.py index b9da9b60..4b43078f 100644 --- a/libreyolo/training/callbacks.py +++ b/libreyolo/training/callbacks.py @@ -1,8 +1,9 @@ """Public training callback types.""" +from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, field from types import MappingProxyType -from typing import Any, Callable, Iterable, Mapping, Protocol +from typing import Any, Protocol @dataclass(frozen=True) @@ -116,8 +117,22 @@ def on_train_exception(self, event: TrainExceptionEvent) -> None: """Handle a training exception.""" +class TrainFitnessCallback(Protocol): + """Optional scorer for best-checkpoint selection and patience. + + ``metrics`` is the same read-only scalar mapping as + ``TrainEpochEvent.val_metrics``. Return a finite real scalar; higher is + better. Scoring runs on rank zero for each validation result used for + selection, before checkpoints and epoch callbacks. Callback code and + state are not saved, so custom-fitness training cannot be resumed. + """ + + def fitness(self, metrics: Mapping[str, float]) -> float: + """Score validation metrics without modifying them.""" + + TrainEpochCallable = Callable[[TrainEpochEvent], None] -TrainCallbackLike = TrainCallback | TrainEpochCallable +TrainCallbackLike = TrainCallback | TrainFitnessCallback | TrainEpochCallable TrainCallbacks = TrainCallbackLike | Iterable[TrainCallbackLike] | None @@ -131,6 +146,21 @@ def __init__(self, callbacks: TrainCallbacks = None): self._callbacks = [callbacks] else: self._callbacks = list(callbacks) + _ = self.fitness # Validate scorers before any training work starts. + + @property + def fitness(self) -> Callable[[Mapping[str, float]], float] | None: + """Return the sole optional fitness method, rejecting ambiguity.""" + scorers = [] + for callback in self._callbacks: + if hasattr(callback, "fitness"): + method = callback.fitness + if not callable(method): + raise TypeError("Train callback attribute fitness must be callable") + scorers.append(method) + if len(scorers) > 1: + raise ValueError("At most one training callback may define fitness") + return scorers[0] if scorers else None def __bool__(self) -> bool: return bool(self._callbacks) @@ -140,6 +170,11 @@ def __len__(self) -> int: def append(self, callback: TrainCallbackLike) -> None: self._callbacks.append(callback) + try: + _ = self.fitness + except (TypeError, ValueError): + self._callbacks.pop() + raise @staticmethod def _is_callback_object(callback) -> bool: @@ -150,6 +185,7 @@ def _is_callback_object(callback) -> bool: "on_train_epoch_end", "on_train_end", "on_train_exception", + "fitness", ) ) diff --git a/libreyolo/training/trainer.py b/libreyolo/training/trainer.py index f25de43a..8dd3b8b1 100644 --- a/libreyolo/training/trainer.py +++ b/libreyolo/training/trainer.py @@ -11,7 +11,9 @@ import time from abc import ABC, abstractmethod from collections.abc import Mapping +from numbers import Real from pathlib import Path +from types import MappingProxyType from typing import Any, Dict, List, Optional, Tuple, Type import torch @@ -236,6 +238,7 @@ def __init__( self.callbacks = TrainCallbackList(callbacks) for logger_callback in resolve_loggers(loggers): self.callbacks.append(logger_callback) + self._fitness_callback = self.callbacks.fitness # TrainingArtifactsCallback is family-gated (results.csv / summary.json # only for opted-in families). TrainingStatusCallback is universal: every # run gets a live status.json + train.log so agents and the `libreyolo @@ -2091,6 +2094,11 @@ def train(self) -> Dict: self.epoch_losses.append(epoch_loss) profile_truncated = bool(getattr(self, "_stop_training", False)) + if ( + not profile_truncated + and getattr(self, "_fitness_callback", None) is not None + ): + val_metrics = self._apply_fitness(val_metrics) is_best = ( False if profile_truncated @@ -2115,6 +2123,8 @@ def train(self) -> Dict: refreshed = self._validate_epoch( epoch, save_plots=save_final_plots ) + if getattr(self, "_fitness_callback", None) is not None: + refreshed = self._apply_fitness(refreshed) if refreshed is not None: val_metrics = refreshed is_best = self._update_best_state(epoch, val_metrics) @@ -2374,6 +2384,8 @@ def _best_metric_value(self, val_metrics: Optional[Dict[str, Any]]) -> float: return scalar if scalar is not None else 0.0 def _best_metric_name(self, val_metrics: Optional[Dict[str, Any]]) -> str: + if getattr(self, "_fitness_callback", None) is not None: + return "fitness/custom" if val_metrics: return str( val_metrics.get( @@ -2383,6 +2395,59 @@ def _best_metric_name(self, val_metrics: Optional[Dict[str, Any]]) -> str: ) return str(getattr(self, "best_metric_key", "metrics/mAP50-95")) + def _apply_fitness( + self, val_metrics: Optional[Dict[str, Any]], *, synchronize: bool = True + ) -> Optional[Dict[str, Any]]: + """Score once, then let all score consumers reuse the selected value. + + The main loop calls this on every rank, even without validation. + Share scorer failures before the next training collective so peer + ranks do not continue alone. Final averaging already runs on rank 0 + and therefore uses ``synchronize=False``. + """ + scorer = getattr(self, "_fitness_callback", None) + if scorer is None: + return val_metrics + error = None + if is_main_process() and val_metrics: + try: + metrics = MappingProxyType( + self._validation_metrics_for_event(val_metrics) + ) + score = scorer(metrics) + if isinstance(score, torch.Tensor): + if ( + score.ndim != 0 + or score.is_complex() + or score.dtype == torch.bool + ): + raise TypeError("fitness must return a finite real scalar") + score = score.detach().item() + if isinstance(score, bool) or not isinstance(score, Real): + raise TypeError("fitness must return a finite real scalar") + score = float(score) + if not math.isfinite(score): + raise ValueError("fitness must return a finite real scalar") + val_metrics = dict( + val_metrics, best_metric=score, best_metric_key="fitness/custom" + ) + except BaseException as exc: + error = exc + if synchronize and self.is_distributed: + import torch.distributed as _dist + + status = [ + f"{type(error).__name__}: {error}" if error is not None else None + ] + _dist.broadcast_object_list(status, src=0) + if status[0] is not None and error is None: + raise RuntimeError( + f"Training fitness callback failed on rank 0: {status[0]}" + ) + if error is not None: + raise error + return val_metrics + def _validation_metrics_for_event( self, val_metrics: Optional[Dict[str, Any]] ) -> Dict[str, float]: @@ -3629,6 +3694,10 @@ def _write_average_checkpoint(self) -> Optional[Path]: logger.info("average_best: validating averaged weights") average_metrics = self._validate_average_state(averaged) + if getattr(self, "_fitness_callback", None) is not None: + average_metrics = self._apply_fitness( + average_metrics, synchronize=False + ) self._average_validation_metrics = average_metrics if average_metrics is None: logger.warning( @@ -3656,6 +3725,9 @@ def _write_average_checkpoint(self) -> Optional[Path]: extra_checkpoint_meta["imgsz_h"] = cp_h extra_checkpoint_meta["imgsz_w"] = cp_w extra_checkpoint_meta.update(self._checkpoint_extra_metadata()) + checkpoint_task = extra_checkpoint_meta.pop( + "task", getattr(getattr(self, "wrapper_model", None), "task", "detect") + ) average_metric_key = ( average_metrics.get("best_metric_key") if average_metrics else None ) @@ -3666,15 +3738,23 @@ def _write_average_checkpoint(self) -> Optional[Path]: averaged, model_family=self.get_model_family(), size=self.config.size, - task=getattr(getattr(self, "wrapper_model", None), "task", "detect"), + task=checkpoint_task, nc=int(getattr(self, "num_classes", self.config.num_classes)), names=names, imgsz=int(checkpoint_imgsz), epoch=self.current_epoch, best_mAP50_95=self.best_mAP50_95, best_mAP50=self.best_mAP50, - best_metric_key=getattr(self, "best_metric_key", "metrics/mAP50-95"), + best_metric_key=( + "fitness/custom" + if getattr(self, "_fitness_callback", None) is not None + else getattr(self, "best_metric_key", "metrics/mAP50-95") + ), best_metric_value=self.best_mAP50_95, + fitness_source=( + "callback" + if getattr(self, "_fitness_callback", None) is not None else None + ), best_epoch=self.best_epoch, is_ema_weights=self.ema_model is not None, averaged_snapshot_count=averager.size, @@ -3849,6 +3929,8 @@ def _save_checkpoint( if val_metrics else getattr(self, "best_metric_key", "metrics/mAP50-95") ) + if getattr(self, "_fitness_callback", None) is not None: + best_metric_key = "fitness/custom" names = ( self.wrapper_model.names if self.wrapper_model is not None and hasattr(self.wrapper_model, "names") @@ -3897,6 +3979,10 @@ def _save_checkpoint( best_mAP50=self.best_mAP50, best_metric_key=best_metric_key, best_metric_value=self.best_mAP50_95, + fitness_source=( + "callback" + if getattr(self, "_fitness_callback", None) is not None else None + ), best_epoch=self.best_epoch, is_ema_weights=self.ema_model is not None, **extra_checkpoint_meta, @@ -3975,6 +4061,11 @@ def _checkpoint_extra_metadata(self) -> Dict[str, Any]: return input_metadata(self.wrapper_model) def resume(self, checkpoint_path: str): + if getattr(self, "_fitness_callback", None) is not None: + raise ValueError( + "Custom fitness does not support resume: callback code and state " + "are not saved. Load weights into a new run with resume=False." + ) if not Path(checkpoint_path).exists(): raise FileNotFoundError(f"Resume checkpoint not found: {checkpoint_path}") @@ -3984,6 +4075,11 @@ def resume(self, checkpoint_path: str): map_location=self.device, context="training resume checkpoint", ) + if checkpoint.get("fitness_source") == "callback": + raise ValueError( + "Cannot resume a custom-fitness checkpoint: its scores cannot be " + "compared with this run. Load weights into a new run with resume=False." + ) if isinstance(checkpoint.get("input_profile"), dict) or isinstance( getattr(getattr(self, "wrapper_model", None), "input_profile", None), dict ): diff --git a/libreyolo/utils/serialization.py b/libreyolo/utils/serialization.py index 928f1ae5..9843f664 100644 --- a/libreyolo/utils/serialization.py +++ b/libreyolo/utils/serialization.py @@ -320,6 +320,14 @@ def validate_checkpoint_metadata( if not isinstance(imgsz, int) or isinstance(imgsz, bool) or imgsz <= 0: errors.append("imgsz must be a positive int.") + if "fitness_source" in checkpoint: + if checkpoint["fitness_source"] != "callback": + errors.append("fitness_source must be 'callback' when present.") + if checkpoint.get("best_metric_key") != "fitness/custom": + errors.append( + "callback fitness requires best_metric_key='fitness/custom'." + ) + if isinstance(checkpoint, dict) and "input_profile" in checkpoint: from .event_histogram import validate_input_profile try: diff --git a/tests/unit/test_train_fitness.py b/tests/unit/test_train_fitness.py new file mode 100644 index 00000000..9d0828bf --- /dev/null +++ b/tests/unit/test_train_fitness.py @@ -0,0 +1,445 @@ +"""Custom fitness controls selection, patience and metadata without changing metrics.""" + +from __future__ import annotations + +import json +from pathlib import Path +from types import SimpleNamespace + +import numpy as np +import pytest +import torch +import torch.distributed as dist +import torch.multiprocessing as mp +from torch import nn + +from libreyolo.training.callbacks import TrainCallbackList +from libreyolo.training.trainer import BaseTrainer +from libreyolo.training.weight_averaging import MetricGatedAverager +from libreyolo.utils.serialization import ( + CheckpointMetadataError, + validate_checkpoint_metadata, + wrap_libreyolo_checkpoint, +) + +pytestmark = pytest.mark.unit + + +class LossFitness: + def __init__(self): + self.calls = [] + + def fitness(self, metrics): + assert not dist.is_initialized() or dist.get_rank() == 0 + with pytest.raises(TypeError): + metrics["metrics/loss"] = 0 + self.calls.append(dict(metrics)) + return -metrics["metrics/loss"] + + +class FitnessTrainer(BaseTrainer): + """Run the real orchestration and checkpoint writer with tiny epoch work.""" + + losses = (3.0, 1.0, 2.0, 1.0, 0.5) + skipped = () + + def get_model_family(self): + return "yolo9" + + def get_model_tag(self): + return "tiny" + + def create_transforms(self): + raise NotImplementedError + + def create_scheduler(self, iters_per_epoch): + raise NotImplementedError + + def get_loss_components(self, outputs): + return {} + + def setup(self): + self.save_dir = self._test_save_dir + self.save_dir.mkdir(parents=True, exist_ok=True) + self.optimizer = torch.optim.SGD(self.model.parameters(), lr=0.01) + self._is_setup = True + + def _train_epoch(self, epoch): + if self.is_distributed: + # A peer proceeding after rank-zero failure/stop would hang here. + dist.all_reduce(torch.ones(1)) + with torch.no_grad(): + self.model.weight.fill_(epoch + 1) + metrics = None if epoch in self.skipped else self._metrics(epoch) + if self.is_distributed and dist.get_rank() != 0: + metrics = None + return 1.0, metrics + + def _metrics(self, epoch): + return { + "mAP50": 0.1 * (epoch + 1), + "mAP50_95": 0.05 * (epoch + 1), + "best_metric": 0.05 * (epoch + 1), + "best_metric_key": "metrics/mAP50-95", + "metrics": { + "metrics/mAP50-95": 0.05 * (epoch + 1), + "metrics/loss": self.losses[epoch], + }, + } + + def _run_validation(self, epoch, *, save_plots=None): + return self._metrics(epoch) + + +def make_trainer(tmp_path, **kwargs): + trainer = FitnessTrainer( + nn.Linear(1, 1, bias=False), + data=None, + device="cpu", + ema=False, + epochs=5, + patience=2, + no_aug_epochs=0, + **kwargs, + ) + trainer._test_save_dir = tmp_path + return trainer + + +def load_checkpoint(path): + checkpoint = torch.load(path, weights_only=True, map_location="cpu") + validate_checkpoint_metadata(checkpoint, strict=True) + return checkpoint + + +def test_custom_fitness_selects_checkpoint_and_patience_before_observers(tmp_path): + scorer = LossFitness() + observed = [] + + def observer(event): + observed.append( + load_checkpoint(Path(event.save_dir) / "weights/best.pt")["epoch"] + ) + + trainer = make_trainer(tmp_path, callbacks=[scorer, observer]) + result = trainer.train() + best = load_checkpoint(result["best_checkpoint"]) + last = load_checkpoint(result["last_checkpoint"]) + + assert len(scorer.calls) == len(result["epoch_metrics"]) == 4 + assert observed == [0, 1, 1, 1] + assert result["best_epoch"] == 2 + assert [e["is_best"] for e in result["epoch_metrics"]] == [True, True, False, False] + assert [e["current_metric"] for e in result["epoch_metrics"]] == [-3, -1, -2, -1] + assert all( + e["current_metric_name"] == "fitness/custom" for e in result["epoch_metrics"] + ) + assert best["epoch"] == 1 and last["epoch"] == 3 + assert best["model"]["weight"].item() == 2 + assert best["best_metric_value"] == -1 + assert best["fitness_source"] == "callback" + assert "callbacks" not in best["config"] + assert result["val_metrics"] == scorer.calls + assert result["val_metrics"][-1]["metrics/mAP50-95"] == pytest.approx(0.2) + + +def test_no_scorer_keeps_default_selection_and_checkpoint_contract(tmp_path): + trainer = make_trainer(tmp_path) + result = trainer.train() + assert result["best_epoch"] == 5 + assert all(event["is_best"] for event in result["epoch_metrics"]) + best = load_checkpoint(result["best_checkpoint"]) + assert "fitness_source" not in best + assert best["best_metric_key"] == "metrics/mAP50-95" + assert best["best_metric_value"] == 0.25 + + +def test_skipped_validation_keeps_score_name_and_counts_patience_in_epochs(tmp_path): + scorer = LossFitness() + trainer = make_trainer(tmp_path, callbacks=scorer) + trainer.skipped = (2, 3) + result = trainer.train() + assert len(scorer.calls) == 2 + assert len(result["epoch_metrics"]) == 4 + assert result["epoch_metrics"][-1]["current_metric"] is None + assert result["epoch_metrics"][-1]["best_metric_name"] == "fitness/custom" + assert result["epoch_metrics"][-1]["val_metrics"] == {} + last = load_checkpoint(result["last_checkpoint"]) + assert last["best_metric_key"] == "fitness/custom" + assert last["best_metric_value"] == -1 + + +@pytest.mark.parametrize( + "score", + [ + None, + True, + "0.5", + [], + complex(1, 0), + float("nan"), + float("inf"), + -float("inf"), + torch.ones(1), + torch.tensor(True), + torch.tensor(1j), + ], +) +def test_invalid_score_does_not_update_best_or_write_checkpoint(tmp_path, score): + trainer = make_trainer( + tmp_path, callbacks=SimpleNamespace(fitness=lambda metrics: score) + ) + with pytest.raises((TypeError, ValueError), match="finite real scalar"): + trainer.train() + assert trainer.best_epoch == 0 + assert not (tmp_path / "weights/last.pt").exists() + assert json.loads((tmp_path / "status.json").read_text())["state"] == "failed" + + +@pytest.mark.parametrize("score", [0, -2.5, np.float32(0.5), torch.tensor(-3.0)]) +def test_real_scalars_are_accepted_without_changing_raw_metrics(tmp_path, score): + trainer = make_trainer( + tmp_path, callbacks=SimpleNamespace(fitness=lambda metrics: score) + ) + original = trainer._metrics(0) + scored = trainer._apply_fitness(original) + assert scored["best_metric"] == float(score) + assert original["best_metric"] == 0.05 + assert scored["metrics"] == original["metrics"] + + +def test_missing_metric_error_propagates_without_checkpoint(tmp_path): + callback = SimpleNamespace(fitness=lambda metrics: metrics["missing"]) + trainer = make_trainer(tmp_path, callbacks=callback) + with pytest.raises(KeyError, match="missing"): + trainer.train() + assert trainer.best_epoch == 0 + assert not (tmp_path / "weights/last.pt").exists() + + +def test_fitness_only_objects_and_multiple_scorer_rejection(): + callback = LossFitness() + callbacks = TrainCallbackList(callback) + callbacks.on_train_start(None) + callbacks.on_train_epoch_end(None) + assert callbacks.fitness == callback.fitness + with pytest.raises(ValueError, match="At most one"): + TrainCallbackList([callback, LossFitness()]) + with pytest.raises(ValueError, match="At most one"): + callbacks.append(LossFitness()) + assert len(callbacks) == 1 + with pytest.raises(TypeError, match="fitness must be callable"): + TrainCallbackList(SimpleNamespace(fitness=None)) + + +def test_separate_vlm_and_vla_trainers_reject_scorers(): + from libreyolo.models.vla.training.trainer import VLATrainer + from libreyolo.models.vlm.training.trainer import VLMDetectionTrainer + + for trainer, wrapper in ( + (VLATrainer, SimpleNamespace()), + (VLMDetectionTrainer, SimpleNamespace(FAMILY="qwen3vl")), + ): + with pytest.raises( + NotImplementedError, match="does not support custom fitness" + ): + trainer(wrapper, data="unused.yaml", callbacks=LossFitness()) + + +def test_custom_resume_rejected_before_loading_weights_and_default_resume_works( + tmp_path, +): + original = make_trainer(tmp_path / "original", callbacks=LossFitness()) + result = original.train() + resumed = make_trainer(tmp_path / "resumed") + before = resumed.model.weight.detach().clone() + with pytest.raises(ValueError, match="Cannot resume a custom-fitness"): + resumed.resume(result["last_checkpoint"]) + assert torch.equal(before, resumed.model.weight) + with pytest.raises(ValueError, match="Custom fitness does not support resume"): + original.resume(result["last_checkpoint"]) + + default = make_trainer(tmp_path / "default") + default_result = default.train() + with pytest.raises(ValueError, match="Custom fitness does not support resume"): + original.resume(default_result["last_checkpoint"]) + resumed.resume(default_result["last_checkpoint"]) + assert resumed.start_epoch == 5 + assert resumed.best_epoch == 5 + assert resumed.best_mAP50_95 == 0.25 + + +def test_average_pool_and_average_checkpoint_use_custom_score(tmp_path): + scorer = LossFitness() + trainer = make_trainer(tmp_path, callbacks=scorer) + trainer._weight_averager = MetricGatedAverager(2) + result = trainer.train() + average = load_checkpoint(result["average_checkpoint"]) + assert sorted(trainer._weight_averager.metrics()) == [-1, -1] + assert average["model"]["weight"].item() == 3 # epochs 2 and 4 + assert ( + average["best_metric_key"] == average["average_metric_key"] == "fitness/custom" + ) + assert average["fitness_source"] == "callback" + assert average["average_metric_value"] == -1 + assert len(scorer.calls) == 5 # four epochs plus averaged-weight validation + + +@pytest.mark.parametrize("custom", [False, True]) +@pytest.mark.parametrize("with_wrapper", [False, True]) +def test_fomo_checkpoint_metadata_preserves_selected_metric( + tmp_path, monkeypatch, custom, with_wrapper +): + from libreyolo import LibreFOMO + from libreyolo.models.fomo.trainer import FOMOTrainer + + wrapper = LibreFOMO(None, size="s", nb_classes=1, device="cpu") + trainer = FOMOTrainer( + wrapper.model, + wrapper_model=wrapper if with_wrapper else None, + size="s", + num_classes=1, + device="cpu", + epochs=1, + ema=False, + no_aug_epochs=0, + callbacks=LossFitness() if custom else None, + ) + metrics = { + "best_metric": 0.7, + "best_metric_key": "metrics/grid_F1", + "metrics": {"metrics/grid_F1": 0.7, "metrics/loss": 2.0}, + } + + def setup(): + trainer.save_dir = tmp_path + trainer.optimizer = torch.optim.SGD(trainer.model.parameters(), lr=0.01) + trainer._is_setup = True + + monkeypatch.setattr(trainer, "setup", setup) + monkeypatch.setattr(trainer, "_train_epoch", lambda epoch: (1.0, metrics)) + monkeypatch.setattr(trainer, "_validate_average_state", lambda state: metrics) + trainer._weight_averager = MetricGatedAverager(2) + results = trainer.train() + + expected_key = "fitness/custom" if custom else "metrics/grid_F1" + for name in ("best_checkpoint", "last_checkpoint", "average_checkpoint"): + checkpoint = load_checkpoint(results[name]) + assert checkpoint["task"] == "point" + assert checkpoint["best_metric_key"] == expected_key + assert checkpoint["best_metric_value"] == (-2.0 if custom else 0.7) + assert checkpoint.get("fitness_source") == ("callback" if custom else None) + average = load_checkpoint(results["average_checkpoint"]) + assert average["average_metric_key"] == expected_key + reloaded = LibreFOMO(results["best_checkpoint"], size="s", device="cpu") + assert reloaded.task == "point" + + +def test_precise_bn_refresh_can_reverse_stop_decision(tmp_path, monkeypatch): + scorer = LossFitness() + trainer = make_trainer(tmp_path, callbacks=scorer) + monkeypatch.setattr( + trainer, "_maybe_precise_bn", lambda epoch, force=False: epoch == 3 + ) + refreshed = trainer._metrics(3) + refreshed["metrics"]["metrics/loss"] = 0.1 + monkeypatch.setattr(trainer, "_validate_epoch", lambda *args, **kwargs: refreshed) + result = trainer.train() + assert len(result["epoch_metrics"]) == 5 + assert result["best_epoch"] == 4 + assert result["epoch_metrics"][3]["current_metric"] == -0.1 + assert len(scorer.calls) == 6 + + +@pytest.mark.parametrize( + "metadata", [{"fitness_source": "unknown"}, {"fitness_source": "callback"}] +) +def test_checkpoint_helper_rejects_invalid_fitness_metadata(metadata): + with pytest.raises(CheckpointMetadataError, match="fitness"): + wrap_libreyolo_checkpoint( + {}, + model_family="yolo9", + size="t", + task="detect", + nc=1, + imgsz=32, + **metadata, + ) + + +def _distributed_worker(rank, init_file, root, failure): + torch.set_num_threads(1) + dist.init_process_group( + "gloo", init_method=f"file://{init_file}", rank=rank, world_size=2 + ) + try: + scorer = LossFitness() + if failure == "invalid": + scorer = SimpleNamespace(fitness=lambda metrics: float("nan")) + elif failure == "exit": + + def exit_fitness(metrics): + raise SystemExit("fitness interrupted") + + scorer = SimpleNamespace(fitness=exit_fitness) + elif failure == "average": + class AverageFailure(LossFitness): + def fitness(self, metrics): + if len(self.calls) == 4: + raise ValueError("averaged fitness failed") + return super().fitness(metrics) + + scorer = AverageFailure() + trainer = make_trainer(Path(root) / "run", callbacks=scorer) + if failure == "average": + trainer._weight_averager = MetricGatedAverager(2) + try: + result = trainer.train() + outcome = { + "epochs": len(result["epoch_metrics"]), + "calls": len(scorer.calls), + } + except (ValueError, RuntimeError, SystemExit) as exc: + outcome = {"error": str(exc), "type": type(exc).__name__} + Path(root, f"rank{rank}.json").write_text(json.dumps(outcome)) + finally: + dist.destroy_process_group() + + +@pytest.mark.distributed +@pytest.mark.timeout(90) +@pytest.mark.skipif( + not dist.is_available() or not dist.is_gloo_available(), reason="requires Gloo" +) +@pytest.mark.parametrize("failure", [None, "invalid", "exit", "average"]) +def test_distributed_fitness_rank_zero_stop_and_failure(tmp_path, failure): + mp.spawn( + _distributed_worker, + args=(str(tmp_path / "init"), str(tmp_path), failure), + nprocs=2, + join=True, + ) + ranks = [ + json.loads((tmp_path / f"rank{rank}.json").read_text()) for rank in range(2) + ] + if failure == "average": + # Rank zero raises after releasing the averaging barrier. Peers have + # no further training collective and can finish, allowing the launcher + # to report the rank-zero error instead of hanging. + assert ranks[0] == {"type": "ValueError", "error": "averaged fitness failed"} + assert ranks[1] == {"epochs": 4, "calls": 0} + assert not (tmp_path / "run/weights/average.pt").exists() + elif failure: + assert [r["type"] for r in ranks] == [ + "ValueError" if failure == "invalid" else "SystemExit", + "RuntimeError", + ] + message = ( + "finite real scalar" if failure == "invalid" else "fitness interrupted" + ) + assert all(message in r["error"] for r in ranks) + assert not (tmp_path / "run/weights/last.pt").exists() + else: + assert [r["epochs"] for r in ranks] == [4, 4] + assert [r["calls"] for r in ranks] == [4, 0] + assert load_checkpoint(tmp_path / "run/weights/best.pt")["best_epoch"] == 2