2823 enhance convert_data/dst_type typing

monai/apps/deepgrow/transforms.py

@@ -652,7 +652,7 @@ def __call__(self, data):
             return d
 
         guidance = d[self.guidance]
-        original_spatial_shape = d[first_key].shape[1:]  # type: ignore
+        original_spatial_shape = d[first_key].shape[1:]
         box_start, box_end = self.bounding_box(np.array(guidance[0] + guidance[1]), original_spatial_shape)
         center = list(np.mean([box_start, box_end], axis=0).astype(int, copy=False))
         spatial_size = self.spatial_size

monai/apps/pathology/transforms/spatial/array.py

@@ -190,8 +190,7 @@ def randomize(self, img_size: Sequence[int]) -> None:
             self.random_idxs = np.array((0,))
 
     def __call__(self, image: NdarrayOrTensor) -> NdarrayOrTensor:
-        img_np: np.ndarray
-        img_np, *_ = convert_data_type(image, np.ndarray)  # type: ignore
+        img_np, *_ = convert_data_type(image, np.ndarray)
 
         # add random offset
         self.randomize(img_size=img_np.shape)

monai/config/type_definitions.py

@@ -63,15 +63,14 @@
 #: Type of datatypes: Adapted from https://github.com/numpy/numpy/blob/v1.21.4/numpy/typing/_dtype_like.py#L121
 DtypeLike = Union[np.dtype, type, str, None]
 
+#: NdarrayOrTensor: Union of numpy.ndarray and torch.Tensor to be used for typing
+NdarrayOrTensor = Union[np.ndarray, torch.Tensor]
+
 #: NdarrayTensor
 #
 # Generic type which can represent either a numpy.ndarray or a torch.Tensor
 # Unlike Union can create a dependence between parameter(s) / return(s)
-NdarrayTensor = TypeVar("NdarrayTensor", np.ndarray, torch.Tensor)
-
-
-#: NdarrayOrTensor: Union of numpy.ndarray and torch.Tensor to be used for typing
-NdarrayOrTensor = Union[np.ndarray, torch.Tensor]
+NdarrayTensor = TypeVar("NdarrayTensor", bound=NdarrayOrTensor)
 
 #: TensorOrList: The TensorOrList type is used for defining `batch-first Tensor` or `list of channel-first Tensor`.
 TensorOrList = Union[torch.Tensor, Sequence[torch.Tensor]]

monai/data/dataloader.py

@@ -81,4 +81,4 @@ def __init__(self, dataset: Dataset, num_workers: int = 0, **kwargs) -> None:
         if "worker_init_fn" not in kwargs:
             kwargs.update({"worker_init_fn": worker_init_fn})
 
-        super().__init__(dataset=dataset, num_workers=num_workers, **kwargs)  # type: ignore[call-overload]
+        super().__init__(dataset=dataset, num_workers=num_workers, **kwargs)

monai/data/dataset.py

