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

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

Author: pre-commit-ci[bot]

deployment/fastapi_inference/app/inference.py

@@ -32,19 +32,17 @@ class InferenceEngine:
 
     def __init__(self):
         """Initialize the inference engine with preprocessing transforms."""
-        self.preprocess = Compose([
-            LoadImage(image_only=True),
-            EnsureChannelFirst(),
-            Spacing(pixdim=(1.5, 1.5, 2.0)),
-            ScaleIntensity(),
-            EnsureType(dtype=torch.float32),
-        ])
-
-    async def process_image(
-        self,
-        image_bytes: bytes,
-        filename: str
-    ) -> Dict:
+        self.preprocess = Compose(
+            [
+                LoadImage(image_only=True),
+                EnsureChannelFirst(),
+                Spacing(pixdim=(1.5, 1.5, 2.0)),
+                ScaleIntensity(),
+                EnsureType(dtype=torch.float32),
+            ]
+        )
+
+    async def process_image(self, image_bytes: bytes, filename: str) -> Dict:
         """
         Process an uploaded image and return predictions.
 
@@ -80,13 +78,15 @@ async def process_image(
             result = {
                 "success": True,
                 "prediction": self._format_prediction(prediction),
-                "segmentation_shape": list(prediction.shape) if isinstance(prediction, (np.ndarray, torch.Tensor)) else None,
+                "segmentation_shape": (
+                    list(prediction.shape) if isinstance(prediction, (np.ndarray, torch.Tensor)) else None
+                ),
                 "metadata": {
                     "image_shape": list(image_tensor.shape),
                     "processing_time": round(processing_time, 3),
                     "device": str(model_loader.device),
                 },
-                "message": f"Inference completed successfully in {processing_time:.3f}s"
+                "message": f"Inference completed successfully in {processing_time:.3f}s",
             }
 
             logger.info(f"Inference completed in {processing_time:.3f}s")
@@ -112,15 +112,12 @@ def _load_image(self, image_buffer: BytesIO, filename: str) -> np.ndarray:
         """
         try:
             # Support NIfTI format (.nii, .nii.gz)
-            if filename.endswith(('.nii', '.nii.gz')):
+            if filename.endswith((".nii", ".nii.gz")):
                 image_buffer.seek(0)
                 img = nib.load(image_buffer)
                 return img.get_fdata()
             else:
-                raise ValueError(
-                    f"Unsupported image format: {filename}. "
-                    "Supported formats: .nii, .nii.gz"
-                )
+                raise ValueError(f"Unsupported image format: {filename}. " "Supported formats: .nii, .nii.gz")
         except Exception as e:
             raise ValueError(f"Failed to load image: {str(e)}")
 
@@ -170,7 +167,7 @@ async def _run_inference(self, image_tensor: torch.Tensor) -> torch.Tensor:
 
             # Run inference with no gradient computation
             with torch.no_grad():
-                if hasattr(model, '__call__'):
+                if hasattr(model, "__call__"):
                     prediction = model(image_tensor)
                 else:
                     raise RuntimeError("Model is not callable")

deployment/fastapi_inference/app/main.py

@@ -17,10 +17,7 @@
 from .schemas import HealthResponse, PredictionResponse, ErrorResponse
 
 # Configure logging
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-)
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
 logger = logging.getLogger(__name__)
 
 
@@ -33,10 +30,7 @@ async def lifespan(app: FastAPI):
     # Startup: Load the MONAI model
     logger.info("Starting up: Loading MONAI model...")
     try:
-        model_loader.load_model(
-            model_name="spleen_ct_segmentation",
-            bundle_dir="./models"
-        )
+        model_loader.load_model(model_name="spleen_ct_segmentation", bundle_dir="./models")
         logger.info("Model loaded successfully!")
     except Exception as e:
         logger.error(f"Failed to load model: {e}")
@@ -77,8 +71,8 @@ async def global_exception_handler(request, exc):
         content={
             "error": "InternalServerError",
             "detail": "An unexpected error occurred. Please try again.",
-            "status_code": 500
-        }
+            "status_code": 500,
+        },
     )
 
 
@@ -97,7 +91,7 @@ async def root():
             "health": "/health",
             "predict": "/predict",
             "docs": "/docs",
-        }
+        },
     }
 
 
@@ -132,14 +126,9 @@ async def health_check():
         200: {"description": "Successful prediction"},
         400: {"model": ErrorResponse, "description": "Bad request"},
         500: {"model": ErrorResponse, "description": "Internal server error"},
-    }
+    },
 )
-async def predict(
-    file: UploadFile = File(
-        ...,
-        description="Medical image file (NIfTI format: .nii or .nii.gz)"
-    )
-):
+async def predict(file: UploadFile = File(..., description="Medical image file (NIfTI format: .nii or .nii.gz)")):
     """
     Run inference on uploaded medical image.
 
@@ -153,64 +142,46 @@ async def predict(
         HTTPException: If file format is invalid or inference fails
     """
     # Validate file format
