Refactor operator imports and enhance code clarity

- Updated operator imports in the MONAI Deploy SDK to streamline the structure and improve readability.
- Refined the NiftiDataLoader to ensure proper handling of various dimensionalities and added logging for unexpected shapes.
- Enhanced the pipeline generator to include new operators and improved the handling of output types.
- Cleaned up whitespace and formatting inconsistencies across multiple files for better code clarity.
- Removed deprecated test files related to GenericDirectoryScanner and VLM operators to maintain a clean codebase.

Signed-off-by: Victor Chang <[email protected]>

Author: mocsharp

monai/deploy/operators/__init__.py

@@ -1,4 +1,4 @@
-# Copyright 2021-2025 MONAI Consortium
+# Copyright 2021-2022 MONAI Consortium
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
@@ -22,57 +22,33 @@
     DICOMTextSRWriterOperator
     EquipmentInfo
     InferenceOperator
-    InfererType
     IOMapping
     ModelInfo
     MonaiBundleInferenceOperator
     MonaiSegInferenceOperator
-    NiftiDataLoader
     PNGConverterOperator
     PublisherOperator
-    SegmentDescription
     STLConversionOperator
     STLConverter
+    NiftiDataLoader
 """
 
 # If needed, can choose to expose some or all of Holoscan SDK built-in operators.
 # from holoscan.operators import *
-from holoscan.operators import (
-    PingRxOp,
-    PingTxOp,
-    VideoStreamRecorderOp,
-    VideoStreamReplayerOp,
-)
+from holoscan.operators import PingRxOp, PingTxOp, VideoStreamRecorderOp, VideoStreamReplayerOp
 
 from .clara_viz_operator import ClaraVizOperator
 from .dicom_data_loader_operator import DICOMDataLoaderOperator
 from .dicom_encapsulated_pdf_writer_operator import DICOMEncapsulatedPDFWriterOperator
-from .dicom_seg_writer_operator import (
-    DICOMSegmentationWriterOperator,
-    SegmentDescription,
-)
+from .dicom_seg_writer_operator import DICOMSegmentationWriterOperator
 from .dicom_series_selector_operator import DICOMSeriesSelectorOperator
 from .dicom_series_to_volume_operator import DICOMSeriesToVolumeOperator
-from .dicom_text_sr_writer_operator import (
-    DICOMTextSRWriterOperator,
-    EquipmentInfo,
-    ModelInfo,
-)
-
+from .dicom_text_sr_writer_operator import DICOMTextSRWriterOperator
+from .dicom_utils import EquipmentInfo, ModelInfo, random_with_n_digits, save_dcm_file, write_common_modules
 from .inference_operator import InferenceOperator
-
-from .monai_bundle_inference_operator import (
-    BundleConfigNames,
-    IOMapping,
-    MonaiBundleInferenceOperator,
-)
-
-from .monai_seg_inference_operator import InfererType, MonaiSegInferenceOperator
-
-
+from .monai_bundle_inference_operator import BundleConfigNames, IOMapping, MonaiBundleInferenceOperator
+from .monai_seg_inference_operator import MonaiSegInferenceOperator
 from .nii_data_loader_operator import NiftiDataLoader
 from .png_converter_operator import PNGConverterOperator
-
 from .publisher_operator import PublisherOperator
 from .stl_conversion_operator import STLConversionOperator, STLConverter
-

monai/deploy/operators/nii_data_loader_operator.py

@@ -81,7 +81,7 @@ def convert_and_save(self, nii_path):
         image_reader.SetFileName(str(nii_path))
         image = image_reader.Execute()
         image_np = SimpleITK.GetArrayFromImage(image)
-        
+
         # Handle different dimensionalities properly
         if image_np.ndim == 3:
             # Standard 3D volume: transpose from (z, y, x) to (x, y, z)
@@ -94,8 +94,10 @@ def convert_and_save(self, nii_path):
             image_np = np.transpose(image_np, [1, 0])
         else:
             # For other dimensions, log a warning and return as-is
-            self._logger.warning(f"Unexpected {image_np.ndim}D NIfTI file shape {image_np.shape} from {nii_path}, returning without transpose")
-        
+            self._logger.warning(
+                f"Unexpected {image_np.ndim}D NIfTI file shape {image_np.shape} from {nii_path}, returning without transpose"
+            )
+
         return image_np
 
 

tools/pipeline-generator/pipeline_generator/cli/run.py

@@ -28,39 +28,39 @@
 
 def _validate_results(output_dir: Path) -> tuple[bool, str]:
     """Validate that the application actually generated results.
-    
+
     Args:
         output_dir: Path to the output directory
-        
+
     Returns:
         Tuple of (success, message) where success is True if validation passed
     """
     if not output_dir.exists():
         return False, f"Output directory does not exist: {output_dir}"
-    
+
     # Check if any files were generated in the output directory
     output_files = list(output_dir.rglob("*"))
     result_files = [f for f in output_files if f.is_file()]
-    
+
     if not result_files:
         return False, f"No result files generated in {output_dir}"
-    
+
     # Count different types of output files
-    json_files = [f for f in result_files if f.suffix.lower() == '.json']
-    nifti_files = [f for f in result_files if f.suffix.lower() in ['.nii', '.gz']]
-    image_files = [f for f in result_files if f.suffix.lower() in ['.png', '.jpg', '.jpeg', '.tiff']]
+    json_files = [f for f in result_files if f.suffix.lower() == ".json"]
+    nifti_files = [f for f in result_files if f.suffix.lower() in [".nii", ".gz"]]
+    image_files = [f for f in result_files if f.suffix.lower() in [".png", ".jpg", ".jpeg", ".tiff"]]
     other_files = [f for f in result_files if f not in json_files + nifti_files + image_files]
-    
+
     file_summary = []
     if json_files:
         file_summary.append(f"{len(json_files)} JSON files")
     if nifti_files:
-        file_summary.append(f"{len(nifti_files)} NIfTI files") 
+        file_summary.append(f"{len(nifti_files)} NIfTI files")
     if image_files:
         file_summary.append(f"{len(image_files)} image files")
     if other_files:
         file_summary.append(f"{len(other_files)} other files")
-    
+
     summary = ", ".join(file_summary) if file_summary else f"{len(result_files)} files"
     return True, f"Generated {summary}"
 

tools/pipeline-generator/pipeline_generator/generator/app_generator.py

@@ -12,6 +12,7 @@
 """Generate MONAI Deploy applications from MONAI Bundles."""
 
 import logging
+import re
 from pathlib import Path
 from typing import Any, Dict, Optional
 
@@ -20,8 +21,6 @@
 from ..config.settings import Settings, load_config
 from .bundle_downloader import BundleDownloader
 
-import re
-
 logger = logging.getLogger(__name__)
 
 
@@ -463,48 +462,43 @@ def _copy_operators(self, output_dir: Path, context: Dict[str, Any]) -> None:
             context: Template context
         """
         import shutil
-        
+
         # Map operator usage based on context
         needed_operators = []
-        
-        input_type = context.get('input_type', '')
-        output_type = context.get('output_type', '')
-        task = context.get('task', '').lower()
-        
+
+        input_type = context.get("input_type", "")
+        output_type = context.get("output_type", "")
+        task = context.get("task", "").lower()
+
         # Determine which operators are needed based on the application type
         if input_type == "image":
-            needed_operators.extend([
-                'generic_directory_scanner_operator.py',
-                'image_file_loader_operator.py'
-            ])
+            needed_operators.extend(["generic_directory_scanner_operator.py", "image_file_loader_operator.py"])
         elif input_type == "custom":
-            needed_operators.extend([
-                'llama3_vila_inference_operator.py',
-                'prompts_loader_operator.py',
-                'vlm_results_writer_operator.py'
-            ])
+            needed_operators.extend(
+                ["llama3_vila_inference_operator.py", "prompts_loader_operator.py", "vlm_results_writer_operator.py"]
+            )
         elif input_type == "nifti":
-            needed_operators.append('generic_directory_scanner_operator.py')
-            
+            needed_operators.append("generic_directory_scanner_operator.py")
+
         if output_type == "json":
-            needed_operators.append('json_results_writer_operator.py')
+            needed_operators.append("json_results_writer_operator.py")
         elif output_type == "image_overlay":
-            needed_operators.append('image_overlay_writer_operator.py')
+            needed_operators.append("image_overlay_writer_operator.py")
         elif output_type == "nifti":
-            needed_operators.append('nifti_writer_operator.py')
-            
+            needed_operators.append("nifti_writer_operator.py")
+
         if "classification" in task and input_type == "image":
-            needed_operators.append('monai_classification_operator.py')
-            
+            needed_operators.append("monai_classification_operator.py")
+
         # Remove duplicates
         needed_operators = list(set(needed_operators))
-        
+
         if needed_operators:
             # Get the operators directory in templates
             operators_dir = Path(__file__).parent.parent / "templates" / "operators"
-            
+
             logger.info(f"Copying {len(needed_operators)} operators to generated application")
-            
+
             for operator_file in needed_operators:
                 src_path = operators_dir / operator_file
                 if src_path.exists():

tools/pipeline-generator/pipeline_generator/templates/app.py.j2

@@ -55,6 +55,7 @@ from vlm_results_writer_operator import VLMResultsWriterOperator
 
 {% else %}
 from generic_directory_scanner_operator import GenericDirectoryScanner
+
 from monai.deploy.operators.nii_data_loader_operator import NiftiDataLoader
 
 {% endif %}

tools/pipeline-generator/pipeline_generator/templates/operators/generic_directory_scanner_operator.py

@@ -18,11 +18,11 @@
 
 class GenericDirectoryScanner(Operator):
     """Scan a directory for files matching specified extensions and emit file paths one by one.
-    
+
     This operator provides a generic way to iterate through files in a directory,
     emitting one file path at a time. It can be chained with file-specific loaders
     to create flexible data loading pipelines.
-    
+
     Named Outputs:
         file_path: Path to the current file being processed
         filename: Name of the current file (without extension)
@@ -51,10 +51,10 @@ def __init__(
         """
         self._logger = logging.getLogger("{}.{}".format(__name__, type(self).__name__))
         self._input_folder = Path(input_folder)
-        self._file_extensions = [ext if ext.startswith('.') else f'.{ext}' for ext in file_extensions]
+        self._file_extensions = [ext if ext.startswith(".") else f".{ext}" for ext in file_extensions]
         self._recursive = bool(recursive)
         self._case_sensitive = bool(case_sensitive)
-        
+
         # State tracking
         self._files = []
         self._current_index = 0
@@ -64,69 +64,65 @@ def __init__(
     def _find_files(self) -> List[Path]:
         """Find all files matching the specified extensions."""
         files = []
-        
+
         # Normalize extensions for comparison
         if not self._case_sensitive:
             extensions = [ext.lower() for ext in self._file_extensions]
         else:
             extensions = self._file_extensions
-        
+
         # Choose search method based on recursive flag
         if self._recursive:
             search_pattern = "**/*"
             search_method = self._input_folder.rglob
         else:
             search_pattern = "*"
             search_method = self._input_folder.glob
-        
+
         # Find all files and filter by extension
         for file_path in search_method(search_pattern):
             if file_path.is_file():
                 # Skip hidden files (starting with .) to avoid macOS metadata files like ._file.nii.gz
-                if file_path.name.startswith('.'):
+                if file_path.name.startswith("."):
                     continue
-                
+
                 # Handle compound extensions like .nii.gz by checking if filename ends with any extension
                 filename = file_path.name
                 if not self._case_sensitive:
                     filename = filename.lower()
-                
+
                 # Check if filename ends with any of the specified extensions
                 for ext in extensions:
                     if filename.endswith(ext):
                         files.append(file_path)
                         break  # Only add once even if multiple extensions match
-        
+
         # Sort files for consistent ordering
         files.sort()
         return files
 
     def setup(self, spec: OperatorSpec):
         """Define the operator outputs."""
         spec.output("file_path")
-        spec.output("filename") 
+        spec.output("filename")
         spec.output("file_index").condition(ConditionType.NONE)
         spec.output("total_files").condition(ConditionType.NONE)
 
         # Pre-initialize the files list
         if not self._input_folder.is_dir():
             raise ValueError(f"Input folder {self._input_folder} is not a directory")
-        
+
         self._files = self._find_files()
         self._current_index = 0
 
         if not self._files:
-            self._logger.warning(
-                f"No files found in {self._input_folder} with extensions {self._file_extensions}"
-            )
+            self._logger.warning(f"No files found in {self._input_folder} with extensions {self._file_extensions}")
         else:
-            self._logger.info(
-                f"Found {len(self._files)} files to process with extensions {self._file_extensions}"
-            )
+            self._logger.info(f"Found {len(self._files)} files to process with extensions {self._file_extensions}")
 
     def compute(self, op_input, op_output, context):
         """Emit the next file path."""
-        
+
         # Check if we have more files to process
         if self._current_index >= len(self._files):
             # No more files to process
@@ -144,9 +140,7 @@ def compute(self, op_input, op_output, context):
             op_output.emit(self._current_index, "file_index")
             op_output.emit(len(self._files), "total_files")
 
-            self._logger.info(
-                f"Emitted file: {file_path.name} ({self._current_index + 1}/{len(self._files)})"
-            )
+            self._logger.info(f"Emitted file: {file_path.name} ({self._current_index + 1}/{len(self._files)})")
 
         except Exception as e:
             self._logger.error(f"Failed to process file {file_path}: {e}")
@@ -162,16 +156,13 @@ def test():
     # Create a temporary directory with test files
     with tempfile.TemporaryDirectory() as temp_dir:
         temp_path = Path(temp_dir)
-        
+
         # Create test files with different extensions
-        test_files = [
-            "test1.jpg", "test2.png", "test3.nii", "test4.nii.gz", 
-            "test5.txt", "test6.jpeg"
-        ]
-        
+        test_files = ["test1.jpg", "test2.png", "test3.nii", "test4.nii.gz", "test5.txt", "test6.jpeg"]
+
         for filename in test_files:
             (temp_path / filename).touch()
-        
+
         # Create a subdirectory with more files
         sub_dir = temp_path / "subdir"
         sub_dir.mkdir()
@@ -181,14 +172,12 @@ def test():
         # Test the operator with image extensions
         fragment = Fragment()
         scanner = GenericDirectoryScanner(
-            fragment, 
-            input_folder=temp_path,
-            file_extensions=['.jpg', '.jpeg', '.png'],
-            recursive=True
+            fragment, input_folder=temp_path, file_extensions=[".jpg", ".jpeg", ".png"], recursive=True
         )
 
         # Simulate setup
         from monai.deploy.core import OperatorSpec
+
         spec = OperatorSpec()
         scanner.setup(spec)