PNG converter operator and the DICOM series to image app final update

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

Author: MMelQin

examples/apps/dicom_series_to_image_app/__init__.py

@@ -1,3 +1,14 @@
+# 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 os
 import sys
 

examples/apps/dicom_series_to_image_app/__main__.py

@@ -1,4 +1,15 @@
+# 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.
+
 from app import App
 
 if __name__ == "__main__":
-    App(do_run=True)
+    App().run()

examples/apps/dicom_series_to_image_app/app.py

@@ -1,4 +1,4 @@
-# Copyright 2021 MONAI Consortium
+# 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
@@ -9,27 +9,44 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from pathlib import Path
 
-from monai.deploy.core import Application
+from monai.deploy.conditions import CountCondition
+from monai.deploy.core import AppContext, Application
+from monai.deploy.logger import load_env_log_level
 from monai.deploy.operators.dicom_data_loader_operator import DICOMDataLoaderOperator
 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.png_converter_operator import PNGConverterOperator
 
 
 class App(Application):
+    """This application loads DICOM files, converts them to 3D image, then to PNG files on disk.
+
+    This showcases the MONAI Deploy application framework
+    """
+
     def compose(self):
-        study_loader_op = DICOMDataLoaderOperator()
-        series_selector_op = DICOMSeriesSelectorOperator()
-        series_to_vol_op = DICOMSeriesToVolumeOperator()
-        png_converter_op = PNGConverterOperator()
+        app_context = AppContext({})  # Let it figure out the data paths using well-known env vars etc.
+        input_dcm_folder = Path(app_context.input_path)
+        output_folder = Path(app_context.output_path)
+        print(f"input_dcm_folder: {input_dcm_folder}")
+
+        # Set the first operator to run only once by setting the count condition to 1
+        study_loader_op = DICOMDataLoaderOperator(
+            self, CountCondition(self, 1), input_folder=input_dcm_folder, name="dcm_loader"
+        )
+        series_selector_op = DICOMSeriesSelectorOperator(self, name="series_selector")
+        series_to_vol_op = DICOMSeriesToVolumeOperator(self, name="series_to_vol")
+        png_converter_op = PNGConverterOperator(self, output_folder=output_folder, name="png_converter")
 
-        self.add_flow(study_loader_op, series_selector_op, {"dicom_study_list": "dicom_study_list"})
+        # Create the execution DAG by linking operators' named output to named input.
+        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"}
+            series_selector_op, series_to_vol_op, {("study_selected_series_list", "study_selected_series_list")}
         )
-        self.add_flow(series_to_vol_op, png_converter_op, {"image": "image"})
+        self.add_flow(series_to_vol_op, png_converter_op, {("image", "image")})
 
 
 if __name__ == "__main__":
-    App(do_run=True)
+    App().run()

monai/deploy/operators/png_converter_operator.py

@@ -1,4 +1,4 @@
-# Copyright 2021 MONAI Consortium
+# 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
@@ -13,39 +13,75 @@
 from os import getcwd, makedirs
 from os.path import join
 from pathlib import Path
+from typing import Union
 
-from monai.deploy.core import Operator, OperatorSpec
+import numpy as np
+
+from monai.deploy.core import Fragment, Image, Operator, OperatorSpec
 from monai.deploy.operators.dicom_data_loader_operator import DICOMDataLoaderOperator
 from monai.deploy.operators.dicom_series_to_volume_operator import DICOMSeriesToVolumeOperator
 from monai.deploy.utils.importutil import optional_import
 
 PILImage, _ = optional_import("PIL", name="Image")
 
 
-# @md.env(pip_packages=["Pillow >= 8.0.0"])
+# @md.env(pip_packages=["Pillow >= 8.0.0", "numpy"])
 class PNGConverterOperator(Operator):
     """
-    This operator writes out a 3D Volumetric Image to disk in a slice by slice manner
+    This operator writes out a 3D Volumetric Image to to a file folder in a slice by slice manner.
+
+    Named input:
+        image: Image object or numpy ndarray
+
+    Named output:
+        None
+
+    File output:
+        Generated PNG image file(s) saved in the provided output folder.
     """
 