@@ -269,7 +269,7 @@ def _pre_transform(self, item_transformed):
             random transform object
 
         """
-        for _transform in self.transform.transforms:  # type:ignore
+        for _transform in self.transform.transforms:
             # execute all the deterministic transforms
             if isinstance(_transform, Randomizable) or not isinstance(_transform, Transform):
                 break

monai/data/image_reader.py

@@ -25,7 +25,7 @@
 from .utils import is_supported_format
 
 if TYPE_CHECKING:
-    import itk  # type: ignore
+    import itk
     import nibabel as nib
     from nibabel.nifti1 import Nifti1Image
     from PIL import Image as PILImage

monai/data/nifti_writer.py

@@ -107,17 +107,15 @@ def write_nifti(
     sr = min(data.ndim, 3)
     if affine is None:
         affine = np.eye(4, dtype=np.float64)
-    affine = to_affine_nd(sr, affine)  # type: ignore
+    affine = to_affine_nd(sr, affine)
 
     if target_affine is None:
-        target_affine = affine  # type: ignore
-    target_affine, *_ = convert_data_type(to_affine_nd(sr, target_affine), np.ndarray)  # type: ignore
+        target_affine = affine
+    target_affine, *_ = convert_data_type(to_affine_nd(sr, target_affine), np.ndarray)
 
-    if allclose(affine, target_affine, atol=1e-3):  # type: ignore
+    if allclose(affine, target_affine, atol=1e-3):
         # no affine changes, save (data, affine)
-        results_img = nib.Nifti1Image(
-            data.astype(output_dtype, copy=False), to_affine_nd(3, target_affine)  # type: ignore
-        )
+        results_img = nib.Nifti1Image(data.astype(output_dtype, copy=False), to_affine_nd(3, target_affine))
         nib.save(results_img, file_name)
         return
 
@@ -128,7 +126,7 @@ def write_nifti(
     data_shape = data.shape
     data = nib.orientations.apply_orientation(data, ornt_transform)
     _affine = affine @ nib.orientations.inv_ornt_aff(ornt_transform, data_shape)
-    if allclose(_affine, target_affine, atol=1e-3) or not resample:  # type: ignore
+    if allclose(_affine, target_affine, atol=1e-3) or not resample:
         results_img = nib.Nifti1Image(data.astype(output_dtype, copy=False), to_affine_nd(3, _affine))  # type: ignore
         nib.save(results_img, file_name)
         return
@@ -139,8 +137,8 @@ def write_nifti(
     )
     transform = np.linalg.inv(_affine) @ target_affine
     if output_spatial_shape is None:
-        output_spatial_shape, _ = compute_shape_offset(data.shape, _affine, target_affine)  # type: ignore
-    output_spatial_shape_ = list(output_spatial_shape) if output_spatial_shape is not None else []  # type: ignore
+        output_spatial_shape, _ = compute_shape_offset(data.shape, _affine, target_affine)
+    output_spatial_shape_ = list(output_spatial_shape) if output_spatial_shape is not None else []
     if data.ndim > 3:  # multi channel, resampling each channel
         while len(output_spatial_shape_) < 3:
             output_spatial_shape_ = output_spatial_shape_ + [1]
@@ -165,8 +163,6 @@ def write_nifti(
         )
         data_np = data_torch.squeeze(0).squeeze(0).detach().cpu().numpy()
 
-    results_img = nib.Nifti1Image(
-        data_np.astype(output_dtype, copy=False), to_affine_nd(3, target_affine)  # type: ignore
-    )
+    results_img = nib.Nifti1Image(data_np.astype(output_dtype, copy=False), to_affine_nd(3, target_affine))
     nib.save(results_img, file_name)
     return

monai/data/png_writer.py

@@ -65,10 +65,10 @@ def write_png(
             data = np.expand_dims(data, 0)  # make a channel
             data = xform(data)[0]  # type: ignore
         if mode != InterpolateMode.NEAREST:
-            data = np.clip(data, _min, _max)  # type: ignore
+            data = np.clip(data, _min, _max)
 
     if scale is not None:
-        data = np.clip(data, 0.0, 1.0)  # type: ignore # png writer only can scale data in range [0, 1]
+        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 = (scale * data).astype(np.uint8, copy=False)
         elif scale == np.iinfo(np.uint16).max:
@@ -77,7 +77,7 @@ def write_png(
             raise ValueError(f"Unsupported scale: {scale}, available options are [255, 65535]")
 
     # PNG data must be int number
-    if data.dtype not in (np.uint8, np.uint16):  # type: ignore
+    if data.dtype not in (np.uint8, np.uint16):
         data = data.astype(np.uint8, copy=False)
 
     data = np.moveaxis(data, 0, 1)

monai/data/samplers.py

@@ -50,7 +50,7 @@ def __init__(
         super().__init__(dataset=dataset, num_replicas=num_replicas, rank=rank, shuffle=shuffle, **kwargs)
 
         if not even_divisible:
-            data_len = len(dataset)  # type: ignore
+            data_len = len(dataset)
             extra_size = self.total_size - data_len
             if self.rank + extra_size >= self.num_replicas:
                 self.num_samples -= 1

monai/data/utils.py

@@ -27,7 +27,7 @@
 import torch
 from torch.utils.data._utils.collate import default_collate
 
-from monai.config.type_definitions import NdarrayOrTensor, PathLike
+from monai.config.type_definitions import NdarrayOrTensor, NdarrayTensor, PathLike
 from monai.networks.layers.simplelayers import GaussianFilter
 from monai.utils import (
     MAX_SEED,
@@ -676,16 +676,16 @@ def compute_shape_offset(
     """
     shape = np.array(spatial_shape, copy=True, dtype=float)
     sr = len(shape)
