Finally...

Signed-off-by: Fabian Klopfer <[email protected]>

Author: SomeUserName1

monai/transforms/__init__.py

@@ -531,6 +531,8 @@
     RandIdentity,
     RandImageFilter,
     RandLambda,
+    RandTorchIO,
+    RandTorchVision,
     RemoveRepeatedChannel,
     RepeatChannel,
     SimulateDelay,
@@ -621,6 +623,9 @@
     RandLambdad,
     RandLambdaD,
     RandLambdaDict,
+    RandTorchIOd,
+    RandTorchIOD,
+    RandTorchIODict,
     RandTorchVisiond,
     RandTorchVisionD,
     RandTorchVisionDict,

monai/transforms/utility/array.py

@@ -18,10 +18,10 @@
 import sys
 import time
 import warnings
-from collections.abc import Mapping, Sequence
+from collections.abc import Hashable, Mapping, Sequence
 from copy import deepcopy
 from functools import partial
-from typing import Any, Callable
+from typing import Any, Callable, Union
 
 import numpy as np
 import torch
@@ -1227,7 +1227,7 @@ def __init__(self, name: str, *args, **kwargs) -> None:
         transform, _ = optional_import("torchio.transforms", "0.18.0", min_version, name=name)
         self.trans = transform(*args, **kwargs)
 
-    def __call__(self, img: NdarrayOrTensor):
+    def __call__(self, img: Union[NdarrayOrTensor, Mapping[Hashable, NdarrayOrTensor]]):
         """
         Args:
             img: an instance of torchio.Subject, torchio.Image, numpy.ndarray, torch.Tensor, SimpleITK.Image,
@@ -1259,7 +1259,7 @@ def __init__(self, name: str, *args, **kwargs) -> None:
         transform, _ = optional_import("torchio.transforms", "0.18.0", min_version, name=name)
         self.trans = transform(*args, **kwargs)
 
-    def __call__(self, img: NdarrayOrTensor):
+    def __call__(self, img: Union[NdarrayOrTensor, Mapping[Hashable, NdarrayOrTensor]]):
         """
         Args:
             img: an instance of torchio.Subject, torchio.Image, numpy.ndarray, torch.Tensor, SimpleITK.Image,

monai/transforms/utility/dictionary.py

