example files for MAP building with required fields and instructions

project/app.py

@@ -0,0 +1,126 @@
+"""
+This file contains the application class for MAP building.
+"""
+
+import monai.deploy.core as md
+import logging
+from monai.deploy.core import Application
+from monai.deploy.operators.dicom_utils import ModelInfo
+from monai.deploy.operators import (
+    DICOMEncapsulatedPDFWriterOperator,
+    DICOMSeriesSelectorOperator,
+)
+from monai.deploy.operators import DICOMDataLoaderOperator
+
+# TODO: Import project-specific operators from map.operators, eg
+
+from map.operators.example_detector import DetectorOperator
+from map.operators.example_classifier import ClassifierOperator
+from map.operators.generate_pdf import GeneratePDFOperator
+
+import os
+import json
+from pathlib import Path
+
+os.environ["root_dir"] = os.path.dirname(os.path.realpath(__file__))
+
+# These files are expected to exist and correctly filled in
+config = json.load(open(Path(__file__).resolve().parent / "map" / "app_config.json"))
+requirements_file = Path(__file__).resolve().parent / "map" / "app_requirements.txt"
+
+
+@md.resource(
+    cpu=config["resources"]["cpu"],
+    gpu=config["resources"]["gpu"],
+    memory=config["resources"]["memory"],
+)
+@md.env(pip_packages=requirements_file.as_posix())
+class myApplication(Application):
+    """Classifies the given image and returns the class name."""
+
+    def __init__(self, *args, **kwargs):
+        """Creates an application instance."""
+        self._logger = logging.getLogger("{}.{}".format(__name__, type(self).__name__))
+        super().__init__(*args, **kwargs)
+
+    def run(self, *args, **kwargs):
+        # This method calls the base class to run. Can be omitted if simply calling through.
+        self._logger.debug(f"Begin {self.run.__name__}")
+        super().run(*args, **kwargs)
+        self._logger.debug(f"End {self.run.__name__}")
+
+    def compose(self):
+        """Creates the app specific operators and chain them up in the processing DAG."""
+
+        self._logger.debug(f"Begin {self.compose.__name__}")
+
+        app_directory = os.path.dirname(os.path.realpath(__file__))
+        model_info = ModelInfo(creator="", name="", version="", uid="")
+
+        # TODO: Define your models and optionally additional configs
+        # Model weights are the output of the training run saved as .pth
+        detector_model = app_directory + "/map/operators/models/detector_version3.pth"
+        mmdet_config = app_directory + "/detector/mmdetector.py"
+        classifier_model = (
+            app_directory + "/map/operators/models/classifier_version2.pth"
+        )
+
+        # Build operators:
+        loader = DICOMDataLoaderOperator()
+        selector = DICOMSeriesSelectorOperator(
+            rules=json.dumps(config["input"]["selection_rules"]), all_matched=True
+        )
+
+        # TODO: Define project-sepcific operators, eg
+        example_detector = DetectorOperator(
+            detector_model=detector_model, mmdet_config=mmdet_config
+        )
+        example_classifier = ClassifierOperator(classifier_model=classifier_model)
+        generate_pdf = GeneratePDFOperator(detector_model=detector_model)
+
+        dicom_encapsulation = DICOMEncapsulatedPDFWriterOperator(
+            copy_tags=True, model_info=model_info
+        )
+
+        # Create flow - stringing together the operators, specifying inputs and outputs
+        # TODO: update flow as appropriate with project specific operators
+        self.add_flow(loader, selector, {"dicom_study_list": "dicom_study_list"})
+        self.add_flow(
+            selector,
+            example_detector,
+            {"study_selected_series_list": "study_selected_series_list"},
+        )
+
+        # EXAMPLE Classifier requires 2 inputs
+        self.add_flow(
+            selector,
+            example_classifier,
+            {"study_selected_series_list": "study_selected_series_list"},
+        )
+        self.add_flow(
+            example_detector, example_classifier, {"bbox_array": "bbox_array"}
+        )
+
+        # EXAMPLE PDF Generator requires 2 inputs
+        self.add_flow(
+            selector,
+            generate_pdf,
+            {"study_selected_series_list": "study_selected_series_list"},
+        )
+        self.add_flow(
+            example_classifier, generate_pdf, {"output_udict": "output_udict"}
+        )
+
+        # EXAMPLE dicom_encapsulation requires 2 inputs
+        self.add_flow(
+            selector,
+            dicom_encapsulation,
+            {"study_selected_series_list": "study_selected_series_list"},
+        )
+        self.add_flow(generate_pdf, dicom_encapsulation, {"pdf_file": "pdf_file"})
+
+        self._logger.debug(f"End {self.compose.__name__}")
+
+
+if __name__ == "__main__":
+    myApplication(do_run=True)

