app.py

import logging
from lung_seg_operator import LungSegOperator

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.publisher_operator import PublisherOperator

Sample_Rules_Text = """
{
    "selections": [
        {
            "name": "CT Series",
            "conditions": {
                "Modality": "(?i)CT",
                "ImageType": ["PRIMARY", "ORIGINAL"],
                "PhotometricInterpretation": "MONOCHROME2"
            }
        }
    ]
}
"""

@resource(cpu=1, gpu=1, memory="7Gi")
class LungSegmentationApp(Application):
	def __init__(self, *args, **kwargs):
		self._logger = logging.getLogger("{}.{}".format(__name__, type(self).__name__))
		super().__init__(*args, **kwargs)

	def run(self, *args, **kwargs):
		self._logger.debug(f"Begin {self.run.__name__}")
		super().run(*args, **kwargs)
		self._logger.debug(f"End {self.run.__name__}")

	def compose(self):
		self._logger.debug(f"Begin {self.compose.__name__}")
		study_loader_op = DICOMDataLoaderOperator()
		series_selector_op = DICOMSeriesSelectorOperator(rules=Sample_Rules_Text)
		series_to_vol_op = DICOMSeriesToVolumeOperator()

		lung_op = LungSegOperator()

		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, lung_op, {"image": "image"})
		print(series_selector_op)

if __name__ == '__main__':
	logging.basicConfig(level=logging.DEBUG)
	app_instance = LungSegmentationApp()
	app_instance.run()

lung_seg_operator.py

import logging
from numpy import uint8

import monai.deploy.core as md 
from monai.deploy.core import DataPath, ExecutionContext, Image, InputContext, IOType, Operator, OutputContext
from monai.deploy.operators.monai_seg_inference_operator import InMemImageReader, MonaiSegInferenceOperator, InferenceOperator
from monai.transforms import (
	Activationsd, 
	AsDiscreted, 
	Compose, 
	CropForegroundd, 
	EnsureChannelFirstd, 
	Invertd, 
	LoadImaged, 
	SaveImaged, 
	ScaleIntensityRanged, 
	Spacingd,)

@md.input("image", Image, IOType.IN_MEMORY)
@md.output("seg_image", Image, IOType.IN_MEMORY)
@md.output("saved_images_folder", DataPath, IOType.DISK)
@md.env(pip_packages=["monai>=1.0.0", "numpy>-1.21", "nibabel"])

class LungSegOperator(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_image = op_input.get("image")
		if not input_image:
			raise ValueError("Input image is not found")
		"""
		op_output_folder_name = DataPath("saved_images_folder")
		op_output.set(op_output_folder_name, "saved_images_folder")
		op_output_folder_path = op_output.get("saved_images_folder").path
		op_output_folder_path.mkdir(parents=True, exist_ok = True)
		print(f"Operator output folder path: {op_output_folder_path}")
		"""
		op_output_folder_path='./'
		_reader = InMemImageReader(input_image)
		pre_transforms = self.pre_process(_reader, op_output_folder_path)
		x = input_image.asnumpy()
		print(x.shape)
		a = pre_transforms(x)
		#post_transforms = self.post_process(pre_transforms, op_output_folder_path)

		#infer_operator = MonaiSegInferenceOperator(
		#				roi_size=(256, 256, 256), 
		#				pre_transforms, post_transforms, model_name="")
	
		#infer_operator = InferenceOperator(

		#infer_operator.input_dataset_key = self._input_dataset_key
		#infer_operator.pred_dataset_key = self._pred_dataset_key

		#infer_operator.compute(op_input, op_output, context)

	def pre_process(self, img_reader, out_dir: str= "./input_images") -> 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), 
				SaveImaged(keys=my_key, 
					output_dir = out_dir, output_postfix = "", resample=False)])

	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),
			AsDiscreted(keys=pred_key, argmax=True), 
			Invertd(keys=pred_key, output_dir=out_dir, output_postfix="seg", output_dtype=uint8, resample=False),]
			)