tools: cloud: application/release CLI

Add subcommands to `cloud` to display applications and releases and
upload releases from a release folder.

Signed-off-by: Jordan Yates <jordan@embeint.com>

Author: JordanYates

src/infuse_iot/tools/cloud.py

@@ -5,13 +5,22 @@
 __author__ = "Jordan Yates"
 __copyright__ = "Copyright 2024, Embeint Holdings Pty Ltd"
 
+import glob
 import sys
 from typing import Any
+from uuid import UUID
 
 from tabulate import tabulate
 
 import infuse_iot.api_client.models as models
 from infuse_iot.api_client import Client
+from infuse_iot.api_client.api.application import (
+    create_application,
+    create_release,
+    get_application_by_organisation_id_and_application_id,
+    get_applications_by_organisation_id,
+    get_releases_by_organisation_id_and_application_id,
+)
 from infuse_iot.api_client.api.board import (
     create_board,
     get_board_by_id,
@@ -30,10 +39,12 @@
     get_all_organisations,
     get_organisation_by_id,
 )
-from infuse_iot.api_client.models import COAPFilesList, Error, NewBoard, NewOrganisation
-from infuse_iot.api_client.types import Unset
+from infuse_iot.api_client.types import File, Unset
 from infuse_iot.commands import InfuseCommand
 from infuse_iot.credentials import get_api_key
+from infuse_iot.util.argparse import ValidRelease
+from infuse_iot.util.console import choose_one, user_confirm, user_response
+from infuse_iot.util.version import Version
 
 
 class CloudSubCommand:
@@ -322,6 +333,188 @@ def list(self, client: Client):
             print("\t" + "\n\t".join(sorted_list))
 
 
