4636 4637 backward compatible types

monai/data/meta_tensor.py

@@ -15,13 +15,15 @@
 from copy import deepcopy
 from typing import Any, Sequence
 
+import numpy as np
 import torch
 
 from monai.config.type_definitions import NdarrayTensor
 from monai.data.meta_obj import MetaObj, get_track_meta
 from monai.data.utils import affine_to_spacing, decollate_batch, list_data_collate, remove_extra_metadata
+from monai.utils import look_up_option
 from monai.utils.enums import PostFix
-from monai.utils.type_conversion import convert_to_tensor
+from monai.utils.type_conversion import convert_data_type, convert_to_tensor
 
 __all__ = ["MetaTensor"]
 
@@ -307,6 +309,33 @@ def as_dict(self, key: str) -> dict:
             PostFix.transforms(key): deepcopy(self.applied_operations),
         }
 
+    def astype(self, dtype, device=None, *unused_args, **unused_kwargs):
+        """
+        Cast to ``dtype``, sharing data whenever possible.
+
+        Args:
+            dtype: dtypes such as np.float32, torch.float, "np.float32", float.
+            device: the device if `dtype` is a torch data type.
+            unused_args: additional args (currently unused).
+            unused_kwargs: additional kwargs (currently unused).
+
+        Returns:
+            data array instance
+        """
+        if isinstance(dtype, str):
+            mod_str, *dtype = dtype.split(".", 1)
+            dtype = mod_str if not dtype else dtype[0]
+        else:
+            mod_str = getattr(dtype, "__module__", "torch")
+        mod_str = look_up_option(mod_str, {"torch", "numpy", "np"}, default="numpy")
+        if mod_str == "torch":
+            out_type = torch.Tensor
+        elif mod_str in ("numpy", "np"):
+            out_type = np.ndarray
+        else:
+            out_type = None
+        return convert_data_type(self, output_type=out_type, device=device, dtype=dtype, wrap_sequence=True)[0]
+
     @property
     def affine(self) -> torch.Tensor:
         """Get the affine."""

monai/transforms/utility/array.py

@@ -435,6 +435,8 @@ class EnsureType(Transform):
         device: for Tensor data type, specify the target device.
         wrap_sequence: if `False`, then lists will recursively call this function, default to `True`.
             E.g., if `False`, `[1, 2]` -> `[tensor(1), tensor(2)]`, if `True`, then `[1, 2]` -> `tensor([1, 2])`.
+        track_meta: whether to convert to `MetaTensor` when `data_type` is "tensor".
+            If False, the output data type will be `torch.Tensor`. Default to the return value of ``get_track_meta``.
 
     """
 
@@ -446,11 +448,16 @@ def __init__(
         dtype: Optional[Union[DtypeLike, torch.dtype]] = None,
         device: Optional[torch.device] = None,
         wrap_sequence: bool = True,
+        track_meta: Optional[bool] = None,
     ) -> None:
         self.data_type = look_up_option(data_type.lower(), {"tensor", "numpy"})
         self.dtype = dtype
         self.device = device
         self.wrap_sequence = wrap_sequence
+        if track_meta is None:
+            self.track_meta = get_track_meta()
+        else:
+            self.track_meta = bool(track_meta)
 
     def __call__(self, data: NdarrayOrTensor):
         """
@@ -461,10 +468,17 @@ def __call__(self, data: NdarrayOrTensor):
                 if applicable and `wrap_sequence=False`.
 
         """
-        output_type = torch.Tensor if self.data_type == "tensor" else np.ndarray
+        if self.data_type == "tensor":
+            output_type = MetaTensor if self.track_meta else torch.Tensor
+        else:
+            output_type = np.ndarray  # type: ignore
         out: NdarrayOrTensor
         out, *_ = convert_data_type(
-            data=data, output_type=output_type, dtype=self.dtype, device=self.device, wrap_sequence=self.wrap_sequence
+            data=data,
+            output_type=output_type,  # type: ignore
+            dtype=self.dtype,
+            device=self.device,
+            wrap_sequence=self.wrap_sequence,
         )
         return out
 

monai/transforms/utility/dictionary.py

@@ -62,7 +62,7 @@
 )
 from monai.transforms.utils import extreme_points_to_image, get_extreme_points
 from monai.transforms.utils_pytorch_numpy_unification import concatenate
-from monai.utils import convert_to_numpy, deprecated, deprecated_arg, ensure_tuple, ensure_tuple_rep
+from monai.utils import deprecated, deprecated_arg, ensure_tuple, ensure_tuple_rep
 from monai.utils.enums import PostFix, TraceKeys, TransformBackends
 from monai.utils.type_conversion import convert_to_dst_type
 
@@ -519,7 +519,7 @@ def inverse(self, data: Mapping[Hashable, NdarrayOrTensor]) -> Dict[Hashable, Nd
         return d
 
 
-class EnsureTyped(MapTransform, InvertibleTransform):
+class EnsureTyped(MapTransform):
     """
     Dictionary-based wrapper of :py:class:`monai.transforms.EnsureType`.
 
