Source code for monailabel.main

# Copyright (c) MONAI Consortium
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import json
import logging
import os
import pathlib
import platform
import shutil
import sys

import uvicorn

from monailabel import print_config
from monailabel.config import settings
from monailabel.utils.others.generic import init_log_config

logger = logging.getLogger(__name__)

[docs]class Main: def __init__(self, loglevel=logging.INFO, actions=("start_server", "apps", "datasets", "plugins")): self.actions = set([actions] if isinstance(actions, str) else actions) logging.basicConfig( level=loglevel, format="[%(asctime)s] [%(process)s] [%(threadName)s] [%(levelname)s] (%(name)s:%(lineno)d) - %(message)s", )
[docs] def args_start_server(self, parser): parser.add_argument("-a", "--app", help="App Directory") parser.add_argument("-s", "--studies", help="Studies Directory") parser.add_argument( "-v", "--verbose", default="INFO", type=str, choices=["DEBUG", "INFO", "WARNING", "ERROR"], help="Log Level" ) # --conf key1 value1 --conf key2 value2 parser.add_argument( "-c", "--conf", nargs=2, action="append", help="config for the app. Example: --conf key1 value1 --conf key2 value2", ) parser.add_argument("-i", "--host", default="", type=str, help="Server IP") parser.add_argument("-p", "--port", default=8000, type=int, help="Server Port") parser.add_argument("--uvicorn_app", default="", type=str, help="Uvicorn App (<path>:<app>)") parser.add_argument("--ssl_keyfile", default=None, type=str, help="SSL key file") parser.add_argument("--ssl_certfile", default=None, type=str, help="SSL certificate file") parser.add_argument("--ssl_keyfile_password", default=None, type=str, help="SSL key file password") parser.add_argument("--ssl_ca_certs", default=None, type=str, help="CA certificates file") parser.add_argument("--workers", default=None, type=int, help="Number of worker processes") parser.add_argument("--limit_concurrency", default=None, type=int, help="Max concurrent connections") parser.add_argument("--access_log", action="store_true", help="Enable access log") parser.add_argument("--root_path", default="", help="Application root path") parser.add_argument("--log_level", default="info", help="Log level") parser.add_argument("-l", "--log_config", default=None, type=str, help="Logging config") parser.add_argument("--dryrun", action="store_true", help="Dry run without starting server")
[docs] def args_apps(self, parser): parser.add_argument("-d", "--download", action="store_true", help="download app") parser.add_argument("-n", "--name", help="Name of the sample app to download", default=None) parser.add_argument("-o", "--output", help="Output path to save the app", default=None) parser.add_argument("--prefix", default=None)
[docs] def args_datasets(self, parser): parser.add_argument("-d", "--download", action="store_true", help="download dataset") parser.add_argument("-n", "--name", help="Name of the dataset to download", default=None) parser.add_argument("-o", "--output", help="Output path to save the dataset", default=None) parser.add_argument("--prefix", default=None)
[docs] def args_plugins(self, parser): parser.add_argument("-d", "--download", action="store_true", help="download plugin") parser.add_argument("-n", "--name", help="Name of the plugin to download", default=None) parser.add_argument("-o", "--output", help="Output path to save the plugin", default=None) parser.add_argument("--prefix", default=None)
[docs] def args_parser(self, name="monailabel"): parser = argparse.ArgumentParser(name) parser.add_argument("-v", "--version", action="store_true", help="print version") subparsers = parser.add_subparsers(help="sub-command help") if "start_server" in self.actions: parser_a = subparsers.add_parser("start_server", help="Start Application Server") self.args_start_server(parser_a) parser_a.set_defaults(action="start_server") if "apps" in self.actions: parser_b = subparsers.add_parser("apps", help="list or download sample apps") self.args_apps(parser_b) parser_b.set_defaults(action="apps") if "datasets" in self.actions: parser_c = subparsers.add_parser("datasets", help="list or download sample datasets") self.args_datasets(parser_c) parser_c.set_defaults(action="datasets") if "plugins" in self.actions: parser_d = subparsers.add_parser("plugins", help="list or download viewer plugins") self.args_plugins(parser_d) parser_d.set_defaults(action="plugins") return parser
[docs] def run(self): parser = self.args_parser() args = parser.parse_args() if args.version: print_config() exit(0) if not hasattr(args, "action"): parser.print_usage() exit(-1) if args.action == "apps": self.action_apps(args) elif args.action == "datasets": self.action_datasets(args) elif args.action == "plugins": self.action_plugins(args) else: self.action_start_server(args)
[docs] def action_apps(self, args): self._action_xyz(args, "sample-apps", "App", None, shutil.ignore_patterns("logs", "model", "__pycache__"))
[docs] def action_plugins(self, args): self._action_xyz(args, "plugins", "Plugin", None, shutil.ignore_patterns("__pycache__"))
[docs] def action_datasets(self, args): from monai.apps.datasets import DecathlonDataset from monai.apps.utils import download_and_extract resource = DecathlonDataset.resource md5 = DecathlonDataset.md5 if not print("Available Datasets are:") print("----------------------------------------------------") for k, v in resource.items(): print(f" {k:<30}: {v}") print("") else: url = resource.get( if else None if not url: print(f"Dataset ({}) NOT Exists.") available = " " + "\n ".join(resource.keys()) print(f"Available Datasets are:: \n{available}") print("----------------------------------------------------") exit(-1) dataset_dir = os.path.join(args.output, if args.output else if os.path.exists(dataset_dir): print(f"Directory already exists: {dataset_dir}") exit(-1) root_dir = os.path.dirname(os.path.realpath(dataset_dir)) os.makedirs(root_dir, exist_ok=True) tarfile_name = f"{dataset_dir}.tar" download_and_extract(resource[], tarfile_name, root_dir, md5.get( junk_files = pathlib.Path(dataset_dir).rglob("._*") for j in junk_files: os.remove(j) os.unlink(tarfile_name) print(f"{} is downloaded at: {dataset_dir}")
def _get_installed_dir(self, prefix, name): project_root_absolute = pathlib.Path(__file__).parent.parent.resolve() # add searched paths to include variant python installation options # Specify --prefix for customized download path installed_dirs = [ os.path.join(project_root_absolute, name), os.path.join(prefix, "monailabel", name) if prefix else None, os.path.join(sys.prefix, "monailabel", name), os.path.join(sys.prefix, "local", "monailabel", name), os.path.join(pathlib.Path.home(), ".local", "monailabel", name), ] for d in installed_dirs: if d and os.path.exists(d): return d raise ValueError( f"Cannot find MONAI Label installed: {name} installed directory. Add '--prefix' of installed path." ) def _action_xyz(self, args, name, title, exclude, ignore): xyz_dir = self._get_installed_dir(args.prefix, name) exclude = [exclude] if isinstance(exclude, str) else exclude xyz = os.listdir(xyz_dir) xyz = [os.path.basename(a) for a in xyz if os.path.isdir(os.path.join(xyz_dir, a))] xyz = [p for p in xyz if p not in exclude] if exclude else xyz xyz.sort() resource = {p: f"{xyz_dir}/{p}" for p in xyz} if not print(f"Available {title}s are:") print("----------------------------------------------------") for k, v in resource.items(): print(f" {k:<30}: {v}") print("") else: xyz_dir = os.path.join(xyz_dir, if not in xyz or not os.path.exists(xyz_dir): print(f"{title} {} => {xyz_dir} not exists") exit(-1) output_dir = os.path.realpath(os.path.join(args.output, if args.output else if os.path.exists(output_dir): print(f"Directory already exists: {output_dir}") exit(-1) if os.path.dirname(output_dir): os.makedirs(os.path.dirname(output_dir), exist_ok=True) shutil.copytree(xyz_dir, output_dir, ignore=ignore) print(f"{} is copied at: {output_dir}")
[docs] def action_start_server(self, args): self.start_server_validate_args(args) self.start_server_init_settings(args) log_config = init_log_config(args.log_config,, "app.log", args.verbose) if args.dryrun: return args.uvicorn_app,, port=args.port, root_path=args.root_path, log_level=args.log_level, log_config=log_config, use_colors=True, access_log=args.access_log, ssl_keyfile=args.ssl_keyfile, ssl_certfile=args.ssl_certfile, ssl_keyfile_password=args.ssl_keyfile_password, ssl_ca_certs=args.ssl_ca_certs, workers=args.workers, limit_concurrency=args.limit_concurrency, )
[docs] def start_server_validate_args(self, args): if not print("APP Directory NOT provided") exit(1) if not args.studies: print("STUDIES Path/Directory NOT provided") exit(1) if not os.path.exists( print(f"APP Directory {} NOT Found") exit(1) if ( not args.studies.startswith("http://") and not args.studies.startswith("https://") and not os.path.exists(args.studies) ): print(f"STUDIES Directory {args.studies} NOT Found; Creating an EMPTY folder/placeholder") os.makedirs(args.studies, exist_ok=True) = os.path.realpath( if not args.studies.startswith("http://") and not args.studies.startswith("https://"): args.studies = os.path.realpath(args.studies) for arg in vars(args):"USING:: {arg} = {getattr(args, arg)}") sensitive = [ "MONAI_LABEL_DICOMWEB_PASSWORD", "MONAI_ZOO_AUTH_TOKEN", "MONAI_LABEL_DATASTORE_PASSWORD", "MONAI_LABEL_DATASTORE_API_KEY", ] for k, v in settings.model_dump().items(): v = f"'{json.dumps(v)}'" if isinstance(v, list) or isinstance(v, dict) else v logger.debug(f"ENV SETTINGS:: {k} = {'*' * len(v) if k in sensitive else v}")"")
[docs] def start_server_init_settings(self, args): # namespace('conf': [['key1','value1'],['key2','value2']]) conf = {c[0]: c[1] for c in args.conf} if args.conf else {} settings.MONAI_LABEL_SERVER_PORT = args.port settings.MONAI_LABEL_APP_DIR = settings.MONAI_LABEL_STUDIES = args.studies settings.MONAI_LABEL_APP_CONF = conf dirs = ["model", "lib", "logs", "bin"] for d in dirs: d = os.path.join(, d) if not os.path.exists(d): os.makedirs(d) os.environ["PATH"] += os.pathsep + os.path.join(, "bin") os.environ["PYTHONPATH"] = os.environ.get("PYTHONPATH", "") + os.pathsep + os.path.join( if args.dryrun: export_key = "set " if any(platform.win32_ver()) else "export " with open("env.bat" if any(platform.win32_ver()) else ".env", "w") as f: for k, v in settings.model_dump().items(): v = f"'{json.dumps(v)}'" if isinstance(v, list) or isinstance(v, dict) else v e = f"{export_key}{k}={v}" f.write(e) f.write(os.linesep) logger.debug(e) py_path = [os.environ.get("PYTHONPATH", "").rstrip(os.pathsep),, os.path.join(, "lib")] py_path = [p for p in py_path if p] others = [ f"{export_key}PYTHONPATH={os.pathsep.join(py_path)}", f"{export_key}PATH={os.environ['PATH']}", ] for o in others: f.write(o) f.write(os.linesep) else: logger.debug("") logger.debug("**********************************************************") logger.debug(" ENV VARIABLES/SETTINGS ") logger.debug("**********************************************************") for k, v in settings.model_dump().items(): if isinstance(v, list) or isinstance(v, dict): v = json.dumps(v) elif v is not None: v = str(v) else: v = None logger.debug(f"{'set' if any(platform.win32_ver()) else 'export'} {k}={v}") if v is not None: os.environ[k] = v logger.debug("**********************************************************") logger.debug("")
if __name__ == "__main__": Main().run()