diff options
author | Diego Russo <diego.russo@arm.com> | 2022-05-30 13:34:14 +0100 |
---|---|---|
committer | Diego Russo <diego.russo@arm.com> | 2022-05-30 13:34:14 +0100 |
commit | 0efca3cadbad5517a59884576ddb90cfe7ac30f8 (patch) | |
tree | abed6cb6fbf3c439fc8d947f505b6a53d5daeb1e /src/aiet/cli | |
parent | 0777092695c143c3a54680b5748287d40c914c35 (diff) | |
download | mlia-0efca3cadbad5517a59884576ddb90cfe7ac30f8.tar.gz |
Add MLIA codebase0.3.0-rc.1
Add MLIA codebase including sources and tests.
Change-Id: Id41707559bd721edd114793618d12ccd188d8dbd
Diffstat (limited to 'src/aiet/cli')
-rw-r--r-- | src/aiet/cli/__init__.py | 28 | ||||
-rw-r--r-- | src/aiet/cli/application.py | 362 | ||||
-rw-r--r-- | src/aiet/cli/common.py | 173 | ||||
-rw-r--r-- | src/aiet/cli/completion.py | 72 | ||||
-rw-r--r-- | src/aiet/cli/system.py | 122 | ||||
-rw-r--r-- | src/aiet/cli/tool.py | 143 |
6 files changed, 900 insertions, 0 deletions
diff --git a/src/aiet/cli/__init__.py b/src/aiet/cli/__init__.py new file mode 100644 index 0000000..bcd17c3 --- /dev/null +++ b/src/aiet/cli/__init__.py @@ -0,0 +1,28 @@ +# 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 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) diff --git a/src/aiet/cli/common.py b/src/aiet/cli/common.py new file mode 100644 index 0000000..1d157b6 --- /dev/null +++ b/src/aiet/cli/common.py @@ -0,0 +1,173 @@ +# 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 new file mode 100644 index 0000000..71f054f --- /dev/null +++ b/src/aiet/cli/completion.py @@ -0,0 +1,72 @@ +# 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 new file mode 100644 index 0000000..f1f7637 --- /dev/null +++ b/src/aiet/cli/system.py @@ -0,0 +1,122 @@ +# 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 new file mode 100644 index 0000000..2c80821 --- /dev/null +++ b/src/aiet/cli/tool.py @@ -0,0 +1,143 @@ +# 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) |