Enhanced the code and documentation

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

Author: MMelQin

examples/apps/simple_imaging_app/__main__.py

@@ -2,4 +2,3 @@
 
 if __name__ == "__main__":
     App().run()
-

monai/deploy/operators/dicom_data_loader_operator.py

@@ -14,7 +14,7 @@
 from pathlib import Path
 from typing import List
 
-from monai.deploy.core import Operator, OperatorSpec
+from monai.deploy.core import Fragment, Operator, OperatorSpec
 from monai.deploy.core.domain.dicom_series import DICOMSeries
 from monai.deploy.core.domain.dicom_study import DICOMStudy
 from monai.deploy.exceptions import ItemNotExistsError
@@ -33,20 +33,23 @@ class DICOMDataLoaderOperator(Operator):
     """This operator loads DICOM studies into memory from a folder containing DICOM instance files.
 
     Input:
-        Path to the folder containing DICOM instance files, set as argument to the object constructor
+        Path to the folder containing DICOM instance files, set as argument to the object constructor.
+        It defaults to the folder, input, in the current working directory.
 
     Output:
         A list of DICOMStudy objects in memory, named `dicom_study_list` by default but can be changed
-        via the object instance attribute, `output_name`.
+        via the object instance attribute, `output_name`. The name can be omitted when linking it to the
+        receiver, simply because it is the only output.
     """
 
     DEFAULT_INPUT_FOLDER = Path.cwd() / "input"
     DEFAULT_OUTPUT_NAME = "dicom_study_list"
 
     # For now, need to have the input folder as an instance attribute, set on init, because the
