Merge branch 'dev' into metatensor-croppad

Author: Nic-Ma

monai/data/png_writer.py

@@ -81,9 +81,9 @@ def write_png(
     if scale is not None:
         data = np.clip(data, 0.0, 1.0)  # png writer only can scale data in range [0, 1]
         if scale == np.iinfo(np.uint8).max:
-            data = convert_data_type((scale * data), np.ndarray, dtype=np.uint8, drop_meta=True)[0]
+            data = convert_data_type((scale * data), np.ndarray, dtype=np.uint8)[0]
         elif scale == np.iinfo(np.uint16).max:
-            data = convert_data_type((scale * data), np.ndarray, dtype=np.uint16, drop_meta=True)[0]
+            data = convert_data_type((scale * data), np.ndarray, dtype=np.uint16)[0]
         else:
             raise ValueError(f"Unsupported scale: {scale}, available options are [255, 65535]")
 

monai/data/utils.py

@@ -856,7 +856,7 @@ def to_affine_nd(r: Union[np.ndarray, int], affine: NdarrayTensor, dtype=np.floa
         an (r+1) x (r+1) matrix (tensor or ndarray depends on the input ``affine`` data type)
 
     """
-    affine_np = convert_data_type(affine, output_type=np.ndarray, dtype=dtype, wrap_sequence=True, drop_meta=True)[0]
+    affine_np = convert_data_type(affine, output_type=np.ndarray, dtype=dtype, wrap_sequence=True)[0]
     affine_np = affine_np.copy()
     if affine_np.ndim != 2:
         raise ValueError(f"affine must have 2 dimensions, got {affine_np.ndim}.")

monai/inferers/utils.py

@@ -175,7 +175,7 @@ def sliding_window_inference(
             for idx in slice_range
         ]
         window_data = torch.cat(
-            [convert_data_type(inputs[win_slice], torch.Tensor, drop_meta=True)[0] for win_slice in unravel_slice]
+            [convert_data_type(inputs[win_slice], torch.Tensor)[0] for win_slice in unravel_slice]
         ).to(sw_device)
         seg_prob_out = predictor(window_data, *args, **kwargs)  # batched patch segmentation
 

monai/metrics/utils.py

@@ -155,8 +155,8 @@ def get_mask_edges(seg_pred, seg_gt, label_idx: int = 1, crop: bool = True) -> T
         seg_pred, seg_gt = np.expand_dims(seg_pred, axis=channel_dim), np.expand_dims(seg_gt, axis=channel_dim)
         box_start, box_end = generate_spatial_bounding_box(np.asarray(seg_pred | seg_gt))
         cropper = SpatialCrop(roi_start=box_start, roi_end=box_end)
-        seg_pred = convert_data_type(np.squeeze(cropper(seg_pred), axis=channel_dim), np.ndarray, drop_meta=True)[0]
-        seg_gt = convert_data_type(np.squeeze(cropper(seg_gt), axis=channel_dim), np.ndarray, drop_meta=True)[0]
+        seg_pred = convert_data_type(np.squeeze(cropper(seg_pred), axis=channel_dim), np.ndarray)[0]
+        seg_gt = convert_data_type(np.squeeze(cropper(seg_gt), axis=channel_dim), np.ndarray)[0]
 
     # Do binary erosion and use XOR to get edges
     edges_pred = binary_erosion(seg_pred) ^ seg_pred

monai/transforms/utils_pytorch_numpy_unification.py

@@ -376,7 +376,7 @@ def mode(x: NdarrayTensor, dim: int = -1, to_long: bool = True) -> NdarrayTensor
         to_long: convert input to long before performing mode.
     """
     dtype = torch.int64 if to_long else None
-    x_t, *_ = convert_data_type(x, torch.Tensor, dtype=dtype, drop_meta=True)
+    x_t, *_ = convert_data_type(x, torch.Tensor, dtype=dtype)
     o_t = torch.mode(x_t, dim).values
     o, *_ = convert_to_dst_type(o_t, x)
     return o

monai/utils/type_conversion.py

@@ -105,7 +105,8 @@ def convert_to_tensor(
     track_meta: bool = False,
 ):
     """
-    Utility to convert the input data to a PyTorch Tensor, if tracking meta, convert to `MetaTensor`.
+    Utility to convert the input data to a PyTorch Tensor, if `track_meta` is True, the output will be a `MetaTensor`,
+    otherwise, the output will be a regular torch Tensor.
     If passing a dictionary, list or tuple, recursively check every item and convert it to PyTorch Tensor.
 
     Args:
@@ -121,13 +122,14 @@ def convert_to_tensor(
 
     """
 
-    def _convert_tensor(tensor):
+    def _convert_tensor(tensor, **kwargs):
         if not isinstance(tensor, torch.Tensor):
-            tensor = torch.as_tensor(tensor)
+            # if input data is not Tensor, convert it to Tensor first
+            tensor = torch.as_tensor(tensor, **kwargs)
         if track_meta and not isinstance(tensor, monai.data.MetaTensor):
             return monai.data.MetaTensor(tensor)
         if not track_meta and isinstance(tensor, monai.data.MetaTensor):
-            return tensor.as_tensor(tensor)
+            return tensor.as_tensor()
         return tensor
 
     if isinstance(data, torch.Tensor):
@@ -142,15 +144,15 @@ def _convert_tensor(tensor):
                 data = np.ascontiguousarray(data)
             return _convert_tensor(data, dtype=dtype, device=device)
     elif (has_cp and isinstance(data, cp_ndarray)) or isinstance(data, (float, int, bool)):
-        return _convert_tensor(data, dtype=dtype, device=device)  # type: ignore
+        return _convert_tensor(data, dtype=dtype, device=device)
     elif isinstance(data, list):
-        list_ret = [convert_to_tensor(i, dtype=dtype, device=device) for i in data]
+        list_ret = [convert_to_tensor(i, dtype=dtype, device=device, track_meta=track_meta) for i in data]
         return _convert_tensor(list_ret, dtype=dtype, device=device) if wrap_sequence else list_ret
     elif isinstance(data, tuple):
-        tuple_ret = tuple(convert_to_tensor(i, dtype=dtype, device=device) for i in data)
-        return _convert_tensor(tuple_ret, dtype=dtype, device=device) if wrap_sequence else tuple_ret  # type: ignore
+        tuple_ret = tuple(convert_to_tensor(i, dtype=dtype, device=device, track_meta=track_meta) for i in data)
+        return _convert_tensor(tuple_ret, dtype=dtype, device=device) if wrap_sequence else tuple_ret
     elif isinstance(data, dict):
-        return {k: convert_to_tensor(v, dtype=dtype, device=device) for k, v in data.items()}
+        return {k: convert_to_tensor(v, dtype=dtype, device=device, track_meta=track_meta) for k, v in data.items()}
 
     return data
 
@@ -230,7 +232,6 @@ def convert_data_type(
     device: Optional[torch.device] = None,
     dtype: Union[DtypeLike, torch.dtype] = None,
     wrap_sequence: bool = False,
-    drop_meta: bool = True,
 ) -> Tuple[NdarrayTensor, type, Optional[torch.device]]:
     """
     Convert to `torch.Tensor`/`np.ndarray` from `torch.Tensor`/`np.ndarray`/`float`/`int` etc.
@@ -244,9 +245,6 @@ def convert_data_type(
             If left blank, it remains unchanged.
         wrap_sequence: if `False`, then lists will recursively call this function.
             E.g., `[1, 2]` -> `[array(1), array(2)]`. If `True`, then `[1, 2]` -> `array([1, 2])`.
-        drop_meta: whether to drop the meta information of the input data, default to `True`.
-            If `True`, then the meta information will be dropped quietly, unless the output type is MetaTensor.
-            If `False`, converting a MetaTensor into a non-tensor instance will raise an error.
 
     Returns:
         modified data, orig_type, orig_device
@@ -278,18 +276,10 @@ def convert_data_type(
 
     dtype_ = get_equivalent_dtype(dtype, output_type)
 
-    if not drop_meta and not issubclass(output_type, monai.data.MetaObj) and isinstance(data, monai.data.MetaObj):
-        # input has a MetaObj, user chose keep the metadata, but the output type cannot take a MetaObj.
-        if issubclass(output_type, torch.Tensor):
-            # user-specified MetaTensor to torch tensor keep the MetaTensor type, for backward compatibility
-            output_type = type(data)  # type: ignore
-        else:
-            raise RuntimeError(f"the specified output_type {output_type} cannot have the metaobj, but drop_meta=False.")
-
     data_: NdarrayTensor
 
     if issubclass(output_type, torch.Tensor):
-        track_meta = True if issubclass(output_type, monai.data.MetaTensor) else False
+        track_meta = issubclass(output_type, monai.data.MetaTensor)
         data_ = convert_to_tensor(data, dtype=dtype_, device=device, wrap_sequence=wrap_sequence, track_meta=track_meta)
         return data_, orig_type, orig_device
     if issubclass(output_type, np.ndarray):
@@ -302,11 +292,7 @@ def convert_data_type(
 
 
 def convert_to_dst_type(
-    src: Any,
-    dst: NdarrayTensor,
-    dtype: Union[DtypeLike, torch.dtype, None] = None,
-    wrap_sequence: bool = False,
-    drop_meta: bool = True,
+    src: Any, dst: NdarrayTensor, dtype: Union[DtypeLike, torch.dtype, None] = None, wrap_sequence: bool = False
 ) -> Tuple[NdarrayTensor, type, Optional[torch.device]]:
     """
     Convert source data to the same data type and device as the destination data.
@@ -320,9 +306,6 @@ def convert_to_dst_type(
         dtype: an optional argument if the target `dtype` is different from the original `dst`'s data type.
         wrap_sequence: if `False`, then lists will recursively call this function. E.g., `[1, 2]` -> `[array(1), array(2)]`.
             If `True`, then `[1, 2]` -> `array([1, 2])`.
-        drop_meta: whether to drop the meta information of the input data, default to `True`.
-            If `True`, then the meta information will be dropped quietly, unless the output type is MetaTensor.
-            If `False`, converting a MetaTensor into a non-tensor instance will raise an error.
 
     See Also:
         :func:`convert_data_type`
@@ -346,7 +329,7 @@ def convert_to_dst_type(
         output_type = type(dst)
     output: NdarrayTensor
     output, _type, _device = convert_data_type(
-        data=src, output_type=output_type, device=device, dtype=dtype, wrap_sequence=wrap_sequence, drop_meta=drop_meta
+        data=src, output_type=output_type, device=device, dtype=dtype, wrap_sequence=wrap_sequence
     )
     if copy_meta and isinstance(output, monai.data.MetaTensor):  # type: ignore
         output.meta, output.applied_operations = deepcopy(dst.meta), deepcopy(dst.applied_operations)  # type: ignore

tests/test_convert_data_type.py

@@ -16,24 +16,25 @@
 import torch
 from parameterized import parameterized
 
+from monai.data import MetaTensor
 from monai.utils.type_conversion import convert_data_type, convert_to_dst_type
-from tests.utils import TEST_NDARRAYS
+from tests.utils import TEST_NDARRAYS_ALL
 
 TESTS: List[Tuple] = []
-for in_type in TEST_NDARRAYS + (int, float):
-    for out_type in TEST_NDARRAYS:
+for in_type in TEST_NDARRAYS_ALL + (int, float):
+    for out_type in TEST_NDARRAYS_ALL:
         TESTS.append((in_type(np.array(1.0)), out_type(np.array(1.0))))  # type: ignore
 
 TESTS_LIST: List[Tuple] = []
-for in_type in TEST_NDARRAYS + (int, float):
-    for out_type in TEST_NDARRAYS:
+for in_type in TEST_NDARRAYS_ALL + (int, float):
+    for out_type in TEST_NDARRAYS_ALL:
         TESTS_LIST.append(
             ([in_type(np.array(1.0)), in_type(np.array(1.0))], out_type(np.array([1.0, 1.0])), True)  # type: ignore
         )
         TESTS_LIST.append(
             (
                 [in_type(np.array(1.0)), in_type(np.array(1.0))],  # type: ignore
-                [out_type(np.array(1.0)), out_type(np.array(1.0))],
+                [out_type(np.array(1.0)), out_type(np.array(1.0))],  # type: ignore
                 False,
             )
         )
@@ -83,14 +84,17 @@ def test_convert_data_type(self, in_image, im_out):
         self.assertEqual(type(in_image), orig_type)
         if isinstance(in_image, torch.Tensor):
             self.assertEqual(in_image.device, orig_device)
+
         # check output is desired type
-        if isinstance(im_out, torch.Tensor):
+        if isinstance(im_out, MetaTensor):
+            output_type = MetaTensor
+        elif isinstance(im_out, torch.Tensor):
             output_type = torch.Tensor
         else:
             output_type = np.ndarray
         self.assertEqual(type(converted_im), output_type)
         # check dtype is unchanged
-        if isinstance(in_type, (np.ndarray, torch.Tensor)):
+        if isinstance(in_type, (np.ndarray, torch.Tensor, MetaTensor)):
             self.assertEqual(converted_im.dtype, im_out.dtype)
 
 

tests/test_utils_pytorch_numpy_unification.py

@@ -36,7 +36,7 @@ def test_percentile(self):
             for p in TEST_NDARRAYS:
                 arr = p(np.arange(100 * 101).reshape(1, 100, 101).astype(np.float32))
                 results.append(percentile(arr, q))
-                assert_allclose(results[0], results[-1], type_test=False, atol=1e-4)
+                assert_allclose(results[0], results[-1], type_test=False, atol=1e-4, rtol=1e-4)
 
     def test_fails(self):
         for p in TEST_NDARRAYS:

tests/utils.py

@@ -38,6 +38,7 @@
 from monai.config.deviceconfig import USE_COMPILED
 from monai.config.type_definitions import NdarrayOrTensor
 from monai.data import create_test_image_2d, create_test_image_3d
+from monai.data.meta_tensor import MetaTensor
 from monai.networks import convert_to_torchscript
 from monai.utils import optional_import
 from monai.utils.module import pytorch_after, version_leq
@@ -712,6 +713,19 @@ def query_memory(n=2):
     gpu_tensor: Callable = partial(torch.as_tensor, device="cuda")
     TEST_NDARRAYS = TEST_NDARRAYS + (gpu_tensor,)  # type: ignore
 
+TEST_TORCH_TENSORS: Tuple[Callable] = (torch.as_tensor,)  # type: ignore
+if torch.cuda.is_available():
+    gpu_tensor: Callable = partial(torch.as_tensor, device="cuda")  # type: ignore
+    TEST_NDARRAYS = TEST_TORCH_TENSORS + (gpu_tensor,)  # type: ignore
+
+DEFAULT_TEST_AFFINE = torch.tensor(
+    [[2.0, 0.0, 0.0, 0.0], [0.0, 2.0, 0.0, 0.0], [0.0, 0.0, 2.0, 0.0], [0.0, 0.0, 0.0, 1.0]]
+)
+_metatensor_creator = partial(MetaTensor, meta={"a": "b", "affine": DEFAULT_TEST_AFFINE})
+TEST_NDARRAYS_NO_META_TENSOR: Tuple[Callable] = (np.array,) + TEST_TORCH_TENSORS  # type: ignore
+TEST_TORCH_AND_META_TENSORS: Tuple[Callable] = TEST_TORCH_TENSORS + (_metatensor_creator,)  # type: ignore
+TEST_NDARRAYS_ALL: Tuple[Callable] = TEST_NDARRAYS_NO_META_TENSOR + (_metatensor_creator,)  # type: ignore
+
 
 TEST_DEVICES = [[torch.device("cpu")]]
 if torch.cuda.is_available():