Mypy check untyped functions (#5673)

Fixes #5657.

### Description

Enable config to also type check untyped functions (not in tests).
- Type annotations were added
- Explicit casts were added
- Variable redefinitions with other types were refactored 
- If no other option was available (or it would be overly verbose) `#
type: ignore` was added

### Types of changes
<!--- Put an `x` in all the boxes that apply, and remove the not
applicable items -->
- [x] Non-breaking change (fix or new feature that would not break
existing functionality).
- [ ] Breaking change (fix or new feature that would cause existing
functionality to change).
- [ ] New tests added to cover the changes.
- [x] Integration tests passed locally by running `./runtests.sh -f -u
--net --coverage`.
- [ ] Quick tests passed locally by running `./runtests.sh --quick
--unittests --disttests`.
- [ ] In-line docstrings updated.
- [ ] Documentation updated, tested `make html` command in the `docs/`
folder.

Signed-off-by: Felix Schnabel <f.schnabel@tum.de>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: f-schnabel

monai/_extensions/loader.py

@@ -30,7 +30,8 @@ def timeout(time, message):
     try:
         timer = Timer(time, interrupt_main)
         timer.daemon = True
-        yield timer.start()
+        timer.start()
+        yield
     except KeyboardInterrupt as e:
         if timer is not None and timer.is_alive():
             raise e  # interrupt from user?

monai/apps/auto3dseg/auto_runner.py

@@ -14,7 +14,7 @@
 import subprocess
 from copy import deepcopy
 from time import sleep
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union, cast
 
 import numpy as np
 import torch
@@ -204,6 +204,8 @@ class AutoRunner:
 
     """
 
+    analyze_params: Optional[Dict]
+
     def __init__(
         self,
         work_dir: str = "./work_dir",
@@ -561,7 +563,7 @@ def _train_algo_in_nni(self, history):
                 nni_config_filename = os.path.abspath(os.path.join(self.work_dir, f"{name}_nni_config.yaml"))
                 ConfigParser.export_config_file(nni_config, nni_config_filename, fmt="yaml", default_flow_style=None)
 
-                max_trial = min(self.hpo_tasks, default_nni_config["maxTrialNumber"])
+                max_trial = min(self.hpo_tasks, cast(int, default_nni_config["maxTrialNumber"]))
                 cmd = "nnictl create --config " + nni_config_filename + " --port 8088"
 
                 if mode_dry_run:
@@ -585,7 +587,7 @@ def run(self):
         Run the AutoRunner pipeline
         """
         # step 1: data analysis
-        if self.analyze:
+        if self.analyze and self.analyze_params is not None:
             logger.info("Running data analysis...")
             da = DataAnalyzer(
                 self.datalist_filename, self.dataroot, output_path=self.datastats_filename, **self.analyze_params

monai/apps/auto3dseg/bundle_gen.py

@@ -246,9 +246,9 @@ def infer(self, image_file):
         configs_path = [os.path.join(config_dir, f) for f in os.listdir(config_dir)]
 
         spec = importlib.util.spec_from_file_location("InferClass", infer_py)
-        infer_class = importlib.util.module_from_spec(spec)
+        infer_class = importlib.util.module_from_spec(spec)  # type: ignore
         sys.modules["InferClass"] = infer_class
-        spec.loader.exec_module(infer_class)
+        spec.loader.exec_module(infer_class)  # type: ignore
         return infer_class.InferClass(configs_path, *args, **kwargs)
 
     def predict(self, predict_files: list, predict_params=None):

monai/apps/auto3dseg/data_analyzer.py

@@ -11,7 +11,7 @@
 
 import warnings
 from os import path
-from typing import Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union, cast
 
 import numpy as np
 import torch
@@ -227,7 +227,7 @@ def get_all_case_stats(self, key="training", transform_list=None):
         files, _ = datafold_read(datalist=self.datalist, basedir=self.dataroot, fold=-1, key=key)
         dataset = Dataset(data=files, transform=transform)
         dataloader = DataLoader(dataset, batch_size=1, shuffle=False, num_workers=self.worker, collate_fn=no_collation)
-        result = {DataStatsKeys.SUMMARY: {}, DataStatsKeys.BY_CASE: []}
+        result: Dict[DataStatsKeys, Any] = {DataStatsKeys.SUMMARY: {}, DataStatsKeys.BY_CASE: []}
 
         if not has_tqdm:
             warnings.warn("tqdm is not installed. not displaying the caching progress.")
@@ -261,7 +261,7 @@ def get_all_case_stats(self, key="training", transform_list=None):
                 )
             result[DataStatsKeys.BY_CASE].append(stats_by_cases)
 
-        result[DataStatsKeys.SUMMARY] = summarizer.summarize(result[DataStatsKeys.BY_CASE])
+        result[DataStatsKeys.SUMMARY] = summarizer.summarize(cast(List, result[DataStatsKeys.BY_CASE]))
 
         if not self._check_data_uniformity([ImageStatsKeys.SPACING], result):
             print("Data spacing is not completely uniform. MONAI transforms may provide unexpected result")

monai/apps/auto3dseg/hpo_gen.py

@@ -12,7 +12,7 @@
 import os
 from abc import abstractmethod
 from copy import deepcopy
-from typing import Optional
+from typing import Optional, cast
 from warnings import warn
 
 from monai.apps.auto3dseg.bundle_gen import BundleAlgo
@@ -147,7 +147,8 @@ def print_bundle_algo_instruction(self):
         logger.info("-" * 140)
         logger.info("If NNI will run in a remote env: ")
         logger.info(
-            f"1. Copy the algorithm_templates folder {self.algo.template_path} to remote {{remote_algorithm_templates_dir}}"
+            f"1. Copy the algorithm_templates folder {cast(BundleAlgo, self.algo).template_path} "
+            f"to remote {{remote_algorithm_templates_dir}}"
         )
         logger.info(f"2. Copy the older {self.algo.get_output_path()} to the remote machine {{remote_algo_dir}}")
         logger.info("Then add the following line to the trialCommand in your NNI config: ")

monai/apps/deepedit/transforms.py

@@ -503,8 +503,8 @@ def __init__(
         self.discrepancy = discrepancy
         self.probability = probability
         self._will_interact = None
-        self.is_pos = None
-        self.is_other = None
+        self.is_pos: Optional[bool] = None
+        self.is_other: Optional[bool] = None
         self.default_guidance = None
         self.guidance: Dict[str, List[List[int]]] = {}
 

monai/apps/deepgrow/dataset.py

@@ -11,7 +11,7 @@
 
 import logging
 import os
-from typing import Dict, List
+from typing import Dict, List, Union
 
 import numpy as np
 
@@ -144,7 +144,7 @@ def _default_transforms(image_key, label_key, pixdim):
 
 
 def _save_data_2d(vol_idx, vol_image, vol_label, dataset_dir, relative_path):
-    data_list = []
+    data_list: List[Dict[str, Union[str, int]]] = []
 
     image_count = 0
     label_count = 0
@@ -211,7 +211,7 @@ def _save_data_2d(vol_idx, vol_image, vol_label, dataset_dir, relative_path):
 
 
 def _save_data_3d(vol_idx, vol_image, vol_label, dataset_dir, relative_path):
-    data_list = []
+    data_list: List[Dict[str, Union[str, int]]] = []
 
     image_count = 0
     label_count = 0

monai/apps/deepgrow/transforms.py

@@ -9,7 +9,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import json
-from typing import Callable, Dict, Hashable, Optional, Sequence, Union
+from typing import Callable, Dict, Hashable, List, Optional, Sequence, Union
 
 import numpy as np
 import torch
@@ -437,8 +437,8 @@ def __call__(self, data):
 
         if np.all(np.less(current_size, self.spatial_size)):
             cropper = SpatialCrop(roi_center=center, roi_size=self.spatial_size)
-            box_start = np.array([s.start for s in cropper.slices])
-            box_end = np.array([s.stop for s in cropper.slices])
+            box_start = np.array([s.start for s in cropper.slices])  # type: ignore
+            box_end = np.array([s.stop for s in cropper.slices])  # type: ignore
         else:
             cropper = SpatialCrop(roi_start=box_start, roi_end=box_end)
 
@@ -523,11 +523,10 @@ def _apply(self, pos_clicks, neg_clicks, factor, slice_num):
         pos = neg = []
 
         if self.dimensions == 2:
-            points = list(pos_clicks)
+            points: List = list(pos_clicks)
             points.extend(neg_clicks)
-            points = np.array(points)
 
-            slices = list(np.unique(points[:, self.axis]))
+            slices = list(np.unique(np.array(points)[:, self.axis]))
             slice_idx = slices[0] if slice_num is None else next(x for x in slices if x == slice_num)
 
             if len(pos_clicks):
@@ -938,8 +937,7 @@ def _apply(self, image, guidance):
         for i, size_i in enumerate(image.shape):
             idx.append(slice_idx) if i == self.axis else idx.append(slice(0, size_i))
 
-        idx = tuple(idx)
-        return image[idx], idx
+        return image[tuple(idx)], tuple(idx)
 
     def __call__(self, data):
         d = dict(data)

monai/apps/nuclick/transforms.py

@@ -15,7 +15,7 @@
 import numpy as np
 import torch
 
-from monai.config import KeysCollection
+from monai.config import KeysCollection, NdarrayOrTensor
 from monai.networks.layers import GaussianFilter
 from monai.transforms import MapTransform, Randomizable, SpatialPad
 from monai.utils import StrEnum, convert_to_numpy, optional_import
@@ -337,14 +337,15 @@ def _apply_gaussion(self, t):
 
     def _seed_point(self, label):
         if distance_transform_cdt is None or not self.use_distance:
+            indices: NdarrayOrTensor
             if hasattr(torch, "argwhere"):
                 indices = torch.argwhere(label > 0)
             else:
                 indices = np.argwhere(convert_to_numpy(label) > 0)
 
             if len(indices) > 0:
-                idx = self.R.randint(0, len(indices))
-                return indices[idx, 0], indices[idx, 1]
+                index = self.R.randint(0, len(indices))
+                return indices[index, 0], indices[index, 1]
             return None
 
         distance = distance_transform_cdt(label).flatten()

monai/apps/pathology/data/datasets.py

@@ -11,7 +11,7 @@
 
 import os
 import sys
-from typing import Callable, Dict, List, Optional, Sequence, Tuple, Union
+from typing import Callable, Dict, List, Optional, Sequence, Tuple, Union, cast
 
 import numpy as np
 
@@ -62,14 +62,14 @@ def __init__(
     ):
         super().__init__(data, transform)
 
-        self.region_size = ensure_tuple_rep(region_size, 2)
-        self.grid_shape = ensure_tuple_rep(grid_shape, 2)
-        self.patch_size = ensure_tuple_rep(patch_size, 2)
+        self.region_size = cast(Tuple[int, int], ensure_tuple_rep(region_size, 2))
+        self.grid_shape = cast(Tuple[int, int], ensure_tuple_rep(grid_shape, 2))
+        self.patch_size = cast(Tuple[int, int], ensure_tuple_rep(patch_size, 2))
 
         self.image_path_list = list({x["image"] for x in self.data})
         self.image_reader_name = image_reader_name.lower()
         self.image_reader = WSIReader(backend=image_reader_name, **kwargs)
-        self.wsi_object_dict = None
+        self.wsi_object_dict: Optional[Dict] = None
         if self.image_reader_name != "openslide":
             # OpenSlide causes memory issue if we prefetch image objects
             self._fetch_wsi_objects()
@@ -85,11 +85,11 @@ def __getitem__(self, index):
         if self.image_reader_name == "openslide":
             img_obj = self.image_reader.read(sample["image"])
         else:
-            img_obj = self.wsi_object_dict[sample["image"]]
+            img_obj = cast(Dict, self.wsi_object_dict)[sample["image"]]
         location = [sample["location"][i] - self.region_size[i] // 2 for i in range(len(self.region_size))]
         images, _ = self.image_reader.get_data(
             img=img_obj,
-            location=location,
+            location=cast(Tuple[int, int], location),
             size=self.region_size,
             grid_shape=self.grid_shape,
             patch_size=self.patch_size,
@@ -209,7 +209,7 @@ def __init__(
     ) -> None:
         super().__init__(data, transform)
 
-        self.patch_size = ensure_tuple_rep(patch_size, 2)
+        self.patch_size = cast(Tuple[int, int], ensure_tuple_rep(patch_size, 2))
 
         # set up whole slide image reader
         self.image_reader_name = image_reader_name.lower()
@@ -309,7 +309,7 @@ def _load_a_patch(self, index):
         this method, first, finds the whole slide image and the patch that should be extracted,
         then it loads the patch and provide it with its image name and the corresponding mask location.
         """
-        sample_num = np.argmax(self.cum_num_patches > index) - 1
+        sample_num = cast(int, np.argmax(self.cum_num_patches > index)) - 1
         sample = self.data[sample_num]
         patch_num = index - self.cum_num_patches[sample_num]
         location_on_image = sample["image_locations"][patch_num]