[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

monai/apps/nnunet/nnunetv2_runner.py

@@ -19,9 +19,9 @@
 import shlex
 import shutil
 import subprocess
-from tempfile import NamedTemporaryFile, TemporaryDirectory
 import warnings
 from concurrent.futures import ThreadPoolExecutor
+from tempfile import NamedTemporaryFile, TemporaryDirectory
 from typing import Any
 
 import monai
@@ -30,11 +30,10 @@
     analyze_data,
     create_new_data_copy,
     create_new_dataset_json,
-    glob_to_datalist,
-    check_existing_data_indices,
-    get_next_available_index,
     get_info_from_dataset_json,
-    move_predictions
+    get_next_available_index,
+    glob_to_datalist,
+    move_predictions,
 )
 from monai.bundle import ConfigParser
 from monai.utils import ensure_tuple, optional_import
@@ -212,8 +211,8 @@ def __init__(
         os.environ["OMP_NUM_THREADS"] = str(1)
 
         # dataset_name_or_id has to be a string
-        if 'dataset_name_or_id' in self.input_info:
-            self.dataset_name_or_id = str(self.input_info['dataset_name_or_id'])
+        if "dataset_name_or_id" in self.input_info:
+            self.dataset_name_or_id = str(self.input_info["dataset_name_or_id"])
         else:
             # we get the next available index
             self.dataset_name_or_id = str(get_next_available_index(self.nnunet_raw))
@@ -285,7 +284,9 @@ def convert_dataset(self, testing=False):
             if isinstance(datalist_json[test_key][0], dict) and "label" in datalist_json[test_key][0]:
                 os.makedirs(os.path.join(raw_data_foldername, "labelsTs"))
 
-        num_input_channels, num_foreground_classes = self.input_info.get('num_input_channels'), self.input_info.get('num_foreground_classes')
+        num_input_channels, num_foreground_classes = self.input_info.get("num_input_channels"), self.input_info.get(
+            "num_foreground_classes"
+        )
 
         if num_input_channels is None or num_foreground_classes is None:
             # can't get num_foreground classes from the data, so should be inserted by user
@@ -296,7 +297,7 @@ def convert_dataset(self, testing=False):
             modality = [modality]
 
         create_new_dataset_json(
-          # pyrefly: ignore [bad-argument-type]
+            # pyrefly: ignore [bad-argument-type]
             modality=modality,
             num_foreground_classes=num_foreground_classes,
             num_input_channels=num_input_channels,
@@ -439,13 +440,7 @@ def preprocess(
         logger.info("Preprocessing...")
 
         plans_name = overwrite_plans_name if overwrite_plans_name is not None else self.plans_identifier
-        preprocess(
-            [int(self.dataset_name_or_id)],
-            plans_name,
-            configurations=c,
-            num_processes=n_proc,
-            verbose=verbose,
-        )
+        preprocess([int(self.dataset_name_or_id)], plans_name, configurations=c, num_processes=n_proc, verbose=verbose)
         self.plans_identifier = plans_name
 
     def plan_and_process(
@@ -1058,7 +1053,7 @@ def predict_datalist(
         modality: str = "CT",
         num_foreground_classes: int | None = None,
         num_input_channels: int | None = None,
-        work_dir: str = 'work_dir',
+        work_dir: str = "work_dir",
     ):
         """Method to run inference on a datalist using a model trained by this runner.
         Handles all nnUNet boilerplate, instantiation of the runner, etc.
@@ -1112,17 +1107,13 @@ def predict_datalist(
         # these things are hardcoded upstream
         raw_data_foldername_prefix = str(int(runner.dataset_name_or_id) + 1000)
         raw_data_foldername_prefix = "Dataset" + raw_data_foldername_prefix[-3:]
-        raw_data_foldername = raw_data_foldername_prefix + "_" + input_config['dataroot'].split(os.sep)[-1]
-        raw_data_foldername = os.path.join(input_config['nnunet_raw'], raw_data_foldername)
+        raw_data_foldername = raw_data_foldername_prefix + "_" + input_config["dataroot"].split(os.sep)[-1]
+        raw_data_foldername = os.path.join(input_config["nnunet_raw"], raw_data_foldername)
 
         with TemporaryDirectory() as pred_work_folder:
             test_images_dir = os.path.join(raw_data_foldername, "imagesTs")  # Also hardcoded upstream
 
-            runner.predict(
-                test_images_dir,
-                output_folder=pred_work_folder,
-                model_training_output_dir=model_dir,
-            )
+            runner.predict(test_images_dir, output_folder=pred_work_folder, model_training_output_dir=model_dir)
             move_predictions(raw_data_foldername, pred_work_folder, output_dir)
 
         # now we can delete the raw data folder too
@@ -1141,18 +1132,18 @@ def predict_files_glob(
         modality: str = "CT",
     ):
         """Method to run inference on a glob of files using a model trained by this runner.
-        
+
         Creates a temporary datalist json file from the glob of files, and then calls predict_datalist.
-        
+
         Args:
             input_files_glob: glob pattern to match input files (e.g., ``/path/to/images/*.nii.gz``)
             input_files_root: root directory for the input files (e.g., ``/path/to/images``)
             model_dir: path to the folder containing the trained model (full path inside the work_dir, e.g., ``work_dir/nnUNet_trained_models/Dataset001_data/nnUNetTrainer__nnUNetPlans__3d_fullres``)
             output_dir: path to the output directory, predictions will be saved here under their original names.
-            work_dir: path to the work_dir created by the runner during training. 
+            work_dir: path to the work_dir created by the runner during training.
             modality: modality of the input data (default: "CT")
         """
-        with NamedTemporaryFile(mode='w+', delete=False, suffix='.json') as temp_json_file:
+        with NamedTemporaryFile(mode="w+", delete=False, suffix=".json") as temp_json_file:
             temp_json_path = temp_json_file.name
             glob_to_datalist(input_files_glob, output_json=temp_json_path, key="testing", dataroot=input_files_root)
 
@@ -1162,23 +1153,25 @@ def predict_files_glob(
                 model_dir=model_dir,
                 work_dir=work_dir,
                 output_dir=output_dir,
-                modality=modality
+                modality=modality,
             )
 
     def _determine_configs(self):
         from nnunetv2.paths import nnUNet_preprocessed
         from nnunetv2.utilities.dataset_name_id_conversion import maybe_convert_to_dataset_name
 
-        preprocessed_dataset_folder_base = join(nnUNet_preprocessed, maybe_convert_to_dataset_name(self.dataset_name_or_id))
-        plans_file = join(preprocessed_dataset_folder_base, self.plans_identifier + '.json')
+        preprocessed_dataset_folder_base = join(
+            nnUNet_preprocessed, maybe_convert_to_dataset_name(self.dataset_name_or_id)
+        )
+        plans_file = join(preprocessed_dataset_folder_base, self.plans_identifier + ".json")
 
-        with open(plans_file, 'r') as f:
+        with open(plans_file) as f:
             plans = json.load(f)
 
-        configurations = plans.get('configurations', [])
+        configurations = plans.get("configurations", [])
         if not configurations:
             raise ValueError(f"No configurations found in plans file: {plans_file}")
-        
+
         config_names = list(configurations.keys())
         return config_names
 

monai/apps/nnunet/utils.py

@@ -171,18 +171,12 @@ def create_new_dataset_json(
     new_json_data["file_ending"] = ".nii.gz"
 
     ConfigParser.export_config_file(
-        config=new_json_data,
-        filepath=output_filepath,
-        fmt="json",
-        sort_keys=True,
-        indent=4,
-        ensure_ascii=False,
+        config=new_json_data, filepath=output_filepath, fmt="json", sort_keys=True, indent=4, ensure_ascii=False
     )
 
     return
 
 
-
 def glob_to_datalist(glob_pattern, output_json="datalist.json", key="testing", dataroot=None):
     files = sorted(glob.glob(glob_pattern, recursive=True))
     if not files:
@@ -232,37 +226,37 @@ def get_info_from_dataset_json(model_dir):
     if not os.path.exists(dataset_json_path):
         raise FileNotFoundError(f"dataset.json not found in model directory '{model_dir}'")
 
-    with open(dataset_json_path, "r") as f:
+    with open(dataset_json_path) as f:
         dataset_info = json.load(f)
 
     channel_names = dataset_info.get("channel_names", [])
     num_input_channels = len(channel_names)
 
     labels = dataset_info.get("labels", {})
-    num_foreground_classes = len(labels) - 1 if 'background' in labels else len(labels)  # Exclude background if present
+    num_foreground_classes = len(labels) - 1 if "background" in labels else len(labels)  # Exclude background if present
 
     return num_input_channels, num_foreground_classes
 
 
 def move_predictions(raw_data_foldername, pred_work_folder, output_dir):
-    # the output is now per 'case'. We need to use the generated datalist to map the output back to the original input files. 
-        # so we have the datalist
+    # the output is now per 'case'. We need to use the generated datalist to map the output back to the original input files.
+    # so we have the datalist
     datalist_path = os.path.join(raw_data_foldername, "datalist.json")
-    with open(datalist_path, "r") as f:
+    with open(datalist_path) as f:
         datalist = json.load(f)
 
-    if 'test' in datalist:
-        key = 'test'
-    elif 'testing' in datalist:
-        key = 'testing'
+    if "test" in datalist:
+        key = "test"
+    elif "testing" in datalist:
+        key = "testing"
     else:
         raise ValueError(f"Warning: Neither 'test' nor 'testing' key found in datalist '{datalist_path}'")
 
     test_cases = datalist[key]
     if not test_cases:
         raise ValueError(f"Warning: No test cases found in datalist '{datalist_path}'")
-        
-    case_to_image_path = {item['new_name']: item['image'] for item in test_cases}
+
+    case_to_image_path = {item["new_name"]: item["image"] for item in test_cases}
 
     os.makedirs(output_dir, exist_ok=True)
 
@@ -273,11 +267,12 @@ def move_predictions(raw_data_foldername, pred_work_folder, output_dir):
             continue
 
         # Copy the prediction file to the output directory with the original image name
-        image_extension = os.path.split(image_path, '.', 1)[1]  # assumes no periods in filename, supports .nii.gz
-        output_prediction_path = os.path.join(output_dir, image_path.replace(image_extension, '_pred.nii.gz'))  # nnunet outputs nii.gz
+        image_extension = os.path.split(image_path, ".", 1)[1]  # assumes no periods in filename, supports .nii.gz
+        output_prediction_path = os.path.join(
+            output_dir, image_path.replace(image_extension, "_pred.nii.gz")
+        )  # nnunet outputs nii.gz
         output_prediction_folder = os.path.dirname(output_prediction_path)
         os.makedirs(output_prediction_folder, exist_ok=True)  # sometimes can be nested folders
 
         shutil.move(prediction_file, output_prediction_path)
         print(f"Moved prediction for case '{case_name}' to '{output_prediction_path}'")
-