list/linfo with Rich + Remove terminaltables and Curses (#269)

pyproject.toml

@@ -33,7 +33,6 @@ dependencies = [
     "docker>=7.0.0",
     "kubernetes>=23.3.0",
     "requests>=2.22.0",
-    "terminaltables>=3.1.0",
     "slug>=2.0",
     "deepdiff==6.2.2",
     "pyroute2",

scripts/autocompletion/requirements.txt

@@ -1,6 +1,5 @@
 binaryornot>=0.4.4;
 requests>=2.22.0;
-terminaltables>=3.1.0;
 slug>=2.0;
 deepdiff>=4.0.9;
 kubernetes>=23.3.0;

setup.py

@@ -20,7 +20,6 @@
         "docker>=7.0.0",
         "kubernetes>=23.3.0",
         "requests>=2.22.0",
-        "terminaltables>=3.1.0",
         "slug>=2.0",
         "deepdiff==6.2.2",
         "pyroute2",

src/Kathara/cli/command/LinfoCommand.py

@@ -1,6 +1,8 @@
 import argparse
 from typing import List
 
+from rich.live import Live
+
 from ..ui.utils import create_panel
 from ..ui.utils import create_table
 from ... import utils
@@ -10,7 +12,6 @@
 from ...model.Link import BRIDGE_LINK_NAME
 from ...parser.netkit.LabParser import LabParser
 from ...strings import strings, wiki_description
-from ...trdparty.curses.curses import Curses
 
 
 class LinfoCommand(Command):
@@ -84,47 +85,38 @@ def run(self, current_path: str, argv: List[str]) -> None:
             return
 
         if args['name']:
-            self.console.print(
-                create_panel(
-                    str(next(Kathara.get_instance().get_machine_stats(args['name'], lab.hash))),
-                    title=f"{args['name']} Information"
-                )
-            )
+            machine_stats = next(Kathara.get_instance().get_machine_stats(args['name'], lab.hash))
+            message = str(machine_stats) if machine_stats else f"Device `{args['name']}` Not Found."
+            style = None if machine_stats else "red bold"
+
+            self.console.print(create_panel(message, title=f"{args['name']} Information", style=style))
         else:
             machines_stats = Kathara.get_instance().get_machines_stats(lab.hash)
-            print(next(create_table(machines_stats)))
+            self.console.print(create_table(machines_stats))
 
     @staticmethod
     def _get_machine_live_info(lab: Lab, machine_name: str) -> None:
-        # TODO: Replace Curses with rich Live
-        Curses.get_instance().init_window()
-
-        try:
+        with Live(None, refresh_per_second=1, screen=True) as live:
             while True:
-                Curses.get_instance().print_string(
-                    create_panel("Device Information") + "\n" +
-                    str(next(Kathara.get_instance().get_machine_stats(machine_name, lab.hash))) + "\n"
-                )
-        finally:
-            Curses.get_instance().close()
+                machine_stats = next(Kathara.get_instance().get_machine_stats(machine_name, lab.hash))
+                message = str(machine_stats) if machine_stats else f"Device `{machine_name}` Not Found."
+                style = None if machine_stats else "red bold"
+
+                live.update(create_panel(message, title=f"{machine_name} Information", style=style))
 
     @staticmethod
     def _get_lab_live_info(lab: Lab) -> None:
         machines_stats = Kathara.get_instance().get_machines_stats(lab.hash)
-        table = create_table(machines_stats)
-
-        Curses.get_instance().init_window()
 
-        try:
+        with Live(None, refresh_per_second=1, screen=True) as live:
             while True:
-                Curses.get_instance().print_string(next(table))
-        except StopIteration:
-            pass
-        finally:
-            Curses.get_instance().close()
+                table = create_table(machines_stats)
+                if not table:
+                    break
 
-    @staticmethod
-    def _get_conf_info(lab: Lab, machine_name: str = None) -> None:
+                live.update(table)
+
+    def _get_conf_info(self, lab: Lab, machine_name: str = None) -> None:
         if machine_name:
             self.console.print(
                 create_panel(

src/Kathara/cli/command/ListCommand.py

@@ -1,13 +1,14 @@
 import argparse
 from typing import List, Optional
 
+from rich.live import Live
+
 from ..ui.utils import create_table
 from ... import utils
 from ...exceptions import PrivilegeError
 from ...foundation.cli.command.Command import Command
 from ...manager.Kathara import Kathara
 from ...strings import strings, wiki_description
-from ...trdparty.curses.curses import Curses
 
 
 class ListCommand(Command):
@@ -62,19 +63,16 @@ def run(self, current_path: str, argv: List[str]) -> None:
             self._get_live_info(machine_name=args['name'], all_users=all_users)
         else:
             machines_stats = Kathara.get_instance().get_machines_stats(machine_name=args['name'], all_users=all_users)
-            print(next(create_table(machines_stats)))
+            self.console.print(create_table(machines_stats))
 
     @staticmethod
     def _get_live_info(machine_name: Optional[str], all_users: bool) -> None:
         machines_stats = Kathara.get_instance().get_machines_stats(machine_name=machine_name, all_users=all_users)
-        table = create_table(machines_stats)
-
-        Curses.get_instance().init_window()
 
-        try:
+        with Live(None, refresh_per_second=1, screen=True) as live:
             while True:
-                Curses.get_instance().print_string(next(table))
-        except StopIteration:
-            pass
-        finally:
-            Curses.get_instance().close()
+                table = create_table(machines_stats)
+                if not table:
+                    break
+
+                live.update(table)

src/Kathara/cli/ui/utils.py

@@ -4,14 +4,15 @@
 import subprocess
 import sys
 from datetime import datetime
-from typing import Any, Dict, Generator
+from typing import Any, Dict, Generator, Optional
 from typing import Callable
 
 from rich import box
+from rich.console import RenderableType, Group
 from rich.panel import Panel
 from rich.prompt import Confirm
+from rich.table import Table
 from rich.text import Text
-from terminaltables import DoubleTable
 
 from ... import utils
 from ...foundation.manager.stats.IMachineStats import IMachineStats
@@ -37,35 +38,37 @@ def create_panel(message: str = "", **kwargs) -> Panel:
             justify=kwargs['justify'] if 'justify' in kwargs else None,
         ),
         title=kwargs['title'] if 'title' in kwargs else None,
-        title_align="center", box=box.SQUARE
+        title_align="center",
+        box=kwargs['box'] if 'box' in kwargs else box.SQUARE,
     )
 
 
-def create_table(streams: Generator[Dict[str, IMachineStats], None, None]) -> \
-        Generator[str, None, None]:
-    table = DoubleTable([])
-    table.inner_row_border = True
+def create_table(streams: Generator[Dict[str, IMachineStats], None, None]) -> Optional[RenderableType]:
+    try:
+        result = next(streams)
+    except StopIteration:
+        return None
 
-    while True:
-        try:
-            result = next(streams)
-        except StopIteration:
-            return
+    ts_header = f"TIMESTAMP: {datetime.now()}"
+    if not result:
+        return Group(
+            Text(ts_header, style="italic", justify="center"),
+            create_panel("No Devices Found", style="red bold", justify="center", box=box.DOUBLE)
+        )
 
-        if not result:
-            return
+    table = Table(title=ts_header, show_lines=True, expand=True, box=box.SQUARE_DOUBLE_HEAD)
 
-        table.table_data = []
-        for item in result.values():
-            row_data = item.to_dict()
-            row_data = dict(filter(lambda x: x[0] not in FORBIDDEN_TABLE_COLUMNS, row_data.items()))
+    for item in result.values():
+        row_data = item.to_dict()
+        row_data = dict(filter(lambda x: x[0] not in FORBIDDEN_TABLE_COLUMNS, row_data.items()))
 
-            if not table.table_data:
-                table.table_data.append(list(map(lambda x: x.replace('_', ' ').upper(), row_data.keys())))
+        if not table.columns:
+            for col in map(lambda x: x.replace('_', ' ').upper(), row_data.keys()):
+                table.add_column(col, header_style="blue")
 
-            table.table_data.append(row_data.values())
+        table.add_row(*map(lambda x: str(x), row_data.values()))
 
-        yield "TIMESTAMP: %s" % datetime.now() + "\n\n" + table.table
+    return table
 
 
 def open_machine_terminal(machine) -> None:

src/Kathara/foundation/manager/IManager.py

@@ -364,7 +364,8 @@ def get_machines_stats(self, lab_hash: Optional[str] = None, lab_name: Optional[
 
     @abstractmethod
     def get_machine_stats(self, machine_name: str, lab_hash: Optional[str] = None, lab_name: Optional[str] = None,
-                          lab: Optional[Lab] = None, all_users: bool = False) -> Generator[IMachineStats, None, None]:
+                          lab: Optional[Lab] = None, all_users: bool = False) \
+            -> Generator[Optional[IMachineStats], None, None]:
         """Return information of the specified device in a specified network scenario.
 
         Args:
@@ -378,7 +379,8 @@ def get_machine_stats(self, machine_name: str, lab_hash: Optional[str] = None, l
             all_users (bool): If True, search the device among all the users devices.
 
         Returns:
-            IMachineStats: IMachineStats object containing the device info.
+            Generator[Optional[IMachineStats], None, None]: A generator containing the IMachineStats object
+            with the device info. Returns None if the device is not found.
 
         Raises:
             InvocationError: If a running network scenario hash or name is not specified.
@@ -408,7 +410,8 @@ def get_links_stats(self, lab_hash: Optional[str] = None, lab_name: Optional[str
 
     @abstractmethod
     def get_link_stats(self, link_name: str, lab_hash: Optional[str] = None, lab_name: Optional[str] = None,
-                       lab: Optional[Lab] = None, all_users: bool = False) -> Generator[ILinkStats, None, None]:
+                       lab: Optional[Lab] = None, all_users: bool = False) \
+            -> Generator[Optional[ILinkStats], None, None]:
         """Return information of the specified deployed network in a specified network scenario.
 
         Args:
@@ -422,8 +425,8 @@ def get_link_stats(self, link_name: str, lab_hash: Optional[str] = None, lab_nam
             all_users (bool): If True, return information about the networks of all users.
 
         Returns:
-             Generator[Dict[str, ILinkStats], None, None]: A generator containing dicts that has API Object
-             identifier as keys and ILinksStats objects as values.
+            Generator[Optional[ILinkStats], None, None]: A generator containing the ILinkStats object
+            with the network info. Returns None if the network is not found.
 
         Raises:
             InvocationError: If a running network scenario hash or name is not specified.

src/Kathara/manager/Kathara.py

@@ -380,7 +380,8 @@ def get_machines_stats(self, lab_hash: Optional[str] = None, lab_name: Optional[
         return self.manager.get_machines_stats(lab_hash, lab_name, lab, machine_name, all_users)
 
     def get_machine_stats(self, machine_name: str, lab_hash: Optional[str] = None, lab_name: Optional[str] = None,
-                          lab: Optional[Lab] = None, all_users: bool = False) -> Generator[IMachineStats, None, None]:
+                          lab: Optional[Lab] = None, all_users: bool = False) \
+            -> Generator[Optional[IMachineStats], None, None]:
         """Return information of the specified device in a specified network scenario.
 
         Args:
@@ -394,7 +395,8 @@ def get_machine_stats(self, machine_name: str, lab_hash: Optional[str] = None, l
             all_users (bool): If True, search the device among all the users devices.
 
         Returns:
-            IMachineStats: IMachineStats object containing the device info.
+            Generator[Optional[IMachineStats], None, None]: A generator containing the IMachineStats object
+            with the device info. Returns None if the device is not found.
 
         Raises:
             InvocationError: If a running network scenario hash or name is not specified.
@@ -422,7 +424,8 @@ def get_links_stats(self, lab_hash: Optional[str] = None, lab_name: Optional[str
         return self.manager.get_links_stats(lab_hash, lab_name, lab, link_name, all_users)
 
     def get_link_stats(self, link_name: str, lab_hash: Optional[str] = None, lab_name: Optional[str] = None,
-                       lab: Optional[Lab] = None, all_users: bool = False) -> Generator[ILinkStats, None, None]:
+                       lab: Optional[Lab] = None, all_users: bool = False) \
+            -> Generator[Optional[ILinkStats], None, None]:
         """Return information of the specified deployed network in a specified network scenario.
 
         Args:
@@ -436,8 +439,8 @@ def get_link_stats(self, link_name: str, lab_hash: Optional[str] = None, lab_nam
             all_users (bool): If True, return information about the networks of all users.
 
         Returns:
-             Generator[Dict[str, ILinkStats], None, None]: A generator containing dicts that has API Object
-             identifier as keys and ILinksStats objects as values.
+            Generator[Optional[ILinkStats], None, None]: A generator containing the ILinkStats object
+            with the network info. Returns None if the network is not found.
 
         Raises:
             InvocationError: If a running network scenario hash or name is not specified.

src/Kathara/manager/docker/DockerLink.py

@@ -13,7 +13,6 @@
 from .stats.DockerLinkStats import DockerLinkStats
 from ... import utils
 from ...event.EventDispatcher import EventDispatcher
-from ...exceptions import LinkNotFoundError
 from ...exceptions import PrivilegeError
 from ...model.ExternalLink import ExternalLink
 from ...model.Lab import Lab
@@ -228,7 +227,7 @@ def get_links_api_objects_by_filters(self, lab_hash: str = None, link_name: str
         if link_name:
             filters["label"].append(f"name={link_name}")
 
-        return self.client.networks.list(filters=filters)
+        return self.client.networks.list(filters=filters, greedy=True)
 
     def get_links_stats(self, lab_hash: str = None, link_name: str = None, user: str = None) -> \
             Generator[Dict[str, DockerLinkStats], None, None]:
@@ -243,38 +242,35 @@ def get_links_stats(self, lab_hash: str = None, link_name: str = None, user: str
         Returns:
            Generator[Dict[str, DockerMachineStats], None, None]: A generator containing network names as keys and
            DockerLinkStats as values.
-
-        Raises:
-            LinkNotFoundError: If the collision domains specified are not found.
         """
-        networks = self.get_links_api_objects_by_filters(lab_hash=lab_hash, link_name=link_name, user=user)
-        if not networks:
-            if not link_name:
-                raise LinkNotFoundError("No collision domains found.")
-            else:
-                raise LinkNotFoundError(f"Collision domains with name {link_name} not found.")
-
-        networks = sorted(networks, key=lambda x: x.name)
-
         networks_stats = {}
 
         def load_link_stats(network):
-            networks_stats[network.name] = DockerLinkStats(network)
+            if network.name not in networks_stats:
+                networks_stats[network.name] = DockerLinkStats(network)
 
-        pool_size = utils.get_pool_size()
-        items = utils.chunk_list(networks, pool_size)
+        while True:
+            networks = self.get_links_api_objects_by_filters(lab_hash=lab_hash, link_name=link_name, user=user)
+            if not networks:
+                yield dict()
 
-        with Pool(pool_size) as links_pool:
-            for chunk in items:
-                links_pool.map(func=load_link_stats, iterable=chunk)
+            pool_size = utils.get_pool_size()
+            items = utils.chunk_list(networks, pool_size)
+            with Pool(pool_size) as links_pool:
+                for chunk in items:
+                    links_pool.map(func=load_link_stats, iterable=chunk)
 
-        while True:
-            for network_stats in networks_stats.values():
+            networks_to_remove = []
+            for network_id, network_stats in networks_stats.items():
                 try:
                     network_stats.update()
                 except StopIteration:
+                    networks_to_remove.append(network_id)
                     continue
 
+            for k in networks_to_remove:
+                networks_stats.pop(k, None)
+
             yield networks_stats
 
     def _delete_link(self, network: docker.models.networks.Network) -> None: