Add test pipeline generator workflow and bump version to 1.0.0

- Introduced a new GitHub Actions workflow for testing the pipeline generator, including steps for setting up Python 3.10, installing dependencies, and running tests.
- Updated the version of the pipeline-generator package from 0.1.0 to 1.0.0 in the uv.lock file.
- Refactored import statements and improved code formatting in various files for better readability and consistency.

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

Author: mocsharp

.github/workflows/pr.yml

@@ -55,3 +55,22 @@ jobs:
         with:
           fail_ci_if_error: false
           files: ./coverage.xml
+
+  test-pipeline-generator:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python 3.10
+        uses: actions/setup-python@v2
+        with:
+          python-version: "3.10"
+      - name: Install uv
+        uses: astral-sh/setup-uv@v6
+      - name: Install dependencies
+        working-directory: tools/pipeline-generator
+        run: |
+          uv sync
+      - name: Run tests
+        working-directory: tools/pipeline-generator
+        run: |
+          uv run pytest

monai/deploy/operators/__init__.py

@@ -68,22 +68,22 @@
     ModelInfo,
 )
 from .image_directory_loader_operator import ImageDirectoryLoader
+from .image_overlay_writer_operator import ImageOverlayWriter
 from .inference_operator import InferenceOperator
 from .json_results_writer_operator import JSONResultsWriter
+from .llama3_vila_inference_operator import Llama3VILAInferenceOperator
 from .monai_bundle_inference_operator import (
     BundleConfigNames,
     IOMapping,
     MonaiBundleInferenceOperator,
 )
 from .monai_classification_operator import MonaiClassificationOperator
 from .monai_seg_inference_operator import InfererType, MonaiSegInferenceOperator
-from .nii_data_loader_operator import NiftiDataLoader
 from .nifti_directory_loader_operator import NiftiDirectoryLoader
 from .nifti_writer_operator import NiftiWriter
+from .nii_data_loader_operator import NiftiDataLoader
 from .png_converter_operator import PNGConverterOperator
+from .prompts_loader_operator import PromptsLoaderOperator
 from .publisher_operator import PublisherOperator
 from .stl_conversion_operator import STLConversionOperator, STLConverter
-from .image_overlay_writer_operator import ImageOverlayWriter
-from .prompts_loader_operator import PromptsLoaderOperator
-from .llama3_vila_inference_operator import Llama3VILAInferenceOperator
 from .vlm_results_writer_operator import VLMResultsWriterOperator

monai/deploy/operators/image_directory_loader_operator.py

@@ -142,6 +142,7 @@ def compute(self, op_input, op_output, context):
 def test():
     """Test the ImageDirectoryLoader operator."""
     import tempfile
+
     from PIL import Image as PILImageCreate
 
     # Create a temporary directory with test images

monai/deploy/operators/image_overlay_writer_operator.py

@@ -21,9 +21,9 @@
 - filename: base name (stem) for output file
 """
 
+import logging
 from pathlib import Path
 from typing import Optional, Tuple
-import logging
 
 import numpy as np
 
@@ -78,9 +78,7 @@ def _to_hwc_uint8(self, image) -> np.ndarray:
         else:
             arr = np.asarray(image)
         if arr.ndim != 3 or arr.shape[2] not in (3, 4):
-            raise ValueError(
-                f"Expected HWC image with 3 or 4 channels, got shape {arr.shape}"
-            )
+            raise ValueError(f"Expected HWC image with 3 or 4 channels, got shape {arr.shape}")
         # Drop alpha if present
         if arr.shape[2] == 4:
             arr = arr[..., :3]
@@ -105,17 +103,14 @@ def _to_mask_uint8(self, pred) -> np.ndarray:
         return arr
 
     @staticmethod
-    def _blend_overlay(
-        img: np.ndarray, mask_u8: np.ndarray, alpha: float, color: Tuple[int, int, int]
-    ) -> np.ndarray:
+    def _blend_overlay(img: np.ndarray, mask_u8: np.ndarray, alpha: float, color: Tuple[int, int, int]) -> np.ndarray:
         # img: HWC uint8, mask_u8: HW uint8
         mask = (mask_u8 > 0).astype(np.float32)[..., None]
         color_img = np.zeros_like(img, dtype=np.uint8)
         color_img[..., 0] = color[0]
         color_img[..., 1] = color[1]
         color_img[..., 2] = color[2]
         blended = (
-            img.astype(np.float32) * (1.0 - alpha * mask)
-            + color_img.astype(np.float32) * (alpha * mask)
+            img.astype(np.float32) * (1.0 - alpha * mask) + color_img.astype(np.float32) * (alpha * mask)
         ).astype(np.uint8)
         return blended

monai/deploy/operators/json_results_writer_operator.py

@@ -128,15 +128,11 @@ def _process_prediction(self, pred: Any, filename: str) -> Dict[str, Any]:
                     }
                 else:
                     # Generic classification
-                    result["probabilities"] = {
-                        f"class_{i}": float(pred_data[i]) for i in range(len(pred_data))
-                    }
+                    result["probabilities"] = {f"class_{i}": float(pred_data[i]) for i in range(len(pred_data))}
 
                 # Add predicted class
                 max_idx = int(np.argmax(pred_data))
-                result["predicted_class"] = list(result["probabilities"].keys())[
-                    max_idx
-                ]
+                result["predicted_class"] = list(result["probabilities"].keys())[max_idx]
                 result["confidence"] = float(pred_data[max_idx])
 
             elif pred_data.ndim == 2:  # 2D array (batch of predictions)
@@ -172,14 +168,13 @@ def _print_classification_summary(self, result: Dict[str, Any]):
         for class_name, prob in probs.items():
             print(f"  {class_name}: {prob:.4f}")
         if "predicted_class" in result:
-            print(
-                f"  Predicted: {result['predicted_class']} (confidence: {result['confidence']:.4f})"
-            )
+            print(f"  Predicted: {result['predicted_class']} (confidence: {result['confidence']:.4f})")
 
 
 def test():
     """Test the JSONResultsWriter operator."""
     import tempfile
+
     import numpy as np
 
     with tempfile.TemporaryDirectory() as temp_dir:

monai/deploy/operators/llama3_vila_inference_operator.py

@@ -116,9 +116,7 @@ def _load_model(self):
             config = AutoConfig.from_pretrained(self.model_path)
 
             # Load tokenizer
-            self.tokenizer = AutoTokenizer.from_pretrained(
-                self.model_path / "llm", use_fast=False
-            )
+            self.tokenizer = AutoTokenizer.from_pretrained(self.model_path / "llm", use_fast=False)
 
             # For LLaVA-style models, we typically need to handle image processing
             # and model loading in a specific way. For now, we'll create a simplified
@@ -156,17 +154,15 @@ def _preprocess_image(self, image: Image) -> torch.Tensor:
         # For now, we'll just convert to tensor
         return torch.from_numpy(image_array).float()
 
-    def _generate_response(
-        self, image_tensor: torch.Tensor, prompt: str, generation_params: Dict[str, Any]
-    ) -> str:
+    def _generate_response(self, image_tensor: torch.Tensor, prompt: str, generation_params: Dict[str, Any]) -> str:
         """Generate text response from the model."""
         if self._mock_mode:
             # Mock response based on common medical VQA patterns
             mock_responses = {
-                "what is this image showing": "This medical image shows anatomical structures with various tissue densities and contrast patterns.",
-                "summarize key findings": "Key findings include: 1) Normal anatomical structures visible, 2) No obvious pathological changes detected, 3) Image quality is adequate for assessment.",
-                "is there a focal lesion": "No focal lesion is identified in the visible field of view.",
-                "describe the image": "This appears to be a medical imaging study showing cross-sectional anatomy with good tissue contrast.",
+                "what is this image showing": "This medical image shows anatomical structures with various tissue densities and contrast patterns.",  # noqa: B950
+                "summarize key findings": "Key findings include: 1) Normal anatomical structures visible, 2) No obvious pathological changes detected, 3) Image quality is adequate for assessment.",  # noqa: B950
+                "is there a focal lesion": "No focal lesion is identified in the visible field of view.",  # noqa: B950
+                "describe the image": "This appears to be a medical imaging study showing cross-sectional anatomy with good tissue contrast.",  # noqa: B950
             }
 
             # Find best matching response
@@ -176,7 +172,7 @@ def _generate_response(
                     return response
 
             # Default response
-            return f"Analysis of the medical image based on the prompt: '{prompt}'. [Mock response - actual model not loaded]"
+            return f"Analysis of the medical image based on the prompt: {prompt!r}. [Mock response - actual model not loaded]"
 
         # In a real implementation, you would:
         # 1. Tokenize the prompt
@@ -189,8 +185,8 @@ def _create_json_result(
         self,
         text_response: str,
         request_id: str,
-        prompt: str = None,
-        image_metadata: Dict = None,
+        prompt: Optional[str] = None,
+        image_metadata: Optional[Dict] = None,
     ) -> Dict[str, Any]:
         """Create a JSON result from the text response."""
         result = {
@@ -276,44 +272,30 @@ def compute(self, op_input, op_output, context):
         request_id = op_input.receive("request_id")
         generation_params = op_input.receive("generation_params")
 
-        self._logger.info(
-            f"Processing request {request_id} with output type '{output_type}'"
-        )
+        self._logger.info(f"Processing request {request_id} with output type {output_type!r}")
 
         try:
             # Preprocess image
             image_tensor = self._preprocess_image(image)
 
             # Generate text response
-            text_response = self._generate_response(
-                image_tensor, prompt, generation_params
-            )
+            text_response = self._generate_response(image_tensor, prompt, generation_params)
 
             # Get image metadata if available
-            image_metadata = (
-                image.metadata()
-                if hasattr(image, "metadata") and callable(image.metadata)
-                else None
-            )
+            image_metadata = image.metadata() if hasattr(image, "metadata") and callable(image.metadata) else None
 
             # Create result based on output type
             if output_type == "json":
-                result = self._create_json_result(
-                    text_response, request_id, prompt, image_metadata
-                )
+                result = self._create_json_result(text_response, request_id, prompt, image_metadata)
             elif output_type == "image":
                 # For now, just return the original image
                 # In future, this could generate new images
                 result = image
             elif output_type == "image_overlay":
                 result = self._create_image_overlay(image, text_response)
             else:
-                self._logger.warning(
-                    f"Unknown output type: {output_type}, defaulting to json"
-                )
-                result = self._create_json_result(
-                    text_response, request_id, prompt, image_metadata
-                )
+                self._logger.warning(f"Unknown output type: {output_type}, defaulting to json")
+                result = self._create_json_result(text_response, request_id, prompt, image_metadata)
 
             # Emit outputs
             op_output.emit(result, "result")
@@ -335,3 +317,4 @@ def compute(self, op_input, op_output, context):
             op_output.emit(error_result, "result")
             op_output.emit(output_type, "output_type")
             op_output.emit(request_id, "request_id")
+            raise e from None