Executing app locally

Executing and Inspecting App in Jupyter Notebook

You can define Operators and Application in Jupyter Notebook.

from monai.deploy.conditions import CountCondition
from monai.deploy.core import (Application, ConditionType, ExecutionContext, InputContext,
                               Operator, OperatorSpec, OutputContext)

class TaskA(Operator):
    def setup(self, spec: OperatorSpec):
        spec.input("input_path").condition(ConditionType.NONE)  # optional input
        spec.output("A")

    def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):
        input_path = op_input.receive("input_path")  # Get input path via named input. Not used here.
        if not input_path:
            print("TaskA receives None at optional input.")
        data = 1
        op_output.emit(data, "A")
        print(f"TaskA emits {data}")

class TaskB(Operator):
    def setup(self, spec: OperatorSpec):
        spec.input("input")
        spec.output("B").condition(ConditionType.NONE)  # optional output, not requiring receiver

    def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):
        input_data = op_input.receive("input")
        print(f"TaskB receives {input_data}")

        output_data = input_data + 1
        op_output.emit(output_data, "B")
        print(f"TaskB emits {output_data}")

class App(Application):
    def compose(self):
        taskA = TaskA(self, CountCondition(self, 1), name="Task A")  # self and name are required
        taskB = TaskB(self, name="Task B")
        self.add_flow(taskA, taskB) # Need not explicitly connect single output and input

if __name__ == "__main__":
    App().run()

Once an Application class (App) is defined, you can instantiate the application and execute with Application.run() method.

Since the above example doesn’t use input or output paths, we need not them, otherwise, environment variables can be used to provide the paths, as in this example.

app = App()
app.run()

app.argv

You can access Graph object through Application.graph

print(app.graph)
graph = app.graph
print(f"graph.context: {graph.context}")
operators = graph.get_nodes()
print(f"get_nodes.get_nodes(): {operators}")
print(f"graph.is_root(operators[0]): {graph.is_root(operators[0])}")
print(f"graph.is_leaf(operators[1]): {graph.is_leaf(operators[1])}")

Executing Application

Once the application is verified inside Jupyter notebook, we can write the whole application as a file(app.py) with the following lines at the end of the file:

if __name__ == "__main__":
    App(do_run=True)

Above lines are needed to execute the application code by using python interpreter.

%%writefile app.py
from monai.deploy.conditions import CountCondition
from monai.deploy.core import (Application, ConditionType, ExecutionContext, InputContext,
                               Operator, OperatorSpec, OutputContext)

class TaskA(Operator):
    def setup(self, spec: OperatorSpec):
        spec.input("input_path").condition(ConditionType.NONE)  # optional input
        spec.output("A")

    def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):
        input_path = op_input.receive("input_path")  # Get input path via named input. Not used here.
        if not input_path:
            print("TaskA receives None at optional input.")
        data = 1
        op_output.emit(data, "A")
        print(f"TaskA emits {data}")

class TaskB(Operator):
    def setup(self, spec: OperatorSpec):
        spec.input("input")
        spec.output("B").condition(ConditionType.NONE)  # optional output, not requiring receiver

    def compute(self, op_input: InputContext, op_output: OutputContext, context: ExecutionContext):
        input_data = op_input.receive("input")
        print(f"TaskB receives {input_data}")

        output_data = input_data + 1
        op_output.emit(output_data, "B")
        print(f"TaskB emits {output_data}")

class App(Application):
    def compose(self):
        taskA = TaskA(self, CountCondition(self, 1), name="Task A")  # self and name are required
        taskB = TaskB(self, name="Task B")
        self.add_flow(taskA, taskB) # Need not explicitly connect single output and input

if __name__ == "__main__":
    App().run()

Let’s run the application

!python app.py

Creating Application class Packaging app