aboutsummaryrefslogtreecommitdiff
path: root/src/aiet/cli/system.py
blob: f1f76376f54c867c771d5c10e59d62942b29c18d (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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)