feat(network): updates spinner and refactors existing commands

riocli/network/__init__.py

@@ -1,4 +1,4 @@
-# Copyright 2021 Rapyuta Robotics
+# Copyright 2023 Rapyuta Robotics
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,6 +14,7 @@
 import click
 from click_help_colors import HelpColorsGroup
 
+from riocli.constants import Colors
 from riocli.network.create import create_network
 from riocli.network.delete import delete_network
 from riocli.network.inspect import inspect_network
@@ -24,8 +25,8 @@
 @click.group(
     invoke_without_command=False,
     cls=HelpColorsGroup,
-    help_headers_color='yellow',
-    help_options_color='green',
+    help_headers_color=Colors.YELLOW,
+    help_options_color=Colors.GREEN,
 )
 def network() -> None:
     """

riocli/network/create.py

@@ -25,14 +25,19 @@
 @click.option('--network', help='Type of Network',
               type=click.Choice(['routed', 'native']), default='routed')
 @click.option('--ros', help='Version of ROS',
-              type=click.Choice(['kinetic', 'melodic', 'noetic']), default='melodic')
-@click.option('--device', 'device_name', help='Device ID of the Device where Network will run (device only)')
+              type=click.Choice(['kinetic', 'melodic', 'noetic']),
+              default='melodic')
+@click.option('--device', 'device_name',
+              help='Device ID of the Device where Network will run (device only)')
 @click.option('--cpu', help='cpu limit for Network (cloud only) ', type=float)
-@click.option('--memory', help='memory limit for Network (cloud only) ', type=int)
+@click.option('--memory', help='memory limit for Network (cloud only) ',
+              type=int)
 @click.option('--network-interface', '-nic', type=str,
               help='Network Interface on which Network will listen (device only)')
-@click.option('--restart-policy', help='Restart policy for the Network (device only)',
-              type=click.Choice(['always', 'no', 'on-failure']), default='always')
+@click.option('--restart-policy',
+              help='Restart policy for the Network (device only)',
+              type=click.Choice(['always', 'no', 'on-failure']),
+              default='always')
 @device_name_to_guid
 def create_network(name: str, network: str, **kwargs: typing.Any) -> None:
     """

riocli/network/delete.py

@@ -1,4 +1,4 @@
-# Copyright 2021 Rapyuta Robotics
+# Copyright 2023 Rapyuta Robotics
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,37 +12,62 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import click
-from click_spinner import spinner
+from click_help_colors import HelpColorsCommand
+from yaspin.api import Yaspin
 
 from riocli.config import new_client
+from riocli.constants import Colors, Symbols
 from riocli.network.util import name_to_guid
+from riocli.utils.spinner import with_spinner
 
 
-@click.command('delete')
-@click.option('--force', '-f', is_flag=True, default=False, help='Skip confirmation', type=bool)
-@click.option('--network', 'network_type', help='Type of Network', default=None,
+@click.command(
+    'delete',
+    cls=HelpColorsCommand,
+    help_headers_color=Colors.YELLOW,
+    help_options_color=Colors.GREEN,
+)
+@click.option('--force', '-f', is_flag=True, default=False,
+              help='Skip confirmation', type=bool)
+@click.option('--network', 'network_type', help='Type of Network',
+              default=None,
               type=click.Choice(['routed', 'native']))
 @click.argument('network-name', type=str)
 @name_to_guid
-def delete_network(force: bool, network_name: str, network_guid: str, network_type: str) -> None:
+@with_spinner(text='Deleting network...')
+def delete_network(
+        force: bool,
+        network_name: str,
+        network_guid: str,
+        network_type: str,
+        spinner: Yaspin = None
+) -> None:
     """
-    Delete the network from the Platform
+    Deletes a network
     """
-
     if not force:
-        click.confirm('Deleting {} network {} ({})'.
-                      format(network_type, network_name, network_guid), abort=True)
+        with spinner.hidden():
+            click.confirm(
+                'Deleting {} network {} ({})'.
+                format(network_type, network_name, network_guid),
+                abort=True)
 
     try:
         client = new_client()
-        with spinner():
-            if network_type == 'routed':
-                # TODO: Implement and use the delete_routed_network of client directly.
-                rn = client.get_routed_network(network_guid)
-                rn.delete()
-            elif network_type == 'native':
-                client.delete_native_network(network_guid)
-        click.secho('{} Network deleted successfully!'.format(network_type.capitalize()), fg='green')
+
+        if network_type == 'routed':
+            client.delete_routed_network(network_guid)
+        elif network_type == 'native':
+            client.delete_native_network(network_guid)
+        else:
+            raise Exception('invalid network type')
+
+        spinner.text = click.style(
+            '{} deleted successfully!'.format(network_type.capitalize()),
+            fg=Colors.GREEN)
+        spinner.green.ok(Symbols.SUCCESS)
     except Exception as e:
-        click.secho(str(e), fg='red')
-        raise SystemExit(1)
+        spinner.text = click.style('Failed to delete network: {}'.format(e),
+                                   fg=Colors.RED)
+        spinner.red.fail(Symbols.ERROR)
+        raise SystemExit(1) from e

riocli/network/inspect.py

@@ -1,4 +1,4 @@
-# Copyright 2021 Rapyuta Robotics
+# Copyright 2023 Rapyuta Robotics
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,19 +12,28 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import click
+from click_help_colors import HelpColorsCommand
 
+from riocli.constants import Colors
 from riocli.network.native_network import inspect_native_network
 from riocli.network.routed_network import inspect_routed_network
 from riocli.network.util import name_to_guid
 from riocli.utils import inspect_with_format
 
 
-@click.command('inspect')
+@click.command(
+    'inspect',
+    cls=HelpColorsCommand,
+    help_headers_color=Colors.YELLOW,
+    help_options_color=Colors.GREEN,
+)
 @click.option('--format', '-f', 'format_type',
-              type=click.Choice(['json', 'yaml'], case_sensitive=False), default='yaml')
+              type=click.Choice(['json', 'yaml'], case_sensitive=False),
+              default='yaml')
 @click.argument('network-name', type=str)
 @name_to_guid