-    in_affine = convert_data_type(to_affine_nd(sr, in_affine), np.ndarray)[0]  # type: ignore
-    out_affine = convert_data_type(to_affine_nd(sr, out_affine), np.ndarray)[0]  # type: ignore
+    in_affine_ = convert_data_type(to_affine_nd(sr, in_affine), np.ndarray)[0]
+    out_affine_ = convert_data_type(to_affine_nd(sr, out_affine), np.ndarray)[0]
     in_coords = [(0.0, dim - 1.0) for dim in shape]
     corners: np.ndarray = np.asarray(np.meshgrid(*in_coords, indexing="ij")).reshape((len(shape), -1))
     corners = np.concatenate((corners, np.ones_like(corners[:1])))
-    corners = in_affine @ corners  # type: ignore
+    corners = in_affine_ @ corners
     try:
-        inv_mat = np.linalg.inv(out_affine)
+        inv_mat = np.linalg.inv(out_affine_)
     except np.linalg.LinAlgError as e:
-        raise ValueError(f"Affine {out_affine} is not invertible") from e
+        raise ValueError(f"Affine {out_affine_} is not invertible") from e
     corners_out = inv_mat @ corners
     corners_out = corners_out[:-1] / corners_out[-1]
     out_shape = np.round(corners_out.ptp(axis=1) + 1.0)
