diff options
Diffstat (limited to 'src/aiet/cli/application.py')
| -rw-r--r-- | src/aiet/cli/application.py | 362 |
1 files changed, 362 insertions, 0 deletions
diff --git a/src/aiet/cli/application.py b/src/aiet/cli/application.py new file mode 100644 index 0000000..59b652d --- /dev/null +++ b/src/aiet/cli/application.py @@ -0,0 +1,362 @@ +# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. +# SPDX-FileCopyrightText: Copyright (c) 2021, Gianluca Gippetto. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause +"""Module to manage the CLI interface of applications.""" +import json +import logging +import re +from pathlib import Path +from typing import Any +from typing import IO +from typing import List +from typing import Optional +from typing import Tuple + +import click +import cloup + +from aiet.backend.application import get_application +from aiet.backend.application import get_available_application_directory_names +from aiet.backend.application import get_unique_application_names +from aiet.backend.application import install_application +from aiet.backend.application import remove_application +from aiet.backend.common import DataPaths +from aiet.backend.execution import execute_application_command +from aiet.backend.execution import run_application +from aiet.backend.system import get_available_systems +from aiet.cli.common import get_format +from aiet.cli.common import middleware_exception_handler +from aiet.cli.common import middleware_signal_handler +from aiet.cli.common import print_command_details +from aiet.cli.common import set_format + + +@click.group(name="application") +@click.option( + "-f", + "--format", + "format_", + type=click.Choice(["cli", "json"]), + default="cli", + show_default=True, +) +@click.pass_context +def application_cmd(ctx: click.Context, format_: str) -> None: + """Sub command to manage applications.""" + set_format(ctx, format_) + + +@application_cmd.command(name="list") +@click.pass_context +@click.option( + "-s", + "--system", + "system_name", + type=click.Choice([s.name for s in get_available_systems()]), + required=False, +) +def list_cmd(ctx: click.Context, system_name: str) -> None: + """List all available applications.""" + unique_application_names = get_unique_application_names(system_name) + unique_application_names.sort() + if get_format(ctx) == "json": + data = {"type": "application", "available": unique_application_names} + print(json.dumps(data)) + else: + print("Available applications:\n") + print(*unique_application_names, sep="\n") + + +@application_cmd.command(name="details") +@click.option( + "-n", + "--name", + "application_name", + type=click.Choice(get_unique_application_names()), + required=True, +) +@click.option( + "-s", + "--system", + "system_name", + type=click.Choice([s.name for s in get_available_systems()]), + required=False, +) +@click.pass_context +def details_cmd(ctx: click.Context, application_name: str, system_name: str) -> None: + """Details of a specific application.""" + applications = get_application(application_name, system_name) + if not applications: + raise click.UsageError( + "Application '{}' doesn't support the system '{}'".format( + application_name, system_name + ) + ) + + if get_format(ctx) == "json": + applications_details = [s.get_details() for s in applications] + print(json.dumps(applications_details)) + else: + for application in applications: + application_details = application.get_details() + application_details_template = ( + 'Application "{name}" details\nDescription: {description}' + ) + + print( + application_details_template.format( + name=application_details["name"], + description=application_details["description"], + ) + ) + + print( + "\nSupported systems: {}".format( + ", ".join(application_details["supported_systems"]) + ) + ) + + command_details = application_details["commands"] + + for command, details in command_details.items(): + print("\n{} commands:".format(command)) + print_command_details(details) + + +# pylint: disable=too-many-arguments +@application_cmd.command(name="execute") +@click.option( + "-n", + "--name", + "application_name", + type=click.Choice(get_unique_application_names()), + required=True, +) +@click.option( + "-s", + "--system", + "system_name", + type=click.Choice([s.name for s in get_available_systems()]), + required=True, +) +@click.option( + "-c", + "--command", + "command_name", + type=click.Choice(["build", "run"]), + required=True, +) +@click.option("-p", "--param", "application_params", multiple=True) +@click.option("--system-param", "system_params", multiple=True) +@click.option("-d", "--deploy", "deploy_params", multiple=True) +@middleware_signal_handler +@middleware_exception_handler +def execute_cmd( + application_name: str, + system_name: str, + command_name: str, + application_params: List[str], + system_params: List[str], + deploy_params: List[str], +) -> None: + """Execute application commands. DEPRECATED! Use 'aiet application run' instead.""" + logging.warning( + "Please use 'aiet application run' instead. Use of 'aiet application " + "execute' is deprecated and might be removed in a future release." + ) + + custom_deploy_data = get_custom_deploy_data(command_name, deploy_params) + + execute_application_command( + command_name, + application_name, + application_params, + system_name, + system_params, + custom_deploy_data, + ) + + +@cloup.command(name="run") +@cloup.option( + "-n", + "--name", + "application_name", + type=click.Choice(get_unique_application_names()), +) +@cloup.option( + "-s", + "--system", + "system_name", + type=click.Choice([s.name for s in get_available_systems()]), +) +@cloup.option("-p", "--param", "application_params", multiple=True) +@cloup.option("--system-param", "system_params", multiple=True) +@cloup.option("-d", "--deploy", "deploy_params", multiple=True) +@click.option( + "-r", + "--report", + "report_file", + type=Path, + help="Create a report file in JSON format containing metrics parsed from " + "the simulation output as specified in the aiet-config.json.", +) +@cloup.option( + "--config", + "config_file", + type=click.File("r"), + help="Read options from a config file rather than from the command line. " + "The config file is a json file.", +) +@cloup.constraint( + cloup.constraints.If( + cloup.constraints.conditions.Not( + cloup.constraints.conditions.IsSet("config_file") + ), + then=cloup.constraints.require_all, + ), + ["system_name", "application_name"], +) +@cloup.constraint( + cloup.constraints.If("config_file", then=cloup.constraints.accept_none), + [ + "system_name", + "application_name", + "application_params", + "system_params", + "deploy_params", + ], +) +@middleware_signal_handler +@middleware_exception_handler +def run_cmd( + application_name: str, + system_name: str, + application_params: List[str], + system_params: List[str], + deploy_params: List[str], + report_file: Optional[Path], + config_file: Optional[IO[str]], +) -> None: + """Execute application commands.""" + if config_file: + payload_data = json.load(config_file) + ( + system_name, + application_name, + application_params, + system_params, + deploy_params, + report_file, + ) = parse_payload_run_config(payload_data) + + custom_deploy_data = get_custom_deploy_data("run", deploy_params) + + run_application( + application_name, + application_params, + system_name, + system_params, + custom_deploy_data, + report_file, + ) + + +application_cmd.add_command(run_cmd) + + +def parse_payload_run_config( + payload_data: dict, +) -> Tuple[str, str, List[str], List[str], List[str], Optional[Path]]: + """Parse the payload into a tuple.""" + system_id = payload_data.get("id") + arguments: Optional[Any] = payload_data.get("arguments") + + if not isinstance(system_id, str): + raise click.ClickException("invalid payload json: no system 'id'") + if not isinstance(arguments, dict): + raise click.ClickException("invalid payload json: no arguments object") + + application_name = arguments.pop("application", None) + if not isinstance(application_name, str): + raise click.ClickException("invalid payload json: no application_id") + + report_path = arguments.pop("report_path", None) + + application_params = [] + system_params = [] + deploy_params = [] + + for (param_key, value) in arguments.items(): + (par, _) = re.subn("^application/", "", param_key) + (par, found_sys_param) = re.subn("^system/", "", par) + (par, found_deploy_param) = re.subn("^deploy/", "", par) + + param_expr = par + "=" + value + if found_sys_param: + system_params.append(param_expr) + elif found_deploy_param: + deploy_params.append(par) + else: + application_params.append(param_expr) + + return ( + system_id, + application_name, + application_params, + system_params, + deploy_params, + report_path, + ) + + +def get_custom_deploy_data( + command_name: str, deploy_params: List[str] +) -> List[DataPaths]: + """Get custom deploy data information.""" + custom_deploy_data: List[DataPaths] = [] + if not deploy_params: + return custom_deploy_data + + for param in deploy_params: + parts = param.split(":") + if not len(parts) == 2 or any(not part.strip() for part in parts): + raise click.ClickException( + "Invalid deploy parameter '{}' for command {}".format( + param, command_name + ) + ) + data_path = DataPaths(Path(parts[0]), parts[1]) + if not data_path.src.exists(): + raise click.ClickException("Path {} does not exist".format(data_path.src)) + custom_deploy_data.append(data_path) + + return custom_deploy_data + + +@application_cmd.command(name="install") +@click.option( + "-s", + "--source", + "source", + required=True, + help="Path to the directory or archive with application definition", +) +def install_cmd(source: str) -> None: + """Install new application.""" + source_path = Path(source) + install_application(source_path) + + +@application_cmd.command(name="remove") +@click.option( + "-d", + "--directory_name", + "directory_name", + type=click.Choice(get_available_application_directory_names()), + required=True, + help="Name of the directory with application", +) +def remove_cmd(directory_name: str) -> None: + """Remove application.""" + remove_application(directory_name) |