project/map/app_config.json

@@ -0,0 +1,27 @@
+{
+    "info": {
+        "name": "myApp name",
+        "intended use": "Image classification.", 
+        "application version": "0.0.1",
+        "detector algorithm version": "1",
+        "classifier algorithm version": "1",
+        "manufacturer": "Clinical Scientific Computing, Guy's and St Thomas' NHS Foundation Trust"
+    },
+    "input": {
+        "selection_rules": {
+            "selections": [
+                {
+                    "name": "Example conditions",
+                    "conditions": {
+                        "Modality": "CR"
+                    }
+                }
+            ]
+        }
+    },
+    "resources": {
+        "cpu": 1,
+        "memory": "4Gi",
+        "gpu": 0
+    }
+}

project/map/app_requirements.txt

@@ -0,0 +1,3 @@
+monai-deploy-app-sdk==0.5.1
+csc-mlops==0.9.18
+highdicom==0.22.0

project/map/operators/example_classifier.py

@@ -0,0 +1,117 @@
+"""This is an example operator to implement an image classifier"""
+
+import monai.deploy.core as md
+import numpy as np
+from monai.deploy.core.domain.dicom_series_selection import StudySelectedSeries
+from typing import List, Optional
+from collections import UserDict
+from monai.deploy.core import (
+    ExecutionContext,
+    InputContext,
+    IOType,
+    Operator,
+    OutputContext,
+    DataPath,
+)
+
+from monai.data import DataLoader, Dataset
+from monai.transforms import Compose
+from transforms import load_and_norm, crop_and_output
+import torch
+import datetime
+
+
+@md.input("study_selected_series_list", List[StudySelectedSeries], IOType.IN_MEMORY)
+@md.input("bbox_array", np.ndarray, IOType.DISK)
+@md.output("output_udict", UserDict, IOType.DISK)
+@md.env(pip_packages=["monai"])
+class ClassifierOperator(Operator):
+    """Classifies the given image and returns the class name."""
+
+    def __init__(self, classifier_model: Optional[str] = "", *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.classifier_model = classifier_model
+
+    def compute(
+        self,
+        op_input: InputContext,
+        op_output: OutputContext,
+        context: ExecutionContext,
+    ):
+
+        # get image from list
+        study_selected_series_list = op_input.get("study_selected_series_list")
+
+        if not study_selected_series_list or len(study_selected_series_list) < 1:
+            raise ValueError("Missing expected input 'study_selected_series_list'")
+
+        selected_series = study_selected_series_list[0].selected_series[0].series
+        dicom_list = [sop._sop.filename for sop in selected_series._sop_instances]
+
+        transforms = Compose(load_and_norm() + crop_and_output())
+
+        bbox_array = op_input.get("bbox_array").tolist()[0]
+
+        if bbox_array[-1] < 0.9:
+            # Update as appropriate to given project
+            output_dict = {
+                "result": "Indeterminate result",
+                "error_message": "Unable to identify the scaphoid bone",
+                "advice": "ScaphX is unable to provide a reliable assessment for this case. Please proceed with standard clinical evaluation, including X-ray interpretation and additional imaging as needed.",
+                "start_time": f"{begin}",
+                "end_time": f"{end}",
+                "bounding_box": bbox_array,
+            }
+        else:
+            dataset = Dataset(
+                data=[{"image": dicom_list, "label": -1, "roi": bbox_array}],
+                transform=transforms,
+            )
+
+            dataloader = DataLoader(dataset, batch_size=1, shuffle=False)
+
+            model = torch.jit.load(self.classifier_model)
+            model.eval()
+
+            begin = str(datetime.datetime.now())
+
+            with torch.inference_mode():
+                for d in dataloader:
+                    pred = model(d["image"].to("cpu"))
+                    confidence = torch.nn.functional.softmax(pred, dim=1)
+
+            result = confidence[0].detach().numpy()[0]
+
+            end = str(datetime.datetime.now())
+
+            # Update as appropriate to specific project
+            if result >= 0.9:
+                output_dict = {
+                    "result": "Scaphoid Fracture Detected",
+                    "error_message": "None",
+                    "advice": "Apply a below elbow POP backslab and refer the patient to fracture clinic as per scaphoid pathway.",
+                    "start_time": f"{begin}",
+                    "end_time": f"{end}",
+                    "bounding_box": bbox_array,
+                }
+            elif result >= 0.9:  # and result < 0.8:
+                output_dict = {
+                    "result": "Indeterminate result",
+                    "error_message": "Unable to classify image",
+                    "advice": "ScaphX is unable to provide a reliable assessment for this case. Please proceed with standard clinical evaluation, including X-ray interpretation and additional imaging as needed.",
+                    "start_time": f"{begin}",
+                    "end_time": f"{end}",
+                    "bounding_box": bbox_array,
+                }
+            else:
+                output_dict = {
+                    "result": "No Scaphoid Fracture Detected",
+                    "error_message": "None",
+                    "advice": "Provide patient with Futura splint and refer the patient as per scaphoid pathway.",
+                    "start_time": f"{begin}",
+                    "end_time": f"{end}",
+                    "bounding_box": bbox_array,
+                }
+
+        output_udict = UserDict(output_dict)
+        op_output.set(output_udict, "output_udict")

project/map/operators/example_detector.py

@@ -0,0 +1,82 @@
+"""This is an example operator to implement an image segmentator"""
+
+from typing import List, Optional
+import numpy as np
+import torch
+import monai.deploy.core as md
+from monai.deploy.core.domain.dicom_series_selection import StudySelectedSeries
+
+from monai.deploy.core import (
+    ExecutionContext,
+    InputContext,
+    IOType,
+    Operator,
+    OutputContext,
+)
+
+from monai.data import Dataset
+from monai.transforms import Compose
+from transforms import load_and_norm
+
+from torch.utils.data import DataLoader
+from transforms.Detector import Detectord
+
+
+@md.input("study_selected_series_list", List[StudySelectedSeries], IOType.IN_MEMORY)
+@md.output("bbox_array", np.ndarray, IOType.DISK)
+class DetectorOperator(Operator):
+    """Apply detector model on the selected image and return bbox coordinates."""
+
+    def __init__(
+        self,
+        detector_model: Optional[str] = "",
+        mmdet_config: Optional[str] = "",
+        *args,
+        **kwargs
+    ):
+        super().__init__(*args, **kwargs)
+        self.detector_model = detector_model
+        self.mmdet_config = mmdet_config
+
+    def compute(
+        self,
+        op_input: InputContext,
+        op_output: OutputContext,
+        context: ExecutionContext,
+    ):
+        """
+        Applies transforms to pixel data necessary for use of detector.
+        Set operator output to bbox coordinates
+        """
+
+        # get image from list
+        study_selected_series_list = op_input.get("study_selected_series_list")
+
+        if not study_selected_series_list or len(study_selected_series_list) < 1:
+            raise ValueError("Missing expected input 'study_selected_series_list'")
+
+        selected_series = study_selected_series_list[0].selected_series[0].series
+        dicom_list = [sop._sop.filename for sop in selected_series._sop_instances]
+
+        transforms = Compose(
+            load_and_norm()
+            + [
+                Detectord(
+                    keys=["image"],
+                    config_file=self.mmdet_config,
+                    checkpoint_file=self.detector_model,
+                    device=torch.device("cpu"),
+                )
+            ]
+        )
+
+        dataset = Dataset(
+            data=[{"image": dicom_list, "label": -1}],
+            transform=transforms,
+        )
+
+        dataloader = DataLoader(dataset, batch_size=1, shuffle=False)
+        for d in dataloader:
+            bbox = d["roi"].numpy()
+
+        op_output.set(bbox, "bbox_array")