Custom Operators

[junxiant]

Hello,

I have created my own operator, which does image preprocessing functions and then saves it in an output folder, like this

However, when i try to add another operator (LoadPILOperator) like this,

        # Loads list of dicom files in a given dir
        study_loader_op = DICOMDataLoaderOperator()
        
        # Select series based on rule
        series_selector_op = DICOMSeriesSelectorOperator()
        
        # Convert into 3D image array of DHW
        series_to_vol_op = DICOMSeriesToVolumeOperator()
        
        # Get RGB images
        vol_to_RGB_op = stacker_operator.RGBConverterOperator()
        
        # Load PIL
        loadpil_op = loadpil_operator.LoadPILOperator()
        
        # # Run the classifier operator
        classifier_op = ClassifierOperator()
        
        # # Outputs
        # sr_writer_op = DICOMTextSRWriterOperator(
        #     copy_tags=True, model_info=model_info, equipment_info=my_equipment, custom_tags=my_special_tags
        # )  # copy_tags=True to use Study and Patient modules of the original 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"}
        )
        self.add_flow(
            series_to_vol_op, vol_to_RGB_op, {"image": "image"}
        )
        self.add_flow(
            vol_to_RGB_op, loadpil_op, {"image_path": "image_path"}
        ) 

I received some errors:

I think this probably has something to do with the @md.input and @md.output that i am defining, but I'm not sure what to change. Does anyone have ideas?

This is part of the RGBConverterOperator, with @md.output set to the DataPath, which i would assume is the op_output?

@md.input("image", Image, IOType.IN_MEMORY)
@md.output("image_path", DataPath, IOType.DISK)
@md.env(pip_packages=["Pillow >= 8.0.0"])
class RGBConverterOperator(Operator):
    """
    This operator writes out a 3D Volumetric Image to disk in a slice by slice manner
    """

    def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):
        input_path = op_input.get("image")        
        print("op_input", input_path)
        
        output_dir = op_output.get().path
        print("Output dir", output_dir)
        output_dir.mkdir(exist_ok=True)
        
        # Get RGB first
        self.convert_and_save(input_path, output_dir)
        print("op_output", op_output)

And for LoadPILOperator:

@md.input("image_path", DataPath, IOType.DISK)
@md.output("image", Image, IOType.IN_MEMORY)
@md.env(pip_packages=["pillow"])
class LoadPILOperator(Operator):
    """Load image from the given input (DataPath) and set numpy array to the output (Image)."""

    def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):
        import numpy as np
        from PIL import Image as PILImage

        input_path_o = op_input.get().path
        print("OG Path", input_path_o)
        input_path = Path(f'{input_path_o}_rgb')
        print("RGB path", input_path)
        
        if input_path.is_dir():
            input_path = next(input_path.glob("*.*"))  # take the first file

        image = PILImage.open(input_path)
        image = image.convert("RGB")  # convert to greyscale image
        image_arr = np.asarray(image)

        output_image = Image(image_arr)  # create Image domain object with a numpy array
        op_output.set(output_image)

[MMelQin]

A quick answer is that the interior operators do not use (or the decorator prevents) DISK type I/O for performance reasons. If indeed, a operator needs to save a file on disk and the downstream operator reads it, it is better to save the file to a well-known path, e.g. a folder under the application's output folder, and then pass the path in the output as str with the downstream operator receiving and parsing it as file path.