Migrated the MEDNIST app

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

Author: MMelQin

examples/apps/breast_density_classifer_app/breast_density_classifier_operator.py

@@ -21,8 +21,6 @@
 )
 
 
-# @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):
     """Performs breast density classification using a DL model with an image converted from a DICOM MG series.
@@ -123,10 +121,6 @@ def compute(self, op_input, op_output, context):
         input_img_metadata = self._convert_dicom_metadata_datatype(input_image.metadata())
         img_name = str(input_img_metadata.get("SeriesInstanceUID", "Img_in_context"))
 
-        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")
 
         # Need to get the model from context, when it is re-implemented, and for now, load it directly here.
@@ -150,12 +144,15 @@ def compute(self, op_input, op_output, context):
             "A " + ":" + str(out[0]) + " B " + ":" + str(out[1]) + " C " + ":" + str(out[2]) + " D " + ":" + str(out[3])
         )
 
-        output_path = output_folder / "output.json"
+        op_output.emit(result_dict, "result_text")
+
+        # Get output folder, with value in optional input port overriding the obj attribute
+        output_folder_on_compute = op_input.receive(self.input_name_output_folder) or self.output_folder
+        Path.mkdir(output_folder_on_compute, parents=True, exist_ok=True)  # Let exception bubble up if raised.
+        output_path = output_folder_on_compute / "output.json"
         with open(output_path, "w") as fp:
             json.dump(result_dict, fp)
 
-        op_output.emit(result_dict, "result_text")
-
     def pre_process(self, image_reader) -> Compose:
         return Compose(
             [

examples/apps/mednist_classifier_monaideploy/mednist_classifier_monaideploy.py

@@ -9,70 +9,157 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Text
-
-import monai.deploy.core as md
-from monai.deploy.core import (
-    Application,
-    DataPath,
-    ExecutionContext,
-    Image,
-    InputContext,
-    IOType,
-    Operator,
-    OutputContext,
-)
-from monai.deploy.operators import DICOMTextSRWriterOperator, EquipmentInfo, ModelInfo
+import logging
+import os
+from pathlib import Path
+from typing import Optional, Text
+
+from monai.deploy.conditions import CountCondition
+from monai.deploy.core import AppContext, Application, ConditionType, Fragment, Image, Operator, OperatorSpec
+from monai.deploy.operators.dicom_text_sr_writer_operator import DICOMTextSRWriterOperator, EquipmentInfo, ModelInfo
 from monai.transforms import AddChannel, Compose, EnsureType, ScaleIntensity
 
 MEDNIST_CLASSES = ["AbdomenCT", "BreastMRI", "CXR", "ChestCT", "Hand", "HeadCT"]
 
 
-@md.input("image", DataPath, IOType.DISK)
-@md.output("image", Image, IOType.IN_MEMORY)
-@md.env(pip_packages=["pillow"])
+# @md.env(pip_packages=["pillow"])
 class LoadPILOperator(Operator):
     """Load image from the given input (DataPath) and set numpy array to the output (Image)."""
 
-    def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):
+    DEFAULT_INPUT_FOLDER = Path.cwd() / "input"
+    DEFAULT_OUTPUT_NAME = "image"
+
+    # For now, need to have the input folder as an instance attribute, set on init.
+    # If dynamically changing the input folder, per compute, then use a (optional) input port to convey the
+    # value of the input folder, which is then emitted by a upstream operator.
+    def __init__(
+        self,
+        fragment: Fragment,
+        *args,
+        input_folder: Path = DEFAULT_INPUT_FOLDER,
+        output_name: str = DEFAULT_OUTPUT_NAME,
+        **kwargs,
+    ):
+        """Creates an loader object with the input folder and the output port name overrides as needed.
+
+        Args:
+            fragment (Fragment): An instance of the Application class which is derived from Fragment.
+            input_folder (Path): Folder from which to load input file(s).
+                                 Defaults to `input` in the current working directory.
+            output_name (str): Name of the output port, which is an image object. Defaults to `image`.
+        """
+
+        self._logger = logging.getLogger("{}.{}".format(__name__, type(self).__name__))
+        self.input_path = input_folder
+        self.index = 0
+        self.output_name_image = (
+            output_name.strip() if output_name and len(output_name.strip()) > 0 else LoadPILOperator.DEFAULT_OUTPUT_NAME
+        )
+
+        super().__init__(fragment, *args, **kwargs)
+
+    def setup(self, spec: OperatorSpec):
+        """Set up the named input and output port(s)"""
+        spec.output(self.output_name_image)
+
+    def compute(self, op_input, op_output, context):
         import numpy as np
         from PIL import Image as PILImage
 
-        input_path = op_input.get().path
+        # Input path is stored in the object attribute, but could change to use a named port if need be.
+        input_path = self.input_path
         if input_path.is_dir():
-            input_path = next(input_path.glob("*.*"))  # take the first file
+            input_path = next(self.input_path.glob("*.*"))  # take the first file
 
         image = PILImage.open(input_path)
         image = image.convert("L")  # convert to greyscale image
         image_arr = np.asarray(image)
 
         output_image = Image(image_arr)  # create Image domain object with a numpy array
-        op_output.set(output_image)
+        op_output.emit(output_image, self.output_name_image)  # cannot omit the name even if single output.
 
 
-@md.input("image", Image, IOType.IN_MEMORY)
-@md.output("result_text", Text, IOType.IN_MEMORY)
-@md.env(pip_packages=["monai"])
+# @md.env(pip_packages=["monai"])
 class MedNISTClassifierOperator(Operator):
-    """Classifies the given image and returns the class name."""
+    """Classifies the given image and returns the class name.
+
+    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.
 
     @property
     def transform(self):
         return Compose([AddChannel(), ScaleIntensity(), EnsureType()])
 