@@ -1466,7 +1466,6 @@ def __init__(self, keys: KeysCollection, name: str, allow_missing_keys: bool = F
             keys: keys of the corresponding items to be transformed.
                 See also: :py:class:`monai.transforms.compose.MapTransform`
             name: The transform name in TorchIO package.
-            apply_same_transform: whether to apply the same transform for all the items specified by `keys`.
             allow_missing_keys: don't raise exception if key is missing.
             args: parameters for the TorchIO transform.
             kwargs: parameters for the TorchIO transform.
@@ -1478,11 +1477,8 @@ def __init__(self, keys: KeysCollection, name: str, allow_missing_keys: bool = F
 
         self.trans = TorchIO(name, *args, **kwargs)
 
-    def __call__(self, data: Mapping[Hashable, NdarrayOrTensor]) -> dict[Hashable, NdarrayOrTensor]:
-        d = dict(data)
-        for key in self.key_iterator(d):
-            d[key] = self.trans(d[key])
-        return d
+    def __call__(self, data: Mapping[Hashable, NdarrayOrTensor]) -> Mapping[Hashable, NdarrayOrTensor]:
+        return dict(self.trans(data))
 
 
 class RandTorchIOd(MapTransform, RandomizableTrait):
@@ -1499,7 +1495,6 @@ def __init__(self, keys: KeysCollection, name: str, allow_missing_keys: bool = F
             keys: keys of the corresponding items to be transformed.
                 See also: :py:class:`monai.transforms.compose.MapTransform`
             name: The transform name in TorchIO package.
-            apply_same_transform: whether to apply the same transform for all the items specified by `keys`.
             allow_missing_keys: don't raise exception if key is missing.
             args: parameters for the TorchIO transform.
             kwargs: parameters for the TorchIO transform.
@@ -1511,12 +1506,8 @@ def __init__(self, keys: KeysCollection, name: str, allow_missing_keys: bool = F
 
         self.trans = TorchIO(name, *args, **kwargs)
 
-    def __call__(self, data: Mapping[Hashable, NdarrayOrTensor]) -> dict[Hashable, NdarrayOrTensor]:
-        d = dict(data)
-        for key in self.key_iterator(d):
-            d[key] = self.trans(d[key])
-        return d
-
+    def __call__(self, data: Mapping[Hashable, NdarrayOrTensor]) -> Mapping[Hashable, NdarrayOrTensor]:
+        return dict(self.trans(data))
 
 
 class MapLabelValued(MapTransform):

tests/test_rand_torchio.py

@@ -0,0 +1,54 @@
+# Copyright (c) MONAI Consortium
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#     http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import unittest
+from unittest import skipUnless
+
+import numpy as np
+import torch
+from parameterized import parameterized
+
+from monai.transforms import RandTorchIO
+from monai.utils import optional_import, set_determinism
+
+_, has_torchio = optional_import("torchio")
+
+TEST_DIMS = [3, 128, 160, 160]
+TESTS = [
+    [{"name": "RandomAffine"}, torch.rand(TEST_DIMS)],
+    [{"name": "RandomElasticDeformation"}, torch.rand(TEST_DIMS)],
+    [{"name": "RandomAnisotropy"}, torch.rand(TEST_DIMS)],
+    [{"name": "RandomMotion"}, torch.rand(TEST_DIMS)],
+    [{"name": "RandomGhosting"}, torch.rand(TEST_DIMS)],
+    [{"name": "RandomSpike"}, torch.rand(TEST_DIMS)],
+    [{"name": "RandomBiasField"}, torch.rand(TEST_DIMS)],
+    [{"name": "RandomBlur"}, torch.rand(TEST_DIMS)],
+    [{"name": "RandomNoise"}, torch.rand(TEST_DIMS)],
+    [{"name": "RandomSwap"}, torch.rand(TEST_DIMS)],
+    [{"name": "RandomGamma"}, torch.rand(TEST_DIMS)],
+]
+
+
+@skipUnless(has_torchio, "Requires torchio")
+class TestRandTorchIO(unittest.TestCase):
+
+    @parameterized.expand(TESTS)
+    def test_value(self, input_param, input_data):
+        set_determinism(seed=0)
+        result = RandTorchIO(**input_param)(input_data)
+        self.assertIsNotNone(result)
+        self.assertFalse(np.array_equal(result.numpy(), input_data.numpy()), f"{input_param} failed")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_rand_torchiod.py

@@ -0,0 +1,44 @@
+# Copyright (c) MONAI Consortium
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#     http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import unittest
+from unittest import skipUnless
+
+import numpy as np
+import torch
+from parameterized import parameterized
+
+from monai.transforms import RandTorchIOd
+from monai.utils import optional_import, set_determinism
+from tests.utils import assert_allclose
+
+_, has_torchio = optional_import("torchio")
+
+TEST_DIMS = [3, 128, 160, 160]
+TEST_TENSOR = torch.rand(TEST_DIMS)
+TEST_PARAMS = [[{"keys": ["img1", "img2"], "name": "RandomAffine"}, {"img1": TEST_TENSOR, "img2": TEST_TENSOR}]]
+
+
+@skipUnless(has_torchio, "Requires torchio")
+class TestRandTorchIOd(unittest.TestCase):
+
+    @parameterized.expand(TEST_PARAMS)
+    def test_random_transform(self, input_param, input_data):
+        set_determinism(seed=0)
+        result = RandTorchIOd(**input_param)(input_data)
+        self.assertFalse(np.allclose(input_data["img1"], result["img1"], atol=1e-6, rtol=1e-6))
+        assert_allclose(result["img1"], result["img2"], atol=1e-4, rtol=1e-4)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_torchio.py

@@ -19,34 +19,19 @@
 from parameterized import parameterized
 
 from monai.transforms import TorchIO
-from monai.utils import optional_import, set_determinism
+from monai.utils import optional_import
 
 _, has_torchio = optional_import("torchio")
 
 TEST_DIMS = [3, 128, 160, 160]
-TESTS = [
-    [{"name": "RescaleIntensity"}, torch.rand(TEST_DIMS)],
-    [{"name": "ZNormalization"}, torch.rand(TEST_DIMS)],
-    [{"name": "RandomAffine"}, torch.rand(TEST_DIMS)],
-    [{"name": "RandomElasticDeformation"}, torch.rand(TEST_DIMS)],
-    [{"name": "RandomAnisotropy"}, torch.rand(TEST_DIMS)],
-    [{"name": "RandomMotion"}, torch.rand(TEST_DIMS)],
-    [{"name": "RandomGhosting"}, torch.rand(TEST_DIMS)],
-    [{"name": "RandomSpike"}, torch.rand(TEST_DIMS)],
-    [{"name": "RandomBiasField"}, torch.rand(TEST_DIMS)],
-    [{"name": "RandomBlur"}, torch.rand(TEST_DIMS)],
-    [{"name": "RandomNoise"}, torch.rand(TEST_DIMS)],
-    [{"name": "RandomSwap"}, torch.rand(TEST_DIMS)],
-    [{"name": "RandomGamma"}, torch.rand(TEST_DIMS)],
-]
+TESTS = [[{"name": "RescaleIntensity"}, torch.rand(TEST_DIMS)], [{"name": "ZNormalization"}, torch.rand(TEST_DIMS)]]
 
 
 @skipUnless(has_torchio, "Requires torchio")
 class TestTorchIO(unittest.TestCase):
 
     @parameterized.expand(TESTS)
     def test_value(self, input_param, input_data):
-        set_determinism(seed=0)
         result = TorchIO(**input_param)(input_data)
         self.assertIsNotNone(result)
         self.assertFalse(np.array_equal(result.numpy(), input_data.numpy()), f"{input_param} failed")

tests/test_torchiod.py

@@ -18,37 +18,29 @@
 from parameterized import parameterized
 
 from monai.transforms import TorchIOd
-from monai.utils import optional_import, set_determinism
+from monai.utils import optional_import
 from tests.utils import assert_allclose
 
 _, has_torchio = optional_import("torchio")
 
 TEST_DIMS = [3, 128, 160, 160]
 TEST_TENSOR = torch.rand(TEST_DIMS)
-TEST1 = [
+TEST_PARAMS = [
     [
         {"keys": "img", "name": "RescaleIntensity", "out_min_max": (0, 42)},
         {"img": TEST_TENSOR},
         ((TEST_TENSOR - TEST_TENSOR.min()) / (TEST_TENSOR.max() - TEST_TENSOR.min())) * 42,
     ]
 ]
-TEST2 = [[{"keys": ["img1", "img2"], "name": "RandomAffine"}, {"img1": TEST_TENSOR, "img2": TEST_TENSOR}]]
 
 
 @skipUnless(has_torchio, "Requires torchio")
 class TestTorchIOd(unittest.TestCase):
 
-    @parameterized.expand(TEST1)
+    @parameterized.expand(TEST_PARAMS)
     def test_value(self, input_param, input_data, expected_value):
-        set_determinism(seed=0)
         result = TorchIOd(**input_param)(input_data)
-        assert_allclose(result["img"], expected_value, atol=1e-4, rtol=1e-4, type_test=False)
-
-    @parameterized.expand(TEST2)
-    def test_random_transform(self, input_param, input_data):
-        set_determinism(seed=0)
-        result = TorchIOd(**input_param)(input_data)
-        assert_allclose(result["img1"], result["img2"], atol=1e-4, rtol=1e-4, type_test=False)
+        assert_allclose(result["img"], expected_value, atol=1e-4, rtol=1e-4)
 
 
 if __name__ == "__main__":