Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion src/spectre/io.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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)
Comment on lines +164 to 168

Expand Down
115 changes: 115 additions & 0 deletions tests/test_hf_modeling.py
Original file line number Diff line number Diff line change
@@ -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
Comment on lines +26 to +30


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")
105 changes: 105 additions & 0 deletions tests/test_io_resample.py
Original file line number Diff line number Diff line change
@@ -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]
Comment on lines +43 to +45
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