-    def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):
+    def compute(self, op_input, op_output, context):
         import json
 
         import torch
 
-        img = op_input.get().asnumpy()  # (64, 64), uint8
+        img = op_input.receive(self.input_name_image).asnumpy()  # (64, 64), uint8. Input validation can be added.
         image_tensor = self.transform(img)  # (1, 64, 64), torch.float64
         image_tensor = image_tensor[None].float()  # (1, 1, 64, 64), torch.float32
 
         device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
         image_tensor = image_tensor.to(device)
 
-        model = context.models.get()  # get a TorchScriptModel object
+        # Need to get the model from context, when it is re-implemented, and for now, load it directly here.
+        # model = context.models.get()  # get a TorchScriptModel object
+        model = torch.jit.load(self.model_path, map_location=device)
 
         with torch.no_grad():
             outputs = model(image_tensor)
@@ -81,37 +168,45 @@ def compute(self, op_input: InputContext, op_output: OutputContext, context: Exe
 
         result = MEDNIST_CLASSES[output_classes[0]]  # get the class name
         print(result)
-        op_output.set(result, "result_text")
-
-        # Get output (folder) path and create the folder if not exists
-        # The following gets the App context's output path, instead the operator's.
-        output_folder = context.output.get().path
-        output_folder.mkdir(parents=True, exist_ok=True)
+        op_output.emit(result, self.output_name_result)
 
-        # Write result to "output.json"
-        output_path = output_folder / "output.json"
+        # Get output folder, with value in optional input port overriding the obj attribute
+        output_folder_on_compute = op_input.receive(self.input_name_output_folder) or self.output_folder
+        Path.mkdir(output_folder_on_compute, parents=True, exist_ok=True)  # Let exception bubble up if raised.
+        output_path = output_folder_on_compute / "output.json"
         with open(output_path, "w") as fp:
             json.dump(result, fp)
 
 
-@md.resource(cpu=1, gpu=1, memory="1Gi")
+# @md.resource(cpu=1, gpu=1, memory="1Gi")
 class App(Application):
     """Application class for the MedNIST classifier."""
 
     def compose(self):
-        load_pil_op = LoadPILOperator()
-        classifier_op = MedNISTClassifierOperator()
+        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)
+        load_pil_op = LoadPILOperator(self, CountCondition(self, 1), input_folder=app_input_path, name="pil_loader_op")
+        classifier_op = MedNISTClassifierOperator(
+            self, output_folder=app_output_path, model_path=model_path, name="classifier_op"
+        )
 
         my_model_info = ModelInfo("MONAI WG Trainer", "MEDNIST Classifier", "0.1", "xyz")
         my_equipment = EquipmentInfo(manufacturer="MOANI Deploy App SDK", manufacturer_model="DICOM SR Writer")
         my_special_tags = {"SeriesDescription": "Not for clinical use. The result is for research use only."}
         dicom_sr_operator = DICOMTextSRWriterOperator(
-            copy_tags=False, model_info=my_model_info, equipment_info=my_equipment, custom_tags=my_special_tags
+            self,
+            copy_tags=False,
+            model_info=my_model_info,
+            equipment_info=my_equipment,
+            custom_tags=my_special_tags,
+            output_folder=app_output_path,
         )
 
-        self.add_flow(load_pil_op, classifier_op)
-        self.add_flow(classifier_op, dicom_sr_operator, {"result_text": "classification_result"})
+        self.add_flow(load_pil_op, classifier_op, {("image", "image")})
+        self.add_flow(classifier_op, dicom_sr_operator, {("result_text", "text")})
 
 
 if __name__ == "__main__":
-    App(do_run=True)
+    App().run()