-    if not file.filename.endswith(('.nii', '.nii.gz')):
+    if not file.filename.endswith((".nii", ".nii.gz")):
         raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="Invalid file format. Supported formats: .nii, .nii.gz"
+            status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid file format. Supported formats: .nii, .nii.gz"
         )
 
     # Check if model is loaded
     if not model_loader.is_loaded():
         raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail="Model not loaded. Please try again later."
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Model not loaded. Please try again later."
         )
 
     try:
         # Read file content
         contents = await file.read()
 
         # Run inference
-        result = await inference_engine.process_image(
-            image_bytes=contents,
-            filename=file.filename
-        )
+        result = await inference_engine.process_image(image_bytes=contents, filename=file.filename)
 
         return PredictionResponse(**result)
 
     except ValueError as e:
         # Client error (bad input)
         logger.warning(f"Bad request: {str(e)}")
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=str(e)
-        )
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
 
     except RuntimeError as e:
         # Server error (inference failed)
         logger.error(f"Inference error: {str(e)}")
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Inference failed: {str(e)}"
-        )
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Inference failed: {str(e)}")
 
     except Exception as e:
         # Unexpected error
         logger.error(f"Unexpected error during prediction: {str(e)}")
         raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="An unexpected error occurred during prediction"
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="An unexpected error occurred during prediction"
         )
 
 
 if __name__ == "__main__":
     import uvicorn
 
     # For development only - use proper ASGI server in production
-    uvicorn.run(
-        "main:app",
-        host="0.0.0.0",
-        port=8000,
-        reload=True,
-        log_level="info"
-    )
+    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True, log_level="info")

deployment/fastapi_inference/app/model_loader.py

@@ -46,11 +46,7 @@ def _setup_device(self):
             self._device = torch.device("cpu")
             logger.info("Using CPU for inference")
 
-    def load_model(
-        self,
-        model_name: str = "spleen_ct_segmentation",
-        bundle_dir: str = "./models"
-    ) -> None:
+    def load_model(self, model_name: str = "spleen_ct_segmentation", bundle_dir: str = "./models") -> None:
         """
         Load a MONAI model bundle.
 
@@ -74,18 +70,14 @@ def load_model(
 
             # Load the model
             logger.info("Loading model into memory...")
-            self._model = load(
-                name=model_name,
-                bundle_dir=bundle_dir,
-                source="monaihosting"
-            )
+            self._model = load(name=model_name, bundle_dir=bundle_dir, source="monaihosting")
 
             # Move model to device
-            if hasattr(self._model, 'to'):
+            if hasattr(self._model, "to"):
                 self._model = self._model.to(self._device)
 
             # Set model to evaluation mode
-            if hasattr(self._model, 'eval'):
+            if hasattr(self._model, "eval"):
                 self._model.eval()
 
             logger.info("Model loaded successfully")
@@ -98,9 +90,7 @@ def load_model(
     def model(self):
         """Get the loaded model instance."""
         if self._model is None:
-            raise RuntimeError(
-                "Model not loaded. Call load_model() first."
-            )
+            raise RuntimeError("Model not loaded. Call load_model() first.")
         return self._model
 
     @property

deployment/fastapi_inference/app/schemas.py

@@ -29,14 +29,8 @@ class PredictionResponse(BaseModel):
     """Response model for inference predictions."""
 
     success: bool = Field(..., description="Whether prediction was successful")
-    prediction: Optional[Dict] = Field(
-        None,
-        description="Prediction results (format depends on model output)"
-    )
-    segmentation_shape: Optional[List[int]] = Field(
-        None,
-        description="Shape of segmentation mask if applicable"
-    )
+    prediction: Optional[Dict] = Field(None, description="Prediction results (format depends on model output)")
+    segmentation_shape: Optional[List[int]] = Field(None, description="Shape of segmentation mask if applicable")
     metadata: PredictionMetadata = Field(..., description="Prediction metadata")
     message: Optional[str] = Field(None, description="Additional information or error message")
 

deployment/fastapi_inference/examples/client.py

@@ -63,24 +63,10 @@ def predict(self, image_path: str) -> dict:
 
 def main():
     """Main function for command-line usage."""
-    parser = argparse.ArgumentParser(
-        description="MONAI FastAPI Inference Client"
-    )
-    parser.add_argument(
-        "--url",
-        default="http://localhost:8000",
-        help="API base URL (default: http://localhost:8000)"
-    )
-    parser.add_argument(
-        "--health",
-        action="store_true",
-        help="Check API health status"
-    )
-    parser.add_argument(
-        "--image",
-        type=str,
-        help="Path to medical image file for prediction"
-    )
+    parser = argparse.ArgumentParser(description="MONAI FastAPI Inference Client")
+    parser.add_argument("--url", default="http://localhost:8000", help="API base URL (default: http://localhost:8000)")
+    parser.add_argument("--health", action="store_true", help="Check API health status")
+    parser.add_argument("--image", type=str, help="Path to medical image file for prediction")
 
     args = parser.parse_args()