Skip to content

MonaiAdapter

Bases: Transform

Wrap a MONAI transform for use in TorchIO pipelines.

Both dictionary transforms (subclasses of MONAI's MapTransform, e.g., NormalizeIntensityd) and array transforms (e.g., NormalizeIntensity) are supported.

Dictionary transforms operate on the full subject dictionary: only the keys specified in the MONAI transform are modified.

Array transforms are applied to each ScalarImage in the subject individually, respecting the include / exclude parameters.

Parameters:

Name Type Description Default
monai_transform Callable

A MONAI transform or any callable. Requires MONAI to be installed: pip install torchio[monai].

 required
**kwargs Any

See Transform for additional keyword arguments.

 {}

Examples:

>>> import torchio as tio
>>> from monai.transforms import NormalizeIntensity
>>> # Array transform: applied to each ScalarImage
>>> adapter = tio.MonaiAdapter(NormalizeIntensity())
>>> result = adapter(subject)
>>> # Inside a Compose pipeline
>>> pipeline = tio.Compose([
...     tio.MonaiAdapter(NormalizeIntensity()),
...     tio.Noise(std=0.1),
... ])
Note

MonaiAdapter does not record itself in the subject's transform history, because MONAI transform objects are not serializable.

Source code in src/torchio/transforms/monai_adapter.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class MonaiAdapter(Transform):
    """Wrap a MONAI transform for use in TorchIO pipelines.

    Both **dictionary transforms** (subclasses of MONAI's
    `MapTransform`, e.g., `NormalizeIntensityd`) and **array
    transforms** (e.g., `NormalizeIntensity`) are supported.

    Dictionary transforms operate on the full subject dictionary:
    only the keys specified in the MONAI transform are modified.

    Array transforms are applied to each
    [`ScalarImage`][torchio.ScalarImage] in the subject individually,
    respecting the `include` / `exclude` parameters.

    Args:
        monai_transform: A MONAI transform or any callable. Requires
            MONAI to be installed: `pip install torchio[monai]`.
        **kwargs: See [`Transform`][torchio.Transform] for additional
            keyword arguments.

    Examples:
        >>> import torchio as tio
        >>> from monai.transforms import NormalizeIntensity
        >>> # Array transform: applied to each ScalarImage
        >>> adapter = tio.MonaiAdapter(NormalizeIntensity())
        >>> result = adapter(subject)
        >>> # Inside a Compose pipeline
        >>> pipeline = tio.Compose([
        ...     tio.MonaiAdapter(NormalizeIntensity()),
        ...     tio.Noise(std=0.1),
        ... ])

    Note:
        `MonaiAdapter` does **not** record itself in the subject's
        transform history, because MONAI transform objects are not
        serializable.
    """

    def __init__(self, monai_transform: Callable, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        if not callable(monai_transform):
            msg = (
                "monai_transform must be callable, "
                f"got {type(monai_transform).__name__}"
            )
            raise TypeError(msg)
        self.monai_transform = monai_transform

    def forward(self, data):
        """Apply without recording history (MONAI transforms are opaque)."""
        batch, unwrap = self._wrap(data)
        if self.copy:
            batch = _copy.deepcopy(batch)
        if torch.rand(1).item() > self.p:
            return unwrap(batch)
        # MONAI transforms operate per-subject
        monai = get_monai()
        subjects = batch.unbatch()
        for subject in subjects:
            is_dict = isinstance(
                self.monai_transform,
                monai.transforms.MapTransform,
            )
            if is_dict:
                _apply_dict_transform(subject, self.monai_transform, monai)
            else:
                images = self._get_subject_images(subject)
                _apply_array_transform(images, self.monai_transform, monai)
        from ..data.batch import SubjectsBatch

        result = SubjectsBatch.from_subjects(subjects)
        result.adopt_history(batch, subjects)
        return unwrap(result)

    def apply_transform(self, batch: Any, params: dict[str, Any]) -> Any:
        # Not used: MonaiAdapter overrides forward directly
        return batch

    def _get_subject_images(self, subject: Subject) -> dict[str, Image]:
        """Filter to ScalarImage, then apply include/exclude."""
        images: dict[str, Image] = {
            k: v for k, v in subject.images.items() if isinstance(v, ScalarImage)
        }
        if self.include is not None:
            images = {k: v for k, v in images.items() if k in self.include}
        if self.exclude is not None:
            images = {k: v for k, v in images.items() if k not in self.exclude}
        return images

supports_per_instance_params property

Whether this transform can sample parameters per batch element.

Defaults to False. Transforms that implement per-instance parameter sampling override this to return True. When False, the transform always uses batch-shared parameters regardless of the per_instance flag, preserving the legacy behavior.

supports_per_instance_p property

Whether this transform can gate each batch element independently.

Defaults to False. Shape-preserving transforms that implement per-element probability override this to return True. Shape-changing transforms must leave it False because masked and unmasked elements would have incompatible shapes.

invertible property

Whether this transform can be inverted.

make_params(batch)

Sample random parameters for this transform.

Override in subclasses that have random behavior.

Parameters:

Name Type Description Default
batch SubjectsBatch

A SubjectsBatch.

 required

Returns:

Type Description
dict[str, Any]

Dict of sampled parameters.
Source code in src/torchio/transforms/transform.py 
382
383
384
385
386
387
388
389
390
391
392
393
def make_params(self, batch: SubjectsBatch) -> dict[str, Any]:
    """Sample random parameters for this transform.

    Override in subclasses that have random behavior.

    Args:
        batch: A `SubjectsBatch`.

    Returns:
        Dict of sampled parameters.
    """
    return {}

inverse(params)

Return a transform that undoes this one.

Override in invertible subclasses. The returned transform, when applied, reverses the effect of the forward pass with the given parameters.

Parameters:

Name Type Description Default
params dict[str, Any]

The parameters recorded in the forward pass.

 required

Returns:

Type Description
Transform

A new Transform instance that inverts this one.
Source code in src/torchio/transforms/transform.py 
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
def inverse(self, params: dict[str, Any]) -> Transform:
    """Return a transform that undoes this one.

    Override in invertible subclasses. The returned transform,
    when applied, reverses the effect of the forward pass with
    the given parameters.

    Args:
        params: The parameters recorded in the forward pass.

    Returns:
        A new `Transform` instance that inverts this one.
    """
    msg = f"{type(self).__name__} is not invertible"
    raise NotImplementedError(msg)

to_hydra()

Export as a Hydra-compatible config dict.

Returns a dict with _target_ set to the fully qualified class name and only non-default field values included.

Returns:

Type Description
dict[str, Any]

Dict suitable for hydra.utils.instantiate().
Source code in src/torchio/transforms/transform.py 
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
def to_hydra(self) -> dict[str, Any]:
    """Export as a Hydra-compatible config dict.

    Returns a dict with `_target_` set to the fully qualified
    class name and only non-default field values included.

    Returns:
        Dict suitable for `hydra.utils.instantiate()`.
    """
    from .parameter_range import _ParameterRange

    cls = type(self)
    target = f"torchio.{cls.__qualname__}"
    cfg: dict[str, Any] = {"_target_": target}

    for name, default in _collect_init_params(cls).items():
        value = getattr(self, name, default)
        if isinstance(value, _ParameterRange):
            if value._original == default:
                continue
            value = _hydra_value(value._original)
        elif value == default:
            continue
        else:
            value = _hydra_value(value)
        cfg[name] = value
    return cfg

forward(data)

Apply without recording history (MONAI transforms are opaque).

Source code in src/torchio/transforms/monai_adapter.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def forward(self, data):
    """Apply without recording history (MONAI transforms are opaque)."""
    batch, unwrap = self._wrap(data)
    if self.copy:
        batch = _copy.deepcopy(batch)
    if torch.rand(1).item() > self.p:
        return unwrap(batch)
    # MONAI transforms operate per-subject
    monai = get_monai()
    subjects = batch.unbatch()
    for subject in subjects:
        is_dict = isinstance(
            self.monai_transform,
            monai.transforms.MapTransform,
        )
        if is_dict:
            _apply_dict_transform(subject, self.monai_transform, monai)
        else:
            images = self._get_subject_images(subject)
            _apply_array_transform(images, self.monai_transform, monai)
    from ..data.batch import SubjectsBatch

    result = SubjectsBatch.from_subjects(subjects)
    result.adopt_history(batch, subjects)
    return unwrap(result)