Create a Segmentation App with MONAI Deploy App SDK 0.5.1 that can read NIFTI file

[fanruifu]

Hello,

I am working on creating a segmentation app based on this example:
https://docs.monai.io/projects/monai-deploy-app-sdk/en/0.5.1/notebooks/tutorials/03_segmentation_app.html
I am using a Windows system, so I am unable to use SDK versions newer than 0.5.1.

The example uses DICOMSeriesToVolumeOperator to pass an image object to the SpleenSegOperator, so I have been looking for an operator that can do the similar operation to NIFTI files. I found the NiftiDataLoader here (https://docs.monai.io/projects/monai-deploy-app-sdk/en/0.5.1/modules/_autosummary/monai.deploy.operators.NiftiDataLoader.html) but it only returns a ndarray.

I also found a discussion related to this question but not sure how to use it.
#421

Any suggestions would be greatly appreciated. Thank you!

[vikashg]

Hi @fanruifu Can you share your app.py file I would like to see how are you chaining the operators ?

[fanruifu]

I also tried changing the input of the segmentation operator to allow it take the datapath as the input, and directly assign it to Loadiamged transform method but it did not work as well.

"""code"""
@md.input("image", DataPath, IOType.DISK)
@md.output("seg_image", Image, IOType.IN_MEMORY)
class SpleenSegOperator(Operator):

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

def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):
    input_path = op_input.get().path

    
    output_path = context.output.get().path
    loader = LoadImaged(keys="image",ensure_channel_first=True)
    data_dict = loader({"image":input_path})
    pre_trans=self.pre_process()
    pre_transforms=pre_trans(data_dict)
    post_transforms = self.post_process(pre_transforms, path.join(output_path, "prediction_output"))

    infer_operator = MonaiSegInferenceOperator(
        (
            96,
            96,
            96,
        ),
        pre_transforms,
        post_transforms,
    )

    # Set the keys for the operator
    infer_operator.input_dataset_key = self._input_dataset_key
    infer_operator.pred_dataset_key = self._pred_dataset_key


    # Invoke the inference
    infer_operator.compute(op_input, op_output, context)

def pre_process(self) -> Compose:
    my_key = self._input_dataset_key
    return Compose(
        [
            Orientationd(keys=my_key, axcodes="RAS"),
            Spacingd(keys=my_key, pixdim=[0.5, 0.5, 0.5], mode=["bilinear"]),
            NormalizeIntensityd(keys='image', nonzero=True),
            EnsureTyped(keys=my_key),
        ]
    )

def post_process(self, pre_transforms: Compose, out_dir: str = "./prediction_output") -> Compose:
    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),
            SaveImaged(
                keys=pred_key,
                output_dir=out_dir,
                output_postfix="seg",
                output_dtype=uint8,
            ),
        ]
    )

[vikashg]

@fanruifu I am trying to run your code and resolve it. Can you please share the error you are getting along with the nifti file.
BTW, @MMelQin the link for the spleen segmentation data is broken. This link https://drive.google.com/uc?id=1Uds8mEvdGNYUuvFpTtCQ8gNU97bAPCaQ in the following tutorial https://github.com/Project-MONAI/monai-deploy-app-sdk/blob/main/notebooks/tutorials/03_segmentation_app.ipynb

[fanruifu]

Sure, the error is :

TypeCheckError: numpy.ndarray is not an instance of monai.deploy.core.domain.image.Image

Here is the complite code, the nifti file has no problem:

import logging
from os import path
from numpy import uint8
import monai.deploy.core as md
from monai.deploy.core import ExecutionContext, Image, InputContext, IOType, Operator, OutputContext
from monai.deploy.operators.monai_seg_inference_operator import InMemImageReader, MonaiSegInferenceOperator
from monai.transforms import (
    Activationsd,
    AsDiscreted,
    Compose,
    EnsureChannelFirstd,
    EnsureTyped,
    Invertd,
    LoadImaged,
    Orientationd,
    SaveImaged,
    ScaleIntensityRanged,
    Spacingd,
    NormalizeIntensityd,
)


@md.input("image", Image, IOType.IN_MEMORY)
@md.output("seg_image", Image, IOType.IN_MEMORY)
@md.env(pip_packages=["monai>=0.8.1", "torch>=1.10.2", "numpy>=1.21", "nibabel"])
class SpleenSegOperator(Operator):
    """Performs Spleen segmentation with a 3D image converted from a DICOM CT series.
    """

    def __init__(self):

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

    def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):

        input_image = op_input.get("image")
        if not input_image:
            raise ValueError("Input image is not found.")

        output_path = context.output.get().path

        # This operator gets an in-memory Image object, so a specialized ImageReader is needed.
        _reader = InMemImageReader(input_image)
        pre_transforms = self.pre_process(_reader)
        post_transforms = self.post_process(pre_transforms, path.join(output_path, "prediction_output"))

        # Delegates inference and saving output to the built-in operator.
        infer_operator = MonaiSegInferenceOperator(
            (
                96,
                96,
                96,
            ),
            pre_transforms,
            post_transforms,
        )

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

        # Now let the built-in operator handles the work with the I/O spec and execution context.
        infer_operator.compute(op_input, op_output, context)

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

        my_key = self._input_dataset_key
        return Compose(
            [
                LoadImaged(keys=my_key, reader=img_reader),
                EnsureChannelFirstd(keys=my_key),
                Orientationd(keys=my_key, axcodes="RAS"),
                Spacingd(keys=my_key, pixdim=[0.5, 0.5, 0.5], mode=["bilinear"]),
                NormalizeIntensityd(keys='image', nonzero=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."""

        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),
                SaveImaged(
                    keys=pred_key,
                    output_dir=out_dir,
                    output_postfix="seg",
                    output_dtype=uint8,
                ),
            ]
        )

# =============================================================================
# 
# =============================================================================

# 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.core import Application, resource
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 import NiftiDataLoader

@resource(cpu=1, gpu=1, memory="7Gi")
class AISpleenSegApp(Application):
    def __init__(self, *args, **kwargs):
        """Creates an application instance."""

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

    def run(self, *args, **kwargs):
        # This method calls the base class to run. Can be omitted if simply calling through.
        self._logger.debug(f"Begin {self.run.__name__}")
        super().run(*args, **kwargs)
        self._logger.debug(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__}")
        nifti_op = NiftiDataLoader()
        # Model specific inference operator, supporting MONAI transforms.

        # Creates the model specific segmentation operator
        spleen_seg_op = SpleenSegOperator()


        self.add_flow(nifti_op, spleen_seg_op, {"image": "image"})


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

# =============================================================================
# 
# =============================================================================
app = AISpleenSegApp()

app.run(input=r"path to nifti file", output=r"path to output folder", model=r"path to model")

[fanruifu]

This issue is similar to my question:
#421

[vikashg]

Hi @fanruifu working on this I got tied up into other things.