Creating a Segmentation App with MONAI Deploy App SDK

This tutorial shows how to create an organ segmentation application for a PyTorch model that has been trained with MONAI. Please note that this one does not require the model be a MONAI Bundle.

Deploying AI models requires the integration with clinical imaging network, even if just in a for-research-use setting. This means that the AI deploy application will need to support standards-based imaging protocols, and specifically for Radiological imaging, DICOM protocol.

Typically, DICOM network communication, either in DICOM TCP/IP network protocol or DICOMWeb, would be handled by DICOM devices or services, e.g. MONAI Deploy Informatics Gateway, so the deploy application itself would only need to use DICOM Part 10 files as input and save the AI result in DICOM Part10 file(s). For segmentation use cases, the DICOM instance file for AI results could be a DICOM Segmentation object or a DICOM RT Structure Set, and for classification, DICOM Structure Report and/or DICOM Encapsulated PDF.

During model training, input and label images are typically in non-DICOM volumetric image format, e.g., NIfTI and PNG, converted from a specific DICOM study series. Furthermore, the voxel spacings most likely have been re-sampled to be uniform for all images. When integrated with imaging networks and receiving DICOM instances from modalities and Picture Archiving and Communications System, PACS, an AI deploy application has to deal with a whole DICOM study with multiple series, whose images’ spacing may not be the same as expected by the trained model. To address these cases consistently and efficiently, MONAI Deploy Application SDK provides classes, called operators, to parse DICOM studies, select specific series with application-defined rules, and convert the selected DICOM series into domain-specific image format along with meta-data representing the pertinent DICOM attributes. The image is then further processed in the pre-processing stage to normalize spacing, orientation, intensity, etc., before pixel data as Tensors are used for inference.

In the following sections, we will demonstrate how to create a MONAI Deploy application package using the MONAI Deploy App SDK.

Note

For local testing, if there is a lack of DICOM Part 10 files, one can use open source programs, e.g. 3D Slicer, to convert NIfTI to DICOM files.

Creating Operators and connecting them in Application class

We will implement an application that consists of five Operators:

• DICOMDataLoaderOperator:

  • Input(dicom_files): a folder path (Path)

  • Output(dicom_study_list): a list of DICOM studies in memory (List[DICOMStudy])

• DICOMSeriesSelectorOperator:

  • Input(dicom_study_list): a list of DICOM studies in memory (List[DICOMStudy])

  • Input(selection_rules): a selection rule (Dict)

  • Output(study_selected_series_list): a DICOM series object in memory (StudySelectedSeries)

• DICOMSeriesToVolumeOperator:

  • Input(study_selected_series_list): a DICOM series object in memory (StudySelectedSeries)

  • Output(image): an image object in memory (Image)

• SpleenSegOperator:

  • Input(image): an image object in memory (Image)

  • Output(seg_image): an image object in memory (Image)

• DICOMSegmentationWriterOperator:

  • Input(seg_image): a segmentation image object in memory (Image)

  • Input(study_selected_series_list): a DICOM series object in memory (StudySelectedSeries)

  • Output(dicom_seg_instance): a file path (Path)

Note

The DICOMSegmentationWriterOperator needs both the segmentation image as well as the original DICOM series meta-data in order to use the patient demographics and the DICOM Study level attributes.

The workflow of the application would look like this.

        %%{init: {"theme": "base", "themeVariables": { "fontSize": "16px"}} }%%

classDiagram
    direction TB

    DICOMDataLoaderOperator --|> DICOMSeriesSelectorOperator : dicom_study_list...dicom_study_list
    DICOMSeriesSelectorOperator --|> DICOMSeriesToVolumeOperator : study_selected_series_list...study_selected_series_list
    DICOMSeriesToVolumeOperator --|> SpleenSegOperator : image...image
    DICOMSeriesSelectorOperator --|> DICOMSegmentationWriterOperator : study_selected_series_list...study_selected_series_list
    SpleenSegOperator --|> DICOMSegmentationWriterOperator : seg_image...seg_image


    class DICOMDataLoaderOperator {
        <in>dicom_files : DISK
        dicom_study_list(out) IN_MEMORY
    }
    class DICOMSeriesSelectorOperator {
        <in>dicom_study_list : IN_MEMORY
        <in>selection_rules : IN_MEMORY
        study_selected_series_list(out) IN_MEMORY
    }
    class DICOMSeriesToVolumeOperator {
        <in>study_selected_series_list : IN_MEMORY
        image(out) IN_MEMORY
    }
    class SpleenSegOperator {
        <in>image : IN_MEMORY
        seg_image(out) IN_MEMORY
    }
    class DICOMSegmentationWriterOperator {
        <in>seg_image : IN_MEMORY
        <in>study_selected_series_list : IN_MEMORY
        dicom_seg_instance(out) DISK
    }
    

Setup environment

# Install MONAI and other necessary image processing packages for the application
!python -c "import monai" || pip install --upgrade -q "monai"
!python -c "import torch" || pip install -q "torch>=1.10.2"
!python -c "import numpy" || pip install -q "numpy>=1.21"
!python -c "import nibabel" || pip install -q "nibabel>=3.2.1"
!python -c "import pydicom" || pip install -q "pydicom>=1.4.2"
!python -c "import highdicom" || pip install -q "highdicom>=0.18.2"
!python -c "import SimpleITK" || pip install -q "SimpleITK>=2.0.0"

# Install MONAI Deploy App SDK package
!python -c "import monai.deploy" || pip install -q "monai-deploy-app-sdk"

Note: you may need to restart the Jupyter kernel to use the updated packages.

Download/Extract ai_spleen_bundle_data from Google Drive

Note: Data files are now access controlled. Please first request permission to access the shared folder on Google Drive. Please download zip file, ai_spleen_seg_bundle_data.zip in the ai_spleen_seg_app folder, to the same folder as the notebook example.

# Download ai_spleen_bundle_data test data zip file. Please request access and download manually.
# !pip install gdown
# !gdown "https://drive.google.com/uc?id=1IwWMpbo2fd38fKIqeIdL8SKTGvkn31tK"

# After downloading ai_spleen_bundle_data zip file from the web browser or using gdown,
!unzip -o "ai_spleen_seg_bundle_data.zip"

# Need to copy the model.ts file to its own clean subfolder for packaging, to work around an issue in the Packager
models_folder = "models"
!rm -rf {models_folder} && mkdir -p {models_folder}/model && cp model.ts {models_folder}/model && ls {models_folder}/model

%env HOLOSCAN_INPUT_PATH dcm
%env HOLOSCAN_MODEL_PATH {models_folder}
%env HOLOSCAN_OUTPUT_PATH output

Setup imports

Let’s import necessary classes/decorators to define Application and Operator.

import logging
from numpy import uint8  # Needed if SaveImaged is enabled
from pathlib import Path

# Required for setting SegmentDescription attributes. Direct import as this is not part of App SDK package.
from pydicom.sr.codedict import codes

from monai.deploy.conditions import CountCondition
from monai.deploy.core import AppContext, Application, ConditionType, Fragment, Operator, OperatorSpec
from monai.deploy.core.domain import Image
from monai.deploy.core.io_type import IOType
from monai.deploy.operators.dicom_data_loader_operator import DICOMDataLoaderOperator
from monai.deploy.operators.dicom_seg_writer_operator import DICOMSegmentationWriterOperator, SegmentDescription
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.monai_seg_inference_operator import InfererType, InMemImageReader, MonaiSegInferenceOperator

from monai.transforms import (
    Activationsd,
    AsDiscreted,
    Compose,
    EnsureChannelFirstd,
    EnsureTyped,
    Invertd,
    LoadImaged,
    Orientationd,
    SaveImaged,
    ScaleIntensityRanged,
    Spacingd,
)

Creating Model Specific Inference Operator classes

Each Operator class inherits the base Operator class. The input/output properties are specified by implementing the setup() method, and the business logic implemented in the compute() method.

The App SDK provides a MonaiSegInferenceOperator class to perform segmentation prediction with a Torch Script model. For consistency, this class uses MONAI dictionary-based transforms, as Compose object, for pre and post transforms. The model-specific inference operator will then only need to create the pre and post transform Compose based on what has been used in the model during training and validation. Note that for deploy application, ignite is not needed nor supported.

SpleenSegOperator

The SpleenSegOperator gets as input an in-memory Image object that has been converted from a DICOM CT series by the preceding DICOMSeriesToVolumeOperator, and as output in-memory segmentation Image object.

The pre_process function creates the pre-transforms Compose object. For LoadImage, a specialized InMemImageReader, derived from MONAI ImageReader, is used to convert the in-memory pixel data and return the numpy array as well as the meta-data. Also, the DICOM input pixel spacings are often not the same as expected by the model, so the Spacingd transform must be used to re-sample the image with the expected spacing.

The post_process function creates the post-transform Compose object. The SaveImageD transform class is used to save the segmentation mask as NIfTI image file, which is optional as the in-memory mask image will be passed down to the DICOM Segmentation writer for creating a DICOM Segmentation instance. The Invertd must also be used to revert the segmentation image’s orientation and spacing to be the same as the input.

When the MonaiSegInferenceOperator object is created, the ROI size is specified, as well as the transform Compose objects. Furthermore, the dataset image key names are set accordingly.

Loading of the model and performing the prediction are encapsulated in the MonaiSegInferenceOperator and other SDK classes. Once the inference is completed, the segmentation Image object is created and set to the output by the SpleenSegOperator.

class SpleenSegOperator(Operator):
    """Performs Spleen segmentation with a 3D image converted from a DICOM CT series.
    """

    DEFAULT_OUTPUT_FOLDER = Path.cwd() / "output/saved_images_folder"

    def __init__(
        self,
        fragment: Fragment,
        *args,
        app_context: AppContext,
        model_path: Path,
        output_folder: Path = DEFAULT_OUTPUT_FOLDER,
        **kwargs,
    ):

        self.logger = logging.getLogger("{}.{}".format(__name__, type(self).__name__))
        self._input_dataset_key = "image"
        self._pred_dataset_key = "pred"

        self.model_path = model_path
        self.output_folder = output_folder
        self.output_folder.mkdir(parents=True, exist_ok=True)
        self.app_context = app_context
        self.input_name_image = "image"
        self.output_name_seg = "seg_image"
        self.output_name_saved_images_folder = "saved_images_folder"

        # The base class has an attribute called fragment to hold the reference to the fragment object
        super().__init__(fragment, *args, **kwargs)

    def setup(self, spec: OperatorSpec):
        spec.input(self.input_name_image)
        spec.output(self.output_name_seg)
        spec.output(self.output_name_saved_images_folder).condition(
            ConditionType.NONE
        )  # Output not requiring a receiver

    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.")

        # This operator gets an in-memory Image object, so a specialized ImageReader is needed.
        _reader = InMemImageReader(input_image)

        pre_transforms = self.pre_process(_reader, str(self.output_folder))
        post_transforms = self.post_process(pre_transforms, str(self.output_folder))

        # Delegates inference and saving output to the built-in operator.
        infer_operator = MonaiSegInferenceOperator(
            self.fragment,
            roi_size=(
                96,
                96,
                96,
            ),
            pre_transforms=pre_transforms,
            post_transforms=post_transforms,
            overlap=0.6,
            app_context=self.app_context,
            model_name="",
            inferer=InfererType.SLIDING_WINDOW,
            sw_batch_size=4,
            model_path=self.model_path,
            name="monai_seg_inference_op",
        )

        # Setting the keys used in the dictionary based transforms may change.
        infer_operator.input_dataset_key = self._input_dataset_key
        infer_operator.pred_dataset_key = self._pred_dataset_key

        # Now emit data to the output ports of this operator
        op_output.emit(infer_operator.compute_impl(input_image, context), self.output_name_seg)
        op_output.emit(self.output_folder, self.output_name_saved_images_folder)

    def pre_process(self, img_reader, out_dir: str = "./input_images") -> Compose:
        """Composes transforms for preprocessing input before predicting on a model."""

        Path(out_dir).mkdir(parents=True, exist_ok=True)
        my_key = self._input_dataset_key

        return Compose(
            [
                LoadImaged(keys=my_key, reader=img_reader),
                EnsureChannelFirstd(keys=my_key),
                # The SaveImaged transform can be commented out to save 5 seconds.
                # Uncompress NIfTI file, nii, is used favoring speed over size, but can be changed to nii.gz
                SaveImaged(
                    keys=my_key,
                    output_dir=out_dir,
                    output_postfix="",
                    resample=False,
                    output_ext=".nii",
                ),
                Orientationd(keys=my_key, axcodes="RAS"),
                Spacingd(keys=my_key, pixdim=[1.5, 1.5, 2.9], mode=["bilinear"]),
                ScaleIntensityRanged(keys=my_key, a_min=-57, a_max=164, b_min=0.0, b_max=1.0, clip=True),
                EnsureTyped(keys=my_key),
            ]
        )

    def post_process(self, pre_transforms: Compose, out_dir: str = "./prediction_output") -> Compose:
        """Composes transforms for postprocessing the prediction results."""

        Path(out_dir).mkdir(parents=True, exist_ok=True)
        pred_key = self._pred_dataset_key

        return Compose(
            [
                Activationsd(keys=pred_key, softmax=True),
                Invertd(
                    keys=pred_key,
                    transform=pre_transforms,
                    orig_keys=self._input_dataset_key,
                    nearest_interp=False,
                    to_tensor=True,
                ),
                AsDiscreted(keys=pred_key, argmax=True),
                # The SaveImaged transform can be commented out to save 5 seconds.
                # Uncompress NIfTI file, nii, is used favoring speed over size, but can be changed to nii.gz
                SaveImaged(
                    keys=pred_key,
                    output_dir=out_dir,
                    output_postfix="seg",
                    output_dtype=uint8,
                    resample=False,
                    output_ext=".nii",
                ),
            ]
        )

Creating Application class

Our application class would look like below.

It defines App class, inheriting the base Application class.

The base class method, compose, is overridden. Objects required for DICOM parsing, series selection, pixel data conversion to volume image, and segmentation instance creation are created, so is the model-specific SpleenSegOperator. The execution pipeline, as a Directed Acyclic Graph (DAG), is created by connecting these objects through the add_flow method.

class AISpleenSegApp(Application):
    def __init__(self, *args, **kwargs):
        """Creates an application instance."""

        super().__init__(*args, **kwargs)
        self._logger = logging.getLogger("{}.{}".format(__name__, type(self).__name__))

    def run(self, *args, **kwargs):
        # This method calls the base class to run. Can be omitted if simply calling through.
        self._logger.info(f"Begin {self.run.__name__}")
        super().run(*args, **kwargs)
        self._logger.info(f"End {self.run.__name__}")

    def compose(self):
        """Creates the app specific operators and chain them up in the processing DAG."""

        self._logger.debug(f"Begin {self.compose.__name__}")
        app_context = Application.init_app_context({})  # Do not pass argv in Jupyter Notebook
        app_input_path = Path(app_context.input_path)
        app_output_path = Path(app_context.output_path)
        model_path = Path(app_context.model_path)

        self._logger.info(f"App input and output path: {app_input_path}, {app_output_path}")

        # instantiates the SDK built-in operator(s).
        study_loader_op = DICOMDataLoaderOperator(
            self, CountCondition(self, 1), input_folder=app_input_path, name="dcm_loader_op"
        )
        series_selector_op = DICOMSeriesSelectorOperator(self, rules=Sample_Rules_Text, name="series_selector_op")
        series_to_vol_op = DICOMSeriesToVolumeOperator(self, name="series_to_vol_op")

        # Model specific inference operator, supporting MONAI transforms.
        spleen_seg_op = SpleenSegOperator(self, app_context=app_context, model_path=model_path, name="seg_op")

        # Create DICOM Seg writer providing the required segment description for each segment with
        # the actual algorithm and the pertinent organ/tissue.
        # The segment_label, algorithm_name, and algorithm_version are limited to 64 chars.
        # https://dicom.nema.org/medical/dicom/current/output/chtml/part05/sect_6.2.html
        # User can Look up SNOMED CT codes at, e.g.
        # https://bioportal.bioontology.org/ontologies/SNOMEDCT

        _algorithm_name = "3D segmentation of the Spleen from a CT series"
        _algorithm_family = codes.DCM.ArtificialIntelligence
        _algorithm_version = "0.1.0"

        segment_descriptions = [
            SegmentDescription(
                segment_label="Spleen",
                segmented_property_category=codes.SCT.Organ,
                segmented_property_type=codes.SCT.Spleen,
                algorithm_name=_algorithm_name,
                algorithm_family=_algorithm_family,
                algorithm_version=_algorithm_version,
            ),
        ]

        custom_tags = {"SeriesDescription": "AI generated Seg, not for clinical use."}

        dicom_seg_writer = DICOMSegmentationWriterOperator(
            self,
            segment_descriptions=segment_descriptions,
            custom_tags=custom_tags,
            output_folder=app_output_path,
            name="dcm_seg_writer_op",
        )

        # Create the processing pipeline, by specifying the source and destination operators, and
        # ensuring the output from the former matches the input of the latter, in both name and type.
        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")}
        )
        self.add_flow(series_to_vol_op, spleen_seg_op, {("image", "image")})

        # Note below the dicom_seg_writer requires two inputs, each coming from a source operator.
        self.add_flow(
            series_selector_op, dicom_seg_writer, {("study_selected_series_list", "study_selected_series_list")}
        )
        self.add_flow(spleen_seg_op, dicom_seg_writer, {("seg_image", "seg_image")})

        self._logger.debug(f"End {self.compose.__name__}")