-    # The default output folder for saveing the generated DICOM instance file.
-    # DEFAULT_OUTPUT_FOLDER = Path(os.path.join(os.path.dirname(__file__))) / "output"
+    # The default output folder for saving the generated DICOM instance file.
     DEFAULT_OUTPUT_FOLDER = Path(getcwd()) / "output"
 
-    DEFAULT_OUTPUT_FOLDER = Path(getcwd()) / "output"
+    def __init__(
+        self,
+        fragment: Fragment,
+        *args,
+        output_folder: Union[str, Path],
+        **kwargs,
+    ):
+        """Class to write out a 3D Volumetric Image to a file folder in a slice by slice manner.
+
+        Args:
+            fragment (Fragment): An instance of the Application class which is derived from Fragment.
+            output_folder (str or Path): The folder for saving the generated DICOM instance file.
+        """
+
+        self.output_folder = output_folder if output_folder else PNGConverterOperator.DEFAULT_OUTPUT_FOLDER
+        self.input_name_image = "image"
+        # Need to call the base class constructor last
+        super().__init__(fragment, *args, **kwargs)
+
+    def setup(self, spec: OperatorSpec):
+        spec.input(self.input_name_image)
 
     def compute(self, op_input, op_output, context):
-        input_path = op_input.receive("image")
-        # output_dir = op_output.get().path
-        output_dir.mkdir(parents=True, exist_ok=True)
-        self.convert_and_save(input_path, output_dir)
+        input_image = op_input.receive(self.input_name_image)
+        self.output_folder.mkdir(parents=True, exist_ok=True)
+        self.convert_and_save(input_image, self.output_folder)
 
     def convert_and_save(self, image, path):
         """
         extracts the slices in originally acquired direction (often axial)
-        and saves then in PNG format slice by slice in the specified directory
+        and saves them in PNG format slice by slice in the specified directory
         """
-        image_data = image.asnumpy()
+
+        if isinstance(image, Image):
+            image_data = image.asnumpy()
+        elif isinstance(image, np.ndarray):
+            image_data = image
+        else:
+            raise ValueError(f"Input is not Image or ndarray, {type(image)}.")
         image_shape = image_data.shape
 
         num_images = image_shape[0]
@@ -62,31 +98,36 @@ def main():
     from pathlib import Path
 
     current_file_dir = Path(__file__).parent.resolve()
-    data_path = current_file_dir.joinpath("../../../examples/ai_spleen_seg_data/dcm")
-    out_path = "png-output"
+    data_path = current_file_dir.joinpath("../../../inputs/spleen_ct/dcm")
+    out_path = "output_png"
     makedirs(out_path, exist_ok=True)
 
     files = []
-    loader = DICOMDataLoaderOperator()
+    fragment = Fragment()
+    loader = DICOMDataLoaderOperator(fragment, name="dcm_loader")
     loader._list_files(
         data_path,
         files,
     )
     study_list = loader._load_data(files)
     series = study_list[0].get_all_series()[0]
 
-    op1 = DICOMSeriesToVolumeOperator()
+    print(f"The loaded series object properties:\n{series}")
+
+    op1 = DICOMSeriesToVolumeOperator(fragment, name="series_to_vol")
     op1.prepare_series(series)
     voxels = op1.generate_voxel_data(series)
     metadata = op1.create_metadata(series)
     image = op1.create_volumetric_image(voxels, metadata)
 
-    op2 = PNGConverterOperator()
-    op2.convert_and_save(image, out_path)
-
-    print(f"The loaded series object properties:\n{series}")
     print(f"The converted Image object metadata:\n{metadata}")
 
+    op2 = PNGConverterOperator(fragment, output_folder=out_path, name="png_converter")
+    # Not mocking the operator context, so bypassing compute
+    op2.convert_and_save(image, op2.output_folder)
+
+    print(f"The converted PNG files are in: {out_path}")
+
 
 if __name__ == "__main__":
     main()