diff --git a/src/torchio/data/batch.py b/src/torchio/data/batch.py index 13b058eb4..6aea3579d 100644 --- a/src/torchio/data/batch.py +++ b/src/torchio/data/batch.py @@ -2,7 +2,10 @@ from __future__ import annotations +import copy as _copy import dataclasses +from collections.abc import Sequence +from typing import TYPE_CHECKING from typing import Any import torch @@ -10,62 +13,133 @@ from typing_extensions import Self from .affine import AffineMatrix +from .batch_schema import _ImageSchema +from .batch_schema import _SubjectSchema +from .bboxes import BoundingBoxes from .image import Image +from .image import LabelMap from .image import ScalarImage from .invertible import Invertible +from .points import Points + +if TYPE_CHECKING: + from .subject import Subject #: Reserved param keys used for per-instance history bookkeeping. _BATCH_META_KEYS = ("_batch_size", "_batched_keys", "_keep") class ImagesBatch(Invertible): - """A batch of images with per-sample affines. + """A batch of images with per-sample affines and private prototypes. Wraps a 5D tensor `(B, C, I, J, K)` and a list of `AffineMatrix` - matrices (one per sample). Created by stacking multiple `Image` - objects or directly from a 5D tensor. + matrices (one per sample). Use `from_images()` for lossless image + round-trips or `from_tensor()` for an existing 5D tensor. Args: data: 5D tensor with shape `(B, C, I, J, K)`. - affines: List of affine matrices, one per sample. + affines: Affine matrices, one per sample. image_class: The `Image` subclass to use when unbatching. """ def __init__( self, data: Tensor, - affines: list[AffineMatrix], + affines: Sequence[AffineMatrix], *, image_class: type[Image] = ScalarImage, ) -> None: + prototypes = _make_prototypes_from_class(data, image_class) + self._initialize(data, affines, prototypes) + + def _initialize( + self, + data: Tensor, + affines: Sequence[AffineMatrix], + prototypes: Sequence[Image], + ) -> None: + """Initialize a validated image batch.""" if data.ndim != 5: msg = f"Expected 5D tensor (B, C, I, J, K), got {data.ndim}D" raise ValueError(msg) + if data.shape[0] == 0: + msg = "Cannot create an empty image batch" + raise ValueError(msg) if len(affines) != data.shape[0]: msg = f"Expected {data.shape[0]} affines, got {len(affines)}" raise ValueError(msg) + if len(prototypes) != data.shape[0]: + msg = f"Expected {data.shape[0]} prototypes, got {len(prototypes)}" + raise ValueError(msg) self._data = data - self._affines = affines - self._image_class = image_class + self._affines = [affine.clone() for affine in affines] + self._prototypes = list(prototypes) self.applied_transforms: list[Any] = [] @classmethod - def from_images(cls, images: list[Image]) -> Self: - """Stack a list of images into a batch. + def _from_parts( + cls, + data: Tensor, + affines: Sequence[AffineMatrix], + prototypes: Sequence[Image], + ) -> Self: + """Build an image batch from validated internal parts.""" + batch = cls.__new__(cls) + batch._initialize(data, affines, prototypes) + return batch + + @classmethod + def from_tensor( + cls, + data: Tensor, + affines: Sequence[AffineMatrix] | None = None, + *, + image_class: type[Image] = ScalarImage, + ) -> Self: + """Build an image batch from a 5D tensor. + + Args: + data: 5D tensor with shape `(B, C, I, J, K)`. + affines: Optional affine matrices, one per element. Identity + matrices are used when omitted. + image_class: Image class used to synthesize private prototypes. + + Returns: + A new image batch. + """ + if data.ndim != 5: + msg = f"Expected 5D tensor (B, C, I, J, K), got {data.ndim}D" + raise ValueError(msg) + resolved_affines = ( + [AffineMatrix().to(data.device) for _ in range(data.shape[0])] + if affines is None + else affines + ) + return cls(data, resolved_affines, image_class=image_class) + + @classmethod + def from_images(cls, images: Sequence[Image]) -> Self: + """Stack images into a lossless batch. - All images must have the same shape. + All images must share the same schema, shape, dtype, and device. Args: - images: List of `Image` instances to stack. + images: Images to stack. + + Returns: + A new image batch. """ if not images: msg = "Cannot create batch from empty list" raise ValueError(msg) - tensors = [img.data for img in images] + schema = _ImageSchema.from_image(images[0]) + for index, image in enumerate(images[1:], 1): + schema.validate(image, index=index, name="image") + tensors = [image.data for image in images] stacked = torch.stack(tensors) - affines = [img.affine.clone() for img in images] - image_class = type(images[0]) - return cls(stacked, affines, image_class=image_class) + affines = [image.affine for image in images] + prototypes = [_make_image_prototype(image) for image in images] + return cls._from_parts(stacked, affines, prototypes) @property def data(self) -> Tensor: @@ -84,6 +158,16 @@ def affines(self) -> list[AffineMatrix]: """List of affine matrices, one per sample.""" return self._affines + @property + def image_class(self) -> type[Image]: + """Image class shared by every batch element.""" + return type(self._prototypes[0]) + + @property + def is_label(self) -> bool: + """Whether the batch contains label images.""" + return issubclass(self.image_class, LabelMap) + @property def batch_size(self) -> int: """Number of samples in the batch.""" @@ -95,18 +179,42 @@ def device(self) -> torch.device: return self._data.device def to(self, *args: Any, **kwargs: Any) -> Self: - """Move batch data to a device and/or cast dtype.""" + """Move batch data and payload to a device or dtype. + + Args: + *args: Positional arguments forwarded to `torch.Tensor.to`. + **kwargs: Keyword arguments forwarded to `torch.Tensor.to`. + + Returns: + `self` (modified in-place). + """ self._data = self._data.to(*args, **kwargs) for affine in self._affines: affine.to(*args, **kwargs) + for prototype in self._prototypes: + prototype.to(*args, **kwargs) return self def __getitem__(self, index: int) -> Image: - """Get a single image from the batch by index.""" - return self._image_class( - self._data[index], + """Get one reconstructed image. + + Args: + index: Batch element index. + + Returns: + The reconstructed image. + """ + prototype = self._prototypes[index] + image = prototype.new_like( + data=self._data[index], affine=self._affines[index].clone(), ) + image._metadata = _copy.deepcopy(prototype.metadata) + image.applied_transforms = [ + *prototype.applied_transforms, + *self.applied_transforms, + ] + return image def __len__(self) -> int: return self.batch_size @@ -115,29 +223,54 @@ def unbatch(self) -> list[Image]: """Split the batch into individual images.""" return [self[i] for i in range(self.batch_size)] + @property + def has_annotations(self) -> bool: + """Whether any image prototype carries annotations.""" + return any( + prototype.points or prototype.bounding_boxes + for prototype in self._prototypes + ) + def __repr__(self) -> str: b, c, i, j, k = self._data.shape - cls = self._image_class.__name__ + cls = self.image_class.__name__ return f"ImagesBatch({cls}, batch_size={b}, shape=({c}, {i}, {j}, {k}))" class SubjectsBatch(Invertible): - """A batch of subjects with stacked image data. + """A batch of image columns and per-element object stores. - Each named image entry becomes an `ImagesBatch`. Metadata is - stored as lists (one value per sample). + Each image field becomes an `ImagesBatch`. Metadata, points, and + bounding boxes are stored as lists with one value per element. Created by `SubjectsLoader` or `SubjectsBatch.from_subjects()`. + + Args: + images: Named image batches. + points: Named subject-level point sets. + bounding_boxes: Named subject-level bounding boxes. + metadata: Named metadata values. """ def __init__( self, - images: dict[str, ImagesBatch], + images: dict[str, ImagesBatch] | None = None, *, + points: dict[str, list[Points]] | None = None, + bounding_boxes: dict[str, list[BoundingBoxes]] | None = None, metadata: dict[str, list[Any]] | None = None, ) -> None: - self._images = images - self._metadata: dict[str, list[Any]] = metadata or {} + self._images = dict(images or {}) + self._points = dict(points or {}) + self._bounding_boxes = dict(bounding_boxes or {}) + self._metadata = dict(metadata or {}) + self._batch_size = _resolve_batch_size( + self._images, + self._points, + self._bounding_boxes, + self._metadata, + ) + self._schema: _SubjectSchema | None = None self.applied_transforms: list[Any] = [] # When per-element branching occurs (e.g. per-instance OneOf), # this stores the frozen per-element history prefix. Transforms @@ -166,73 +299,130 @@ def set_per_element_history(self, histories: list[list[Any]]) -> None: self.applied_transforms = [] @classmethod - def from_subjects(cls, subjects: list[Any]) -> Self: - """Stack a list of subjects into a batch. + def from_subjects(cls, subjects: Sequence[Any]) -> Self: + """Stack subjects into a lossless batch. Args: - subjects: List of `Subject` instances. - """ - from .subject import Subject - - if not subjects: - msg = "Cannot create batch from empty list" - raise ValueError(msg) - - # Collect image names and types from the first subject - first: Subject = subjects[0] - image_names = list(first.images.keys()) - - # Stack images - images: dict[str, ImagesBatch] = {} - for name in image_names: - img_list = [sub.images[name] for sub in subjects] - images[name] = ImagesBatch.from_images(img_list) + subjects: Subjects to stack. - # Collect metadata (non-image, non-annotation entries) - metadata: dict[str, list[Any]] = {} - for key in first.metadata: - metadata[key] = [sub.metadata[key] for sub in subjects] - - return cls(images, metadata=metadata) + Returns: + A new subject batch. + """ + schema = _validate_subjects(subjects) + batch = cls( + _stack_subject_images(subjects, schema), + points=_collect_subject_points(subjects, schema), + bounding_boxes=_collect_subject_boxes(subjects, schema), + metadata=_collect_subject_metadata(subjects, schema), + ) + batch._schema = schema + return batch @property def batch_size(self) -> int: """Number of samples in the batch.""" - first = next(iter(self._images.values())) - return first.batch_size + return self._batch_size @property def images(self) -> dict[str, ImagesBatch]: """Dict of named image batches.""" return self._images + @property + def points(self) -> dict[str, list[Points]]: + """Subject-level point sets, one value per element.""" + return self._points + + @property + def bounding_boxes(self) -> dict[str, list[BoundingBoxes]]: + """Subject-level bounding boxes, one value per element.""" + return self._bounding_boxes + @property def metadata(self) -> dict[str, list[Any]]: """Metadata lists (one value per sample).""" return self._metadata + @property + def has_annotations(self) -> bool: + """Whether the batch contains subject- or image-level annotations.""" + return bool( + self._points + or self._bounding_boxes + or any(image.has_annotations for image in self._images.values()) + ) + @property def device(self) -> torch.device: """Device of the batch data.""" - first = next(iter(self._images.values())) - return first.device + devices = [image.device for image in self._images.values()] + devices.extend( + points.device for values in self._points.values() for points in values + ) + devices.extend( + boxes.device for values in self._bounding_boxes.values() for boxes in values + ) + if not devices: + return torch.device("cpu") + reference = devices[0] + if any(device != reference for device in devices[1:]): + msg = f"Inconsistent devices in SubjectsBatch: {devices}" + raise RuntimeError(msg) + return reference def to(self, *args: Any, **kwargs: Any) -> Self: - """Move all data to a device and/or cast dtype.""" + """Move all spatial data to a device or dtype. + + Args: + *args: Positional arguments forwarded to each field's `to` + method. + **kwargs: Keyword arguments forwarded to each field's `to` + method. + + Returns: + `self` (modified in-place). + """ for batch in self._images.values(): batch.to(*args, **kwargs) + for values in self._points.values(): + for points in values: + points.to(*args, **kwargs) + for values in self._bounding_boxes.values(): + for boxes in values: + boxes.to(*args, **kwargs) return self - def __getitem__(self, key: str) -> ImagesBatch: - """Get a named image batch.""" - return self._images[key] + def __getitem__(self, key: str) -> Any: + """Get a named batched field. + + Args: + key: Field name. - def __getattr__(self, name: str) -> ImagesBatch: - """Attribute-style access to image batches.""" + Returns: + The corresponding batched field. + """ + for store in ( + self._images, + self._points, + self._bounding_boxes, + self._metadata, + ): + if key in store: + return store[key] + raise KeyError(key) + + def __getattr__(self, name: str) -> Any: + """Access a named batched field as an attribute.""" if name.startswith("_"): raise AttributeError(name) - if name in self._images: - return self._images[name] + for store in ( + self._images, + self._points, + self._bounding_boxes, + self._metadata, + ): + if name in store: + return store[name] msg = f"SubjectsBatch has no attribute {name!r}" raise AttributeError(msg) @@ -246,14 +436,17 @@ def unbatch(self) -> list[Any]: """ from .subject import Subject - n = self.batch_size subjects = [] - for i in range(n): + for i in range(self.batch_size): kwargs: dict[str, Any] = {} for name, img_batch in self._images.items(): kwargs[name] = img_batch[i] + for name, values in self._points.items(): + kwargs[name] = _copy.deepcopy(values[i]) + for name, values in self._bounding_boxes.items(): + kwargs[name] = _copy.deepcopy(values[i]) for key, values in self._metadata.items(): - kwargs[key] = values[i] + kwargs[key] = _copy.deepcopy(values[i]) sub = Subject(**kwargs) suffix = _slice_history(self.applied_transforms, i) if self._per_element_history is not None: @@ -326,14 +519,165 @@ def apply_inverse_transform(self, **kwargs: Any) -> SubjectsBatch: return super().apply_inverse_transform(**kwargs) def __repr__(self) -> str: - names = ", ".join(self._images.keys()) - return f"SubjectsBatch(batch_size={self.batch_size}, images=[{names}])" + fields = [] + for label, store in ( + ("images", self._images), + ("points", self._points), + ("bboxes", self._bounding_boxes), + ("metadata", self._metadata), + ): + if store: + fields.append(f"{label}=[{', '.join(store)}]") + return f"SubjectsBatch(batch_size={self.batch_size}, {', '.join(fields)})" # Alias for radiology users (see Subject/Study note in subject.py). StudiesBatch = SubjectsBatch +def _validate_subjects( + subjects: Sequence[Any], +) -> _SubjectSchema: + """Validate subject inputs and return their shared schema.""" + from .subject import Subject + + if not subjects: + msg = "Cannot create batch from empty list" + raise ValueError(msg) + for index, subject in enumerate(subjects): + if not isinstance(subject, Subject): + msg = f"Expected Subject at index {index}, got {type(subject).__name__}" + raise TypeError(msg) + first = subjects[0] + schema = _SubjectSchema.from_subject(first) + for index, subject in enumerate(subjects[1:], 1): + schema.validate(subject, index=index) + return schema + + +def _stack_subject_images( + subjects: Sequence[Subject], + schema: _SubjectSchema, +) -> dict[str, ImagesBatch]: + """Stack each image field across subjects.""" + return { + name: ImagesBatch.from_images([subject.images[name] for subject in subjects]) + for name in schema.images + } + + +def _collect_subject_points( + subjects: Sequence[Subject], + schema: _SubjectSchema, +) -> dict[str, list[Points]]: + """Collect independent subject-level point values.""" + return { + name: [_copy.deepcopy(subject.points[name]) for subject in subjects] + for name in schema.points + } + + +def _collect_subject_boxes( + subjects: Sequence[Subject], + schema: _SubjectSchema, +) -> dict[str, list[BoundingBoxes]]: + """Collect independent subject-level bounding boxes.""" + return { + name: [_copy.deepcopy(subject.bounding_boxes[name]) for subject in subjects] + for name in schema.bounding_boxes + } + + +def _collect_subject_metadata( + subjects: Sequence[Subject], + schema: _SubjectSchema, +) -> dict[str, list[Any]]: + """Collect independent subject metadata values.""" + return { + key: [_copy.deepcopy(subject.metadata[key]) for subject in subjects] + for key in schema.metadata_keys + } + + +def _make_prototypes_from_class( + data: Tensor, + image_class: type[Image], +) -> list[Image]: + """Create minimal private prototypes for an existing tensor batch.""" + if data.ndim != 5: + msg = f"Expected 5D tensor (B, C, I, J, K), got {data.ndim}D" + raise ValueError(msg) + if not isinstance(image_class, type) or not issubclass(image_class, Image): + msg = f"Expected an Image subclass, got {image_class!r}" + raise TypeError(msg) + channels = data.shape[1] + return [ + image_class( + torch.empty( + channels, + 1, + 1, + 1, + dtype=data.dtype, + device=data.device, + ) + ) + for _ in range(data.shape[0]) + ] + + +def _make_image_prototype(image: Image) -> Image: + """Create a lightweight prototype preserving image payload.""" + data = torch.empty( + image.shape[0], + 1, + 1, + 1, + dtype=image.data.dtype, + device=image.data.device, + ) + prototype = image.new_like(data=data) + prototype._metadata = _copy.deepcopy(image.metadata) + prototype.applied_transforms = list(image.applied_transforms) + return prototype + + +def _resolve_batch_size( + images: dict[str, ImagesBatch], + points: dict[str, list[Points]], + bounding_boxes: dict[str, list[BoundingBoxes]], + metadata: dict[str, list[Any]], +) -> int: + """Resolve and validate the shared length of all batch fields.""" + sizes: list[tuple[str, int]] = [] + sizes.extend( + (f"image {name!r}", image.batch_size) for name, image in images.items() + ) + sizes.extend((f"point {name!r}", len(values)) for name, values in points.items()) + sizes.extend( + (f"bounding box {name!r}", len(values)) + for name, values in bounding_boxes.items() + ) + sizes.extend( + (f"metadata {name!r}", len(values)) for name, values in metadata.items() + ) + if not sizes: + msg = "A SubjectsBatch must contain at least one batched field" + raise ValueError(msg) + reference_name, reference_size = sizes[0] + if reference_size == 0: + msg = "Cannot create an empty SubjectsBatch" + raise ValueError(msg) + for name, size in sizes[1:]: + if size != reference_size: + msg = ( + f"Inconsistent batch size: {reference_name} has" + f" {reference_size} elements, but {name} has {size}" + ) + raise ValueError(msg) + return reference_size + + def _slice_params( params: dict[str, Any], index: int, diff --git a/src/torchio/data/batch_schema.py b/src/torchio/data/batch_schema.py new file mode 100644 index 000000000..a16f1c4f4 --- /dev/null +++ b/src/torchio/data/batch_schema.py @@ -0,0 +1,217 @@ +"""Schemas used to validate image and subject batches.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import torch + +from .bboxes import BoundingBoxes +from .image import Image +from .points import Points +from .subject import Subject + + +@dataclass(frozen=True) +class _AnnotationSchema: + """Describe one named annotation field.""" + + value_type: type[Points] | type[BoundingBoxes] + metadata_keys: tuple[str, ...] + + @classmethod + def from_value(cls, value: Points | BoundingBoxes) -> _AnnotationSchema: + """Build a schema from one annotation value.""" + return cls(type(value), tuple(value.metadata)) + + def validate( + self, + value: Points | BoundingBoxes, + *, + index: int, + context: str, + ) -> None: + """Validate one annotation against this schema.""" + if type(value) is not self.value_type: + msg = ( + f"{context} at index {index} has type {type(value).__name__}," + f" expected {self.value_type.__name__}" + ) + raise ValueError(msg) + _validate_keys( + self.metadata_keys, + value.metadata, + index=index, + context=f"{context} metadata", + ) + + +@dataclass(frozen=True) +class _ImageSchema: + """Describe one named image field.""" + + value_type: type[Image] + shape: tuple[int, ...] + dtype: str + device: torch.device + metadata_keys: tuple[str, ...] + points: dict[str, _AnnotationSchema] + bounding_boxes: dict[str, _AnnotationSchema] + + @classmethod + def from_image(cls, image: Image) -> _ImageSchema: + """Build a schema from one image.""" + return cls( + value_type=type(image), + shape=tuple(image.shape), + dtype=_normalize_dtype(image.dtype), + device=image.device, + metadata_keys=tuple(image.metadata), + points={ + name: _AnnotationSchema.from_value(value) + for name, value in image.points.items() + }, + bounding_boxes={ + name: _AnnotationSchema.from_value(value) + for name, value in image.bounding_boxes.items() + }, + ) + + def validate(self, image: Image, *, index: int, name: str) -> None: + """Validate one image against this schema.""" + context = f"Image {name!r}" + if type(image) is not self.value_type: + msg = ( + f"{context} at index {index} has type {type(image).__name__}," + f" expected {self.value_type.__name__}" + ) + raise ValueError(msg) + for attribute in ("shape", "device"): + expected = getattr(self, attribute) + actual = getattr(image, attribute) + if actual != expected: + msg = ( + f"{context} at index {index} has {attribute} {actual}," + f" expected {expected}" + ) + raise ValueError(msg) + actual_dtype = _normalize_dtype(image.dtype) + if actual_dtype != self.dtype: + msg = ( + f"{context} at index {index} has dtype {actual_dtype}," + f" expected {self.dtype}" + ) + raise ValueError(msg) + _validate_keys( + self.metadata_keys, + image.metadata, + index=index, + context=f"{context} metadata", + ) + _validate_annotations( + self.points, + image.points, + index=index, + context=f"{context} points", + ) + _validate_annotations( + self.bounding_boxes, + image.bounding_boxes, + index=index, + context=f"{context} bounding boxes", + ) + + +@dataclass(frozen=True) +class _SubjectSchema: + """Describe the fields shared by all subjects in a batch.""" + + images: dict[str, _ImageSchema] + metadata_keys: tuple[str, ...] + points: dict[str, _AnnotationSchema] + bounding_boxes: dict[str, _AnnotationSchema] + + @classmethod + def from_subject(cls, subject: Subject) -> _SubjectSchema: + """Build a schema from the first subject in a batch.""" + return cls( + images={ + name: _ImageSchema.from_image(image) + for name, image in subject.images.items() + }, + metadata_keys=tuple(subject.metadata), + points={ + name: _AnnotationSchema.from_value(value) + for name, value in subject.points.items() + }, + bounding_boxes={ + name: _AnnotationSchema.from_value(value) + for name, value in subject.bounding_boxes.items() + }, + ) + + def validate(self, subject: Subject, *, index: int) -> None: + """Validate one subject against this schema.""" + _validate_keys( + self.images, subject.images, index=index, context="Subject images" + ) + _validate_keys( + self.metadata_keys, + subject.metadata, + index=index, + context="Subject metadata", + ) + _validate_annotations( + self.points, + subject.points, + index=index, + context="Subject points", + ) + _validate_annotations( + self.bounding_boxes, + subject.bounding_boxes, + index=index, + context="Subject bounding boxes", + ) + for name, schema in self.images.items(): + schema.validate(subject.images[name], index=index, name=name) + + +def _validate_annotations( + reference: dict[str, _AnnotationSchema], + current: dict[str, Points] | dict[str, BoundingBoxes], + *, + index: int, + context: str, +) -> None: + """Validate a named annotation store.""" + _validate_keys(reference, current, index=index, context=context) + for name, schema in reference.items(): + schema.validate(current[name], index=index, context=f"{context} {name!r}") + + +def _validate_keys( + reference: Any, + current: Any, + *, + index: int, + context: str, +) -> None: + """Validate equivalent key sets while allowing reordered keys.""" + reference_set = set(reference) + current_set = set(current) + if reference_set == current_set: + return + missing = sorted(reference_set - current_set) + unexpected = sorted(current_set - reference_set) + msg = ( + f"{context} at index {index} has incompatible keys:" + f" missing {missing}, unexpected {unexpected}" + ) + raise ValueError(msg) + + +def _normalize_dtype(dtype: Any) -> str: + """Return one comparable dtype name for Torch and NumPy dtypes.""" + return str(dtype).removeprefix("torch.") diff --git a/src/torchio/data/bboxes.py b/src/torchio/data/bboxes.py index 4228fe790..389b8ccf0 100644 --- a/src/torchio/data/bboxes.py +++ b/src/torchio/data/bboxes.py @@ -310,14 +310,23 @@ def device(self) -> torch.device: return self._data.device def to(self, *args: Any, **kwargs: Any) -> Self: - """Move bounding box data to a device and/or cast to a dtype. + """Move box coordinates and affine to a device or dtype. + + Coordinate dtype casts are applied to the box tensor. Labels + preserve their dtype and move only to the coordinate device. + The affine moves to supported devices but remains `float64`. + + Args: + *args: Positional arguments forwarded to `torch.Tensor.to`. + **kwargs: Keyword arguments forwarded to `torch.Tensor.to`. Returns: `self` (modified in-place). """ self._data = self._data.to(*args, **kwargs) if self._labels is not None: - self._labels = self._labels.to(*args, **kwargs) + self._labels = self._labels.to(device=self._data.device) + self._affine.to(*args, **kwargs) return self # --- Methods --- diff --git a/src/torchio/data/image.py b/src/torchio/data/image.py index 76837653c..000ee4366 100644 --- a/src/torchio/data/image.py +++ b/src/torchio/data/image.py @@ -642,9 +642,16 @@ def device(self) -> torch.device: return self.data.device def to(self, *args: Any, **kwargs: Any) -> Self: - """Move image data and affine to a device and/or cast to a dtype. + """Move image data, affine, and annotations to a device or dtype. Accepts the same arguments as `torch.Tensor.to()`. + Dtype casts apply to image data, point coordinates, and bounding-box + coordinates. Bounding-box labels preserve their dtype. Affines move + to supported devices but always remain `float64`. + + Args: + *args: Positional arguments forwarded to `torch.Tensor.to`. + **kwargs: Keyword arguments forwarded to `torch.Tensor.to`. Returns: `self` (modified in-place). @@ -652,6 +659,10 @@ def to(self, *args: Any, **kwargs: Any) -> Self: self._data = self.data.to(*args, **kwargs) if self._affine is not None: self._affine.to(*args, **kwargs) + for points in self._points.values(): + points.to(*args, **kwargs) + for boxes in self._bounding_boxes.values(): + boxes.to(*args, **kwargs) self._refresh_backend_from_data() return self diff --git a/src/torchio/data/points.py b/src/torchio/data/points.py index 003f0ca0e..d5afe9e5b 100644 --- a/src/torchio/data/points.py +++ b/src/torchio/data/points.py @@ -108,12 +108,20 @@ def device(self) -> torch.device: return self._data.device def to(self, *args: Any, **kwargs: Any) -> Self: - """Move point data to a device and/or cast to a dtype. + """Move point coordinates and affine to a device or dtype. + + Coordinate dtype casts are applied to the point tensor. The + affine moves to supported devices but always remains `float64`. + + Args: + *args: Positional arguments forwarded to `torch.Tensor.to`. + **kwargs: Keyword arguments forwarded to `torch.Tensor.to`. Returns: `self` (modified in-place). """ self._data = self._data.to(*args, **kwargs) + self._affine.to(*args, **kwargs) return self # --- Methods --- diff --git a/src/torchio/transforms/intensity/labels_to_image.py b/src/torchio/transforms/intensity/labels_to_image.py index a17952535..858e34155 100644 --- a/src/torchio/transforms/intensity/labels_to_image.py +++ b/src/torchio/transforms/intensity/labels_to_image.py @@ -10,7 +10,6 @@ from torch import Tensor from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ...data.image import ScalarImage from ..parameter_range import to_range from ..transform import Transform @@ -173,7 +172,7 @@ def _find_label_batch(self, batch: SubjectsBatch) -> Any: return batch.images[self.label_key] # Auto-detect first LabelMap. for _name, img_batch in batch.images.items(): - if issubclass(img_batch._image_class, LabelMap): + if img_batch.is_label: return img_batch msg = "No LabelMap found in the subject" raise KeyError(msg) diff --git a/src/torchio/transforms/intensity/mask.py b/src/torchio/transforms/intensity/mask.py index c24b07ed4..2a5858d5d 100644 --- a/src/torchio/transforms/intensity/mask.py +++ b/src/torchio/transforms/intensity/mask.py @@ -9,7 +9,6 @@ from torch import Tensor from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ..transform import IntensityTransform @@ -85,7 +84,7 @@ def _resolve_mask(self, batch: SubjectsBatch) -> Tensor: ) raise KeyError(msg) mask_batch = batch.images[key] - if not issubclass(mask_batch._image_class, LabelMap): + if not mask_batch.is_label: msg = f'Masking method "{key}" must refer to a LabelMap.' raise TypeError(msg) mask_data = mask_batch.data[0] diff --git a/src/torchio/transforms/intensity/normalize.py b/src/torchio/transforms/intensity/normalize.py index fddfae64a..5d1860783 100644 --- a/src/torchio/transforms/intensity/normalize.py +++ b/src/torchio/transforms/intensity/normalize.py @@ -12,7 +12,6 @@ from ...data.batch import ImagesBatch from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from .._statistics import compute_quantile from ..parameter_range import Choice from ..parameter_range import _ParameterRange @@ -221,7 +220,7 @@ def _get_mask( ) raise KeyError(msg) mask_batch = batch.images[key] - if not issubclass(mask_batch._image_class, LabelMap): + if not mask_batch.is_label: msg = f'Masking method "{key}" must refer to a LabelMap.' raise TypeError(msg) return mask_batch.data[0].bool() diff --git a/src/torchio/transforms/intensity/standardize.py b/src/torchio/transforms/intensity/standardize.py index 6a16b0a1b..a1f16285d 100644 --- a/src/torchio/transforms/intensity/standardize.py +++ b/src/torchio/transforms/intensity/standardize.py @@ -10,7 +10,6 @@ from ...data.batch import ImagesBatch from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ..transform import IntensityTransform @@ -166,7 +165,7 @@ def _get_mask( ) raise KeyError(msg) mask_batch = batch.images[masking_method] - if not issubclass(mask_batch._image_class, LabelMap): + if not mask_batch.is_label: msg = f'Masking method "{masking_method}" must refer to a LabelMap.' raise TypeError(msg) return mask_batch.data[0].bool() diff --git a/src/torchio/transforms/intensity/swap.py b/src/torchio/transforms/intensity/swap.py index 920987a72..5c055c940 100644 --- a/src/torchio/transforms/intensity/swap.py +++ b/src/torchio/transforms/intensity/swap.py @@ -10,7 +10,6 @@ from torch import Tensor from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ..parameter_range import to_nonneg_range from ..transform import IntensityTransform @@ -63,7 +62,7 @@ def make_params(self, batch: SubjectsBatch) -> dict[str, Any]: """Sample swap locations (per element when batched).""" # Warn if label maps are present. for _name, img_batch in batch.images.items(): - if issubclass(img_batch._image_class, LabelMap): + if img_batch.is_label: warnings.warn( "Swap is applied to a subject containing LabelMap " "images. The spatial rearrangement will make labels " diff --git a/src/torchio/transforms/label/contour.py b/src/torchio/transforms/label/contour.py index 0991329c5..348a2ecec 100644 --- a/src/torchio/transforms/label/contour.py +++ b/src/torchio/transforms/label/contour.py @@ -8,7 +8,6 @@ import torch.nn.functional as functional from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ..transform import Transform @@ -43,7 +42,7 @@ def apply_transform( ) -> SubjectsBatch: """Replace each label map with its boundary voxels.""" for _name, img_batch in batch.images.items(): - if not issubclass(img_batch._image_class, LabelMap): + if not img_batch.is_label: continue img_batch.data = _extract_contour(img_batch.data) return batch diff --git a/src/torchio/transforms/label/keep_largest.py b/src/torchio/transforms/label/keep_largest.py index 7e6891f7e..4d22821d0 100644 --- a/src/torchio/transforms/label/keep_largest.py +++ b/src/torchio/transforms/label/keep_largest.py @@ -10,7 +10,6 @@ from torch import Tensor from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ..transform import Transform @@ -67,7 +66,7 @@ def apply_transform( ) -> SubjectsBatch: """Keep only the largest connected component per label.""" for _name, img_batch in batch.images.items(): - if not issubclass(img_batch._image_class, LabelMap): + if not img_batch.is_label: continue b, c = img_batch.data.shape[:2] if c != 1: diff --git a/src/torchio/transforms/label/one_hot.py b/src/torchio/transforms/label/one_hot.py index af13a0780..3c91e31a6 100644 --- a/src/torchio/transforms/label/one_hot.py +++ b/src/torchio/transforms/label/one_hot.py @@ -7,7 +7,6 @@ import torch.nn.functional as functional from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ..transform import Transform @@ -58,7 +57,7 @@ def apply_transform( """One-hot encode each label map in the batch.""" num_classes = params["num_classes"] for _name, img_batch in batch.images.items(): - if not issubclass(img_batch._image_class, LabelMap): + if not img_batch.is_label: continue # (B, 1, I, J, K) -> (B, num_classes, I, J, K) data = img_batch.data.long() @@ -90,7 +89,7 @@ def apply_transform( params: dict[str, Any], ) -> SubjectsBatch: for _name, img_batch in batch.images.items(): - if not issubclass(img_batch._image_class, LabelMap): + if not img_batch.is_label: continue if img_batch.data.shape[1] > 1: img_batch.data = img_batch.data.argmax(dim=1, keepdim=True).float() diff --git a/src/torchio/transforms/label/remap_labels.py b/src/torchio/transforms/label/remap_labels.py index 7e4ed4370..0dbf414bc 100644 --- a/src/torchio/transforms/label/remap_labels.py +++ b/src/torchio/transforms/label/remap_labels.py @@ -5,7 +5,6 @@ from typing import Any from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ..transform import Transform @@ -49,7 +48,7 @@ def apply_transform( """Remap labels in each label map.""" remapping = params["remapping"] for _name, img_batch in batch.images.items(): - if not issubclass(img_batch._image_class, LabelMap): + if not img_batch.is_label: continue data = img_batch.data.clone() for old, new in remapping.items(): diff --git a/src/torchio/transforms/label/remove_labels.py b/src/torchio/transforms/label/remove_labels.py index 6e580513a..5d5b43e5e 100644 --- a/src/torchio/transforms/label/remove_labels.py +++ b/src/torchio/transforms/label/remove_labels.py @@ -6,7 +6,6 @@ from typing import Any from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ..transform import Transform @@ -52,7 +51,7 @@ def apply_transform( ) -> SubjectsBatch: """Set specified labels to the background value.""" for _name, img_batch in batch.images.items(): - if not issubclass(img_batch._image_class, LabelMap): + if not img_batch.is_label: continue data = img_batch.data.clone() for label in self.labels: diff --git a/src/torchio/transforms/label/sequential_labels.py b/src/torchio/transforms/label/sequential_labels.py index 8e1ed15dd..4ccaa9178 100644 --- a/src/torchio/transforms/label/sequential_labels.py +++ b/src/torchio/transforms/label/sequential_labels.py @@ -7,7 +7,6 @@ import torch from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ..transform import Transform @@ -38,7 +37,7 @@ def make_params(self, batch: SubjectsBatch) -> dict[str, Any]: """Compute the remapping from the first sample's labels.""" remappings: dict[str, dict[int, int]] = {} for name, img_batch in batch.images.items(): - if not issubclass(img_batch._image_class, LabelMap): + if not img_batch.is_label: continue unique = sorted(int(v) for v in img_batch.data[0].unique().tolist()) remappings[name] = {old: new for new, old in enumerate(unique)} diff --git a/src/torchio/transforms/lambda_transform.py b/src/torchio/transforms/lambda_transform.py index 8b54c866e..997c0fcbb 100644 --- a/src/torchio/transforms/lambda_transform.py +++ b/src/torchio/transforms/lambda_transform.py @@ -59,7 +59,7 @@ def apply_transform( ) -> SubjectsBatch: """Apply the callable to each matching image.""" for _name, img_batch in batch.images.items(): - if not self._should_apply(img_batch._image_class): + if not self._should_apply(img_batch.image_class): continue for i in range(img_batch.batch_size): img_batch.data[i] = self.function(img_batch.data[i]) diff --git a/src/torchio/transforms/spatial/_padding.py b/src/torchio/transforms/spatial/_padding.py index 228de3c05..cb98ff491 100644 --- a/src/torchio/transforms/spatial/_padding.py +++ b/src/torchio/transforms/spatial/_padding.py @@ -4,6 +4,7 @@ import warnings from typing import Literal +from typing import TypeGuard from typing import get_args import torch @@ -27,9 +28,13 @@ _STATISTIC_PADDING_MODES = "mean", "median", "minimum" +def _is_padding_mode(value: str) -> TypeGuard[PaddingMode]: + return value in _PADDING_MODES + + def parse_padding_mode(padding_mode: str) -> PaddingMode: """Validate and return a padding mode.""" - if padding_mode not in _PADDING_MODES: + if not _is_padding_mode(padding_mode): msg = f"padding_mode must be one of {_PADDING_MODES}, got {padding_mode!r}" raise ValueError(msg) return padding_mode diff --git a/src/torchio/transforms/spatial/anisotropy.py b/src/torchio/transforms/spatial/anisotropy.py index adc45ab06..f1324accc 100644 --- a/src/torchio/transforms/spatial/anisotropy.py +++ b/src/torchio/transforms/spatial/anisotropy.py @@ -9,7 +9,6 @@ from einops import rearrange from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ..parameter_range import to_nonneg_range from ..transform import Transform @@ -106,7 +105,7 @@ def apply_transform( """Downsample then upsample along the chosen axis.""" per_instance = self._is_per_instance_params(params) for _name, img_batch in batch.images.items(): - is_label = issubclass(img_batch._image_class, LabelMap) + is_label = img_batch.is_label mode = "nearest" if is_label else self.image_interpolation if per_instance: data = img_batch.data diff --git a/src/torchio/transforms/spatial/crop_or_pad.py b/src/torchio/transforms/spatial/crop_or_pad.py index e958c48ba..0ea453891 100644 --- a/src/torchio/transforms/spatial/crop_or_pad.py +++ b/src/torchio/transforms/spatial/crop_or_pad.py @@ -483,9 +483,10 @@ def _forward_lazy(self, data: Subject | Image) -> Subject | Image: if self.copy: subject = _copy.deepcopy(subject) - if torch.rand(1).item() > self.p: + if torch.rand(1).item() >= self.p: return subject.tio_default_image if is_image else subject + self._check_spatial_annotations(subject) first_image = next(iter(subject.images.values())) current_shape: TypeThreeInts = first_image.spatial_shape target_voxels = _to_voxels( diff --git a/src/torchio/transforms/spatial/resize.py b/src/torchio/transforms/spatial/resize.py index 22eef2488..a709ed04a 100644 --- a/src/torchio/transforms/spatial/resize.py +++ b/src/torchio/transforms/spatial/resize.py @@ -7,7 +7,6 @@ import torch.nn.functional as functional from ...data.batch import SubjectsBatch -from ...data.image import LabelMap from ..transform import SpatialTransform @@ -64,7 +63,7 @@ def apply_transform( """Resize each image to the target shape.""" target = list(params["target_shape"]) for _name, img_batch in batch.images.items(): - is_label = issubclass(img_batch._image_class, LabelMap) + is_label = img_batch.is_label mode = self.label_interpolation if is_label else self.image_interpolation torch_mode = "nearest" if mode == "nearest" else "trilinear" old_shape = img_batch.data.shape[2:] diff --git a/src/torchio/transforms/spatial/spatial.py b/src/torchio/transforms/spatial/spatial.py index 117c46c3c..2a4a1cec0 100644 --- a/src/torchio/transforms/spatial/spatial.py +++ b/src/torchio/transforms/spatial/spatial.py @@ -37,7 +37,6 @@ from ...data.batch import ImagesBatch from ...data.batch import SubjectsBatch from ...data.image import Image -from ...data.image import LabelMap from ...data.image import ScalarImage from ...types import TypeSpacing from ...types import TypeThreeInts @@ -1227,7 +1226,7 @@ def _resample_image_batch( """ original_data = img_batch.data original_affines = list(img_batch.affines) - is_label = issubclass(img_batch._image_class, LabelMap) + is_label = img_batch.is_label interpolation = _interpolation_for_batch( img_batch, image_interpolation=image_interpolation, @@ -2038,7 +2037,7 @@ def _batch_fill_value( default_pad_label: float, ) -> float | Tensor: """Compute a single fill value for the whole image batch.""" - if issubclass(img_batch._image_class, LabelMap): + if img_batch.is_label: return float(default_pad_label) if isinstance(default_pad_value, Number): @@ -2419,7 +2418,7 @@ def _interpolation_for_batch( label_interpolation: TypeLabelInterpolation, ) -> str: """Choose the interpolation mode based on the image class.""" - if issubclass(img_batch._image_class, LabelMap): + if img_batch.is_label: return label_interpolation return image_interpolation diff --git a/src/torchio/transforms/transform.py b/src/torchio/transforms/transform.py index d7d9f3bc0..2fbb4882f 100644 --- a/src/torchio/transforms/transform.py +++ b/src/torchio/transforms/transform.py @@ -21,8 +21,10 @@ from ..data.batch import ImagesBatch from ..data.batch import SubjectsBatch +from ..data.bboxes import BoundingBoxes from ..data.image import Image from ..data.image import ScalarImage +from ..data.points import Points from ..data.subject import Subject @@ -66,6 +68,40 @@ def _copy_optional_list(value: list[str] | None) -> list[str] | None: return None if value is None else list(value) +def _image_has_annotations(image: Image) -> bool: + return bool(image.points or image.bounding_boxes) + + +def _subject_has_annotations(subject: Subject) -> bool: + return bool( + subject.points + or subject.bounding_boxes + or any(_image_has_annotations(image) for image in subject.images.values()) + ) + + +def _value_has_annotations(value: Any) -> bool: + if isinstance(value, (Points, BoundingBoxes)): + return True + if isinstance(value, Image): + return _image_has_annotations(value) + return False + + +def _data_has_annotations(data: Any) -> bool: + match data: + case SubjectsBatch() | ImagesBatch(): + return data.has_annotations + case Subject(): + return _subject_has_annotations(data) + case Image(): + return _image_has_annotations(data) + case dict(): + return any(_value_has_annotations(value) for value in data.values()) + case _: + return False + + class Transform(nn.Module): """Abstract class for all TorchIO transforms. @@ -227,6 +263,8 @@ def forward(self, data: Any) -> Any: if not self._per_instance_p_active(batch) and torch.rand(1).item() >= self.p: return unwrap(batch) params = self.make_params(batch) + if not _all_elements_gated_out(params): + self._check_spatial_annotations(batch) batch = self.apply_transform(batch, params) # Record history on the batch, unless every element was gated out by # per-element probability: that is an exact no-op, and recording it @@ -253,6 +291,16 @@ def forward(self, data: Any) -> Any: result.applied_transforms = list(batch.applied_transforms) return result + def _check_spatial_annotations(self, data: Any) -> None: + """Reject spatial transforms that would leave stale annotations.""" + if isinstance(self, SpatialTransform) and _data_has_annotations(data): + msg = ( + "Spatial transforms do not yet support Points or BoundingBoxes." + " Remove the annotations before applying the transform, or apply" + " an annotation-aware spatial operation." + ) + raise NotImplementedError(msg) + @property def supports_per_instance_params(self) -> bool: """Whether this transform can sample parameters per batch element. @@ -668,9 +716,9 @@ def _unwrap_dict(batch: SubjectsBatch, keys: list[str]) -> dict[str, Any]: class SpatialTransform(Transform): """Base for transforms that modify spatial geometry. - Spatial transforms apply to all images (ScalarImage and LabelMap), - and also transform any Points and BoundingBoxes attached to the - Subject. + Spatial transforms apply to all images (ScalarImage and LabelMap). + Coordinate updates for `Points` and `BoundingBoxes` are not yet + supported, so annotated inputs raise before mutation. """ @@ -684,7 +732,9 @@ class IntensityTransform(Transform): def _get_images(self, batch: SubjectsBatch) -> dict[str, ImagesBatch]: """Filter to ScalarImage batches only, then apply include/exclude.""" images = { - k: v for k, v in batch.images.items() if v._image_class is ScalarImage + k: v + for k, v in batch.images.items() + if issubclass(v.image_class, ScalarImage) } if self.include is not None: images = {k: v for k, v in images.items() if k in self.include} diff --git a/tests/test_batch.py b/tests/test_batch.py index d1a3b3f2f..e9331cdc0 100644 --- a/tests/test_batch.py +++ b/tests/test_batch.py @@ -2,6 +2,8 @@ from __future__ import annotations +import nibabel as nib +import numpy as np import pytest import torch @@ -16,6 +18,94 @@ def test_from_images(self) -> None: batch = ImagesBatch.from_images(images) assert batch.data.shape == (4, 1, 8, 8, 8) + def test_from_tensor(self) -> None: + data = torch.rand(3, 1, 4, 5, 6) + + batch = ImagesBatch.from_tensor(data) + + assert batch.data is data + assert batch.batch_size == 3 + assert all(affine.spacing == (1.0, 1.0, 1.0) for affine in batch.affines) + + def test_from_tensor_default_affines_follow_data_device(self) -> None: + batch = ImagesBatch.from_tensor(torch.empty(2, 1, 2, 3, 4, device="meta")) + + assert all(affine.device.type == "meta" for affine in batch.affines) + + def test_image_class_properties(self) -> None: + scalar = ImagesBatch.from_tensor( + torch.rand(2, 1, 4, 4, 4), + image_class=tio.ScalarImage, + ) + label = ImagesBatch.from_tensor( + torch.zeros(2, 1, 4, 4, 4), + image_class=tio.LabelMap, + ) + + assert scalar.image_class is tio.ScalarImage + assert scalar.is_label is False + assert label.image_class is tio.LabelMap + assert label.is_label is True + + def test_payload_round_trip(self) -> None: + images = [ + tio.ScalarImage( + torch.rand(1, 4, 4, 4), + protocol=f"protocol-{index}", + points={"landmarks": tio.Points(torch.rand(index + 1, 3))}, + bounding_boxes={ + "tumors": tio.BoundingBoxes( + torch.rand(index + 1, 6), + format=tio.BoundingBoxFormat.IJKIJK, + ) + }, + ) + for index in range(2) + ] + + restored = ImagesBatch.from_images(images).unbatch() + + assert [image.protocol for image in restored] == [ + "protocol-0", + "protocol-1", + ] + assert restored[0].points["landmarks"].num_points == 1 + assert restored[1].bounding_boxes["tumors"].num_boxes == 2 + + def test_per_image_history_round_trip(self) -> None: + images = [ + tio.ScalarImage(torch.rand(1, 4, 4, 4)), + tio.ScalarImage(torch.rand(1, 4, 4, 4)), + ] + images[0].applied_transforms = [tio.AppliedTransform("Flip", {"axes": (0,)})] + images[1].applied_transforms = [tio.AppliedTransform("Flip", {"axes": (1,)})] + + restored = ImagesBatch.from_images(images).unbatch() + + assert restored[0].applied_transforms[0].params["axes"] == (0,) + assert restored[1].applied_transforms[0].params["axes"] == (1,) + + def test_custom_image_subclass_round_trip(self) -> None: + class CustomScalarImage(tio.ScalarImage): + pass + + images = [ + CustomScalarImage(torch.rand(1, 4, 4, 4), sequence="custom") + for _ in range(2) + ] + + restored = ImagesBatch.from_images(images).unbatch() + + assert all(type(image) is CustomScalarImage for image in restored) + assert all(image.sequence == "custom" for image in restored) + + def test_template_does_not_share_affine(self) -> None: + image = tio.ScalarImage(torch.rand(1, 4, 4, 4)) + + batch = ImagesBatch.from_images([image]) + + assert batch._prototypes[0].affine is not image.affine + def test_batch_size(self) -> None: batch = ImagesBatch( data=torch.rand(4, 1, 8, 8, 8), @@ -142,6 +232,122 @@ def test_metadata_preserved(self) -> None: assert batch.metadata["age"] == [42, 43, 44] assert batch.metadata["name"] == ["sub_0", "sub_1", "sub_2"] + def test_reordered_schema_is_accepted(self) -> None: + first = tio.Subject( + t1=tio.ScalarImage(torch.rand(1, 4, 4, 4)), + seg=tio.LabelMap(torch.zeros(1, 4, 4, 4)), + age=42, + name="first", + ) + second = tio.Subject( + seg=tio.LabelMap(torch.zeros(1, 4, 4, 4)), + t1=tio.ScalarImage(torch.rand(1, 4, 4, 4)), + name="second", + age=43, + ) + + batch = SubjectsBatch.from_subjects([first, second]) + + assert list(batch.images) == ["t1", "seg"] + assert list(batch.metadata) == ["age", "name"] + + def test_loaded_and_lazy_equivalent_dtypes_are_accepted(self) -> None: + loaded = tio.ScalarImage(torch.zeros(1, 4, 4, 4, dtype=torch.float32)) + lazy = tio.ScalarImage( + nib.Nifti1Image( + np.zeros((4, 4, 4), dtype=np.float32), + np.eye(4), + ) + ) + + batch = SubjectsBatch.from_subjects( + [tio.Subject(image=loaded), tio.Subject(image=lazy)] + ) + + assert batch.image.data.dtype == torch.float32 + + @pytest.mark.parametrize( + ("first", "second", "message"), + [ + ( + tio.Subject(t1=tio.ScalarImage(torch.rand(1, 4, 4, 4))), + tio.Subject( + t1=tio.ScalarImage(torch.rand(1, 4, 4, 4)), + t2=tio.ScalarImage(torch.rand(1, 4, 4, 4)), + ), + "Subject images.*unexpected.*t2", + ), + ( + tio.Subject( + t1=tio.ScalarImage(torch.rand(1, 4, 4, 4)), + age=42, + ), + tio.Subject(t1=tio.ScalarImage(torch.rand(1, 4, 4, 4))), + "Subject metadata.*missing.*age", + ), + ( + tio.Subject( + t1=tio.ScalarImage(torch.rand(1, 4, 4, 4)), + landmarks=tio.Points(torch.rand(2, 3)), + ), + tio.Subject(t1=tio.ScalarImage(torch.rand(1, 4, 4, 4))), + "Subject points.*missing.*landmarks", + ), + ], + ) + def test_incompatible_schema_raises( + self, + first: tio.Subject, + second: tio.Subject, + message: str, + ) -> None: + with pytest.raises(ValueError, match=message): + SubjectsBatch.from_subjects([first, second]) + + def test_metadata_only_batch(self) -> None: + batch = SubjectsBatch.from_subjects([tio.Subject(age=42), tio.Subject(age=43)]) + + assert batch.batch_size == 2 + assert batch.device.type == "cpu" + assert [subject.age for subject in batch.unbatch()] == [42, 43] + + def test_annotation_only_batch(self) -> None: + subjects = [ + tio.Subject(landmarks=tio.Points(torch.rand(index + 1, 3))) + for index in range(2) + ] + + restored = SubjectsBatch.from_subjects(subjects).unbatch() + + assert restored[0].landmarks.num_points == 1 + assert restored[1].landmarks.num_points == 2 + + def test_subject_annotations_round_trip(self) -> None: + subject = tio.Subject( + t1=tio.ScalarImage(torch.rand(1, 4, 4, 4)), + landmarks=tio.Points(torch.rand(2, 3), metadata={"source": "manual"}), + tumors=tio.BoundingBoxes( + torch.rand(3, 6), + format=tio.BoundingBoxFormat.IJKWHD, + metadata={"reader": "test"}, + ), + ) + + restored = SubjectsBatch.from_subjects([subject]).unbatch()[0] + + assert restored.landmarks.metadata == {"source": "manual"} + assert restored.tumors.metadata == {"reader": "test"} + assert restored.tumors.format == tio.BoundingBoxFormat.IJKWHD + + def test_metadata_remains_mutable(self) -> None: + batch = SubjectsBatch.from_subjects([tio.Subject(age=42), tio.Subject(age=43)]) + + batch.metadata["age"][0] = 50 + batch.metadata["site"] = ["A", "B"] + + assert batch.unbatch()[0].age == 50 + assert batch.unbatch()[1].site == "B" + class TestBatchTransforms: def test_flip_images_batch(self) -> None: @@ -196,6 +402,59 @@ def test_batch_copy_preserves_original(self) -> None: # Original should be unchanged (copy=True default) torch.testing.assert_close(batch.data, original) + def test_intensity_transform_preserves_annotations(self) -> None: + subject = tio.Subject( + t1=tio.ScalarImage( + torch.rand(1, 4, 4, 4) + 0.1, + points={"image_landmarks": tio.Points(torch.rand(2, 3))}, + ), + landmarks=tio.Points(torch.rand(3, 3)), + ) + + result = tio.Gamma(log_gamma=0.2)(subject) + + assert set(result.points) == {"landmarks"} + assert set(result.t1.points) == {"image_landmarks"} + + @pytest.mark.parametrize( + "data", + [ + tio.Subject( + t1=tio.ScalarImage(torch.rand(1, 4, 4, 4)), + landmarks=tio.Points(torch.rand(2, 3)), + ), + { + "t1": torch.rand(1, 4, 4, 4), + "landmarks": tio.Points(torch.rand(2, 3)), + }, + ], + ) + def test_spatial_transform_rejects_annotations(self, data: object) -> None: + with pytest.raises(NotImplementedError, match="annotations"): + tio.Flip(axes=(0,))(data) + + @pytest.mark.parametrize( + "transform", + [ + tio.Flip(axes=(0,), p=0), + tio.CropOrPad(2, p=0), + ], + ) + def test_skipped_spatial_transform_allows_annotations( + self, + transform: tio.Transform, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + subject = tio.Subject( + image=tio.ScalarImage(torch.rand(1, 4, 4, 4)), + landmarks=tio.Points(torch.rand(2, 3)), + ) + monkeypatch.setattr(torch, "rand", lambda *args, **kwargs: torch.zeros(1)) + + result = transform(subject) + + assert set(result.points) == {"landmarks"} + # ── Coverage gap tests ─────────────────────────────────────────────── diff --git a/tests/test_device.py b/tests/test_device.py index cfa701d41..736f7098a 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -14,6 +14,27 @@ HAS_MPS = torch.backends.mps.is_available() +class TestAnnotationTo: + def test_points_to_moves_affine(self) -> None: + points = Points(torch.rand(2, 3)) + + points.to("meta") + + assert points.device.type == "meta" + assert points.affine.device.type == "meta" + + def test_bounding_boxes_to_moves_affine(self) -> None: + boxes = BoundingBoxes( + torch.rand(2, 6), + format=BoundingBoxFormat.IJKIJK, + ) + + boxes.to("meta") + + assert boxes.device.type == "meta" + assert boxes.affine.device.type == "meta" + + class TestImageTo: def test_to_returns_self(self) -> None: image = tio.ScalarImage(torch.rand(1, 4, 4, 4)) @@ -29,6 +50,26 @@ def test_to_dtype(self) -> None: result = image.to(torch.float16) assert result.data.dtype == torch.float16 + def test_moves_image_annotations(self) -> None: + image = tio.ScalarImage( + torch.rand(1, 4, 4, 4), + points={"landmarks": Points(torch.rand(2, 3))}, + bounding_boxes={ + "tumors": BoundingBoxes( + torch.rand(2, 6), + format=BoundingBoxFormat.IJKIJK, + labels=torch.tensor([1, 2]), + ) + }, + ) + + result = image.to(torch.float64) + + assert result.points["landmarks"].data.dtype == torch.float64 + assert result.bounding_boxes["tumors"].data.dtype == torch.float64 + assert result.bounding_boxes["tumors"].labels is not None + assert result.bounding_boxes["tumors"].labels.dtype == torch.int64 + @pytest.mark.skipif(not HAS_CUDA, reason="No CUDA") def test_to_cuda(self) -> None: image = tio.ScalarImage(torch.rand(1, 4, 4, 4)) diff --git a/tests/test_transforms_base.py b/tests/test_transforms_base.py index 8907b2bce..7965224d5 100644 --- a/tests/test_transforms_base.py +++ b/tests/test_transforms_base.py @@ -32,6 +32,14 @@ def _make_subject() -> tio.Subject: ) +def _make_image_subject() -> tio.Subject: + return tio.Subject( + t1=tio.ScalarImage(torch.rand(1, 8, 8, 8)), + seg=tio.LabelMap(torch.randint(0, 3, (1, 8, 8, 8))), + age=42, + ) + + class _IdentityTransform(tio.Transform): """Transform that does nothing (for testing the base flow).""" @@ -232,7 +240,7 @@ def test_scalar_image_modified(self) -> None: class TestSpatialTransform: def test_all_images_modified(self) -> None: - subject = _make_subject() + subject = _make_image_subject() result = _FlipSpatial()(subject) # Both t1 and seg should be flipped assert result.t1.data.shape == (1, 8, 8, 8) @@ -366,7 +374,7 @@ def test_add_not_implemented_for_non_transform(self) -> None: tio.Flip(axes=(0,)) + 42 # type: ignore[operator] def test_add_produces_working_pipeline(self) -> None: - subject = _make_subject() + subject = _make_image_subject() pipeline = tio.Flip(axes=(0,)) + tio.Noise(std=0.01) result = pipeline(subject) assert result.t1.shape == subject.t1.shape @@ -393,7 +401,7 @@ def test_or_not_implemented_for_non_transform(self) -> None: tio.Flip(axes=(0,)) | "bad" # type: ignore[operator] def test_or_produces_working_pipeline(self) -> None: - subject = _make_subject() + subject = _make_image_subject() pipeline = tio.Flip(axes=(0,)) | tio.Noise(std=0.01) result = pipeline(subject) assert result.t1.shape == subject.t1.shape