# This is a sample series selection rule in JSON, simply selecting CT series.
# If the study has more than 1 CT series, then all of them will be selected.
# Please see more detail in DICOMSeriesSelectorOperator.
# For list of string values, e.g. "ImageType": ["PRIMARY", "ORIGINAL"], it is a match if all elements
# are all in the multi-value attribute of the DICOM series.

Sample_Rules_Text = """
{
    "selections": [
        {
            "name": "CT Series",
            "conditions": {
                "StudyDescription": "(.*?)",
                "Modality": "(?i)CT",
                "SeriesDescription": "(.*?)",
                "ImageType": ["PRIMARY", "ORIGINAL"]
            }
        }
    ]
}
"""

Executing app locally

We can execute the app in Jupyter notebook. Note that the DICOM files of the CT Abdomen series must be present in the dcm folder and the TorchScript, model.ts, in the folder pointed to by the environment variables.

!rm -rf $HOLOSCAN_OUTPUT_PATH
app = AISpleenSegApp()
app.run()

Once the application is verified inside Jupyter notebook, we can write the above Python code into Python files in an application folder.

The application folder structure would look like below:

my_app
├── __main__.py
├── app.py
└── spleen_seg_operator.py

Note

We can create a single application Python file (such as spleen_app.py) that includes the content of the files, instead of creating multiple files. You will see such an example in MedNist Classifier Tutorial.

