From c9b4089b3037b5943565d76242d3016b8776f8d2 Mon Sep 17 00:00:00 2001 From: Benjamin Klimczak Date: Tue, 28 Jun 2022 10:29:35 +0100 Subject: MLIA-546 Merge AIET into MLIA Merge the deprecated AIET interface for backend execution into MLIA: - Execute backends directly (without subprocess and the aiet CLI) - Fix issues with the unit tests - Remove src/aiet and tests/aiet - Re-factor code to replace 'aiet' with 'backend' - Adapt and improve unit tests after re-factoring - Remove dependencies that are not needed anymore (click and cloup) Change-Id: I450734c6a3f705ba9afde41862b29e797e511f7c --- src/aiet/cli/__init__.py | 28 ---- src/aiet/cli/application.py | 362 -------------------------------------------- src/aiet/cli/common.py | 173 --------------------- src/aiet/cli/completion.py | 72 --------- src/aiet/cli/system.py | 122 --------------- src/aiet/cli/tool.py | 143 ----------------- 6 files changed, 900 deletions(-) delete mode 100644 src/aiet/cli/__init__.py delete mode 100644 src/aiet/cli/application.py delete mode 100644 src/aiet/cli/common.py delete mode 100644 src/aiet/cli/completion.py delete mode 100644 src/aiet/cli/system.py delete mode 100644 src/aiet/cli/tool.py (limited to 'src/aiet/cli') diff --git a/src/aiet/cli/__init__.py b/src/aiet/cli/__init__.py deleted file mode 100644 index bcd17c3..0000000 --- a/src/aiet/cli/__init__.py +++ /dev/null @@ -1,28 +0,0 @@ -# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. -# SPDX-License-Identifier: Apache-2.0 -"""Module to mange the CLI interface.""" -import click - -from aiet import __version__ -from aiet.cli.application import application_cmd -from aiet.cli.completion import completion_cmd -from aiet.cli.system import system_cmd -from aiet.cli.tool import tool_cmd -from aiet.utils.helpers import set_verbosity - - -@click.group() -@click.version_option(__version__) -@click.option( - "-v", "--verbose", default=0, count=True, callback=set_verbosity, expose_value=False -) -@click.pass_context -def cli(ctx: click.Context) -> None: # pylint: disable=unused-argument - """AIET: AI Evaluation Toolkit.""" - # Unused arguments must be present here in definition to pass click context. - - -cli.add_command(application_cmd) -cli.add_command(system_cmd) -cli.add_command(tool_cmd) -cli.add_command(completion_cmd) 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) diff --git a/src/aiet/cli/common.py b/src/aiet/cli/common.py deleted file mode 100644 index 1d157b6..0000000 --- a/src/aiet/cli/common.py +++ /dev/null @@ -1,173 +0,0 @@ -# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. -# SPDX-License-Identifier: Apache-2.0 -"""Common functions for cli module.""" -import enum -import logging -from functools import wraps -from signal import SIG_IGN -from signal import SIGINT -from signal import signal as signal_handler -from signal import SIGTERM -from typing import Any -from typing import Callable -from typing import cast -from typing import Dict - -from click import ClickException -from click import Context -from click import UsageError - -from aiet.backend.common import ConfigurationException -from aiet.backend.execution import AnotherInstanceIsRunningException -from aiet.backend.execution import ConnectionException -from aiet.backend.protocol import SSHConnectionException -from aiet.utils.proc import CommandFailedException - - -class MiddlewareExitCode(enum.IntEnum): - """Middleware exit codes.""" - - SUCCESS = 0 - # exit codes 1 and 2 are used by click - SHUTDOWN_REQUESTED = 3 - BACKEND_ERROR = 4 - CONCURRENT_ERROR = 5 - CONNECTION_ERROR = 6 - CONFIGURATION_ERROR = 7 - MODEL_OPTIMISED_ERROR = 8 - INVALID_TFLITE_FILE_ERROR = 9 - - -class CustomClickException(ClickException): - """Custom click exception.""" - - def show(self, file: Any = None) -> None: - """Override show method.""" - super().show(file) - - logging.debug("Execution failed with following exception: ", exc_info=self) - - -class MiddlewareShutdownException(CustomClickException): - """Exception indicates that user requested middleware shutdown.""" - - exit_code = int(MiddlewareExitCode.SHUTDOWN_REQUESTED) - - -class BackendException(CustomClickException): - """Exception indicates that command failed.""" - - exit_code = int(MiddlewareExitCode.BACKEND_ERROR) - - -class ConcurrentErrorException(CustomClickException): - """Exception indicates concurrent execution error.""" - - exit_code = int(MiddlewareExitCode.CONCURRENT_ERROR) - - -class BackendConnectionException(CustomClickException): - """Exception indicates that connection could not be established.""" - - exit_code = int(MiddlewareExitCode.CONNECTION_ERROR) - - -class BackendConfigurationException(CustomClickException): - """Exception indicates some configuration issue.""" - - exit_code = int(MiddlewareExitCode.CONFIGURATION_ERROR) - - -class ModelOptimisedException(CustomClickException): - """Exception indicates input file has previously been Vela optimised.""" - - exit_code = int(MiddlewareExitCode.MODEL_OPTIMISED_ERROR) - - -class InvalidTFLiteFileError(CustomClickException): - """Exception indicates input TFLite file is misformatted.""" - - exit_code = int(MiddlewareExitCode.INVALID_TFLITE_FILE_ERROR) - - -def print_command_details(command: Dict) -> None: - """Print command details including parameters.""" - command_strings = command["command_strings"] - print("Commands: {}".format(command_strings)) - user_params = command["user_params"] - for i, param in enumerate(user_params, 1): - print("User parameter #{}".format(i)) - print("\tName: {}".format(param.get("name", "-"))) - print("\tDescription: {}".format(param["description"])) - print("\tPossible values: {}".format(param.get("values", "-"))) - print("\tDefault value: {}".format(param.get("default_value", "-"))) - print("\tAlias: {}".format(param.get("alias", "-"))) - - -def raise_exception_at_signal( - signum: int, frame: Any # pylint: disable=unused-argument -) -> None: - """Handle signals.""" - # Disable both SIGINT and SIGTERM signals. Further SIGINT and SIGTERM - # signals will be ignored as we allow a graceful shutdown. - # Unused arguments must be present here in definition as used in signal handler - # callback - - signal_handler(SIGINT, SIG_IGN) - signal_handler(SIGTERM, SIG_IGN) - raise MiddlewareShutdownException("Middleware shutdown requested") - - -def middleware_exception_handler(func: Callable) -> Callable: - """Handle backend exceptions decorator.""" - - @wraps(func) - def wrapper(*args: Any, **kwargs: Any) -> Any: - try: - return func(*args, **kwargs) - except (MiddlewareShutdownException, UsageError, ClickException) as error: - # click should take care of these exceptions - raise error - except ValueError as error: - raise ClickException(str(error)) from error - except AnotherInstanceIsRunningException as error: - raise ConcurrentErrorException( - "Another instance of the system is running" - ) from error - except (SSHConnectionException, ConnectionException) as error: - raise BackendConnectionException(str(error)) from error - except ConfigurationException as error: - raise BackendConfigurationException(str(error)) from error - except (CommandFailedException, Exception) as error: - raise BackendException( - "Execution failed. Please check output for the details." - ) from error - - return wrapper - - -def middleware_signal_handler(func: Callable) -> Callable: - """Handle signals decorator.""" - - @wraps(func) - def wrapper(*args: Any, **kwargs: Any) -> Any: - # Set up signal handlers for SIGINT (ctrl-c) and SIGTERM (kill command) - # The handler ignores further signals and it raises an exception - signal_handler(SIGINT, raise_exception_at_signal) - signal_handler(SIGTERM, raise_exception_at_signal) - - return func(*args, **kwargs) - - return wrapper - - -def set_format(ctx: Context, format_: str) -> None: - """Save format in click context.""" - ctx_obj = ctx.ensure_object(dict) - ctx_obj["format"] = format_ - - -def get_format(ctx: Context) -> str: - """Get format from click context.""" - ctx_obj = cast(Dict[str, str], ctx.ensure_object(dict)) - return ctx_obj["format"] diff --git a/src/aiet/cli/completion.py b/src/aiet/cli/completion.py deleted file mode 100644 index 71f054f..0000000 --- a/src/aiet/cli/completion.py +++ /dev/null @@ -1,72 +0,0 @@ -# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. -# SPDX-License-Identifier: Apache-2.0 -""" -Add auto completion to different shells with these helpers. - -See: https://click.palletsprojects.com/en/8.0.x/shell-completion/ -""" -import click - - -def _get_package_name() -> str: - return __name__.split(".", maxsplit=1)[0] - - -# aiet completion bash -@click.group(name="completion") -def completion_cmd() -> None: - """Enable auto completion for your shell.""" - - -@completion_cmd.command(name="bash") -def bash_cmd() -> None: - """ - Enable auto completion for bash. - - Use this command to activate completion in the current bash: - - eval "`aiet completion bash`" - - Use this command to add auto completion to bash globally, if you have aiet - installed globally (requires starting a new shell afterwards): - - aiet completion bash >> ~/.bashrc - """ - package_name = _get_package_name() - print(f'eval "$(_{package_name.upper()}_COMPLETE=bash_source {package_name})"') - - -@completion_cmd.command(name="zsh") -def zsh_cmd() -> None: - """ - Enable auto completion for zsh. - - Use this command to activate completion in the current zsh: - - eval "`aiet completion zsh`" - - Use this command to add auto completion to zsh globally, if you have aiet - installed globally (requires starting a new shell afterwards): - - aiet completion zsh >> ~/.zshrc - """ - package_name = _get_package_name() - print(f'eval "$(_{package_name.upper()}_COMPLETE=zsh_source {package_name})"') - - -@completion_cmd.command(name="fish") -def fish_cmd() -> None: - """ - Enable auto completion for fish. - - Use this command to activate completion in the current fish: - - eval "`aiet completion fish`" - - Use this command to add auto completion to fish globally, if you have aiet - installed globally (requires starting a new shell afterwards): - - aiet completion fish >> ~/.config/fish/completions/aiet.fish - """ - package_name = _get_package_name() - print(f'eval "(env _{package_name.upper()}_COMPLETE=fish_source {package_name})"') diff --git a/src/aiet/cli/system.py b/src/aiet/cli/system.py deleted file mode 100644 index f1f7637..0000000 --- a/src/aiet/cli/system.py +++ /dev/null @@ -1,122 +0,0 @@ -# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. -# SPDX-License-Identifier: Apache-2.0 -"""Module to manage the CLI interface of systems.""" -import json -from pathlib import Path -from typing import cast - -import click - -from aiet.backend.application import get_available_applications -from aiet.backend.system import get_available_systems -from aiet.backend.system import get_available_systems_directory_names -from aiet.backend.system import get_system -from aiet.backend.system import install_system -from aiet.backend.system import remove_system -from aiet.backend.system import System -from aiet.cli.common import get_format -from aiet.cli.common import print_command_details -from aiet.cli.common import set_format - - -@click.group(name="system") -@click.option( - "-f", - "--format", - "format_", - type=click.Choice(["cli", "json"]), - default="cli", - show_default=True, -) -@click.pass_context -def system_cmd(ctx: click.Context, format_: str) -> None: - """Sub command to manage systems.""" - set_format(ctx, format_) - - -@system_cmd.command(name="list") -@click.pass_context -def list_cmd(ctx: click.Context) -> None: - """List all available systems.""" - available_systems = get_available_systems() - system_names = [system.name for system in available_systems] - if get_format(ctx) == "json": - data = {"type": "system", "available": system_names} - print(json.dumps(data)) - else: - print("Available systems:\n") - print(*system_names, sep="\n") - - -@system_cmd.command(name="details") -@click.option( - "-n", - "--name", - "system_name", - type=click.Choice([s.name for s in get_available_systems()]), - required=True, -) -@click.pass_context -def details_cmd(ctx: click.Context, system_name: str) -> None: - """Details of a specific system.""" - system = cast(System, get_system(system_name)) - applications = [ - s.name for s in get_available_applications() if s.can_run_on(system.name) - ] - system_details = system.get_details() - if get_format(ctx) == "json": - system_details["available_application"] = applications - print(json.dumps(system_details)) - else: - system_details_template = ( - 'System "{name}" details\n' - "Description: {description}\n" - "Data Transfer Protocol: {protocol}\n" - "Available Applications: {available_application}" - ) - print( - system_details_template.format( - name=system_details["name"], - description=system_details["description"], - protocol=system_details["data_transfer_protocol"], - available_application=", ".join(applications), - ) - ) - - if system_details["annotations"]: - print("Annotations:") - for ann_name, ann_value in system_details["annotations"].items(): - print("\t{}: {}".format(ann_name, ann_value)) - - command_details = system_details["commands"] - for command, details in command_details.items(): - print("\n{} commands:".format(command)) - print_command_details(details) - - -@system_cmd.command(name="install") -@click.option( - "-s", - "--source", - "source", - required=True, - help="Path to the directory or archive with system definition", -) -def install_cmd(source: str) -> None: - """Install new system.""" - source_path = Path(source) - install_system(source_path) - - -@system_cmd.command(name="remove") -@click.option( - "-d", - "--directory_name", - "directory_name", - type=click.Choice(get_available_systems_directory_names()), - required=True, - help="Name of the directory with system", -) -def remove_cmd(directory_name: str) -> None: - """Remove system by given name.""" - remove_system(directory_name) diff --git a/src/aiet/cli/tool.py b/src/aiet/cli/tool.py deleted file mode 100644 index 2c80821..0000000 --- a/src/aiet/cli/tool.py +++ /dev/null @@ -1,143 +0,0 @@ -# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates. -# SPDX-License-Identifier: Apache-2.0 -"""Module to manage the CLI interface of tools.""" -import json -from typing import Any -from typing import List -from typing import Optional - -import click - -from aiet.backend.execution import execute_tool_command -from aiet.backend.tool import get_tool -from aiet.backend.tool import get_unique_tool_names -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="tool") -@click.option( - "-f", - "--format", - "format_", - type=click.Choice(["cli", "json"]), - default="cli", - show_default=True, -) -@click.pass_context -def tool_cmd(ctx: click.Context, format_: str) -> None: - """Sub command to manage tools.""" - set_format(ctx, format_) - - -@tool_cmd.command(name="list") -@click.pass_context -def list_cmd(ctx: click.Context) -> None: - """List all available tools.""" - # raise NotImplementedError("TODO") - tool_names = get_unique_tool_names() - tool_names.sort() - if get_format(ctx) == "json": - data = {"type": "tool", "available": tool_names} - print(json.dumps(data)) - else: - print("Available tools:\n") - print(*tool_names, sep="\n") - - -def validate_system( - ctx: click.Context, - _: click.Parameter, # param is not used - value: Any, -) -> Any: - """Validate provided system name depending on the the tool name.""" - tool_name = ctx.params["tool_name"] - tools = get_tool(tool_name, value) - if not tools: - supported_systems = [tool.supported_systems[0] for tool in get_tool(tool_name)] - raise click.BadParameter( - message="'{}' is not one of {}.".format( - value, - ", ".join("'{}'".format(system) for system in supported_systems), - ), - ctx=ctx, - ) - return value - - -@tool_cmd.command(name="details") -@click.option( - "-n", - "--name", - "tool_name", - type=click.Choice(get_unique_tool_names()), - required=True, -) -@click.option( - "-s", - "--system", - "system_name", - callback=validate_system, - required=False, -) -@click.pass_context -@middleware_signal_handler -@middleware_exception_handler -def details_cmd(ctx: click.Context, tool_name: str, system_name: Optional[str]) -> None: - """Details of a specific tool.""" - tools = get_tool(tool_name, system_name) - if get_format(ctx) == "json": - tools_details = [s.get_details() for s in tools] - print(json.dumps(tools_details)) - else: - for tool in tools: - tool_details = tool.get_details() - tool_details_template = 'Tool "{name}" details\nDescription: {description}' - - print( - tool_details_template.format( - name=tool_details["name"], - description=tool_details["description"], - ) - ) - - print( - "\nSupported systems: {}".format( - ", ".join(tool_details["supported_systems"]) - ) - ) - - command_details = tool_details["commands"] - - for command, details in command_details.items(): - print("\n{} commands:".format(command)) - print_command_details(details) - - -# pylint: disable=too-many-arguments -@tool_cmd.command(name="execute") -@click.option( - "-n", - "--name", - "tool_name", - type=click.Choice(get_unique_tool_names()), - required=True, -) -@click.option("-p", "--param", "tool_params", multiple=True) -@click.option( - "-s", - "--system", - "system_name", - callback=validate_system, - required=False, -) -@middleware_signal_handler -@middleware_exception_handler -def execute_cmd( - tool_name: str, tool_params: List[str], system_name: Optional[str] -) -> None: - """Execute tool commands.""" - execute_tool_command(tool_name, tool_params, system_name) -- cgit v1.2.1