diff --git a/CHANGELOG.md b/CHANGELOG.md index 170ba83..90246d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,35 +8,43 @@ and this project adheres to ## [4.3.0] - 2026-09-02 -### Changed +### Added + +- `deeplc.calibration` is now a package. `deeplc.calibration.simple` holds the naive, + single-series calibrations unchanged (`Calibration` ABC, `IdentityCalibration`, + `PiecewiseLinearCalibration`, `SplineTransformerCalibration`). New + `deeplc.calibration.multihead` holds `MultiHeadCalibration`, the ABC for a calibration that + takes the whole `(n, n_heads)` prediction matrix `predict(..., return_matrix=True)` returns and + selects its own head(s): `MultiHeadPiecewiseLinearCalibration` and `MultiHeadSplineCalibration` + rank heads by Pearson correlation and delegate to the matching naive class on the winner, and + `MultiHeadRidgeCalibration` combines several best-correlating heads with a ridge fit. All three + are re-exported from `deeplc.calibration`, unchanged import path for existing names. -- Calibration of a multitask model now combines several LC-setup heads instead of keeping only - the best-correlating one. `calibrate` and `predict_and_calibrate` default to the new - `MultiHeadRidgeCalibration`: heads are ranked by Pearson correlation to the reference as - before, the 80 best are each calibrated with `SplineTransformerCalibration`, and a ridge - regression maps the calibrated estimates onto the observed retention times. + On the eight PRIDE setups that no DeepLC model was trained on, `MultiHeadRidgeCalibration` + lowered the held-out error on all eight, by a median of 13 % relative to the observed gradient. - On the eight PRIDE setups that no DeepLC model was trained on this lowered the held-out error - on all eight, by a median of 13 % relative to the observed gradient (0.01248 to 0.01090 - MAE/span). Fitting is faster than the previous path because the head ranking is vectorised - (median 1.0 s against 2.3 s), and prediction is unchanged since the full head matrix is - computed either way. +- `deeplc.calibration.upgrade_calibration`, wrapping an unfitted naive `Calibration` in its + `MultiHead*Calibration` counterpart. `calibrate()` and `predict_and_calibrate()` call it on + whatever `calibration` they are given, so passing a naive instance (e.g. + `SplineTransformerCalibration()`) still works, calibrated on the best-correlating head instead + of head 0. A fitted naive instance is rejected: it carries no record of which head it was fit + on, so fit a `MultiHead*Calibration` instead in that case. - Single-task models keep the previous default (`SplineTransformerCalibration`), and passing - a calibration instance restores the old behaviour on any model: +### Changed - ```python - from deeplc import predict_and_calibrate - from deeplc.calibration import SplineTransformerCalibration +- `calibrate()` and `predict_and_calibrate()` default to `MultiHeadRidgeCalibration()` for every + model, single-task included (it reduces to a spline plus a linear rescaling on one column). Pass + `MultiHeadSplineCalibration()` or `MultiHeadPiecewiseLinearCalibration()` for a lighter + calibration. `Calibration.selected_model_head`/`uses_all_heads` are gone: head selection now + lives entirely in the `MultiHeadCalibration` implementations. - rt = predict_and_calibrate(psms, psm_list_reference=reference, - calibration=SplineTransformerCalibration()) - ``` + **Breaking**: an already-fitted naive `Calibration` passed directly to + `predict_and_calibrate()` is now rejected; fit a `MultiHead*Calibration` instead. -### Added +### Fixed -- `Calibration.uses_all_heads`, telling `calibrate` and `predict_and_calibrate` to hand a - calibration the whole `(n, n_heads)` prediction matrix rather than a single column. +- `IdentityCalibration()` could not be instantiated: it never overrode the abstract + `Calibration.__init__`. ## [4.2.0] - 2026-08-28 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index a95f286..679b37d 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -9,6 +9,48 @@ If you have an idea for a feature, use case to add or an approach for a bugfix, it is best to communicate with the community by creating an issue in [GitHub issues](https://github.com/compomics/DeepLC/issues). +## Codebase overview + +DeepLC predicts peptide retention times using a 1D-CNN model trained on atomic composition +features. + +- **Feature extraction** (`deeplc/_features.py`) — `encode_peptidoform()` converts a ProForma + peptidoform string into per-position atomic composition arrays (C, H, N, O, S, P) plus a 20-AA + one-hot encoding. Padding to 60 residues. Uses `psm_utils.Peptidoform` as the canonical peptide + representation. +- **Dataset** (`deeplc/data.py`) — `DeepLCDataset` wraps lists of `Peptidoform` objects into a + PyTorch `Dataset`. Features are encoded lazily in `__getitem__`. `split_datasets()` handles + train/validation splits. +- **Model architecture** (`deeplc/_architecture.py`) — `DeepLCModel`: Conv1D branches (atomic, + summed-atomic, global, one-hot) feed a shared dense trunk into `BatchedHeads` returning + `[batch, n_heads]`. An optional fine-tuning adapter (`self.adapter`, attached via + `add_adapter()`) maps head output to `[batch, 1]`. +- **Training/inference** (`deeplc/_model_ops.py`) — `load_model()`, `train()`, `predict()`, + `evaluate()`. Checkpoints are plain state dicts loaded with `weights_only=True`. +- **Calibration** (`deeplc/calibration/` package) — `predict(..., return_matrix=True)` always + returns a `(n, n_heads)` matrix, `n_heads=1` for a single-setup model. + `deeplc/calibration/simple.py` holds naive, single-series calibrations (`Calibration` ABC, + `IdentityCalibration`, `PiecewiseLinearCalibration`, `SplineTransformerCalibration`) that map + one column onto observed RT space and know nothing about heads. + `deeplc/calibration/multihead.py` holds `MultiHeadCalibration` ABC and the classes that pick + their own head(s) from the matrix: `MultiHeadPiecewiseLinearCalibration` and + `MultiHeadSplineCalibration` (rank heads by correlation, delegate to the matching `simple.py` + class) and `MultiHeadRidgeCalibration` (combines the best-correlating heads with a ridge fit). + `core.calibrate()`/`predict_and_calibrate()` only accept `MultiHeadCalibration` instances and + always hand over the full matrix; the default is `MultiHeadRidgeCalibration`. +- **Reference selection** (`deeplc/_reference_selection.py`) — selects high-confidence PSMs from + input for auto-calibration. +- **Core API** (`deeplc/core.py`) — top-level functions: `predict()`, `calibrate()`, + `predict_and_calibrate()`, `finetune_and_predict()`. +- **CLI** (`deeplc/__main__.py`) — two subcommands, `predict` and `gui`. Reads PSM files via + `psm_utils.io.read_file()`. +- **GUI** (`deeplc/gui.py`) — NiceGUI-based web UI, launchable as browser app or native desktop + window via `pywebview`. + +Conventions: peptide sequences use ProForma notation throughout (via `psm_utils.Peptidoform`); +PSM collections are `psm_utils.PSMList`; line length is 99 characters (ruff); Python >= 3.11 with +`from __future__ import annotations`. + ## How to contribute - Fork [DeepLC](https://github.com/compomics/DeepLC) on GitHub to diff --git a/deeplc/calibration/__init__.py b/deeplc/calibration/__init__.py new file mode 100644 index 0000000..65375bd --- /dev/null +++ b/deeplc/calibration/__init__.py @@ -0,0 +1,29 @@ +"""Calibration utilities.""" + +from deeplc.calibration.multihead import ( + MultiHeadCalibration, + MultiHeadPiecewiseLinearCalibration, + MultiHeadRidgeCalibration, + MultiHeadSplineCalibration, + upgrade_calibration, +) +from deeplc.calibration.simple import ( + Calibration, + IdentityCalibration, + PiecewiseLinearCalibration, + SplineTransformerCalibration, +) +from deeplc.exceptions import CalibrationError + +__all__ = [ + "Calibration", + "CalibrationError", + "IdentityCalibration", + "MultiHeadCalibration", + "MultiHeadPiecewiseLinearCalibration", + "MultiHeadRidgeCalibration", + "MultiHeadSplineCalibration", + "PiecewiseLinearCalibration", + "SplineTransformerCalibration", + "upgrade_calibration", +] diff --git a/deeplc/calibration/multihead.py b/deeplc/calibration/multihead.py new file mode 100644 index 0000000..6eeda42 --- /dev/null +++ b/deeplc/calibration/multihead.py @@ -0,0 +1,365 @@ +""" +Multitask-aware calibration utilities. + +Every class here maps the full ``(n, n_heads)`` prediction matrix of a multitask model onto a +single series of observed values, and is responsible for picking which head(s) to use itself: +``fit(target: (n,), source: (n, n_heads))`` / ``transform(source: (n, n_heads)) -> (n,)``. A +single-setup model still produces a matrix, just with ``n_heads=1``, so these classes are the +default way to calibrate any DeepLC model. See :mod:`deeplc.calibration.simple` for the naive, +single-series calibrations these delegate to. +""" + +from __future__ import annotations + +import logging +from abc import ABC, abstractmethod +from typing import cast + +import numpy as np +from sklearn.linear_model import RidgeCV # type: ignore[import] + +from deeplc.calibration.simple import ( + Calibration, + PiecewiseLinearCalibration, + SplineTransformerCalibration, +) +from deeplc.exceptions import CalibrationError + +LOGGER = logging.getLogger(__name__) + + +class MultiHeadCalibration(ABC): + """Abstract base class for a calibration that selects its own head(s) from a matrix.""" + + #: The best-correlating head, set by ``fit``. Meaningful for every implementer, including + #: :class:`MultiHeadRidgeCalibration`, which reports the single best head even though it + #: fits on several. + selected_model_head: int + + @abstractmethod + def __init__(self, *args, **kwargs): + """Initialize the calibration model.""" + super().__init__() + + @property + @abstractmethod + def is_fitted(self) -> bool: + """Indicates whether the calibration model has been fitted.""" + ... + + @abstractmethod + def fit(self, target: np.ndarray, source: np.ndarray) -> None: + """Fit the calibration from the source matrix to target.""" + ... + + @abstractmethod + def transform(self, source: np.ndarray) -> np.ndarray: + """Transform the source matrix into the calibrated target space.""" + ... + + +class _SingleHeadCalibration(MultiHeadCalibration): + """ + Shared logic for wrapping one naive, single-series :class:`Calibration`. + + Ranks the heads of the matrix by Pearson correlation to the target, fits ``self._inner`` on + the winner, and transforms by picking that same column. Subclasses only need to construct + ``self._inner``. + """ + + _inner: Calibration + + @property + def is_fitted(self) -> bool: + """True if the wrapped calibration has been fitted.""" + return self._inner.is_fitted + + def fit(self, target: np.ndarray, source: np.ndarray) -> None: + """ + Select the best-correlating head and fit the wrapped calibration on it. + + Parameters + ---------- + target + Observed retention times of the reference, shape ``(n,)``. + source + Reference predictions for every head, shape ``(n, n_heads_total)``. A 1-D array is + accepted and treated as a single head, so a single-task model still works. + + """ + source = np.asarray(source, dtype=np.float64) + if source.ndim == 1: + source = source[:, None] + target = np.asarray(target, dtype=np.float64).ravel() + if source.shape[0] != target.shape[0]: + raise CalibrationError( + f"source has {source.shape[0]} rows and target {target.shape[0]}" + ) + finite = np.isfinite(target) & np.isfinite(source).all(axis=1) + if int(finite.sum()) < 3: + raise CalibrationError("Fewer than three reference points with finite values.") + source, target = source[finite], target[finite] + + order = _rank_heads_by_correlation(source, target) + self.selected_model_head = int(order[0]) + self._inner.fit( + target=target.astype(np.float32), + source=source[:, self.selected_model_head].astype(np.float32), + ) + + def transform(self, source: np.ndarray) -> np.ndarray: + """ + Select the fitted head from the matrix and transform it. + + Parameters + ---------- + source + Predictions for every head, shape ``(n, n_heads_total)``, as returned by + ``predict(..., return_matrix=True)``. + + """ + if not self.is_fitted: + raise CalibrationError("The model has not been fitted yet. Call fit() first.") + source = np.asarray(source, dtype=np.float64) + if source.ndim == 1: + source = source[:, None] + head = self.selected_model_head + if source.shape[1] <= head: + raise CalibrationError( + f"source has {source.shape[1]} heads, but the calibration was fitted on a model " + f"with at least {head + 1}." + ) + if source.shape[0] == 0: + return np.array([]) + return np.asarray( + self._inner.transform(source[:, head].astype(np.float32)), dtype=np.float64 + ) + + +class MultiHeadPiecewiseLinearCalibration(_SingleHeadCalibration): + """Piece-wise linear calibration on whichever head correlates best with the reference.""" + + def __init__( + self, + number_of_splits: int = 10, + extrapolate: bool = True, + use_median: bool = False, + min_samples_per_segment: int = 20, + ) -> None: + """ + Piece-wise linear calibration on whichever head correlates best with the reference. + + Parameters + ---------- + number_of_splits : int + Number of segments to split the source value range into. + More segments allow more flexibility but may lead to overfitting. + extrapolate : bool + If True, allows extrapolation outside the fitted source value range. + If False, clips input values to the fitted range. + use_median : bool + If True, uses the median of each segment to define anchors. If False, uses the mean. + min_samples_per_segment : int + Minimum number of samples required for a segment to contribute an anchor. + Segments with fewer samples are skipped, which helps avoid unstable anchors in + sparse regions when using many splits. + + """ + super().__init__() + self._inner = PiecewiseLinearCalibration( + number_of_splits=number_of_splits, + extrapolate=extrapolate, + use_median=use_median, + min_samples_per_segment=min_samples_per_segment, + ) + + +class MultiHeadSplineCalibration(_SingleHeadCalibration): + """Spline-based calibration on whichever head correlates best with the reference.""" + + def __init__(self) -> None: + """Initialize MultiHeadSplineCalibration.""" + super().__init__() + self._inner = SplineTransformerCalibration() + + +class MultiHeadRidgeCalibration(MultiHeadCalibration): + """ + Calibrate a multitask model against several of its LC-setup heads at once. + + Heads are ranked by Pearson correlation to the reference, the ``n_heads`` best are each + calibrated with :class:`~deeplc.calibration.simple.SplineTransformerCalibration`, and a ridge + regression maps the calibrated estimates onto the observed retention times. Never fits more + head weights than half the reference size. For a single-task model (one head) this reduces to + a spline followed by a linear rescaling. + + Parameters + ---------- + n_heads + How many of the best-correlating heads to combine. + alphas + Ridge strengths offered to the internal cross-validation. + + """ + + def __init__(self, n_heads: int = 80, alphas: np.ndarray | None = None) -> None: + """Initialize MultiHeadRidgeCalibration.""" + super().__init__() + if n_heads < 1: + raise ValueError(f"n_heads must be at least 1, got {n_heads}") + self.n_heads = n_heads + self.alphas = np.logspace(-3, 6, 19) if alphas is None else np.asarray(alphas) + self._head_idx: np.ndarray | None = None + self._head_calibrations: list[SplineTransformerCalibration] = [] + self._ridge = None + + @property + def is_fitted(self) -> bool: + """True once the heads are selected, calibrated and weighted.""" + return self._head_idx is not None and self._ridge is not None + + def fit(self, target: np.ndarray, source: np.ndarray) -> None: + """ + Select, calibrate and weight the heads. + + Parameters + ---------- + target + Observed retention times of the reference, shape ``(n,)``. + source + Reference predictions for every head, shape ``(n, n_heads_total)``. A 1-D array is + accepted and treated as a single head, so a single-task model still works. + + """ + source = np.asarray(source, dtype=np.float64) + if source.ndim == 1: + source = source[:, None] + target = np.asarray(target, dtype=np.float64).ravel() + if source.shape[0] != target.shape[0]: + raise CalibrationError( + f"source has {source.shape[0]} rows and target {target.shape[0]}" + ) + finite = np.isfinite(target) & np.isfinite(source).all(axis=1) + if int(finite.sum()) < 3: + raise CalibrationError("Fewer than three reference points with finite values.") + source, target = source[finite], target[finite] + + order = _rank_heads_by_correlation(source, target) + # never fit more weights than half the reference: a 230-peptide reference cannot support + # eighty of them, and the ridge would be extrapolating its own regularisation + n_heads = int(min(self.n_heads, source.shape[1], max(1, len(target) // 2))) + self._head_idx = order[:n_heads] + self.selected_model_head = int(order[0]) + + calibrated = np.empty((len(target), n_heads), dtype=np.float64) + self._head_calibrations = [] + for position, head in enumerate(self._head_idx): + head_calibration = SplineTransformerCalibration() + column = source[:, head].astype(np.float32) + head_calibration.fit(target=target.astype(np.float32), source=column) + calibrated[:, position] = np.asarray( + head_calibration.transform(column), dtype=np.float64 + ) + self._head_calibrations.append(head_calibration) + + n_splits = int(min(5, max(2, len(target) // 20))) + self._ridge = RidgeCV(alphas=self.alphas, cv=n_splits).fit(calibrated, target) + LOGGER.info( + "Calibrated on %d of %d heads with ridge strength %.4g; head %d correlates best.", + n_heads, + source.shape[1], + float(getattr(self._ridge, "alpha_", float("nan"))), + self.selected_model_head, + ) + + def transform(self, source: np.ndarray) -> np.ndarray: + """ + Calibrate predictions of the model this calibration was fitted with. + + Parameters + ---------- + source + Predictions for every head, shape ``(n, n_heads_total)``, as returned by + ``predict(..., return_matrix=True)``. + + """ + if not self.is_fitted: + raise CalibrationError("The model has not been fitted yet. Call fit() first.") + source = np.asarray(source, dtype=np.float64) + if source.ndim == 1: + source = source[:, None] + head_idx = cast(np.ndarray, self._head_idx) + if source.shape[1] <= int(head_idx.max()): + raise CalibrationError( + f"source has {source.shape[1]} heads, but the calibration was fitted on a model " + f"with at least {int(head_idx.max()) + 1}." + ) + if source.shape[0] == 0: + return np.array([]) + calibrated = np.column_stack( + [ + np.asarray(cal.transform(source[:, head].astype(np.float32)), dtype=np.float64) + for cal, head in zip(self._head_calibrations, head_idx, strict=True) + ] + ) + return np.asarray(self._ridge.predict(calibrated), dtype=np.float64) + + +def upgrade_calibration(calibration: Calibration | MultiHeadCalibration) -> MultiHeadCalibration: + """ + Wrap a naive, single-series calibration in its matching ``MultiHeadCalibration``. + + A ``MultiHeadCalibration`` instance is returned unchanged. A naive + :class:`~deeplc.calibration.simple.Calibration` must not be fitted yet: a fitted instance + carries no record of which head it was fit on, so there is nothing to wrap it around. + + Parameters + ---------- + calibration + The calibration to wrap, naive or already multi-head. + + Returns + ------- + MultiHeadCalibration + ``calibration`` itself if it already is one, otherwise a new wrapper around it. + + """ + if isinstance(calibration, MultiHeadCalibration): + return calibration + if not isinstance(calibration, Calibration): + raise ValueError( + f"Expected calibration to be of type `Calibration` or `MultiHeadCalibration`, got " + f"{type(calibration)}" + ) + if not isinstance(calibration, (PiecewiseLinearCalibration, SplineTransformerCalibration)): + raise ValueError( + f"No MultiHeadCalibration counterpart is known for {type(calibration).__name__}." + ) + if calibration.is_fitted: + raise CalibrationError( + "A fitted, naive Calibration cannot be upgraded to a MultiHeadCalibration: it " + "carries no record of which head it was fit on. Fit a MultiHead*Calibration instead." + ) + if isinstance(calibration, PiecewiseLinearCalibration): + return MultiHeadPiecewiseLinearCalibration( + number_of_splits=calibration.number_of_splits, + extrapolate=calibration.extrapolate, + use_median=calibration.use_median, + min_samples_per_segment=calibration.min_samples_per_segment, + ) + return MultiHeadSplineCalibration() + + +def _rank_heads_by_correlation(source: np.ndarray, target: np.ndarray) -> np.ndarray: + """ + Head indices by decreasing Pearson correlation with the target, in one pass. + + Vectorised because a fused-trunk multitask model can have thousands of heads to rank at once. + """ + centred = source - source.mean(axis=0) + target_centred = target - target.mean() + with np.errstate(invalid="ignore", divide="ignore"): + denominator = np.sqrt((centred**2).sum(axis=0) * (target_centred**2).sum()) + correlation = (centred * target_centred[:, None]).sum(axis=0) / denominator + correlation = np.where(np.isfinite(correlation), correlation, -np.inf) + return np.argsort(-correlation) diff --git a/deeplc/calibration.py b/deeplc/calibration/simple.py similarity index 67% rename from deeplc/calibration.py rename to deeplc/calibration/simple.py index d61867a..f49257f 100644 --- a/deeplc/calibration.py +++ b/deeplc/calibration/simple.py @@ -1,4 +1,11 @@ -"""Calibration utilities.""" +""" +Naive calibration utilities. + +Every class here maps a single series of raw predictions onto a single series of observed +values: ``fit(target: (n,), source: (n,))`` / ``transform(source: (n,)) -> (n,)``. None of them +know about multitask models or LC-setup heads; see :mod:`deeplc.calibration.multihead` for the +classes that select a head from a ``(n, n_heads)`` prediction matrix and delegate to one of these. +""" from __future__ import annotations @@ -7,7 +14,7 @@ from typing import cast import numpy as np -from sklearn.linear_model import LinearRegression, RidgeCV # type: ignore[import] +from sklearn.linear_model import LinearRegression # type: ignore[import] from sklearn.pipeline import Pipeline, make_pipeline # type: ignore[import] from sklearn.preprocessing import SplineTransformer # type: ignore[import] @@ -17,13 +24,7 @@ class Calibration(ABC): - """Abstract base class for calibration.""" - - selected_model_head: int | None = None - - #: Whether ``fit`` and ``transform`` take the full ``(n, n_heads)`` prediction matrix - #: instead of a single series. - uses_all_heads: bool = False + """Abstract base class for a single-series calibration.""" @abstractmethod def __init__(self, *args, **kwargs): @@ -50,6 +51,10 @@ def transform(self, source: np.ndarray) -> np.ndarray: class IdentityCalibration(Calibration): """No calibration; returns inputs unchanged.""" + def __init__(self) -> None: + """Initialize IdentityCalibration.""" + super().__init__() + @property def is_fitted(self) -> bool: """Always fitted; identity calibration requires no fitting.""" @@ -329,146 +334,6 @@ def transform(self, source: np.ndarray) -> np.ndarray: return np.array(cal_preds) -class MultiHeadRidgeCalibration(Calibration): - """ - Calibrate a multitask model against several of its LC-setup heads at once. - - Heads are ranked by Pearson correlation to the reference, the ``n_heads`` best are each - calibrated with :class:`SplineTransformerCalibration`, and a ridge regression maps the - calibrated estimates onto the observed retention times. Never fits more head weights than - half the reference size. For a single-task model (one head) this reduces to a spline - followed by a linear rescaling. - - Parameters - ---------- - n_heads - How many of the best-correlating heads to combine. - alphas - Ridge strengths offered to the internal cross-validation. - - """ - - uses_all_heads = True - - def __init__(self, n_heads: int = 80, alphas: np.ndarray | None = None) -> None: - """Initialize MultiHeadRidgeCalibration.""" - super().__init__() - if n_heads < 1: - raise ValueError(f"n_heads must be at least 1, got {n_heads}") - self.n_heads = n_heads - self.alphas = np.logspace(-3, 6, 19) if alphas is None else np.asarray(alphas) - self._head_idx: np.ndarray | None = None - self._head_calibrations: list[SplineTransformerCalibration] = [] - self._ridge = None - - @property - def is_fitted(self) -> bool: - """True once the heads are selected, calibrated and weighted.""" - return self._head_idx is not None and self._ridge is not None - - def fit(self, target: np.ndarray, source: np.ndarray) -> None: - """ - Select, calibrate and weight the heads. - - Parameters - ---------- - target - Observed retention times of the reference, shape ``(n,)``. - source - Reference predictions for every head, shape ``(n, n_heads_total)``. A 1-D array is - accepted and treated as a single head, so a single-task model still works. - - """ - source = np.asarray(source, dtype=np.float64) - if source.ndim == 1: - source = source[:, None] - target = np.asarray(target, dtype=np.float64).ravel() - if source.shape[0] != target.shape[0]: - raise CalibrationError( - f"source has {source.shape[0]} rows and target {target.shape[0]}" - ) - finite = np.isfinite(target) & np.isfinite(source).all(axis=1) - if int(finite.sum()) < 3: - raise CalibrationError("Fewer than three reference points with finite values.") - source, target = source[finite], target[finite] - - order = _rank_heads_by_correlation(source, target) - # never fit more weights than half the reference: a 230-peptide reference cannot support - # eighty of them, and the ridge would be extrapolating its own regularisation - n_heads = int(min(self.n_heads, source.shape[1], max(1, len(target) // 2))) - self._head_idx = order[:n_heads] - self.selected_model_head = int(order[0]) - - calibrated = np.empty((len(target), n_heads), dtype=np.float64) - self._head_calibrations = [] - for position, head in enumerate(self._head_idx): - head_calibration = SplineTransformerCalibration() - column = source[:, head].astype(np.float32) - head_calibration.fit(target=target.astype(np.float32), source=column) - calibrated[:, position] = np.asarray( - head_calibration.transform(column), dtype=np.float64 - ) - self._head_calibrations.append(head_calibration) - - n_splits = int(min(5, max(2, len(target) // 20))) - self._ridge = RidgeCV(alphas=self.alphas, cv=n_splits).fit(calibrated, target) - LOGGER.info( - "Calibrated on %d of %d heads with ridge strength %.4g; head %d correlates best.", - n_heads, - source.shape[1], - float(getattr(self._ridge, "alpha_", float("nan"))), - self.selected_model_head, - ) - - def transform(self, source: np.ndarray) -> np.ndarray: - """ - Calibrate predictions of the model this calibration was fitted with. - - Parameters - ---------- - source - Predictions for every head, shape ``(n, n_heads_total)``, as returned by - ``predict(..., return_matrix=True)``. - - """ - if not self.is_fitted: - raise CalibrationError("The model has not been fitted yet. Call fit() first.") - source = np.asarray(source, dtype=np.float64) - if source.ndim == 1: - source = source[:, None] - head_idx = cast(np.ndarray, self._head_idx) - if source.shape[1] <= int(head_idx.max()): - raise CalibrationError( - f"source has {source.shape[1]} heads, but the calibration was fitted on a model " - f"with at least {int(head_idx.max()) + 1}." - ) - if source.shape[0] == 0: - return np.array([]) - calibrated = np.column_stack( - [ - np.asarray(cal.transform(source[:, head].astype(np.float32)), dtype=np.float64) - for cal, head in zip(self._head_calibrations, head_idx, strict=True) - ] - ) - return np.asarray(self._ridge.predict(calibrated), dtype=np.float64) - - -def _rank_heads_by_correlation(source: np.ndarray, target: np.ndarray) -> np.ndarray: - """ - Head indices by decreasing Pearson correlation with the target, in one pass. - - The same criterion as :func:`deeplc.core._best_correlating_head`, which takes the first - element of this order, but vectorised because thousands of heads are ranked at once. - """ - centred = source - source.mean(axis=0) - target_centred = target - target.mean() - with np.errstate(invalid="ignore", divide="ignore"): - denominator = np.sqrt((centred**2).sum(axis=0) * (target_centred**2).sum()) - correlation = (centred * target_centred[:, None]).sum(axis=0) / denominator - correlation = np.where(np.isfinite(correlation), correlation, -np.inf) - return np.argsort(-correlation) - - def _prepare_series( target: np.ndarray, source: np.ndarray, diff --git a/deeplc/core.py b/deeplc/core.py index 105855f..f3e1714 100644 --- a/deeplc/core.py +++ b/deeplc/core.py @@ -15,8 +15,9 @@ from deeplc._reference_selection import deduplicate_psms, select_reference_psms from deeplc.calibration import ( Calibration, + MultiHeadCalibration, MultiHeadRidgeCalibration, - SplineTransformerCalibration, + upgrade_calibration, ) from deeplc.data import DeepLCDataset, split_datasets @@ -142,11 +143,11 @@ def _default_task_idx(model: torch.nn.Module) -> int: def calibrate( psm_list_reference: PSMList, model: torch.nn.Module | PathLike | str | None = None, - calibration: Calibration | None = None, + calibration: Calibration | MultiHeadCalibration | None = None, predict_kwargs: dict | None = None, -) -> Calibration: +) -> MultiHeadCalibration: """ - Return a `Calibration` instance fitted to the reference dataset. + Return a `MultiHeadCalibration` instance fitted to the reference dataset. Parameters ---------- @@ -155,15 +156,19 @@ def calibrate( model Trained model or path to model file. calibration - Calibration instance to use. If None, a multitask model gets - MultiHeadRidgeCalibration (combining the best-correlating setup heads) and a - single-task model gets SplineTransformerCalibration. + Calibration instance to use. If None, MultiHeadRidgeCalibration is used. A model with a + single LC setup still predicts a one-column matrix, so this default also covers + single-task models. See ``deeplc.calibration.multihead`` for lighter alternatives (e.g. + ``MultiHeadSplineCalibration``) when the ridge over many heads isn't the right fit, for + example for a custom, non-multitask model. An unfitted naive + :class:`~deeplc.calibration.simple.Calibration` is accepted too and upgraded to its + ``MultiHead*Calibration`` counterpart; see :func:`deeplc.calibration.upgrade_calibration`. predict_kwargs Additional keyword arguments to pass to the prediction function. Returns ------- - Calibration + MultiHeadCalibration Fitted calibration instance. """ @@ -171,16 +176,17 @@ def calibrate( # peptidoform once per spectrum it was identified in, each time with a different observed # retention time, which gives the fit conflicting targets and weighs peptidoforms by how # often they happened to be identified. A caller who wants the repeats to count fits a - # Calibration itself and passes it in already fitted. + # MultiHeadCalibration itself and passes it in already fitted. psm_list_reference = deduplicate_psms(psm_list_reference) - if calibration is not None and not isinstance(calibration, Calibration): - raise ValueError( - f"Expected calibration to be of type `Calibration`, got {type(calibration)}" - ) - if calibration is not None and calibration.is_fitted: + if calibration is None: + LOGGER.debug("No calibration provided, using MultiHeadRidgeCalibration by default.") + calibration = MultiHeadRidgeCalibration() + else: + calibration = upgrade_calibration(calibration) + if calibration.is_fitted: LOGGER.warning( - "Provided Calibration is already fitted. Refitting will overwrite existing fit." + "Provided calibration is already fitted. Refitting will overwrite existing fit." ) if any(psm_list_reference["is_decoy"]): @@ -198,30 +204,11 @@ def calibrate( return_matrix=True, ) - # The default depends on the model: a multitask model is calibrated against its - # best-correlating setup heads combined, a single-task model against its one output. - if calibration is None: - if source_rt_cal.shape[1] > 1: - calibration = MultiHeadRidgeCalibration() - else: - calibration = SplineTransformerCalibration() - LOGGER.debug("No calibration provided, using %s.", type(calibration).__name__) - - # Fit calibration + # Fit calibration; every MultiHeadCalibration takes the whole matrix and selects its own + # head(s), setting selected_model_head itself for callers that want to know which setup came + # out on top. LOGGER.debug("Fitting calibration...") target_rt_cal = np.array(psm_list_reference["retention_time"], dtype=np.float32) - - # A calibration that combines heads is given the whole matrix and picks its own; it sets - # selected_model_head itself, for callers that want to know which setup came out on top. - if getattr(calibration, "uses_all_heads", False): - calibration.fit(target=target_rt_cal, source=source_rt_cal) - return calibration - - # Select the best head for calibration if the model predicts for multiple LC setups - if source_rt_cal.shape[1] > 1: - calibration.selected_model_head = _best_correlating_head(source_rt_cal, target_rt_cal) - source_rt_cal = source_rt_cal[:, calibration.selected_model_head or 0] - calibration.fit(target=target_rt_cal, source=source_rt_cal) return calibration @@ -231,7 +218,7 @@ def predict_and_calibrate( psm_list: PSMList | list[PSM | Peptidoform | str], psm_list_reference: PSMList | list[PSM | Peptidoform | str] | None = None, model: torch.nn.Module | PathLike | str | None = None, - calibration: Calibration | None = None, + calibration: Calibration | MultiHeadCalibration | None = None, predict_kwargs: dict | None = None, ) -> np.ndarray: """ @@ -249,9 +236,12 @@ def predict_and_calibrate( model Trained model or path to model file. calibration - Calibration instance to use. If None, a multitask model gets - MultiHeadRidgeCalibration (combining the best-correlating setup heads) and a - single-task model gets SplineTransformerCalibration. + Calibration instance to use. If None, MultiHeadRidgeCalibration is used. See + ``deeplc.calibration.multihead`` for lighter alternatives. An unfitted naive + :class:`~deeplc.calibration.simple.Calibration` is accepted too and upgraded to its + ``MultiHead*Calibration`` counterpart; see :func:`deeplc.calibration.upgrade_calibration`. + A fitted one is not, since it carries no record of which head it was fit on: fit a + ``MultiHead*Calibration`` instead to pass in an already fitted calibration. predict_kwargs Additional keyword arguments to pass to the prediction function. @@ -278,10 +268,8 @@ def predict_and_calibrate( return_matrix=True, ) - if calibration is not None and not isinstance(calibration, Calibration): - raise ValueError( - f"Expected calibration to be of type `Calibration`, got {type(calibration)}" - ) + if calibration is not None: + calibration = upgrade_calibration(calibration) # Fit calibration if not already fitted if calibration is None or not calibration.is_fitted: @@ -294,25 +282,8 @@ def predict_and_calibrate( else: LOGGER.info("Calibration is already fitted, skipping fitting step.") - if getattr(calibration, "uses_all_heads", False): - # the calibration combines several heads, so it takes the matrix as it is - return calibration.transform(predicted_rt) - - if predicted_rt.shape[1] > 1: - if calibration.selected_model_head is None: - raise ValueError( - "Calibration has no selected_model_head. Either use calibrate() to fit it, " - "or set calibration.selected_model_head manually before calling " - "predict_and_calibrate() with a multitask model." - ) - predicted_rt = predicted_rt[:, calibration.selected_model_head] - else: - predicted_rt = predicted_rt[:, 0] - - # Apply calibration to predictions - calibrated_rt = calibration.transform(predicted_rt) - - return calibrated_rt + # Every MultiHeadCalibration selects its own head(s), so it takes the matrix as it is. + return calibration.transform(predicted_rt) def finetune_and_predict( @@ -368,9 +339,10 @@ def finetune_and_predict( psm_list=parsed_psm_list, model=finetuned_model, predict_kwargs=predict_kwargs, + return_matrix=True, ) - # Fit calibration with simple PiecewiseLinearCalibration to the fine-tuned model predictions + # Fit calibration to the fine-tuned model predictions LOGGER.info("Fitting calibration with fine-tuned model predictions...") calibration = calibrate( psm_list_reference=parsed_psm_list_ref, @@ -736,25 +708,3 @@ def _parse_psms(psm_list: PSMList | list[PSM | Peptidoform | str]) -> PSMList: raise ValueError("List must contain either PSMs, Peptidoforms, or strings.") else: raise ValueError("Input must be a PSMList or a list of PSMs, Peptidoforms, or strings.") - - -def _best_correlating_head(predictions: np.ndarray, targets: np.ndarray) -> int: - """Return the head index with highest valid Pearson correlation to targets.""" - best_idx = 0 - best_corr = float("-inf") - - for idx in range(predictions.shape[1]): - pred_col = predictions[:, idx] - mask = np.isfinite(pred_col) & np.isfinite(targets) - if mask.sum() < 3: - continue - pred_masked = pred_col[mask] - target_masked = targets[mask] - if np.std(pred_masked) < 1e-8 or np.std(target_masked) < 1e-8: - continue - corr = np.corrcoef(pred_masked, target_masked)[0, 1] - if np.isfinite(corr) and corr > best_corr: - best_corr = corr - best_idx = idx - - return best_idx diff --git a/docs/source/models.rst b/docs/source/models.rst index 64e4265..a898605 100644 --- a/docs/source/models.rst +++ b/docs/source/models.rst @@ -25,28 +25,44 @@ bundled as :data:`deeplc.core.LEGACY_MULTITASK_MODEL` and can be passed as Calibrating against several setups at once ========================================== -Since 4.3.0 a multitask model is calibrated with -:class:`~deeplc.calibration.MultiHeadRidgeCalibration` by default: every head is ranked by -Pearson correlation to the reference, the 80 best are calibrated individually, and a ridge -regression maps those calibrated estimates onto the observed retention times, so several setups -contribute. The number of heads is the one parameter worth changing: 80 sits on a flat optimum -between roughly 40 and 320, and the class never fits more weights than half the reference allows. -Prediction costs nothing extra, because the full head matrix is computed either way. - -The previous behaviour, a spline on the single best-correlating head, remains available by -passing the calibration explicitly (it is also still the default for single-task models): +:func:`deeplc.calibrate` and :func:`deeplc.predict_and_calibrate` calibrate against the full +``(n, n_heads)`` prediction matrix of a model (``n_heads=1`` for a single-setup model), and the +calibration instance is responsible for selecting which head(s) to use. +:class:`~deeplc.calibration.MultiHeadCalibration` is the base for such calibrations; the default +is :class:`~deeplc.calibration.MultiHeadRidgeCalibration`: every head is ranked by Pearson +correlation to the reference, the 80 best are calibrated individually, and a ridge regression maps +those calibrated estimates onto the observed retention times, so several setups contribute. The +number of heads is the one parameter worth changing: 80 sits on a flat optimum between roughly 40 +and 320, and the class never fits more weights than half the reference allows. Prediction costs +nothing extra, because the full head matrix is computed either way. + +A lighter alternative, a single naive calibration on the best-correlating head, is available via +:class:`~deeplc.calibration.MultiHeadSplineCalibration` or +:class:`~deeplc.calibration.MultiHeadPiecewiseLinearCalibration`: .. code-block:: python from deeplc import predict_and_calibrate - from deeplc.calibration import SplineTransformerCalibration + from deeplc.calibration import MultiHeadSplineCalibration calibrated_rt = predict_and_calibrate( psm_list, psm_list_reference=reference, - calibration=SplineTransformerCalibration(), + calibration=MultiHeadSplineCalibration(), ) +The naive, single-series calibrations in :mod:`deeplc.calibration.simple` +(:class:`~deeplc.calibration.SplineTransformerCalibration`, +:class:`~deeplc.calibration.PiecewiseLinearCalibration`, +:class:`~deeplc.calibration.IdentityCalibration`) know nothing about heads; the ``MultiHead*`` +classes above delegate to them once a head is picked. An unfitted +:class:`~deeplc.calibration.SplineTransformerCalibration` or +:class:`~deeplc.calibration.PiecewiseLinearCalibration` passed to ``calibrate`` or +``predict_and_calibrate`` is upgraded to its ``MultiHead*`` counterpart automatically via +:func:`deeplc.calibration.upgrade_calibration`. An already fitted naive calibration is not +accepted, since it carries no record of which head it was fit on; fit a ``MultiHead*Calibration`` +instead in that case. + Training a model from scratch ============================== diff --git a/tests/test_deduplication.py b/tests/test_deduplication.py index 4c8b14e..00af54c 100644 --- a/tests/test_deduplication.py +++ b/tests/test_deduplication.py @@ -9,7 +9,7 @@ from deeplc import core from deeplc._reference_selection import deduplicate_psms -from deeplc.calibration import SplineTransformerCalibration +from deeplc.calibration import MultiHeadSplineCalibration _PEPTIDES = [ "AAGPSLSHTSGGTQSK", @@ -164,10 +164,7 @@ def test_calibrate_always_uses_the_first_observations(): calibration = core.calibrate(reference, predict_kwargs={"device": "cpu"}) predicted = core.predict(targets, return_matrix=True) - if calibration.uses_all_heads: - calibrated = calibration.transform(predicted) - else: - calibrated = calibration.transform(predicted[:, calibration.selected_model_head or 0]) + calibrated = calibration.transform(predicted) assert np.isfinite(calibrated).all() clean_low, clean_high = 5.0, 5.0 + 3.0 * (len(_PEPTIDES) - 1) @@ -180,16 +177,15 @@ def test_a_prefitted_calibration_is_the_way_to_keep_the_repeats(): The escape hatch for the rare caller who wants every reference PSM to count. ``calibrate`` deduplicates unconditionally, so a caller who wants the repeats weighed fits - a ``Calibration`` on its own targets and passes it in; ``predict_and_calibrate`` then uses - it as given instead of fitting one. + a ``MultiHeadCalibration`` on its own targets and passes it in; ``predict_and_calibrate`` then + uses it as given instead of fitting one. """ reference = _reference_with_duplicates() psm_list = _psms([(s, None) for s in _PEPTIDES]) source = core.predict(reference, predict_kwargs={"device": "cpu"}, return_matrix=True) - own = SplineTransformerCalibration() - own.selected_model_head = 0 - own.fit(target=np.array(reference["retention_time"], dtype=np.float32), source=source[:, 0]) + own = MultiHeadSplineCalibration() + own.fit(target=np.array(reference["retention_time"], dtype=np.float32), source=source) kept = core.predict_and_calibrate( psm_list, psm_list_reference=reference, calibration=own, predict_kwargs={"device": "cpu"} diff --git a/tests/test_multihead_calibration.py b/tests/test_multihead_calibration.py index ecab51d..702895b 100644 --- a/tests/test_multihead_calibration.py +++ b/tests/test_multihead_calibration.py @@ -7,7 +7,17 @@ from psm_utils import PSM, PSMList from deeplc import core -from deeplc.calibration import MultiHeadRidgeCalibration, SplineTransformerCalibration +from deeplc.calibration import ( + Calibration, + IdentityCalibration, + MultiHeadCalibration, + MultiHeadPiecewiseLinearCalibration, + MultiHeadRidgeCalibration, + MultiHeadSplineCalibration, + PiecewiseLinearCalibration, + SplineTransformerCalibration, + upgrade_calibration, +) from deeplc.exceptions import CalibrationError _PEPTIDES = [ @@ -23,6 +33,12 @@ "LNLSPLGEEMR", ] +#: (multi-head class, plain single-series class it wraps), for the shared behavioral tests below. +_SELECTOR_CLASSES = [ + (MultiHeadPiecewiseLinearCalibration, PiecewiseLinearCalibration), + (MultiHeadSplineCalibration, SplineTransformerCalibration), +] + def _synthetic(n: int = 200, n_heads: int = 12, seed: int = 0): """ @@ -42,10 +58,13 @@ def _synthetic(n: int = 200, n_heads: int = 12, seed: int = 0): return target, source -def test_declares_that_it_takes_the_whole_matrix(): - """The flag is what makes core hand over every head instead of one column.""" - assert MultiHeadRidgeCalibration().uses_all_heads is True - assert SplineTransformerCalibration().uses_all_heads is False +def test_multihead_calibrations_share_a_common_type(): + """`core` dispatches on this type; every head-selecting calibration must be one.""" + assert isinstance(MultiHeadRidgeCalibration(), MultiHeadCalibration) + assert isinstance(MultiHeadPiecewiseLinearCalibration(), MultiHeadCalibration) + assert isinstance(MultiHeadSplineCalibration(), MultiHeadCalibration) + assert not isinstance(SplineTransformerCalibration(), MultiHeadCalibration) + assert not isinstance(PiecewiseLinearCalibration(), MultiHeadCalibration) def test_beats_a_single_head_when_the_target_mixes_two(): @@ -129,6 +148,81 @@ def test_empty_source_returns_empty(): assert calibration.transform(np.zeros((0, source.shape[1]))).shape == (0,) +@pytest.mark.parametrize(("selector_cls", "inner_cls"), _SELECTOR_CLASSES) +def test_selector_matches_a_manual_fit_of_its_inner_calibration( + selector_cls: type[MultiHeadCalibration], inner_cls: type[Calibration] +): + """The wrapper picks a head and defers to its inner calibration, nothing more.""" + target, source = _synthetic() + selector = selector_cls() + selector.fit(target=target, source=source) + assert selector.selected_model_head in (0, 1) + + inner = inner_cls() + head_column = source[:, selector.selected_model_head] + inner.fit(target=target, source=head_column) + + # Exclude the exact extreme points: SplineTransformerCalibration switches between the spline + # and its linear trail model right at the fitted min/max, so a sub-ULP float32 rounding + # difference there can flip the branch and jump the output. That instability is inherent to + # the wrapped calibration, not something the selector introduces. + not_extreme = (head_column != head_column.min()) & (head_column != head_column.max()) + np.testing.assert_allclose( + selector.transform(source)[not_extreme], + inner.transform(head_column)[not_extreme], + rtol=1e-4, + atol=1e-4, + ) + + +@pytest.mark.parametrize(("selector_cls", "_inner_cls"), _SELECTOR_CLASSES) +def test_selector_is_fitted_and_transform_guard(selector_cls, _inner_cls): + """Transforming before fitting is an error, not silent nonsense.""" + calibration = selector_cls() + assert not calibration.is_fitted + with pytest.raises(CalibrationError, match="not been fitted"): + calibration.transform(np.zeros((3, 5))) + + +@pytest.mark.parametrize(("selector_cls", "_inner_cls"), _SELECTOR_CLASSES) +def test_selector_rejects_a_model_with_fewer_heads_than_it_was_fitted_on(selector_cls, _inner_cls): + """A calibration is tied to the model it was fitted on.""" + target, source = _synthetic(n_heads=12) + calibration = selector_cls() + calibration.fit(target=target, source=source) + with pytest.raises(CalibrationError, match="heads"): + # 0 columns is fewer than the fitted head's index whichever head that turned out to be. + calibration.transform(source[:, :0]) + + +@pytest.mark.parametrize(("selector_cls", "_inner_cls"), _SELECTOR_CLASSES) +def test_selector_single_head_input_is_accepted(selector_cls, _inner_cls): + """A single-task model gives a 1-D series; the calibration still works.""" + target, source = _synthetic(n_heads=1) + calibration = selector_cls() + calibration.fit(target=target, source=source[:, 0]) + out = calibration.transform(source[:, 0]) + assert out.shape == target.shape + assert np.isfinite(out).all() + + +@pytest.mark.parametrize(("selector_cls", "_inner_cls"), _SELECTOR_CLASSES) +def test_selector_too_few_finite_points(selector_cls, _inner_cls): + """Two points cannot support a fit.""" + calibration = selector_cls() + with pytest.raises(CalibrationError, match="three reference points"): + calibration.fit(target=np.array([1.0, np.nan]), source=np.zeros((2, 4))) + + +@pytest.mark.parametrize(("selector_cls", "_inner_cls"), _SELECTOR_CLASSES) +def test_selector_empty_source_returns_empty(selector_cls, _inner_cls): + """No PSMs in, no predictions out.""" + target, source = _synthetic() + calibration = selector_cls() + calibration.fit(target=target, source=source) + assert calibration.transform(np.zeros((0, source.shape[1]))).shape == (0,) + + def _psm_list(rts: list[float] | None = None) -> PSMList: return PSMList( psm_list=[ @@ -177,13 +271,81 @@ def test_default_calibration_combines_heads_for_the_multitask_model(): assert calibration.selected_model_head is not None -def test_single_head_calibration_remains_available(): - """Passing SplineTransformerCalibration restores the one-head behaviour.""" +def test_lighter_calibration_can_be_passed_explicitly(): + """Passing MultiHeadSplineCalibration opts out of the ridge combination.""" reference = _psm_list([5.0 + 3.0 * i for i in range(len(_PEPTIDES))]) calibration = core.calibrate( reference, - calibration=SplineTransformerCalibration(), + calibration=MultiHeadSplineCalibration(), predict_kwargs={"device": "cpu"}, ) - assert calibration.uses_all_heads is False + assert isinstance(calibration, MultiHeadSplineCalibration) assert calibration.selected_model_head is not None + + +def test_upgrade_calibration_returns_a_multihead_instance_unchanged(): + """A MultiHeadCalibration is not touched: nothing to upgrade.""" + calibration = MultiHeadRidgeCalibration(n_heads=3) + assert upgrade_calibration(calibration) is calibration + + +def test_upgrade_calibration_wraps_a_naive_spline_calibration(): + """A naive SplineTransformerCalibration becomes a MultiHeadSplineCalibration.""" + upgraded = upgrade_calibration(SplineTransformerCalibration()) + assert isinstance(upgraded, MultiHeadSplineCalibration) + assert not upgraded.is_fitted + + +def test_upgrade_calibration_carries_over_piecewise_linear_parameters(): + """The wrapped PiecewiseLinearCalibration keeps the constructor arguments it was given.""" + naive = PiecewiseLinearCalibration(number_of_splits=25, use_median=True) + upgraded = upgrade_calibration(naive) + assert isinstance(upgraded, MultiHeadPiecewiseLinearCalibration) + assert upgraded._inner.number_of_splits == 25 + assert upgraded._inner.use_median is True + + +def test_upgrade_calibration_rejects_an_already_fitted_naive_calibration(): + """A fitted naive calibration carries no record of which head it was fit on.""" + naive = SplineTransformerCalibration() + naive.fit(target=np.linspace(0, 10, 50), source=np.linspace(0, 10, 50)) + with pytest.raises(CalibrationError, match="fitted, naive Calibration"): + upgrade_calibration(naive) + + +def test_upgrade_calibration_rejects_a_calibration_with_no_multihead_counterpart(): + """IdentityCalibration has no MultiHead* counterpart; upgrading it is a clear error.""" + with pytest.raises(ValueError, match="No MultiHeadCalibration counterpart"): + upgrade_calibration(IdentityCalibration()) + + +def test_upgrade_calibration_rejects_a_nonsensical_type(): + """Neither a Calibration nor a MultiHeadCalibration cannot be upgraded.""" + with pytest.raises(ValueError, match="Expected calibration to be of type"): + upgrade_calibration(object()) + + +def test_core_accepts_an_unfitted_naive_calibration_for_backward_compatibility(): + """`calibrate()` upgrades a plain SplineTransformerCalibration instead of rejecting it.""" + reference = _psm_list([5.0 + 3.0 * i for i in range(len(_PEPTIDES))]) + calibration = core.calibrate( + reference, + calibration=SplineTransformerCalibration(), + predict_kwargs={"device": "cpu"}, + ) + assert isinstance(calibration, MultiHeadSplineCalibration) + assert calibration.is_fitted + + +def test_core_rejects_a_fitted_naive_calibration(): + """predict_and_calibrate() cannot recover the head a naive calibration was fit on.""" + reference = _psm_list([5.0 + 3.0 * i for i in range(len(_PEPTIDES))]) + naive = SplineTransformerCalibration() + naive.fit(target=np.array([5.0 + 3.0 * i for i in range(len(_PEPTIDES))]), source=np.zeros(10)) + with pytest.raises(CalibrationError, match="fitted, naive Calibration"): + core.predict_and_calibrate( + _psm_list(), + psm_list_reference=reference, + calibration=naive, + predict_kwargs={"device": "cpu"}, + )