Creating Application class¶
Application Class¶
The Application class is perhaps the most important class that MONAI Deploy App developers will interact with. A developer will inherit a new Application from the monai.deploy.core.Application base class. The base application class provides support for chaining up operators, as well as a mechanism to execute the application. The compose() method of this class needs to be implemented in the inherited class to instantiate Operators and connect them to form a Directed Acyclic Graph.
The following code shows an example Application (app.py
) code:
1from monai.deploy.core import Application, env, resource
2
3
4@resource(cpu=1, gpu=1, memory="2Gi")
5# pip_packages can be a string that is a path(str) to requirements.txt file or a list of packages.
6@env(pip_packages=["scikit-image >= 0.17.2"])
7class App(Application):
8 """This is a very basic application.
9
10 This showcases the MONAI Deploy application framework.
11 """
12
13 # App's name. <class name>('App') if not specified.
14 name = "my_app"
15 # App's description. <class docstring> if not specified.
16 description = "This is a reference application."
17 # App's version. <git version tag> or '0.0.0' if not specified.
18 version = "0.1.0"
19
20 def compose(self):
21 # Execute `self.add_flow()` or `self.add_operator()` methods here.
22 pass
23
24if __name__ == "__main__":
25 App(do_run=True)
Decorators¶
The resource requirements (such as cpu
, memory
, and gpu
) for the application can be specified by using @resource decorator. This information is used only when the packaged app (Docker image) is executed.
@env accepts pip_packages
parameter as a string that is a path to requirements.txt file or a list of packages to install. If pip_packages
is specified, the definition will be aggregated with the package dependency list of other operators. The aggregated requirement definitions are stored as a “requirements.txt” file and it would be installed in packaging time.
compose() method¶
In compose()
method, operators are instantiated and connected through self.add_flow().
add_flow(source_op, destination_op, io_map=None)
io_map
is a dictionary of mapping from the source operator’s label to the destination operator’s label(s) and its type is Dict[str, str|Set[str]]
.
We can skip specifying io_map
if both the number of source_op
’s outputs and the number of destination_op
’s inputs are one.
For example, if Operators named task1
and task2
has only one input and output (with the label image
), self.add_flow(task1, task2)
is same with self.add_flow(task1, task2, {"image": "image"})
or self.add_flow(task1, task2, {"image": {"image"}})
.
def compose(self):
task1 = Task1()
task2 = Task2()
self.add_flow(task1, task2)
# self.add_flow(task1, task2, {"image": "image"})
# self.add_flow(task1, task2, {"image": {"image"}})
add_operator(operator)
If an operator in the workflow graph is both a root node and a leaf node, you can execute self.add_operator() for adding the operator to the workflow graph of the application.
def compose(self):
single_op = SingleOperator()
self.add_operator(single_op)
if __name__ == “__main__”:¶
if __name__ == "__main__":
App(do_run=True)
The above lines in app.py
are needed to execute the application code by using python
interpreter.
__main__.py file¶
__main__.py file is needed for MONAI Application Packager to detect main application code (app.py
) when the application is executed with the application folder path (e.g., python app_folder/
).
1from app import App
2
3if __name__ == "__main__":
4 App(do_run=True)
Creating a Reusable Application¶
Like Operator class, an Application class can be implemented in a way that the common Application class can be reusable.
Complex compose() Example¶
⠀⠀A complex workflow¶
The above workflow can be expressed like below
def compose(self):
reader1 = Reader1()
reader2 = Reader2()
processor1 = Processor1()
processor2 = Processor2()
processor3 = Processor3()
notifier = Notifier()
writer = Writer()
self.add_flow(reader1, processor1, {"image": {"image1", "image2"},
"metadata": "metadata"})
self.add_flow(reader2, processor2, {"roi": "roi"})
self.add_flow(processor1, processor2, {"image": "image"})
self.add_flow(processor1, writer, {"image": "image"})
self.add_flow(processor2, notifier)
self.add_flow(processor2, processor3)
self.add_flow(processor3, writer, {"seg_image": "seg_image"})