-    # compute function does not get file I/O path in the context. Enhancement has been requested.
+    # compute function does not get file I/O path in the execution context. Enhancement requested.
     def __init__(
         self,
+        fragment: Fragment,
         *args,
         input_folder: Path = DEFAULT_INPUT_FOLDER,
         output_name: str = DEFAULT_OUTPUT_NAME,
@@ -56,6 +59,7 @@ def __init__(
         """Creates an instance of this class
 
         Args:
+            fragment (Fragment): An instance of the Application class which is derived from Fragment.
             input_folder (Path): Folder containing DICOM instance files to load from.
                                  Defaults to `input` in the current working directory.
             output_name (str): The name for the output, which is list of DICOMStudy objects.
@@ -67,13 +71,16 @@ def __init__(
         self._must_load = must_load
         self.input_path = input_folder
         self.index = 0
-        self.output_name = output_name.strip() if output_name and len(output_name.strip()) > 0 else DEFAULT_OUTPUT_NAME
+        self.output_name = (
+            output_name.strip()
+            if output_name and len(output_name.strip()) > 0
+            else DICOMDataLoaderOperator.DEFAULT_OUTPUT_NAME
+        )
 
-        super().__init__(*args, **kwargs)
+        super().__init__(fragment, *args, **kwargs)
 
     def setup(self, spec: OperatorSpec):
         spec.output(self.output_name)
-        # spec.param("input_path", Path("."))
 
     def compute(self, op_input, op_output, context):
         """Performs computation for this operator and handlesI/O."""
@@ -88,7 +95,7 @@ def compute(self, op_input, op_output, context):
         # MQ
 
         dicom_study_list = self.load_data_to_studies(input_path)
-        op_output.emit(dicom_study_list, "dicom_study_list")
+        op_output.emit(dicom_study_list, self.output_name)
 
     def load_data_to_studies(self, input_path: Path):
         """Load DICOM data from files into DICOMStudy objects in a list.
@@ -314,7 +321,7 @@ def test():
     current_file_dir = Path(__file__).parent.resolve()
     data_path = current_file_dir.joinpath("../../../inputs/spleen_ct/dcm")
 
-    loader = DICOMDataLoaderOperator()
+    loader = DICOMDataLoaderOperator(Fragment())
     study_list = loader.load_data_to_studies(data_path.absolute())
 
     for study in study_list:
@@ -363,7 +370,7 @@ def test():
     except ItemNotExistsError as ex:
         print(f"Test passed: exception when no studies loaded & must_load flag is True: {ex}")
 
-    relaxed_loader = DICOMDataLoaderOperator(must_load=False)
+    relaxed_loader = DICOMDataLoaderOperator(Fragment(), must_load=False)
     study_list = relaxed_loader.load_data_to_studies(non_dcm_dir)
     print(f"Test passed: {len(study_list)} study loaded and is OK when must_load flag is False.")
 

monai/deploy/operators/dicom_encapsulated_pdf_writer_operator.py

@@ -11,10 +11,9 @@
 
 import logging
 import os
-from ast import Bytes
 from io import BytesIO
 from pathlib import Path
-from typing import Dict, List, Optional, Union
+from typing import Dict, Optional, Union
 
 from monai.deploy.utils.importutil import optional_import
 
@@ -27,24 +26,36 @@
 Sequence, _ = optional_import("pydicom.sequence", name="Sequence")
 PdfReader, _ = optional_import("PyPDF2", name="PdfReader")
 
-from monai.deploy.core import Operator, OperatorSpec
+from monai.deploy.core import ConditionType, Fragment, Operator, OperatorSpec
 from monai.deploy.core.domain.dicom_series import DICOMSeries
 from monai.deploy.core.domain.dicom_series_selection import StudySelectedSeries
-from monai.deploy.exceptions import ItemNotExistsError
 from monai.deploy.operators.dicom_utils import EquipmentInfo, ModelInfo, save_dcm_file, write_common_modules
 from monai.deploy.utils.version import get_sdk_semver
 
 
 # @md.env(pip_packages=["pydicom >= 1.4.2", "PyPDF2 >= 2.11.1", "monai"])
 class DICOMEncapsulatedPDFWriterOperator(Operator):
+    """Class to write DICOM Encapsulated PDF Instance with provided PDF bytes in memory.
+
+    Named inputs:
+        pdf_bytes: Bytes of the the PDF content.
+        study_selected_series_list: Optional, DICOM series for copying metadata from.
+
+    Named output:
+        None
+
+    File output:
+        Generaed DICOM instance file in the provided output folder.
+    """
+
     # File extension for the generated DICOM Part 10 file.
     DCM_EXTENSION = ".dcm"
     # The default output folder for saveing the generated DICOM instance file.
-    # DEFAULT_OUTPUT_FOLDER = Path(os.path.join(os.path.dirname(__file__))) / "output"
     DEFAULT_OUTPUT_FOLDER = Path(os.getcwd()) / "output"
 
     def __init__(
         self,
+        fragment: Fragment,
         *args,
         output_folder: Union[str, Path],
         copy_tags: bool,
@@ -56,6 +67,7 @@ def __init__(
         """Class to write DICOM Encapsulated PDF Instance with PDF bytes in memory or in a file.
 
         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.
             copy_tags (bool): True for copying DICOM attributes from a provided DICOMSeries.
                               If True and no DICOMSeries obj provided, runtime exception is thrown.
@@ -81,6 +93,8 @@ def __init__(
         self.model_info = model_info if model_info else ModelInfo()
         self.equipment_info = equipment_info if equipment_info else EquipmentInfo()
         self.custom_tags = custom_tags
+        self.input_name_bytes = "pdf_bytes"
+        self.input_name_dcm_series = "study_selected_series_list"
 
         # Set own Modality and SOP Class UID
         # Modality, e.g.,
@@ -101,7 +115,7 @@ def __init__(
         except Exception:
             self.software_version_number = ""
         self.operators_name = f"AI Algorithm {self.model_info.name}"
-        super().__init__(*args, **kwargs)
+        super().__init__(fragment, *args, **kwargs)
 
     def setup(self, spec: OperatorSpec):
         """Set up the named input(s), and output(s) if applicable.
@@ -112,8 +126,8 @@ def setup(self, spec: OperatorSpec):
             spec (OperatorSpec): The Operator specification for inputs and outputs etc.
         """
 
-        spec.input("pdf_bytes")
-        spec.input("study_selected_series_list")
+        spec.input(self.input_name_bytes)
+        spec.input(self.input_name_dcm_series).condition(ConditionType.NONE)  # Optional input
 
     def compute(self, op_input, op_output, context):
         """Performs computation for this operator and handles I/O.
@@ -133,24 +147,13 @@ def compute(self, op_input, op_output, context):
 
         # Gets the input, prepares the output folder, and then delegates the processing.
         pdf_bytes: bytes = b""
-        try:
-            pdf_bytes = op_input.receive("pdf_bytes")
-        except ItemNotExistsError:
-            # try:
-            #     file_path = op_input.receive("pdf_file")
-            # except ItemNotExistsError:
-            #     raise ValueError("None of the named inputs can be found.") from None
-            # # Read file, and if exception, let it bubble up
-            # with open(file_path.path, "rb") as f:
-            #     pdf_bytes = f.read().strip()
-            pass
-
+        pdf_bytes = op_input.receive(self.input_name_bytes)
         if not pdf_bytes or not len(pdf_bytes.strip()):
             raise IOError("Input is read but blank.")
 
         try:
-            study_selected_series_list = op_input.receive("study_selected_series_list")
-        except ItemNotExistsError:
+            study_selected_series_list = op_input.receive(self.input_name_dcm_series)
+        except Exception:
             study_selected_series_list = None
 
         dicom_series = None  # It can be None if not to copy_tags.
@@ -245,24 +248,27 @@ def _is_pdf_bytes(self, content: bytes):
         return True
 
 
-def test():
+def test(test_copy_tags: bool = True):
     from monai.deploy.operators.dicom_data_loader_operator import DICOMDataLoaderOperator
     from monai.deploy.operators.dicom_series_selector_operator import DICOMSeriesSelectorOperator
 
     current_file_dir = Path(__file__).parent.resolve()
     dcm_folder = current_file_dir.joinpath("../../../inputs/livertumor_ct/dcm/1-CT_series_liver_tumor_from_nii014")
     pdf_file = current_file_dir.joinpath("../../../inputs/pdf/TestPDF.pdf")
-    out_path = "output_pdf_op"
+    out_path = Path("output_pdf_op").absolute()
     pdf_bytes = b"Not PDF bytes."
-    test_copy_tags = False
 
-    loader = DICOMDataLoaderOperator()
-    series_selector = DICOMSeriesSelectorOperator()
+    fragment = Fragment()
+    loader = DICOMDataLoaderOperator(fragment, name="loader_op")
+    series_selector = DICOMSeriesSelectorOperator(fragment, name="selector_op")
     sr_writer = DICOMEncapsulatedPDFWriterOperator(
+        fragment,
+        output_folder=out_path,
         copy_tags=test_copy_tags,
         model_info=None,
         equipment_info=EquipmentInfo(),
         custom_tags={"SeriesDescription": "Report from AI algorithm. Not for clinical use."},
+        name="writer_op",
     )
 
     # Testing with the main entry functions
@@ -284,8 +290,9 @@ def test():
     with open(pdf_file, "rb") as f:
         pdf_bytes = f.read()
 
-    sr_writer.write(pdf_bytes, dicom_series, Path(out_path).absolute())
+    sr_writer.write(pdf_bytes, dicom_series, out_path)
 
 
 if __name__ == "__main__":
-    test()
+    test(test_copy_tags=True)
+    test(test_copy_tags=False)

monai/deploy/operators/dicom_series_selector_operator.py

@@ -13,21 +13,23 @@
 import numbers
 import re
 from json import loads as json_loads
-from typing import Dict, List, Text
+from typing import List
 
-from monai.deploy.core import Operator, OperatorSpec  # ExecutionContext, InputContext, IOType, Operator, OutputContext
+from monai.deploy.core import ConditionType, Fragment, Operator, OperatorSpec
 from monai.deploy.core.domain.dicom_series import DICOMSeries
 from monai.deploy.core.domain.dicom_series_selection import SelectedSeries, StudySelectedSeries
 from monai.deploy.core.domain.dicom_study import DICOMStudy
-from monai.deploy.exceptions import ItemNotExistsError
 
 
-# @md.input("dicom_study_list", List[DICOMStudy], IOType.IN_MEMORY)
-# @md.input("selection_rules", Dict, IOType.IN_MEMORY)  # This overrides the rules in the instance.
-# @md.output("study_selected_series_list", List[StudySelectedSeries], IOType.IN_MEMORY)
 class DICOMSeriesSelectorOperator(Operator):
     """This operator selects a list of DICOM Series in a DICOM Study for a given set of selection rules.
 
+    Named input:
+        dicom_study_list: A list of DICOMStudy objects.
+
+    Named output:
+        study_selected_series_list: A list of StudySelectedSeries objects. Downstream receiver optional.
+
     This class can be considered a base class, and a derived class can override the 'filer' function to with
     custom logics.
 
@@ -67,50 +69,41 @@ class DICOMSeriesSelectorOperator(Operator):
     }
     """
 
-    def __init__(self, *args, rules_json_str: str = "", all_matched: bool = False, **kwargs) -> None:
+    def __init__(
+        self, fragment: Fragment, *args, rules_json_str: str = "", all_matched: bool = False, **kwargs
+    ) -> None:
         """Instantiate an instance.
 
         Args:
+            fragment (Fragment): An instance of the Application class which is derived from Fragment.
             rules (Text): Selection rules in JSON string.
             all_matched (bool): Gets all matched series in a study. Defaults to False for first match only.
         """
 
         # rules: Text = "", all_matched: bool = False,
 
         # Delay loading the rules as JSON string till compute time.
-        self._rules_json_str = rules_json_str  # rules if rules and rules.strip() else None
+        self._rules_json_str = rules_json_str
         self._all_matched = all_matched  # all_matched
+        self.input_name_study_list = "dicom_study_list"
+        self.output_name_selected_series = "study_selected_series_list"
 
-        super().__init__(*args, **kwargs)
+        super().__init__(fragment, *args, **kwargs)
 
     def setup(self, spec: OperatorSpec):
-        spec.input("dicom_study_list")
-        spec.output("study_selected_series_list")
+        spec.input(self.input_name_study_list)
+        spec.output(self.output_name_selected_series).condition(ConditionType.NONE)  # Receiver optional
 
         # Can use the config file to alter the selection rules per app run
         # spec.param("selection_rules")
 
     def compute(self, op_input, op_output, context):
         """Performs computation for this operator."""
 
-        dicom_study_list = None
-        selection_rules = None
-        try:
-            dicom_study_list = op_input.receive("dicom_study_list")  # op_input.get("dicom_study_list")
-        except ItemNotExistsError as ex:
-            logging.exception(f"Failed to find input 'dicom_study_list', {ex}")
-            raise
-
-        try:
-            selection_rules = self._rules_json_str  # static for now # op_input.receive("selection_rules")
-        except ItemNotExistsError:
-            # OK for not providing selection rules.
-            pass
-
-        if not selection_rules:
-            selection_rules = self._load_rules() if self._rules_json_str else None
+        dicom_study_list = op_input.receive(self.input_name_study_list)
+        selection_rules = self._load_rules() if self._rules_json_str else None
         study_selected_series = self.filter(selection_rules, dicom_study_list, self._all_matched)
-        op_output.emit(study_selected_series, "study_selected_series_list")
+        op_output.emit(study_selected_series, self.output_name_selected_series)
 
     def filter(self, selection_rules, dicom_study_list, all_matched: bool = False) -> List[StudySelectedSeries]:
         """Selects the series with the given matching rules.
@@ -304,11 +297,12 @@ def test():
     from monai.deploy.operators.dicom_data_loader_operator import DICOMDataLoaderOperator
 
     current_file_dir = Path(__file__).parent.resolve()
-    data_path = current_file_dir.joinpath("../../../inputs/spleen_ct/dcm")
+    data_path = current_file_dir.joinpath("../../../inputs/spleen_ct/dcm").absolute()
 
-    loader = DICOMDataLoaderOperator()
-    study_list = loader.load_data_to_studies(data_path.absolute())
-    selector = DICOMSeriesSelectorOperator()
+    fragment = Fragment()
+    loader = DICOMDataLoaderOperator(fragment, name="loader_op")
+    selector = DICOMSeriesSelectorOperator(fragment, name="selector_op")
+    study_list = loader.load_data_to_studies(data_path)
     sample_selection_rule = json_loads(Sample_Rules_Text)
     print(f"Selection rules in JSON:\n{sample_selection_rule}")
     study_selected_seriee_list = selector.filter(sample_selection_rule, study_list)