-def inspect_network(format_type: str, network_name: str, network_guid: str, network_type: str) -> None:
+def inspect_network(format_type: str, network_name: str, network_guid: str,
+                    network_type: str) -> None:
     """
     Inspect the network resource
     """
@@ -36,5 +45,5 @@ def inspect_network(format_type: str, network_name: str, network_guid: str, netw
 
         inspect_with_format(data, format_type)
     except Exception as e:
-        click.secho(str(e), fg='red')
-        raise SystemExit(1)
+        click.secho(str(e), fg=Colors.RED)
+        raise SystemExit(1) from e

riocli/network/list.py

@@ -1,4 +1,4 @@
-# Copyright 2021 Rapyuta Robotics
+# Copyright 2023 Rapyuta Robotics
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,15 +14,22 @@
 import typing
 
 import click
+from click_help_colors import HelpColorsCommand
 from rapyuta_io import DeploymentPhaseConstants
 from rapyuta_io.clients.native_network import NativeNetwork
 from rapyuta_io.clients.routed_network import RoutedNetwork
 
 from riocli.config import new_client
+from riocli.constants import Colors
 from riocli.utils import tabulate_data
 
 
-@click.command('list')
+@click.command(
+    'list',
+    cls=HelpColorsCommand,
+    help_headers_color=Colors.YELLOW,
+    help_options_color=Colors.GREEN,
+)
 @click.option('--network', help='Type of Network',
               type=click.Choice(['routed', 'native', 'both']), default='both')
 def list_networks(network: str) -> None:
@@ -31,6 +38,7 @@ def list_networks(network: str) -> None:
     """
     try:
         client = new_client()
+
         networks = []
         if network in ['routed', 'both']:
             networks += client.get_all_routed_networks()
@@ -39,6 +47,7 @@ def list_networks(network: str) -> None:
             networks += client.list_native_networks()
 
         networks = sorted(networks, key=lambda n: n.name.lower())
+
         _display_network_list(networks, show_header=True)
     except Exception as e:
         click.secho(str(e), fg='red')
@@ -66,6 +75,7 @@ def _display_network_list(
 
         if phase and phase == DeploymentPhaseConstants.DEPLOYMENT_STOPPED.value:
             continue
-        data.append([network.guid, network.name, network.runtime, network_type, phase])
+        data.append(
+            [network.guid, network.name, network.runtime, network_type, phase])
 
     tabulate_data(data, headers)

riocli/network/model.py

@@ -1,4 +1,4 @@
-# Copyright 2022 Rapyuta Robotics
+# Copyright 2023 Rapyuta Robotics
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -27,21 +27,21 @@
 
 
 class Network(Model):
-
     def __init__(self, *args, **kwargs):
         self.update(*args, **kwargs)
 
     def find_object(self, client: Client) -> bool:
         try:
-            network, _ = find_network_name(client, self.metadata.name, self.spec.type, is_resolve_conflict=False)
+            network, _ = find_network_name(client, self.metadata.name,
+                                           self.spec.type,
+                                           is_resolve_conflict=False)
             return network
         except NetworkNotFound:
             return False
 
     def create_object(self, client: Client, **kwargs) -> Union[NativeNetwork, RoutedNetwork]:
         retry_count = int(kwargs.get('retry_count'))
         retry_interval = int(kwargs.get('retry_interval'))
-
         if self.spec.type == 'routed':
             network = self._create_routed_network(client)
             network.poll_routed_network_till_ready(retry_count=retry_count, sleep_interval=retry_interval)
@@ -51,13 +51,8 @@ def create_object(self, client: Client, **kwargs) -> Union[NativeNetwork, Routed
         network.poll_native_network_till_ready(retry_count=retry_count, sleep_interval=retry_interval)
         return network
 
-    def update_object(self, client: Client, obj: Union[RoutedNetwork, NativeNetwork]) -> Any:
-        # try:
-        #     obj.delete()
-        #     self.create_object(client)
-        # except Exception as e:
-        #     click.secho(str(e), fg='red')
-        #     raise SystemExit(1)
+    def update_object(self, client: Client,
+                      obj: Union[RoutedNetwork, NativeNetwork]) -> Any:
         pass
 
     def delete_object(self, client: Client, obj: typing.Any) -> typing.Any:
@@ -67,20 +62,26 @@ def delete_object(self, client: Client, obj: typing.Any) -> typing.Any:
     def pre_process(cls, client: Client, d: Dict) -> None:
         pass
 
-
     def to_v1(self, client: Client) -> NativeNetwork:
         if self.spec.runtime == 'cloud':
             limits = self._get_limits()
             parameters = NativeNetworkParameters(limits=limits)
         else:
             device = client.get_device(self.spec.deviceGUID)
-            parameters = NativeNetworkParameters(device=device,
-                                                 network_interface=self.spec.networkInterface)
+            parameters = NativeNetworkParameters(
+                device=device,
+                network_interface=self.spec.networkInterface)
 
-        return NativeNetwork(self.metadata.name, self.spec.runtime.lower(), self.spec.rosDistro, parameters=parameters)
+        return NativeNetwork(
+            self.metadata.name,
+            self.spec.runtime.lower(),
+            self.spec.rosDistro,
+            parameters=parameters
+        )
 
     def _get_limits(self):
-        return Limits(self.spec.resourceLimits['cpu'], self.spec.resourceLimits['memory'])
+        return Limits(self.spec.resourceLimits['cpu'],
+                      self.spec.resourceLimits['memory'])
 
     def _create_routed_network(self, client: Client) -> RoutedNetwork:
         if self.spec.runtime == 'cloud':
@@ -93,13 +94,19 @@ def _create_routed_network(self, client: Client) -> RoutedNetwork:
     def _create_cloud_routed_network(self, client: Client) -> RoutedNetwork:
         limits = self._get_limits()
         parameters = RoutedNetworkParameters(limits)
-        return client.create_cloud_routed_network(self.metadata.name, self.spec.rosDistro, True, parameters=parameters)
+        return client.create_cloud_routed_network(self.metadata.name,
+                                                  self.spec.rosDistro, True,
+                                                  parameters=parameters)
 
     def _create_device_routed_network(self, client: Client) -> RoutedNetwork:
         device = client.get_device(self.spec.deviceGUID)
-        return client.create_device_routed_network(name=self.metadata.name, ros_distro=self.spec.rosDistro, shared=True,
-                                                   device=device,
-                                                   network_interface=self.spec.networkInterface)
+        return client.create_device_routed_network(
+            name=self.metadata.name,
+            ros_distro=self.spec.rosDistro,
+            shared=True,
+            device=device,
+            network_interface=self.spec.networkInterface,
+        )
 
     @staticmethod
     def validate(data):

riocli/network/native_network.py

@@ -14,15 +14,18 @@
 import typing
 
 import click
-from click_spinner import spinner
+from rapyuta_io.clients.common_models import Limits
 from rapyuta_io.clients.native_network import NativeNetwork, Parameters
 from rapyuta_io.clients.package import Runtime, ROSDistro
-from rapyuta_io.clients.common_models import Limits
+
 from riocli.config import new_client
 
 
-def create_native_network(name: str, ros: str, device_guid: str = None, network_interface: str = None,
-                          cpu: float = 0, memory: int = 0, restart_policy: str = None, **kwargs: typing.Any) -> None:
+def create_native_network(name: str, ros: str, device_guid: str = None,
+                          network_interface: str = None,
+                          cpu: float = 0, memory: int = 0,
+                          restart_policy: str = None,
+                          **kwargs: typing.Any) -> None:
     client = new_client()
 
     ros_distro = ROSDistro(ros)
@@ -31,20 +34,22 @@ def create_native_network(name: str, ros: str, device_guid: str = None, network_
     limit = None
     if cpu or memory:
         if device_guid:
-            raise Exception('Native network for device does not support cpu or memory')
+            raise Exception(
+                'Native network for device does not support cpu or memory')
         limit = Limits(cpu, memory)
 
     device = None
     if device_guid:
         runtime = Runtime.DEVICE
         device = client.get_device(device_id=device_guid)
 
-    parameters = Parameters(limits=limit, device=device, network_interface=network_interface,
+    parameters = Parameters(limits=limit, device=device,
+                            network_interface=network_interface,
                             restart_policy=restart_policy)
-    with spinner():
-        client.create_native_network(NativeNetwork(name, runtime=runtime,
-                                                   ros_distro=ros_distro,
-                                                   parameters=parameters))
+
+    client.create_native_network(NativeNetwork(name, runtime=runtime,
+                                               ros_distro=ros_distro,
+                                               parameters=parameters))
 
     click.secho('Native Network created successfully!', fg='green')