Deploying a MedNIST Classifier App with MONAI Deploy App SDK#
This tutorial demos the process of packaging up a trained model using MONAI Deploy App SDK into an artifact which can be run as a local program performing inference, a workflow job doing the same, and a Docker containerized workflow execution.
In this tutorial, we will train a MedNIST classifier like the MONAI tutorial here and then implement & package the inference application, executing the application locally.
Train a MedNIST classifier model with MONAI Core#
Setup environment#
# Install necessary packages for MONAI Core
!python -c "import monai" || pip install -q "monai[pillow, tqdm]"
!python -c "import ignite" || pip install -q "monai[ignite]"
!python -c "import gdown" || pip install -q "monai[gdown]"
!python -c "import pydicom" || pip install -q "pydicom>=1.4.2"
!python -c "import highdicom" || pip install -q "highdicom>=0.18.2" # for the use of DICOM Writer operators
# Install MONAI Deploy App SDK package
!python -c "import monai.deploy" || pip install -q "monai-deploy-app-sdk"
Setup imports#
# Copyright 2020 MONAI Consortium
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import shutil
import tempfile
import glob
import PIL.Image
import torch
import numpy as np
from ignite.engine import Events
from monai.apps import download_and_extract
from monai.config import print_config
from monai.networks.nets import DenseNet121
from monai.engines import SupervisedTrainer
from monai.transforms import (
EnsureChannelFirst,
Compose,
LoadImage,
RandFlip,
RandRotate,
RandZoom,
ScaleIntensity,
EnsureType,
)
from monai.utils import set_determinism
set_determinism(seed=0)
print_config()
Download dataset#
The MedNIST dataset was gathered from several sets from TCIA, the RSNA Bone Age Challenge(https://www.rsna.org/education/ai-resources-and-training/ai-image-challenge/rsna-pediatric-bone-age-challenge-2017), and the NIH Chest X-ray dataset.
The dataset is kindly made available by Dr. Bradley J. Erickson M.D., Ph.D. (Department of Radiology, Mayo Clinic) under the Creative Commons CC BY-SA 4.0 license.
If you use the MedNIST dataset, please acknowledge the source.
Note: Data files are now access controlled. Please first request permission to access the shared folder on Google Drive, then download and extract the zip file to a folder called MedNIST_DATA at the same level as this notebook.
directory = os.environ.get("MONAI_DATA_DIRECTORY")
root_dir = directory if directory else os.path.join(os.path.curdir, "MedNIST_DATA")
print(root_dir)
resource = "https://drive.google.com/uc?id=1QsnnkvZyJPcbRoV_ArW8SnE1OTuoVbKE"
md5 = "0bc7306e7427e00ad1c5526a6677552d"
compressed_file = os.path.join(root_dir, "MedNIST.tar.gz")
data_dir = os.path.join(root_dir, "MedNIST")
if not os.path.exists(data_dir):
download_and_extract(resource, compressed_file, root_dir, md5)
subdirs = sorted(glob.glob(f"{data_dir}/*/"))
class_names = [os.path.basename(sd[:-1]) for sd in subdirs]
image_files = [glob.glob(f"{sb}/*") for sb in subdirs]
image_files_list = sum(image_files, [])
image_class = sum(([i] * len(f) for i, f in enumerate(image_files)), [])
image_width, image_height = PIL.Image.open(image_files_list[0]).size
print(f"Label names: {class_names}")
print(f"Label counts: {list(map(len, image_files))}")
print(f"Total image count: {len(image_class)}")
print(f"Image dimensions: {image_width} x {image_height}")
Setup and train#
Here we’ll create a transform sequence and train the network, omitting validation and testing since we know this does indeed work and it’s not needed here:
train_transforms = Compose(
[
LoadImage(image_only=True),
EnsureChannelFirst(channel_dim="no_channel"),
ScaleIntensity(),
RandRotate(range_x=np.pi / 12, prob=0.5, keep_size=True),
RandFlip(spatial_axis=0, prob=0.5),
RandZoom(min_zoom=0.9, max_zoom=1.1, prob=0.5),
EnsureType(),
]
)
class MedNISTDataset(torch.utils.data.Dataset):
def __init__(self, image_files, labels, transforms):
self.image_files = image_files
self.labels = labels
self.transforms = transforms
def __len__(self):
return len(self.image_files)
def __getitem__(self, index):
return self.transforms(self.image_files[index]), self.labels[index]
# just one dataset and loader, we won't bother with validation or testing
train_ds = MedNISTDataset(image_files_list, image_class, train_transforms)
train_loader = torch.utils.data.DataLoader(train_ds, batch_size=300, shuffle=True, num_workers=10)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
net = DenseNet121(spatial_dims=2, in_channels=1, out_channels=len(class_names)).to(device)
loss_function = torch.nn.CrossEntropyLoss()
opt = torch.optim.Adam(net.parameters(), 1e-5)
max_epochs = 5
def _prepare_batch(batch, device, non_blocking):
return tuple(b.to(device) for b in batch)
trainer = SupervisedTrainer(device, max_epochs, train_loader, net, opt, loss_function, prepare_batch=_prepare_batch)
@trainer.on(Events.EPOCH_COMPLETED)
def _print_loss(engine):
print(f"Epoch {engine.state.epoch}/{engine.state.max_epochs} Loss: {engine.state.output[0]['loss']}")
trainer.run()
The network will be saved out here as a Torchscript object named classifier.zip
torch.jit.script(net).save("classifier.zip")
Implementing and Packaging Application with MONAI Deploy App SDK#
Based on the Torchscript model(classifier.zip), we will implement an application that process an input Jpeg image and write the prediction(classification) result as JSON file(output.json).
Creating Operators and connecting them in Application class#
We used the following train transforms as pre-transforms during the training.
Train transforms used in training#
1train_transforms = Compose(
2 [
3 LoadImage(image_only=True),
4 EnsureChannelFirst(channel_dim="no_channel"),
5 ScaleIntensity(),
6 RandRotate(range_x=np.pi / 12, prob=0.5, keep_size=True),
7 RandFlip(spatial_axis=0, prob=0.5),
8 RandZoom(min_zoom=0.9, max_zoom=1.1, prob=0.5),
9 EnsureType(),
10 ]
11)
RandRotate, RandFlip, and RandZoom transforms are used only for training and those are not necessary during the inference.
In our inference application, we will define two operators:
LoadPILOperator- Load a JPEG image from the input path and pass the loaded image object to the next operator.This Operator does similar job with
LoadImage(image_only=True)transform in train_transforms, but handles only one image.Input: a file path (
Path)Output: an image object in memory (
Image)
MedNISTClassifierOperator- Pre-transform the given image by using MONAI’sComposeclass, feed to the Torchscript model (classifier.zip), and write the prediction into JSON file(output.json)Pre-transforms consist of three transforms –
EnsureChannelFirst,ScaleIntensity, andEnsureType.Input: an image object in memory (
Image)Output: a folder path that the prediction result(
output.json) would be written (DataPath)
The workflow of the application would look like this.
%%{init: {"theme": "base", "themeVariables": { "fontSize": "16px"}} }%%
classDiagram
direction LR
LoadPILOperator --|> MedNISTClassifierOperator : image...image
class LoadPILOperator {
<in>image : DISK
image(out) IN_MEMORY
}
class MedNISTClassifierOperator {
<in>image : IN_MEMORY
output(out) DISK
}
Set up environment variables#
Before proceeding to the application building and packaging, we first need to set the well-known environment variables, because the application parses them for the input, output, and model folders. Defaults are used if these environment variable are absent.
Set the environment variables corresponding to the extracted data path.
input_folder = "input"
output_foler = "output"
models_folder = "models"
# Choose a file as test input
test_input_path = image_files[0][0]
!rm -rf {input_folder} && mkdir -p {input_folder} && cp {test_input_path} {input_folder} && ls {input_folder}
# Need to copy the model file to its own clean subfolder for packaging, to workaround an issue in the Packager
!rm -rf {models_folder} && mkdir -p {models_folder}/model && cp classifier.zip {models_folder}/model && ls {models_folder}/model
%env HOLOSCAN_INPUT_PATH {input_folder}
%env HOLOSCAN_OUTPUT_PATH {output_foler}
%env HOLOSCAN_MODEL_PATH {models_folder}
Setup imports#
Let’s import necessary classes/decorators and define MEDNIST_CLASSES.
import logging
import os
from pathlib import Path
from typing import Optional
import torch
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 EnsureChannelFirst, Compose, EnsureType, ScaleIntensity
MEDNIST_CLASSES = ["AbdomenCT", "BreastMRI", "CXR", "ChestCT", "Hand", "HeadCT"]
Creating Operator classes#
LoadPILOperator#
class LoadPILOperator(Operator):
"""Load image from the given input (DataPath) and set numpy array to the output (Image)."""
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 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(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.emit(output_image, self.output_name_image) # cannot omit the name even if single output.
MedNISTClassifierOperator#
class MedNISTClassifierOperator(Operator):
"""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,
fragment: Fragment,
*args,
app_context: AppContext,
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 overridden 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
self.app_context = app_context
self.model = self._get_model(self.app_context, self.model_path, self._model_name)
# This needs to be at the end of the constructor.
super().__init__(fragment, *args, **kwargs)
def _get_model(self, app_context: AppContext, model_path: Path, model_name: str):
"""Load the model with the given name from context or model path
Args:
app_context (AppContext): The application context object holding the model(s)
model_path (Path): The path to the model file, as a backup to load model directly
model_name (str): The name of the model, when multiples are loaded in the context
"""
if app_context.models:
# `app_context.models.get(model_name)` returns a model instance if exists.
# If model_name is not specified and only one model exists, it returns that model.
model = app_context.models.get(model_name)
else:
model = torch.jit.load(
MedNISTClassifierOperator.MODEL_LOCAL_PATH,
map_location=torch.device("cuda" if torch.cuda.is_available() else "cpu"),
)
return model
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([EnsureChannelFirst(channel_dim="no_channel"), ScaleIntensity(), EnsureType()])
def compute(self, op_input, op_output, context):
import json
import torch
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)
with torch.no_grad():
outputs = self.model(image_tensor)
_, output_classes = outputs.max(dim=1)
result = MEDNIST_CLASSES[output_classes[0]] # get the class name
print(result)
op_output.emit(result, self.output_name_result)
# 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)
Creating Application class#
Our application class would look like below.
It defines App class inheriting Application class.
LoadPILOperator is connected to MedNISTClassifierOperator by using self.add_flow() in compose() method of App.
class App(Application):
"""Application class for the MedNIST classifier."""
def compose(self):
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)
load_pil_op = LoadPILOperator(self, CountCondition(self, 1), input_folder=app_input_path, name="pil_loader_op")
classifier_op = MedNISTClassifierOperator(
self, app_context=app_context, 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(
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, {("image", "image")})
self.add_flow(classifier_op, dicom_sr_operator, {("result_text", "text")})
Executing app locally#
The test input file file, output path, and model have been prepared, and the paths set in the environment variables, so we can go ahead and execute the application Jupyter notebook with a clean output folder.
!rm -rf $HOLOSCAN_OUTPUT_PATH
app = App().run()
!cat $HOLOSCAN_OUTPUT_PATH/output.json
Once the application is verified inside Jupyter notebook, we can write the whole application as a file(mednist_classifier_monaideploy.py) by concatenating code above, then add the following lines:
if __name__ == "__main__":
App()
The above lines are needed to execute the application code by using python interpreter.
# Create an application folder
!mkdir -p mednist_app
!rm -rf mednist_app/*
%%writefile mednist_app/mednist_classifier_monaideploy.py
# Copyright 2021-2023 MONAI Consortium
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import os
from pathlib import Path
from typing import Optional
import torch
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 EnsureChannelFirst, Compose, EnsureType, ScaleIntensity
MEDNIST_CLASSES = ["AbdomenCT", "BreastMRI", "CXR", "ChestCT", "Hand", "HeadCT"]
# @md.env(pip_packages=["pillow"])
class LoadPILOperator(Operator):
"""Load image from the given input (DataPath) and set numpy array to the output (Image)."""
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 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(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.emit(output_image, self.output_name_image) # cannot omit the name even if single output.
# @md.env(pip_packages=["monai"])
class MedNISTClassifierOperator(Operator):
"""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,
fragment: Fragment,
*args,
app_context: AppContext,
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 overridden 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
self.app_context = app_context
self.model = self._get_model(self.app_context, self.model_path, self._model_name)
# This needs to be at the end of the constructor.
super().__init__(fragment, *args, **kwargs)
def _get_model(self, app_context: AppContext, model_path: Path, model_name: str):
"""Load the model with the given name from context or model path
Args:
app_context (AppContext): The application context object holding the model(s)
model_path (Path): The path to the model file, as a backup to load model directly
model_name (str): The name of the model, when multiples are loaded in the context
"""
if app_context.models:
# `app_context.models.get(model_name)` returns a model instance if exists.
# If model_name is not specified and only one model exists, it returns that model.
model = app_context.models.get(model_name)
else:
model = torch.jit.load(
MedNISTClassifierOperator.MODEL_LOCAL_PATH,
map_location=torch.device("cuda" if torch.cuda.is_available() else "cpu"),
)
return model
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([EnsureChannelFirst(channel_dim="no_channel"), ScaleIntensity(), EnsureType()])
def compute(self, op_input, op_output, context):
import json
import torch
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)
with torch.no_grad():
outputs = self.model(image_tensor)
_, output_classes = outputs.max(dim=1)
result = MEDNIST_CLASSES[output_classes[0]] # get the class name
print(result)
op_output.emit(result, self.output_name_result)
# 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")
class App(Application):
"""Application class for the MedNIST classifier."""
def compose(self):
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, app_context=app_context, 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(
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, {("image", "image")})
self.add_flow(classifier_op, dicom_sr_operator, {("result_text", "text")})
if __name__ == "__main__":
App().run()
This time, let’s execute the app in the command line.
!python "mednist_app/mednist_classifier_monaideploy.py"
!cat $HOLOSCAN_OUTPUT_PATH/output.json
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 mednist_app/app.yaml
%YAML 1.2
---
application:
title: MONAI Deploy App Package - MedNIST Classifier App
version: 1.0
inputFormats: ["file"]
outputFormats: ["file"]
resources:
cpu: 1
gpu: 1
memory: 1Gi
gpuMemory: 1Gi
%%writefile mednist_app/requirements.txt
monai>=1.2.0
Pillow>=8.4.0
pydicom>=2.3.0
highdicom>=0.18.2
SimpleITK>=2.0.0
setuptools>=59.5.0 # for pkg_resources
tag_prefix = "mednist_app"
!monai-deploy package "mednist_app/mednist_classifier_monaideploy.py" -m {models_folder} -c "mednist_app/app.yaml" -t {tag_prefix}:1.0 --platform x86_64 -l DEBUG
Note
Building a MONAI Application Package (Docker image) can take time. Use -l DEBUG option if you want to see the progress.
We can see that the Docker image is created.
!docker image ls | grep {tag_prefix}
Executing packaged app locally#
We can choose to display and export the MAP manifests, but in this example, we will just run the MAP through MONAI Application Runner.
# Clear the output folder and run the MAP. The input is expected to be a folder.
!rm -rf $HOLOSCAN_OUTPUT_PATH
!monai-deploy run -i$HOLOSCAN_INPUT_PATH -o $HOLOSCAN_OUTPUT_PATH mednist_app-x64-workstation-dgpu-linux-amd64:1.0
!cat $HOLOSCAN_OUTPUT_PATH/output.json
Note: Please execute the following script once the exercise is done.
# Remove data files which is in the temporary folder
if directory is None:
shutil.rmtree(root_dir)