Migrated the breast density classification app, and fixed a bug.

Signed-off-by: M Q <mingmelvinq@nvidia.com>

Author: MMelQin

examples/apps/breast_density_classifer_app/__main__.py

@@ -1,4 +1,11 @@
+import logging
+
 from app import BreastClassificationApp
 
+from monai.deploy.logger import load_env_log_level
+
 if __name__ == "__main__":
-    BreastClassificationApp(do_run=True)
+    load_env_log_level()
+    logging.info(f"Begin {__name__}")
+    BreastClassificationApp().run()
+    logging.info(f"End {__name__}")

examples/apps/breast_density_classifer_app/app.py

@@ -1,42 +1,78 @@
+import logging
+from pathlib import Path
+
 from breast_density_classifier_operator import ClassifierOperator
 
-from monai.deploy.core import Application, env
+from monai.deploy.conditions import CountCondition
+from monai.deploy.core import AppContext, Application
+from monai.deploy.logger import load_env_log_level
 from monai.deploy.operators.dicom_data_loader_operator import DICOMDataLoaderOperator
 from monai.deploy.operators.dicom_series_selector_operator import DICOMSeriesSelectorOperator
 from monai.deploy.operators.dicom_series_to_volume_operator import DICOMSeriesToVolumeOperator
 from monai.deploy.operators.dicom_text_sr_writer_operator import DICOMTextSRWriterOperator, EquipmentInfo, ModelInfo
 
 
-@env(pip_packages=["highdicom>=0.18.2"])
+# @env(pip_packages=["monai~=1.1.0", "highdicom>=0.18.2", "pydicom >= 2.3.0"])
 class BreastClassificationApp(Application):
+    """This is an AI breast density classification application.
+
+    The DL model was trained by Center for Augmented Intelligence in Imaging, Mayo Clinic, Florida,
+    and published on MONAI Model Zoo at
+        https://github.com/Project-MONAI/model-zoo/tree/dev/models/breast_density_classification
+    """
+
     def __init__(self, *args, **kwargs):
+        self._logger = logging.getLogger("{}.{}".format(__name__, type(self).__name__))
         super().__init__(*args, **kwargs)
 
     def compose(self):
+        """Creates the app specific operators and chain them up in the processing DAG."""
+        logging.info(f"Begin {self.compose.__name__}")
+
+        app_context = AppContext({})  # Let it figure out all the attributes without overriding
+        app_input_path = Path(app_context.input_path)
+        app_output_path = Path(app_context.output_path)
+        model_path = Path(app_context.model_path)
+
         model_info = ModelInfo(
             "MONAI Model for Breast Density",
             "BreastDensity",
             "0.1",
             "Center for Augmented Intelligence in Imaging, Mayo Clinic, Florida",
         )
+
         my_equipment = EquipmentInfo(manufacturer="MONAI Deploy App SDK", manufacturer_model="DICOM SR Writer")
         my_special_tags = {"SeriesDescription": "Not for clinical use"}
-        study_loader_op = DICOMDataLoaderOperator()
-        series_selector_op = DICOMSeriesSelectorOperator(rules=Sample_Rules_Text)
-        series_to_vol_op = DICOMSeriesToVolumeOperator()
-        classifier_op = ClassifierOperator()
+        study_loader_op = DICOMDataLoaderOperator(
+            self, CountCondition(self, 1), input_folder=app_input_path, name="study_loader_op"
+        )
+        series_selector_op = DICOMSeriesSelectorOperator(
+            self, rules_json_str=Sample_Rules_Text, name="series_selector_op"
+        )
+        series_to_vol_op = DICOMSeriesToVolumeOperator(self, name="series_to_vol_op")
+        classifier_op = ClassifierOperator(
+            self, output_folder=app_output_path, model_path=model_path, name="classifier_op"
+        )
         sr_writer_op = DICOMTextSRWriterOperator(
-            copy_tags=True, model_info=model_info, equipment_info=my_equipment, custom_tags=my_special_tags
+            self,
+            copy_tags=True,
+            model_info=model_info,
+            equipment_info=my_equipment,
+            custom_tags=my_special_tags,
+            output_folder=app_output_path,
+            name="sr_writer_op",
         )  # copy_tags=True to use Study and Patient modules of the original input
 