# Create an application folder
!mkdir -p my_app && rm -rf my_app/*

spleen_seg_operator.py

%%writefile my_app/spleen_seg_operator.py
import logging

from numpy import uint8
from pathlib import Path

from monai.deploy.core import AppContext, ConditionType, Fragment, Operator, OperatorSpec
from monai.deploy.operators.monai_seg_inference_operator import InfererType, InMemImageReader, MonaiSegInferenceOperator
from monai.transforms import (
    Activationsd,
    AsDiscreted,
    Compose,
    EnsureChannelFirstd,
    EnsureTyped,
    Invertd,
    LoadImaged,
    Orientationd,
    SaveImaged,
    ScaleIntensityRanged,
    Spacingd,
)

class SpleenSegOperator(Operator):
    """Performs Spleen segmentation with a 3D image converted from a DICOM CT series.
    """

    DEFAULT_OUTPUT_FOLDER = Path.cwd() / "output/saved_images_folder"

    def __init__(
        self,
        fragment: Fragment,
        *args,
        app_context: AppContext,
        model_path: Path,
        output_folder: Path = DEFAULT_OUTPUT_FOLDER,
        **kwargs,
    ):

        self.logger = logging.getLogger("{}.{}".format(__name__, type(self).__name__))
        self._input_dataset_key = "image"
        self._pred_dataset_key = "pred"

        self.model_path = model_path
        self.output_folder = output_folder
        self.output_folder.mkdir(parents=True, exist_ok=True)
        self.app_context = app_context
        self.input_name_image = "image"
        self.output_name_seg = "seg_image"
        self.output_name_saved_images_folder = "saved_images_folder"

        # The base class has an attribute called fragment to hold the reference to the fragment object
        super().__init__(fragment, *args, **kwargs)

    def setup(self, spec: OperatorSpec):
        spec.input(self.input_name_image)
        spec.output(self.output_name_seg)
        spec.output(self.output_name_saved_images_folder).condition(
            ConditionType.NONE
        )  # Output not requiring a receiver

    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.")

        # This operator gets an in-memory Image object, so a specialized ImageReader is needed.
        _reader = InMemImageReader(input_image)

        pre_transforms = self.pre_process(_reader, str(self.output_folder))
        post_transforms = self.post_process(pre_transforms, str(self.output_folder))

        # Delegates inference and saving output to the built-in operator.
        infer_operator = MonaiSegInferenceOperator(
            self.fragment,
            roi_size=(
                96,
                96,
                96,
            ),
            pre_transforms=pre_transforms,
            post_transforms=post_transforms,
            overlap=0.6,
            app_context=self.app_context,
            model_name="",
            inferer=InfererType.SLIDING_WINDOW,
            sw_batch_size=4,
            model_path=self.model_path,
            name="monai_seg_inference_op",
        )

        # Setting the keys used in the dictionary based transforms may change.
        infer_operator.input_dataset_key = self._input_dataset_key
        infer_operator.pred_dataset_key = self._pred_dataset_key

        # Now emit data to the output ports of this operator
        op_output.emit(infer_operator.compute_impl(input_image, context), self.output_name_seg)
        op_output.emit(self.output_folder, self.output_name_saved_images_folder)

    def pre_process(self, img_reader, out_dir: str = "./input_images") -> Compose:
        """Composes transforms for preprocessing input before predicting on a model."""

        Path(out_dir).mkdir(parents=True, exist_ok=True)
        my_key = self._input_dataset_key

        return Compose(
            [
                LoadImaged(keys=my_key, reader=img_reader),
                EnsureChannelFirstd(keys=my_key),
                # The SaveImaged transform can be commented out to save 5 seconds.
                # Uncompress NIfTI file, nii, is used favoring speed over size, but can be changed to nii.gz
                SaveImaged(
                    keys=my_key,
                    output_dir=out_dir,
                    output_postfix="",
                    resample=False,
                    output_ext=".nii",
                ),
                Orientationd(keys=my_key, axcodes="RAS"),
                Spacingd(keys=my_key, pixdim=[1.5, 1.5, 2.9], mode=["bilinear"]),
                ScaleIntensityRanged(keys=my_key, a_min=-57, a_max=164, b_min=0.0, b_max=1.0, clip=True),
                EnsureTyped(keys=my_key),
            ]
        )

    def post_process(self, pre_transforms: Compose, out_dir: str = "./prediction_output") -> Compose:
        """Composes transforms for postprocessing the prediction results."""

        Path(out_dir).mkdir(parents=True, exist_ok=True)
        pred_key = self._pred_dataset_key

        return Compose(
            [
                Activationsd(keys=pred_key, softmax=True),
                Invertd(
                    keys=pred_key,
                    transform=pre_transforms,
                    orig_keys=self._input_dataset_key,
                    nearest_interp=False,
                    to_tensor=True,
                ),
                AsDiscreted(keys=pred_key, argmax=True),
                # The SaveImaged transform can be commented out to save 5 seconds.
                # Uncompress NIfTI file, nii, is used favoring speed over size, but can be changed to nii.gz
                SaveImaged(
                    keys=pred_key,
                    output_dir=out_dir,
                    output_postfix="seg",
                    output_dtype=uint8,
                    resample=False,
                    output_ext=".nii",
                ),
            ]
        )

app.py

%%writefile my_app/app.py
import logging
from pathlib import Path

from spleen_seg_operator import SpleenSegOperator

from pydicom.sr.codedict import codes  # Required for setting SegmentDescription attributes.

from monai.deploy.conditions import CountCondition
from monai.deploy.core import AppContext, Application
from monai.deploy.operators.dicom_data_loader_operator import DICOMDataLoaderOperator
from monai.deploy.operators.dicom_seg_writer_operator import DICOMSegmentationWriterOperator, SegmentDescription
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.stl_conversion_operator import STLConversionOperator

class AISpleenSegApp(Application):
    def __init__(self, *args, **kwargs):
        """Creates an application instance."""

        super().__init__(*args, **kwargs)
        self._logger = logging.getLogger("{}.{}".format(__name__, type(self).__name__))

    def run(self, *args, **kwargs):
        # This method calls the base class to run. Can be omitted if simply calling through.
        self._logger.info(f"Begin {self.run.__name__}")
        super().run(*args, **kwargs)
        self._logger.info(f"End {self.run.__name__}")

    def compose(self):
        """Creates the app specific operators and chain them up in the processing DAG."""

        # Use Commandline options over environment variables to init context.
        app_context = Application.init_app_context(self.argv)
        self._logger.debug(f"Begin {self.compose.__name__}")
        app_input_path = Path(app_context.input_path)
        app_output_path = Path(app_context.output_path)
        model_path = Path(app_context.model_path)

        self._logger.info(f"App input and output path: {app_input_path}, {app_output_path}")

        # instantiates the SDK built-in operator(s).
        study_loader_op = DICOMDataLoaderOperator(
            self, CountCondition(self, 1), input_folder=app_input_path, name="dcm_loader_op"
        )
        series_selector_op = DICOMSeriesSelectorOperator(self, rules=Sample_Rules_Text, name="series_selector_op")
        series_to_vol_op = DICOMSeriesToVolumeOperator(self, name="series_to_vol_op")

        # Model specific inference operator, supporting MONAI transforms.
        spleen_seg_op = SpleenSegOperator(self, app_context=app_context, model_path=model_path, name="seg_op")

        # Create DICOM Seg writer providing the required segment description for each segment with
        # the actual algorithm and the pertinent organ/tissue.
        # The segment_label, algorithm_name, and algorithm_version are limited to 64 chars.
        # https://dicom.nema.org/medical/dicom/current/output/chtml/part05/sect_6.2.html
        # User can Look up SNOMED CT codes at, e.g.
        # https://bioportal.bioontology.org/ontologies/SNOMEDCT

        _algorithm_name = "3D segmentation of the Spleen from a CT series"
        _algorithm_family = codes.DCM.ArtificialIntelligence
        _algorithm_version = "0.1.0"

        segment_descriptions = [
            SegmentDescription(
                segment_label="Spleen",
                segmented_property_category=codes.SCT.Organ,
                segmented_property_type=codes.SCT.Spleen,
                algorithm_name=_algorithm_name,
                algorithm_family=_algorithm_family,
                algorithm_version=_algorithm_version,
            ),
        ]

        custom_tags = {"SeriesDescription": "AI generated Seg, not for clinical use."}

        dicom_seg_writer = DICOMSegmentationWriterOperator(
            self,
            segment_descriptions=segment_descriptions,
            custom_tags=custom_tags,
            output_folder=app_output_path,
            name="dcm_seg_writer_op",
        )

        # Create the processing pipeline, by specifying the source and destination operators, and
        # ensuring the output from the former matches the input of the latter, in both name and type.
        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")}
        )
        self.add_flow(series_to_vol_op, spleen_seg_op, {("image", "image")})

        # Note below the dicom_seg_writer requires two inputs, each coming from a source operator.
        self.add_flow(
            series_selector_op, dicom_seg_writer, {("study_selected_series_list", "study_selected_series_list")}
        )
        self.add_flow(spleen_seg_op, dicom_seg_writer, {("seg_image", "seg_image")})

        self._logger.debug(f"End {self.compose.__name__}")


# This is a sample series selection rule in JSON, simply selecting CT series.
# If the study has more than 1 CT series, then all of them will be selected.
# Please see more detail in DICOMSeriesSelectorOperator.
# For list of string values, e.g. "ImageType": ["PRIMARY", "ORIGINAL"], it is a match if all elements
# are all in the multi-value attribute of the DICOM series.

Sample_Rules_Text = """
{
    "selections": [
        {
            "name": "CT Series",
            "conditions": {
                "StudyDescription": "(.*?)",
                "Modality": "(?i)CT",
                "SeriesDescription": "(.*?)",
                "ImageType": ["PRIMARY", "ORIGINAL"]
            }
        }
    ]
}
"""

if __name__ == "__main__":
    # Creates the app and test it standalone.
    AISpleenSegApp().run()

if __name__ == "__main__":
    AISpleenSegApp().run()

The above lines are needed to execute the application code by using python interpreter.

__main__.py

__main__.py is needed for MONAI Application Packager to detect the main application code (app.py) when the application is executed with the application folder path (e.g., python simple_imaging_app).

%%writefile my_app/__main__.py
from app import AISpleenSegApp

if __name__ == "__main__":
    AISpleenSegApp().run()

!ls my_app

In this time, let’s execute the app in the command line.

Note

Since the environment variables have been set and contain the correct paths, it is not necessary to provide the command line options on running the application.

!rm -rf $HOLOSCAN_OUTPUT_PATH
!python my_app

!ls -R $HOLOSCAN_OUTPUT_PATH

Packaging app

Let’s package the app with MONAI Application Packager.

In this version of the App SDK, we need to write out the configuration yaml file as well as the package requirements file, in the application folder.

%%writefile my_app/app.yaml
%YAML 1.2
---
application:
  title: MONAI Deploy App Package - MONAI Bundle AI App
  version: 1.0
  inputFormats: ["file"]
  outputFormats: ["file"]

resources:
  cpu: 1
  gpu: 1
  memory: 1Gi
  gpuMemory: 6Gi

%%writefile my_app/requirements.txt
highdicom>=0.18.2
monai>=1.0
nibabel>=3.2.1
numpy>=1.21.6
pydicom>=2.3.0
setuptools>=59.5.0 # for pkg_resources
SimpleITK>=2.0.0
torch>=1.12.0

Now we can use the CLI package command to build the MONAI Application Package (MAP) container image based on a supported base image.

Note

Building a MONAI Application Package (Docker image) can take time. Use -l DEBUG option to see the progress.

tag_prefix = "my_app"

!monai-deploy package my_app -m {models_folder} -c my_app/app.yaml -t {tag_prefix}:1.0 --platform x86_64 -l DEBUG

We can see that the MAP Docker image is created.

We can choose to display and inspect the MAP manifests by running the container with the show command, as well as extracting the manifests and other contents in the MAP by using the extract command, but not demonstrated in this example.

!docker image ls | grep {tag_prefix}

Executing packaged app locally

The packaged app can be run locally through MONAI Application Runner.

# Clear the output folder and run the MAP. The input is expected to be a folder.
!echo $HOLOSCAN_OUTPUT_PATH
!echo $HOLOSCAN_INPUT_PATH
!rm -rf $HOLOSCAN_OUTPUT_PATH
!monai-deploy run -i $HOLOSCAN_INPUT_PATH -o $HOLOSCAN_OUTPUT_PATH my_app-x64-workstation-dgpu-linux-amd64:1.0

!ls -R $HOLOSCAN_OUTPUT_PATH