+class Applications(CloudSubCommand):
+    @classmethod
+    def add_parser(cls, parser):
+        parser_coap = parser.add_parser("apps", help="Application release management")
+        parser_coap.set_defaults(command_class=cls)
+
+        tool_parser = parser_coap.add_subparsers(title="commands", metavar="<command>", required=True)
+
+        list_parser = tool_parser.add_parser("list", help="List all application releases")
+        list_parser.add_argument("--org", "-o", type=str, required=True, help="Organisation ID")
+        list_parser.set_defaults(command_fn=cls.list)
+
+        info_parser = tool_parser.add_parser("info", help="Display summary of application releases")
+        info_parser.add_argument("--org", "-o", type=str, required=True, help="Organisation ID")
+        info_parser.add_argument("--app", "-a", type=lambda x: int(x, 16), required=True, help="Application ID (hex)")
+        info_parser.set_defaults(command_fn=cls.info)
+
+        upload_parser = tool_parser.add_parser("upload", help="Upload application release")
+        upload_parser.add_argument("--org", "-o", type=str, help="Organisation ID")
+        upload_parser.add_argument("--board", "-b", type=str, help="Board ID")
+        upload_parser.add_argument("--release", "-r", type=ValidRelease, required=True, help="Release to upload")
+        upload_parser.set_defaults(command_fn=cls.upload)
+
+    def run(self):
+        with self.client() as client:
+            self.args.command_fn(self, client)
+
+    def list(self, client: Client):
+        applications = get_applications_by_organisation_id.sync(client=client, id=UUID(self.args.org))
+
+        if not isinstance(applications, list):
+            print(f"Failed to retrieve application list {applications}")
+            return
+
+        app_list = []
+        for app in applications:
+            app_list.append(
+                [
+                    f"0x{app.id:08X}",
+                    app.name,
+                    app.description,
+                ]
+            )
+        print(
+            tabulate(
+                app_list,
+                headers=["ID", "Name", "Description"],
+            )
+        )
+
+    def info(self, client: Client):
+        releases = get_releases_by_organisation_id_and_application_id.sync(
+            client=client, id=UUID(self.args.org), application_id=self.args.app
+        )
+
+        if releases is None:
+            sys.exit("Failed to retrieve release list (No response)")
+        elif isinstance(releases, models.Error):
+            sys.exit(f"<{releases.code}>: {releases.message}")
+
+        release_list = []
+        for release in releases:
+            version = release.version
+            version_str = f"{version.major}.{version.minor}.{version.revision}+{version.build_num:08x}"
+            release_list.append(
+                [
+                    f"{release.board_target}",
+                    version_str,
+                    f"{release.id}",
+                    f"{release.file.len_ / 1024:.2f} kB",
+                    str(release.created_at),
+                ]
+            )
+        print(
+            tabulate(
+                release_list,
+                headers=["Board Target", "Version", "ID", "Full OTA", "Created"],
+            )
+        )
+
+    def upload(self, client: Client):
+        try:
+            self._board = UUID(self.args.board) if self.args.board else None
+        except ValueError:
+            sys.exit(f"Board ID: '{self.args.board}' is not a valid UUID")
+        try:
+            self._org = UUID(self.args.org) if self.args.org else None
+        except ValueError:
+            sys.exit(f"Organisation ID: '{self.args.org}' is not a valid UUID")
+
+        release: ValidRelease = self.args.release
+        release_app_meta = release.metadata["application"]
+        name = release_app_meta["primary"]
+        app_id = release_app_meta["id"]
+        board_target = release_app_meta["board"]
+        version = Version.from_string(release_app_meta["version"])
+
+        if self._org is None:
+            orgs = get_all_organisations.sync(client=client)
+            if isinstance(orgs, models.Error) or orgs is None:
+                sys.exit(f"Organisation query failed {orgs}")
+            options = [f"{o.name:20s} ({o.id})" for o in orgs]
+
+            idx, _val = choose_one("Organisation", options)
+            self._org = orgs[idx].id
+            self._org_name = orgs[idx].name
+        else:
+            org = get_organisation_by_id.sync(client=client, id=self._org)
+            if not isinstance(org, models.Organisation):
+                sys.exit(f"Failed to query org for ID {self._org}")
+            self._org_name = org.name
+
+        if self._board is None:
+            boards = get_boards.sync(client=client, organisation_id=self._org)
+            if isinstance(boards, models.Error) or boards is None:
+                sys.exit(f"Board query failed {boards}")
+            options = [f"{b.name:20s} ({b.id})" for b in boards]
+
+            idx, _val = choose_one("Board", options)
+            self._board = boards[idx].id
+            self._board_name = boards[idx].name
+        else:
+            board = get_board_by_id.sync(client=client, id=self._board)
+            if not isinstance(org, models.Board):
+                sys.exit(f"Failed to query board for ID {self._board}")
+            self._board_name = board.name
+
+        application = get_application_by_organisation_id_and_application_id.sync(
+            client=client, id=self._org, application_id=app_id
+        )
+
+        if application is None:
+            dialog = f"Application 0x{app_id:08x} does not exist in organisation {self.args.org}, create?"
+            if not user_confirm(dialog):
+                return
+            print(f"Creating application 0x{app_id:08x} in organisation {self.args.org}")
+            description = user_response("Application description:")
+            body = models.NewApplication(id=app_id, name=name, description=description)
+            application = create_application.sync(client=client, id=self._org, body=body)
+
+        if not isinstance(application, models.Application):
+            sys.exit(f"Unexpected internal type {type(application)}")
+
+        ota_files = glob.glob(str(release.dir / "ota-*.bin"))
+        if len(ota_files) != 1:
+            sys.exit(f"Unexpected OTA file search result {ota_files}")
+
+        dialog = (
+            f"Create release for application '0x{app_id:08x} {str(version)}'"
+            + f" in organisation '{self._org_name}' for board '{self._board_name}'?"
+        )
+        if not user_confirm(dialog):
+            return
+
+        with open(ota_files[0], "rb") as f:
+            ota_file = File(f, ota_files[0], None)
+
+            release_obj = models.CreateReleaseBody(
+                file=ota_file,
+                file_diff_len=str(0),
+                version_major=str(version.major),
+                version_minor=str(version.minor),
+                version_revision=str(version.revision),
+                version_build_num=str(version.build_num),
+                board_id=self._board,
+                board_target=board_target,
+            )
+
+            rsp = create_release.sync(
+                client=client,
+                id=self._org,
+                application_id=app_id,
+                body=release_obj,
+            )
+            if rsp is None:
+                sys.exit("Create release: No response")
+            elif isinstance(rsp, models.Error):
+                sys.exit(f"<{rsp.code}>: {rsp.message}")
+            else:
+                print(f"Release created with ID '{rsp.id}'")
+
+
 class SubCommand(InfuseCommand):
     NAME = "cloud"
     HELP = "Infuse-IoT cloud interaction"
@@ -335,6 +528,7 @@ def add_parser(cls, parser):
         Boards.add_parser(subparser)
         Device.add_parser(subparser)
         Coap.add_parser(subparser)
+        Applications.add_parser(subparser)
 
     def __init__(self, args):
         self.tool = args.command_class(args)