aboutsummaryrefslogtreecommitdiff
path: root/src/aiet/cli
diff options
context:
space:
mode:
Diffstat (limited to 'src/aiet/cli')
-rw-r--r--src/aiet/cli/__init__.py28
-rw-r--r--src/aiet/cli/application.py362
-rw-r--r--src/aiet/cli/common.py173
-rw-r--r--src/aiet/cli/completion.py72
-rw-r--r--src/aiet/cli/system.py122
-rw-r--r--src/aiet/cli/tool.py143
6 files changed, 0 insertions, 900 deletions
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)