diff options
Diffstat (limited to 'src/aiet/cli/application.py')
| -rw-r--r-- | src/aiet/cli/application.py | 362 |
1 files changed, 0 insertions, 362 deletions
diff --git a/src/aiet/cli/application.py b/src/aiet/cli/application.py deleted file mode 100644 index 59b652d..0000000 --- a/src/aiet/cli/application.py +++ /dev/null @@ -1,362 +0,0 @@ -# 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) |