-        self.add_flow(study_loader_op, series_selector_op, {"dicom_study_list": "dicom_study_list"})
+        self.add_flow(study_loader_op, series_selector_op, {("dicom_study_list", "dicom_study_list")})
         self.add_flow(
-            series_selector_op, series_to_vol_op, {"study_selected_series_list": "study_selected_series_list"}
+            series_selector_op, series_to_vol_op, {("study_selected_series_list", "study_selected_series_list")}
         )
-        self.add_flow(series_to_vol_op, classifier_op, {"image": "image"})
-        self.add_flow(classifier_op, sr_writer_op, {"result_text": "classification_result"})
+        self.add_flow(series_to_vol_op, classifier_op, {("image", "image")})
+        self.add_flow(classifier_op, sr_writer_op, {("result_text", "text")})
         # Pass the Study series to the SR writer for copying tags
-        self.add_flow(series_selector_op, sr_writer_op, {"study_selected_series_list": "study_selected_series_list"})
+        self.add_flow(series_selector_op, sr_writer_op, {("study_selected_series_list", "study_selected_series_list")})
+
+        logging.info(f"End {self.compose.__name__}")
 
 
 # This is a sample series selection rule in JSON, simply selecting a MG series.
@@ -69,4 +105,7 @@ def test():
 
 
 if __name__ == "__main__":
-    app = BreastClassificationApp(do_run=True)
+    load_env_log_level()
+    logging.info(f"Begin {__name__}")
+    BreastClassificationApp().run()
+    logging.info(f"End {__name__}")

examples/apps/breast_density_classifer_app/breast_density_classifier_operator.py

@@ -1,11 +1,12 @@
 import json
-from typing import Dict, Text
+import os
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Sequence, Text, Tuple, Union
 
 import torch
 
-import monai.deploy.core as md
 from monai.data import DataLoader, Dataset
-from monai.deploy.core import ExecutionContext, Image, InputContext, IOType, Operator, OutputContext
+from monai.deploy.core import ConditionType, Fragment, Image, Operator, OperatorSpec
 from monai.deploy.operators.monai_seg_inference_operator import InMemImageReader
 from monai.transforms import (
     Activations,
@@ -20,14 +21,70 @@
 )
 
 
-@md.input("image", Image, IOType.IN_MEMORY)
-@md.output("result_text", Text, IOType.IN_MEMORY)
+# @md.input("image", Image, IOType.IN_MEMORY)
+# @md.output("result_text", Text, IOType.IN_MEMORY)
+# @env(pip_packages=["monai~=1.1.0"])
 class ClassifierOperator(Operator):
-    def __init__(self):
-        super().__init__()
+    """Performs breast density classification using a DL model with an image converted from a DICOM MG series.
+
+    Named inputs:
+        image: Image object for which to generate the classification.
+        output_folder: Optional, the path to save the results JSON file, overridingthe the one set on __init__
+
+    Named output:
+        result_text: The classification results in text.
+    """
+
+    DEFAULT_OUTPUT_FOLDER = Path.cwd() / "classification_results"
+    # For testing the app directly, the model should be at the following path.
+    MODEL_LOCAL_PATH = Path(os.environ.get("HOLOSCAN_MODEL_PATH", Path.cwd() / "model/model.ts"))
+
+    def __init__(
+        self,
+        frament: Fragment,
+        *args,
+        model_name: Optional[str] = "",
+        model_path: Path = MODEL_LOCAL_PATH,
+        output_folder: Path = DEFAULT_OUTPUT_FOLDER,
+        **kwargs,
+    ):
+        """Creates an instance with the reference back to the containing application/fragment.
+
+        fragment (Fragment): An instance of the Application class which is derived from Fragment.
+        model_name (str, optional): Name of the model. Default to "" for single model app.
+        model_path (Path): Path to the model file. Defaults to model/models.ts of current working dir.
+        output_folder (Path, optional): output folder for saving the classification results JSON file.
+        """
+
+        # the names used for the model inference input and output
         self._input_dataset_key = "image"
         self._pred_dataset_key = "pred"
 
