From c92720557d116a8bfd053decd3e0644d7b0a10be Mon Sep 17 00:00:00 2001 From: cclaess Date: Fri, 17 Jul 2026 17:34:12 +0200 Subject: [PATCH] Fix resampling bug and add tests --- src/spectre/io.py | 6 +- tests/test_hf_modeling.py | 115 ++++++++++++++++++++++++++++++++++++++ tests/test_io_resample.py | 105 ++++++++++++++++++++++++++++++++++ 3 files changed, 225 insertions(+), 1 deletion(-) create mode 100644 tests/test_hf_modeling.py create mode 100644 tests/test_io_resample.py diff --git a/src/spectre/io.py b/src/spectre/io.py index bbf9fff..d3b0e80 100644 --- a/src/spectre/io.py +++ b/src/spectre/io.py @@ -149,6 +149,7 @@ def resample( """ _require_nibabel() try: + from monai.data import MetaTensor from monai.transforms import Spacing except ImportError as e: raise ImportError( @@ -158,8 +159,11 @@ def resample( if len(spacing) != 3: raise ValueError(f"spacing must have 3 elements, got {tuple(spacing)}.") + # Spacing reads the source spacing from the input's affine, so the affine has to travel on the + # tensor as a MetaTensor - it is not a call argument (older MONAI took an `affine=` kwarg). + volume = MetaTensor(x, affine=torch.as_tensor(meta.affine, dtype=torch.float64)) spacer = Spacing(pixdim=tuple(float(s) for s in spacing), mode=mode) - resampled = spacer(x, affine=torch.as_tensor(meta.affine, dtype=torch.float64)) + resampled = spacer(volume) affine = getattr(resampled, "affine", meta.affine) resampled = resampled.as_tensor() if hasattr(resampled, "as_tensor") else torch.as_tensor(resampled) diff --git a/tests/test_hf_modeling.py b/tests/test_hf_modeling.py new file mode 100644 index 0000000..5067da0 --- /dev/null +++ b/tests/test_hf_modeling.py @@ -0,0 +1,115 @@ +"""The `transformers` wrapper around SpectreImageFeatureExtractor. + +Uses a small randomly-initialised preset: these tests are about the wrapper's call contract, not +embedding quality, so no weights are downloaded. +""" +import importlib.util +import sys +from pathlib import Path + +import pytest +import torch + +ROOT = Path(__file__).resolve().parents[1] + +pytest.importorskip("transformers") + +from transformers.modeling_outputs import BaseModelOutput # noqa: E402 + + +def _load(name): + """Load an hf_export module without putting hf_export on sys.path. + + hf_export/ holds a generated `spectre/` copy from the last export run; adding it to sys.path + would shadow the real package with that stale snapshot. + """ + spec = importlib.util.spec_from_file_location(name, ROOT / "hf_export" / f"{name}.py") + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +SpectreConfig = _load("configuration_spectre").SpectreConfig +SpectreModel = _load("modeling_spectre").SpectreModel + + +@pytest.fixture(scope="module") +def model(): + return SpectreModel(SpectreConfig(preset="spectre-small")).eval() + + +@pytest.fixture(scope="module") +def crops(): + return torch.randn(1, 2, 1, 128, 128, 64) + + +GRID = (2, 1, 1) + + +def test_forward_returns_a_tensor_by_default(model, crops): + with torch.no_grad(): + out = model(crops, grid_size=GRID) + assert isinstance(out, torch.Tensor) + + +def test_return_dict_gives_a_model_output(model, crops): + with torch.no_grad(): + out = model(crops, grid_size=GRID, return_dict=True) + assert isinstance(out, BaseModelOutput) + assert isinstance(out.last_hidden_state, torch.Tensor) + + +def test_raw_scan_is_windowed_internally(model): + scan = torch.rand(1, 256, 128, 64) * 3000 - 1200 + with torch.no_grad(): + out = model(scan) + assert out.ndim == 2 + + +def test_crop_size_is_exposed(model): + assert model.crop_size == (128, 128, 64) + + +# --- the call contract with transformers-style kwargs ----------------------------------------- + +@pytest.mark.parametrize("kwarg", ["use_cache", "output_norms", "some_future_kwarg"]) +def test_unknown_transformers_kwargs_are_ignored(model, crops, kwarg): + """Generic transformers callers pass these; they must not blow up a feature extractor. + + Regression: forwarding **kwargs into the extractor turned these into a TypeError. + """ + with torch.no_grad(): + out = model(crops, grid_size=GRID, **{kwarg: False}) + assert isinstance(out, torch.Tensor) + + +def test_max_crops_per_forward_is_honoured(model, crops): + """The one extractor kwarg worth exposing; it must reach the extractor, not the sink.""" + with torch.no_grad(): + whole = model(crops, grid_size=GRID) + chunked = model(crops, grid_size=GRID, max_crops_per_forward=1) + assert torch.allclose(whole, chunked, atol=1e-5) + + +def test_output_hidden_states_is_refused_not_ignored(model, crops): + """Silently returning no hidden states would be worse than saying we cannot.""" + with pytest.raises(NotImplementedError, match="hidden states"): + model(crops, grid_size=GRID, output_hidden_states=True) + + +def test_output_attentions_is_refused_not_ignored(model, crops): + with pytest.raises(NotImplementedError, match="attention weights"): + model(crops, grid_size=GRID, output_attentions=True) + + +@pytest.mark.parametrize("falsy", [False, None]) +def test_falsy_output_flags_are_fine(model, crops, falsy): + with torch.no_grad(): + out = model(crops, grid_size=GRID, output_hidden_states=falsy, output_attentions=falsy) + assert isinstance(out, torch.Tensor) + + +def test_the_refusal_points_at_something_that_exists(model): + """The error tells people to use forward_intermediates; make sure it is really there.""" + assert hasattr(model.model.backbone, "forward_intermediates") diff --git a/tests/test_io_resample.py b/tests/test_io_resample.py new file mode 100644 index 0000000..f3f44ba --- /dev/null +++ b/tests/test_io_resample.py @@ -0,0 +1,105 @@ +"""Reading and resampling CT scans (`spectre.io`). + +The resampling path had no coverage, which is how a call to a nonexistent `Spacing(affine=...)` +argument shipped. These exercise it against real NIfTI files. +""" +import numpy as np +import pytest +import torch + +pytest.importorskip("nibabel", reason="spectre.io needs the [inference] extra") +pytest.importorskip("monai", reason="resampling needs MONAI") +import nibabel as nib # noqa: E402 + +from spectre.io import CTMeta, load_and_window, load_ct, resample # noqa: E402 + +SPACING = (0.7, 0.7, 1.5) + + +def write_scan(path, shape=(60, 50, 40), spacing=SPACING, seed=0): + rng = np.random.default_rng(seed) + data = (rng.random(shape) * 3000 - 1200).astype(np.float32) + affine = np.diag([*spacing, 1.0]) + nib.save(nib.Nifti1Image(data, affine), str(path)) + return path + + +def test_load_ct_reports_native_spacing(tmp_path): + scan = write_scan(tmp_path / "scan.nii.gz") + volume, meta = load_ct(scan) + assert volume.shape[0] == 1 # channel-first + assert isinstance(meta, CTMeta) + assert meta.spacing == pytest.approx(SPACING, abs=1e-4) + assert meta.orientation == "RAS" + + +def test_resample_changes_shape_and_spacing(tmp_path): + """The regression: this used to raise TypeError on Spacing(affine=...).""" + scan = write_scan(tmp_path / "scan.nii.gz", shape=(60, 50, 40)) + volume, meta = load_ct(scan) + + resampled, new_meta = resample(volume, meta, (0.5, 0.5, 1.0)) + + assert new_meta.spacing == (0.5, 0.5, 1.0) + # Finer spacing on two axes -> more voxels; coarser depth spacing (1.0 < 1.5) -> more too. + assert resampled.shape[1] > volume.shape[1] + assert resampled.shape[2] > volume.shape[2] + assert torch.isfinite(resampled).all() + + +def test_resample_to_native_spacing_is_close_to_identity(tmp_path): + scan = write_scan(tmp_path / "scan.nii.gz") + volume, meta = load_ct(scan) + resampled, _ = resample(volume, meta, SPACING) + assert tuple(resampled.shape) == tuple(volume.shape) + + +def test_resample_scale_factor_matches_spacing_ratio(tmp_path): + scan = write_scan(tmp_path / "scan.nii.gz", shape=(60, 60, 60), spacing=(1.0, 1.0, 1.0)) + volume, meta = load_ct(scan) + resampled, _ = resample(volume, meta, (0.5, 0.5, 0.5)) + # Halving the spacing doubles each axis (allow a voxel of rounding). + for axis in (1, 2, 3): + assert resampled.shape[axis] == pytest.approx(2 * volume.shape[axis], abs=1) + + +def test_resample_returns_a_plain_tensor(tmp_path): + """Downstream windowing expects a torch.Tensor, not a MetaTensor.""" + scan = write_scan(tmp_path / "scan.nii.gz") + volume, meta = load_ct(scan) + resampled, _ = resample(volume, meta, (0.5, 0.5, 1.0)) + assert type(resampled) is torch.Tensor + + +def test_resample_rejects_wrong_length_spacing(tmp_path): + scan = write_scan(tmp_path / "scan.nii.gz") + volume, meta = load_ct(scan) + with pytest.raises(ValueError, match="3 elements"): + resample(volume, meta, (0.5, 0.5)) + + +def test_load_and_window_without_spacing_keeps_native(tmp_path): + scan = write_scan(tmp_path / "scan.nii.gz", shape=(256, 128, 64)) + crops, grid, meta = load_and_window(scan) + assert crops.shape[1:] == (1, 128, 128, 64) + assert grid == (2, 1, 1) + assert meta.spacing == pytest.approx(SPACING, abs=1e-4) + + +def test_load_and_window_with_spacing_resamples_then_crops(tmp_path): + """The full CLI --spacing path: load -> scale -> resample -> crop -> patch.""" + scan = write_scan(tmp_path / "scan.nii.gz", shape=(200, 200, 100), spacing=(1.0, 1.0, 1.0)) + crops, grid, _ = load_and_window(scan, spacing=(0.5, 0.5, 1.0)) + assert crops.shape[1:] == (1, 128, 128, 64) + # 200 @ 1.0 -> 400 @ 0.5 gives 3 crops of 128 across H and W; depth 100 stays 100 -> 1 crop. + assert grid == (3, 3, 1) + assert torch.isfinite(crops).all() + + +def test_load_and_window_scales_before_resampling(tmp_path): + """Intensity must already be in [0, 1] after windowing, even on the resample path.""" + scan = write_scan(tmp_path / "scan.nii.gz", shape=(140, 140, 70), spacing=(1.0, 1.0, 1.0)) + crops, _, _ = load_and_window(scan, spacing=(0.8, 0.8, 1.0)) + # Interpolation can overshoot slightly past the clamp, but not by much. + assert float(crops.min()) >= -0.05 + assert float(crops.max()) <= 1.05