@@ -700,7 +700,7 @@ def compute_shape_offset(
     return out_shape.astype(int, copy=False), offset
 
 
-def to_affine_nd(r: Union[np.ndarray, int], affine: NdarrayOrTensor, dtype=np.float64) -> NdarrayOrTensor:
+def to_affine_nd(r: Union[np.ndarray, int], affine: NdarrayTensor, dtype=np.float64) -> NdarrayTensor:
     """
     Using elements from affine, to create a new affine matrix by
     assigning the rotation/zoom/scaling matrix and the translation vector.
@@ -728,8 +728,7 @@ def to_affine_nd(r: Union[np.ndarray, int], affine: NdarrayOrTensor, dtype=np.fl
         an (r+1) x (r+1) matrix (tensor or ndarray depends on the input ``affine`` data type)
 
     """
-    affine_np: np.ndarray
-    affine_np = convert_data_type(affine, output_type=np.ndarray, dtype=dtype, wrap_sequence=True)[0]  # type: ignore
+    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}.")
@@ -743,8 +742,8 @@ def to_affine_nd(r: Union[np.ndarray, int], affine: NdarrayOrTensor, dtype=np.fl
     new_affine[:d, :d] = affine_np[:d, :d]
     if d > 1:
         new_affine[:d, -1] = affine_np[:d, -1]
-    new_affine, *_ = convert_to_dst_type(new_affine, affine, dtype=dtype)  # type: ignore
-    return new_affine
+    output, *_ = convert_to_dst_type(new_affine, affine, dtype=dtype)
+    return output
 
 
 def reorient_spatial_axes(
@@ -874,7 +873,7 @@ def compute_importance_map(
 
     """
     mode = look_up_option(mode, BlendMode)
-    device = torch.device(device)  # type: ignore[arg-type]
+    device = torch.device(device)
     if mode == BlendMode.CONSTANT:
         importance_map = torch.ones(patch_size, device=device).float()
     elif mode == BlendMode.GAUSSIAN:

monai/handlers/roc_auc.py

@@ -16,7 +16,7 @@
 from monai.utils import Average
 
 
-class ROCAUC(IgniteMetric):  # type: ignore[valid-type, misc]  # due to optional_import
+class ROCAUC(IgniteMetric):
     """
     Computes Area Under the Receiver Operating Characteristic Curve (ROC AUC).
     accumulating predictions and the ground-truth during an epoch and applying `compute_roc_auc`.

monai/metrics/confusion_matrix.py

@@ -102,7 +102,7 @@ def _compute_tensor(self, y_pred: torch.Tensor, y: torch.Tensor):  # type: ignor
 
         return get_confusion_matrix(y_pred=y_pred, y=y, include_background=self.include_background)
 
-    def aggregate(self):  # type: ignore
+    def aggregate(self):
         """
         Execute reduction for the confusion matrix values.
 

monai/metrics/cumulative_average.py

@@ -51,7 +51,7 @@ def reset(self):
         self.sum = None
         self.not_nans = None
 
-    def aggregate(self):  # type: ignore
+    def aggregate(self):
         """
         Sync data from all the ranks and compute the average value with previous sum value.
 

monai/metrics/hausdorff_distance.py

@@ -99,7 +99,7 @@ def _compute_tensor(self, y_pred: torch.Tensor, y: torch.Tensor):  # type: ignor
             directed=self.directed,
         )
 
-    def aggregate(self):  # type: ignore
+    def aggregate(self):
         """
         Execute reduction logic for the output of `compute_hausdorff_distance`.
 

monai/metrics/meandice.py

@@ -79,7 +79,7 @@ def _compute_tensor(self, y_pred: torch.Tensor, y: torch.Tensor):  # type: ignor
         # compute dice (BxC) for each channel for each batch
         return compute_meandice(y_pred=y_pred, y=y, include_background=self.include_background)
 
-    def aggregate(self):  # type: ignore
+    def aggregate(self):
         """
         Execute reduction logic for the output of `compute_meandice`.
 

monai/metrics/metric.py

@@ -201,8 +201,7 @@ def extend(self, *data) -> None:
             self._buffers = [[] for _ in data]
         for b, d in zip(self._buffers, data):
             # converting to pytorch tensors so that we can use the distributed API
-            d_t: torch.Tensor
-            d_t, *_ = convert_data_type(d, output_type=torch.Tensor, wrap_sequence=True)  # type: ignore
+            d_t, *_ = convert_data_type(d, output_type=torch.Tensor, wrap_sequence=True)
             try:
                 b.extend([x[0] for x in torch.split(d_t, 1, dim=0)])
             except (AttributeError, IndexError, RuntimeError) as e:
@@ -228,8 +227,7 @@ def append(self, *data) -> None:
             self._buffers = [[] for _ in data]
         for b, d in zip(self._buffers, data):
             # converting to pytorch tensors so that we can use the distributed API
-            d_t: torch.Tensor
-            d_t, *_ = convert_data_type(d, output_type=torch.Tensor, wrap_sequence=True)  # type: ignore
+            d_t, *_ = convert_data_type(d, output_type=torch.Tensor, wrap_sequence=True)
             b.append(d_t)
         self._synced = False
 

monai/metrics/regression.py

@@ -45,7 +45,7 @@ def __init__(
         self.reduction = reduction
         self.get_not_nans = get_not_nans
 
-    def aggregate(self):  # type: ignore
+    def aggregate(self):
         data = self.get_buffer()
         if not isinstance(data, torch.Tensor):
             raise ValueError("the data to aggregate must be PyTorch Tensor.")

monai/metrics/rocauc.py

@@ -49,7 +49,7 @@ def __init__(self, average: Union[Average, str] = Average.MACRO) -> None:
     def _compute_tensor(self, y_pred: torch.Tensor, y: torch.Tensor):  # type: ignore
         return y_pred, y
 
-    def aggregate(self):  # type: ignore
+    def aggregate(self):
         """
         As AUC metric needs to execute on the overall data, so usually users accumulate `y_pred` and `y`
         of every iteration, then execute real computation and reduction on the accumulated data.

monai/metrics/surface_distance.py

@@ -91,7 +91,7 @@ def _compute_tensor(self, y_pred: torch.Tensor, y: torch.Tensor):  # type: ignor
             distance_metric=self.distance_metric,
         )
 
-    def aggregate(self):  # type: ignore
+    def aggregate(self):
         """
         Execute reduction logic for the output of `compute_average_surface_distance`.
 
@@ -158,15 +158,15 @@ def compute_average_surface_distance(
         if surface_distance.shape == (0,):
             avg_surface_distance = np.nan
         else:
-            avg_surface_distance = surface_distance.mean()  # type: ignore
+            avg_surface_distance = surface_distance.mean()
         if not symmetric:
             asd[b, c] = avg_surface_distance
         else:
             surface_distance_2 = get_surface_distance(edges_gt, edges_pred, distance_metric=distance_metric)
             if surface_distance_2.shape == (0,):
                 avg_surface_distance_2 = np.nan
             else:
-                avg_surface_distance_2 = surface_distance_2.mean()  # type: ignore
+                avg_surface_distance_2 = surface_distance_2.mean()
             asd[b, c] = np.mean((avg_surface_distance, avg_surface_distance_2))
 
     return torch.from_numpy(asd)

monai/networks/blocks/acti_norm.py

@@ -98,4 +98,4 @@ def __init__(
             if item not in op_dict:
                 raise ValueError(f"ordering must be a string of {op_dict}, got {item} in it.")
             if op_dict[item] is not None:
-                self.add_module(item, op_dict[item])  # type: ignore
+                self.add_module(item, op_dict[item])

monai/networks/layers/spatial_transforms.py

@@ -115,7 +115,7 @@ def grid_pull(
         for i in ensure_tuple(interpolation)
     ]
     out: torch.Tensor
-    out = _GridPull.apply(input, grid, interpolation, bound, extrapolate)  # type: ignore
+    out = _GridPull.apply(input, grid, interpolation, bound, extrapolate)
     return out
 
 

monai/networks/nets/transchex.py

@@ -217,7 +217,7 @@ class MultiModal(BertPreTrainedModel):
     """
 
     def __init__(
-        self, num_language_layers: int, num_vision_layers: int, num_mixed_layers: int, bert_config: dict  # type: ignore
+        self, num_language_layers: int, num_vision_layers: int, num_mixed_layers: int, bert_config: dict
     ) -> None:
         """
         Args:
@@ -254,8 +254,8 @@ class Transchex(torch.nn.Module):
     def __init__(
         self,
         in_channels: int,
-        img_size: Union[Sequence[int], int],  # type: ignore
-        patch_size: Union[int, Tuple[int, int]],  # type: ignore
+        img_size: Union[Sequence[int], int],
+        patch_size: Union[int, Tuple[int, int]],
         num_classes: int,
         num_language_layers: int,
         num_vision_layers: int,
@@ -352,10 +352,7 @@ def __init__(
         self.patch_size = patch_size
         self.num_patches = (img_size[0] // self.patch_size[0]) * (img_size[1] // self.patch_size[1])  # type: ignore
         self.vision_proj = nn.Conv2d(
-            in_channels=in_channels,
-            out_channels=hidden_size,
-            kernel_size=self.patch_size,  # type: ignore
-            stride=self.patch_size,  # type: ignore
+            in_channels=in_channels, out_channels=hidden_size, kernel_size=self.patch_size, stride=self.patch_size
         )
         self.norm_vision_pos = nn.LayerNorm(hidden_size)
         self.pos_embed_vis = nn.Parameter(torch.zeros(1, self.num_patches, hidden_size))

monai/transforms/croppad/array.py

@@ -394,13 +394,13 @@ def __init__(
                     data=roi_center, output_type=torch.Tensor, dtype=torch.int16, wrap_sequence=True
                 )
                 roi_size, *_ = convert_to_dst_type(src=roi_size, dst=roi_center, wrap_sequence=True)
-                _zeros = torch.zeros_like(roi_center)  # type: ignore
+                _zeros = torch.zeros_like(roi_center)
                 roi_start_torch = maximum(roi_center - floor_divide(roi_size, 2), _zeros)  # type: ignore
                 roi_end_torch = maximum(roi_start_torch + roi_size, roi_start_torch)
             else:
                 if roi_start is None or roi_end is None:
                     raise ValueError("Please specify either roi_center, roi_size or roi_start, roi_end.")
-                roi_start_torch, *_ = convert_data_type(  # type: ignore
+                roi_start_torch, *_ = convert_data_type(
                     data=roi_start, output_type=torch.Tensor, dtype=torch.int16, wrap_sequence=True
                 )
                 roi_start_torch = maximum(roi_start_torch, torch.zeros_like(roi_start_torch))  # type: ignore
@@ -1162,7 +1162,7 @@ def __call__(self, img: NdarrayOrTensor, mode: Optional[Union[NumpyPadMode, str]
                 If None, defaults to the ``mode`` in construction.
                 See also: https://numpy.org/doc/1.18/reference/generated/numpy.pad.html
         """
-        return self.padder(self.cropper(img), mode=mode)  # type: ignore
+        return self.padder(self.cropper(img), mode=mode)
 
 
 class BoundingRect(Transform):