@@ -541,6 +541,7 @@ def __init__(
         dtype: Union[DtypeLike, torch.dtype] = None,
         device: Optional[torch.device] = None,
         wrap_sequence: bool = True,
+        track_meta: Optional[bool] = None,
         allow_missing_keys: bool = False,
     ) -> None:
         """
@@ -552,28 +553,21 @@ def __init__(
             device: for Tensor data type, specify the target device.
             wrap_sequence: if `False`, then lists will recursively call this function, default to `True`.
                 E.g., if `False`, `[1, 2]` -> `[tensor(1), tensor(2)]`, if `True`, then `[1, 2]` -> `tensor([1, 2])`.
+            track_meta: whether to convert to `MetaTensor` when `data_type` is "tensor".
+                If False, the output data type will be `torch.Tensor`. Default to the return value of `get_track_meta`.
             allow_missing_keys: don't raise exception if key is missing.
         """
         super().__init__(keys, allow_missing_keys)
-        self.converter = EnsureType(data_type=data_type, dtype=dtype, device=device, wrap_sequence=wrap_sequence)
+        self.converter = EnsureType(
+            data_type=data_type, dtype=dtype, device=device, wrap_sequence=wrap_sequence, track_meta=track_meta
+        )
 
     def __call__(self, data: Mapping[Hashable, NdarrayOrTensor]) -> Dict[Hashable, NdarrayOrTensor]:
         d = dict(data)
         for key in self.key_iterator(d):
-            self.push_transform(d, key)
             d[key] = self.converter(d[key])
         return d
 