+        # The names used for the operator input and output
+        self.input_name_image = "image"
+        self.output_name_result = "result_text"
+
+        # The name of the optional input port for passing data to override the output folder path.
+        self.input_name_output_folder = "output_folder"
+
+        # The output folder set on the object can be overriden at each compute by data in the optional named input
+        self.output_folder = output_folder
+
+        # Need the name when there are multiple models loaded
+        self._model_name = model_name.strip() if isinstance(model_name, str) else ""
+        # Need the path to load the models when they are not loaded in the execution context
+        self.model_path = model_path
+
+        # This needs to be at the end of the constructor.
+        super().__init__(frament, *args, **kwargs)
+
+    def setup(self, spec: OperatorSpec):
+        """Set up the operator named input and named output, both are in-memory objects."""
+
+        spec.input(self.input_name_image)
+        spec.input(self.input_name_output_folder).condition(ConditionType.NONE)  # Optional for overriding.
+        spec.output(self.output_name_result).condition(ConditionType.NONE)  # Not forcing a downstream receiver.
+
     def _convert_dicom_metadata_datatype(self, metadata: Dict):
         if not metadata:
             return metadata
@@ -55,16 +112,26 @@ def _convert_dicom_metadata_datatype(self, metadata: Dict):
 
         return metadata
 
-    def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):
-        input_image = op_input.get("image")
+    def compute(self, op_input, op_output, context):
+        input_image = op_input.receive(self.input_name_image)
+        if not input_image:
+            raise ValueError("Input image is not found.")
+        if not isinstance(input_image, Image):
+            raise ValueError(f"Input is not the required type: {type(Image)!r}")
+
         _reader = InMemImageReader(input_image)
         input_img_metadata = self._convert_dicom_metadata_datatype(input_image.metadata())
         img_name = str(input_img_metadata.get("SeriesInstanceUID", "Img_in_context"))
 
-        output_path = context.output.get().path
+        output_folder_on_compute = op_input.receive(self.input_name_output_folder)
+        output_folder = output_folder_on_compute if output_folder_on_compute else self.output_folder
+        Path.mkdir(output_folder, parents=True, exist_ok=True)  # Let exception bubble up if raised.
 
         device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-        model = context.models.get()
+
+        # Need to get the model from context, when it is re-implemented, and for now, load it directly here.
+        # model = context.models.get()
+        model = torch.jit.load(self.model_path, map_location=device)
 
         pre_transforms = self.pre_process(_reader)
         post_transforms = self.post_process()
@@ -82,15 +149,12 @@ def compute(self, op_input: InputContext, op_output: OutputContext, context: Exe
         result_dict = (
             "A " + ":" + str(out[0]) + " B " + ":" + str(out[1]) + " C " + ":" + str(out[2]) + " D " + ":" + str(out[3])
         )
-        result_dict_out = {"A": str(out[0]), "B": str(out[1]), "C": str(out[2]), "D": str(out[3])}
-        output_folder = context.output.get().path
-        output_folder.mkdir(parents=True, exist_ok=True)
 
         output_path = output_folder / "output.json"
         with open(output_path, "w") as fp:
             json.dump(result_dict, fp)
 
-        op_output.set(result_dict, "result_text")
+        op_output.emit(result_dict, "result_text")
 
     def pre_process(self, image_reader) -> Compose:
         return Compose(

monai/deploy/operators/dicom_encapsulated_pdf_writer_operator.py

@@ -86,7 +86,7 @@ def __init__(
 
         # Need to init the output folder until the execution context supports dynamic FS path
         # Not trying to create the folder to avoid exception on init
-        self.output_dir = (
+        self.output_folder = (
             Path(output_folder) if output_folder else DICOMEncapsulatedPDFWriterOperator.DEFAULT_OUTPUT_FOLDER
         )
         self.copy_tags = copy_tags

monai/deploy/operators/dicom_text_sr_writer_operator.py

@@ -82,7 +82,7 @@ def __init__(
 
         # Need to init the output folder until the execution context supports dynamic FS path
         # Not trying to create the folder to avoid exception on init
-        self.output_dir = Path(output_folder) if output_folder else DICOMTextSRWriterOperator.DEFAULT_OUTPUT_FOLDER
+        self.output_folder = Path(output_folder) if output_folder else DICOMTextSRWriterOperator.DEFAULT_OUTPUT_FOLDER
         self.copy_tags = copy_tags
         self.model_info = model_info if model_info else ModelInfo()
         self.equipment_info = equipment_info if equipment_info else EquipmentInfo()

monai/deploy/operators/monai_seg_inference_operator.py

@@ -84,6 +84,7 @@ def __init__(
             pre_transforms (Compose): MONAI Compose object used for pre-transforms.
             post_transforms (Compose): MONAI Compose object used for post-transforms.
             model_name (str, optional): Name of the model. Default to "" for single model app.
+            model_path (Path): Path to the model file. Defaults to model/models.ts of current working dir.
             overlap (float): The overlap used in sliding window inference.
         """