-    def inverse(self, data: Mapping[Hashable, Any]) -> Dict[Hashable, Any]:
-        d = deepcopy(dict(data))
-        for key in self.key_iterator(d):
-            # FIXME: currently, only convert tensor data to numpy array or scalar number,
-            # need to also invert numpy array but it's not easy to determine the previous data type
-            d[key] = convert_to_numpy(d[key])
-            # Remove the applied transform
-            self.pop_transform(d, key)
-        return d
-
 
 class ToNumpyd(MapTransform):
     """

monai/utils/type_conversion.py

@@ -239,8 +239,8 @@ def convert_data_type(
 
     Args:
         data: data to be converted
-        output_type: `torch.Tensor` or `np.ndarray` (if `None`, unchanged)
-        device: if output is `torch.Tensor`, select device (if `None`, unchanged)
+        output_type: `monai.data.MetaTensor`, `torch.Tensor`, or `np.ndarray` (if `None`, unchanged)
+        device: if output is `MetaTensor` or `torch.Tensor`, select device (if `None`, unchanged)
         dtype: dtype of output data. Converted to correct library type (e.g.,
             `np.float32` is converted to `torch.float32` if output type is `torch.Tensor`).
             If left blank, it remains unchanged.

tests/test_ensure_type.py

@@ -14,6 +14,7 @@
 import numpy as np
 import torch
 
+from monai.data import MetaTensor
 from monai.transforms import EnsureType
 from tests.utils import assert_allclose
 
@@ -59,9 +60,9 @@ def test_string(self):
 
     def test_list_tuple(self):
         for dtype in ("tensor", "numpy"):
-            result = EnsureType(data_type=dtype, wrap_sequence=False)([[1, 2], [3, 4]])
+            result = EnsureType(data_type=dtype, wrap_sequence=False, track_meta=True)([[1, 2], [3, 4]])
             self.assertTrue(isinstance(result, list))
-            self.assertTrue(isinstance(result[0][1], torch.Tensor if dtype == "tensor" else np.ndarray))
+            self.assertTrue(isinstance(result[0][1], MetaTensor if dtype == "tensor" else np.ndarray))
             torch.testing.assert_allclose(result[1][0], torch.as_tensor(3))
             # tuple of numpy arrays
             result = EnsureType(data_type=dtype, wrap_sequence=False)((np.array([1, 2]), np.array([3, 4])))
@@ -77,7 +78,7 @@ def test_dict(self):
             "extra": None,
         }
         for dtype in ("tensor", "numpy"):
-            result = EnsureType(data_type=dtype)(test_data)
+            result = EnsureType(data_type=dtype, track_meta=False)(test_data)
             self.assertTrue(isinstance(result, dict))
             self.assertTrue(isinstance(result["img"], torch.Tensor if dtype == "tensor" else np.ndarray))
             torch.testing.assert_allclose(result["img"], torch.as_tensor([1.0, 2.0]))

tests/test_ensure_typed.py

@@ -14,6 +14,7 @@
 import numpy as np
 import torch
 
+from monai.data import MetaTensor
 from monai.transforms import EnsureTyped
 from tests.utils import assert_allclose
 
@@ -61,9 +62,11 @@ def test_string(self):
 
     def test_list_tuple(self):
         for dtype in ("tensor", "numpy"):
-            result = EnsureTyped(keys="data", data_type=dtype, wrap_sequence=False)({"data": [[1, 2], [3, 4]]})["data"]
+            result = EnsureTyped(keys="data", data_type=dtype, wrap_sequence=False, track_meta=True)(
+                {"data": [[1, 2], [3, 4]]}
+            )["data"]
             self.assertTrue(isinstance(result, list))
-            self.assertTrue(isinstance(result[0][1], torch.Tensor if dtype == "tensor" else np.ndarray))
+            self.assertTrue(isinstance(result[0][1], MetaTensor if dtype == "tensor" else np.ndarray))
             torch.testing.assert_allclose(result[1][0], torch.as_tensor(3))
             # tuple of numpy arrays
             result = EnsureTyped(keys="data", data_type=dtype, wrap_sequence=False)(

tests/test_meta_tensor.py

@@ -20,6 +20,7 @@
 from multiprocessing.reduction import ForkingPickler
 from typing import Optional, Union
 
+import numpy as np
 import torch
 import torch.multiprocessing
 from parameterized import parameterized
@@ -433,6 +434,14 @@ def test_str(self):
         for s in (s1, s2):
             self.assertEqual(s, expected_out)
 
+    def test_astype(self):
+        t = MetaTensor([1.0], affine=torch.tensor(1), meta={"fname": "filename"})
+        for np_types in ("float32", "np.float32", "numpy.float32", np.float32, float, "int", np.compat.long):
+            self.assertIsInstance(t.astype(np_types), np.ndarray)
+        for pt_types in ("torch.float", torch.float, "torch.float64"):
+            self.assertIsInstance(t.astype(pt_types), torch.Tensor)
+        self.assertIsInstance(t.astype("torch.float", device="cpu"), torch.Tensor)
+
     def test_transforms(self):
         key = "im"
         _, im = self.get_im()
@@ -441,7 +450,6 @@ def test_transforms(self):
         data = {key: im, PostFix.meta(key): {"affine": torch.eye(4)}}
 
         # apply one at a time
-        is_meta = isinstance(im, MetaTensor)
         for i, _tr in enumerate(tr.transforms):
             data = _tr(data)
             is_meta = isinstance(_tr, (ToMetaTensord, BorderPadd, DivisiblePadd))
@@ -458,7 +466,6 @@ def test_transforms(self):
             self.assertEqual(n_applied, i + 1)
 
         # inverse one at a time
-        is_meta = isinstance(im, MetaTensor)
         for i, _tr in enumerate(tr.transforms[::-1]):
             data = _tr.inverse(data)
             is_meta = isinstance(_tr, (FromMetaTensord, BorderPadd, DivisiblePadd))