fix: Fix initial configuration for stats plot (#114)

.github/workflows/main.yaml

@@ -71,14 +71,17 @@ jobs:
         auto-update-conda: true
         conda-solver: libmamba
 
+    - name: check poetry lock
+      run: poetry check
+
     - name: Install dependencies
       run: poetry install --verbose
 
     - name: run unit tests
-      run: makim tests.unit --verbose
+      run: makim --verbose tests.unit
 
     - name: CLI tests
-      run: makim tests.smoke --verbose
+      run: makim --verbose tests.smoke
 
     - name: Setup tmate session
       if: "${{ failure() && (contains(github.event.pull_request.labels.*.name, 'ci:enable-debugging')) }}"

pyproject.toml

@@ -46,7 +46,7 @@ mkdocs-literate-nav = ">=0.6.0"
 mkdocs-macros-plugin = ">=0.7.0,<1"
 mkdocs-material = ">=9.1.15"
 mkdocstrings = {version=">=0.19.0", extras=["python"]}
-makim = "1.8.3"
+makim = "1.12.0"
 compose-go = ">=1.23.0"
 
 [build-system]
@@ -66,10 +66,13 @@ ignore_missing_imports = true
 line-length = 79
 force-exclude = true
 src = ["./src/sugar", "./tests"]
-ignore = ["RUF012"]
 exclude = [
   "docs",
 ]
+fix = true
+
+[tool.ruff.lint]
+ignore = ["RUF012"]
 select = [
   "E",   # pycodestyle
   "F",   # pyflakes
@@ -79,13 +82,11 @@ select = [
   "RUF", # Ruff-specific rules
   "I001", # isort
 ]
-# fixable = ["I001"]
-fix = true
 
-[tool.ruff.pydocstyle]
+[tool.ruff.lint.pydocstyle]
 convention = "numpy"
 
-[tool.ruff.isort]
+[tool.ruff.lint.isort]
 # Use a single line between direct and from import
 lines-between-types = 1
 

src/sugar/console.py

@@ -0,0 +1,19 @@
+"""Functions about console."""
+import os
+
+
+def get_terminal_size():
+    """Return the height (number of lines) of the terminal using os module."""
+    size = os.get_terminal_size()
+    try:
+        height = size.lines
+    except OSError:
+        # Default to 24 lines if the terminal size cannot be determined.
+        height = 24
+
+    try:
+        width = size.columns
+    except OSError:
+        # Default to 24 lines if the terminal size cannot be determined.
+        height = 80
+    return width, height

src/sugar/inspect.py

@@ -1,4 +1,7 @@
 """Functions for inspecting and retrieving information from containers."""
+from __future__ import annotations
+
+import re
 import subprocess  # nosec B404
 
 
@@ -12,3 +15,34 @@ def get_container_name(container_id: str) -> str:
         raise Exception('No container name found for the given ID')
     # Removing the leading slash from the container name
     return result.stdout.strip().lstrip('/')
+
+
+def get_container_stats(container_name: str) -> tuple[float, float]:
+    """
+    Fetch the current memory and CPU usage of a given Docker container.
+
+    Parameters
+    ----------
+        container_name (str): Name of the Docker container.
+
+    Returns
+    -------
+    tuple:
+        The current memory usage of the container in MB and CPU usage as
+        a percentage.
+    """
+    command = (
+        f'docker stats {container_name} --no-stream --format '
+        f"'{{{{.MemUsage}}}} {{{{.CPUPerc}}}}'"
+    )
+    result = subprocess.run(  # nosec B602, B603
+        command, capture_output=True, text=True, shell=True, check=False
+    )
+    output = result.stdout.strip().split()
+    mem_usage_str = output[0].split('/')[0].strip()
+    cpu_usage_str = output[-1].strip('%')
+
+    mem_usage = float(re.sub(r'[^\d.]', '', mem_usage_str))
+    cpu_usage = float(cpu_usage_str)
+
+    return mem_usage, cpu_usage

src/sugar/plugins/base.py

@@ -324,7 +324,7 @@ def _load_service_names(self):
             )
         elif self.args.get('services'):
             self.service_names = self.args.get('services').split(',')
-        elif 'default' in services and services['default']:
+        elif services.get('default'):
             self.service_names = services['default'].split(',')
 
     def _verify_args(self):

src/sugar/plugins/stats.py

@@ -1,10 +1,11 @@
 """Sugar Plugin for Containers Statics."""
 from __future__ import annotations
 
+import datetime
 import io
-import re
-import subprocess  # nosec B404
-import time
+
+from itertools import tee
+from typing import Iterable
 
 import plotille
 
@@ -14,40 +15,13 @@
 from textual.widget import Widget
 from textual.widgets import Header
 
-from sugar.inspect import get_container_name
+from sugar.console import get_terminal_size
+from sugar.inspect import get_container_name, get_container_stats
 from sugar.logs import KxgrErrorType, KxgrLogs
 from sugar.plugins.base import SugarDockerCompose
 
-
-def get_container_stats(container_name: str) -> tuple[float, float]:
-    """
-    Fetch the current memory and CPU usage of a given Docker container.
-
-    Parameters
-    ----------
-        container_name (str): Name of the Docker container.
-
-    Returns
-    -------
-    tuple:
-        The current memory usage of the container in MB and CPU usage as
-        a percentage.
-    """
-    command = (
-        f'docker stats {container_name} --no-stream --format '
-        f"'{{{{.MemUsage}}}} {{{{.CPUPerc}}}}'"
-    )
-    result = subprocess.run(  # nosec B602, B603
-        command, capture_output=True, text=True, shell=True, check=False
-    )
-    output = result.stdout.strip().split()
-    mem_usage_str = output[0].split('/')[0].strip()
-    cpu_usage_str = output[-1].strip('%')
-
-    mem_usage = float(re.sub(r'[^\d.]', '', mem_usage_str))
-    cpu_usage = float(cpu_usage_str)
-
-    return mem_usage, cpu_usage
+CHART_WINDOW_DURATION = 60
+CHART_TIME_INTERVAL = 1
 
 
 class StatsPlot:
@@ -56,8 +30,8 @@ class StatsPlot:
     def __init__(
         self,
         container_names: list[str],
-        window_duration: int = 60,
-        interval: int = 1,
+        window_duration: int = CHART_WINDOW_DURATION,
+        interval: int = CHART_TIME_INTERVAL,
     ):
         """
         Initialize StatsPlot.
@@ -74,51 +48,125 @@ def __init__(
         self.container_names = container_names
         self.window_duration = window_duration
         self.interval = interval
-        self.start_time = time.time()
 
+        self.create_chart()
+        self.reset_data()
+
+    def create_chart(self):
+        """Create a new chart."""
         self.fig_mem = plotille.Figure()
         self.fig_cpu = plotille.Figure()
 
+        self.resize_chart()
+
+        self.chart_colors: tuple[Iterable, Iterable] = {
+            'mem': tee(self.fig_mem._color_seq),
+            'cpu': tee(self.fig_cpu._color_seq),
+        }
+
         self.stats: dict[str, dict[str, list[str]]] = {
             name: {'times': [], 'mem_usages': [], 'cpu_usages': []}
-            for name in container_names
+            for name in self.container_names
         }
 
+        for name in self.container_names:
+            container_stats = self.stats[name]
+            # Add data to plots
+            self.fig_mem.plot(
+                container_stats['times'],
+                container_stats['mem_usages'],
+                label=name,
+            )
+            self.fig_cpu.plot(
+                container_stats['times'],
+                container_stats['cpu_usages'],
+                label=name,
+            )
+
+    def resize_chart(self):
+        """Resize chart."""
+        console_width, console_height = get_terminal_size()
+
+        chart_height = min((console_height - 20) // 2, 10)
+        chart_width = console_width - 30
+
+        self.fig_mem.width = chart_width
+        self.fig_mem.height = chart_height
+        self.fig_cpu.width = chart_width
+        self.fig_cpu.height = chart_height
+
+    def reset_chart(self):
+        """Reset chart state."""
+        self.fig_mem._plots.clear()
+        self.fig_cpu._plots.clear()
+
+        self.fig_mem._color_seq = tee(self.chart_colors['mem'][0])[1]
+        self.fig_cpu._color_seq = tee(self.chart_colors['cpu'][0])[1]
+
+        self.resize_chart()
+
+    def reset_data(self):
+        """Generate a clean data."""
+        current_time = datetime.datetime.now()
+
+        for name in self.container_names:
+            container_stats = self.stats[name]
+
+            container_stats['mem_usages'].clear()
+            container_stats['cpu_usages'].clear()
+            container_stats['times'].clear()
+
+            container_stats['mem_usages'].extend([0.0] * CHART_WINDOW_DURATION)
+            container_stats['cpu_usages'].extend([0.0] * CHART_WINDOW_DURATION)
+            container_stats['times'].extend(
+                [
+                    current_time
+                    - datetime.timedelta(seconds=i * CHART_TIME_INTERVAL)
+                    for i in range(CHART_WINDOW_DURATION)
+                ][::-1]
+            )
+
+            # Add data to plots
+            self.fig_mem.plot(
+                container_stats['times'],
+                container_stats['mem_usages'],
+                label=name,
+            )
+            self.fig_cpu.plot(
+                container_stats['times'],
+                container_stats['cpu_usages'],
+                label=name,
+            )
+
     def plot_stats(self):
         """
         Plot containers statistic.
 
         Plots the memory and CPU usage of multiple Docker containers over
         time in a single chart for each metric.
         """
-        self.fig_mem = plotille.Figure()
-        self.fig_mem.width = 50
-        self.fig_mem.height = 5
-        self.fig_cpu = plotille.Figure()
-        self.fig_cpu.width = 50
-        self.fig_cpu.height = 5
-
-        current_time = time.time() - self.start_time
+        current_time = datetime.datetime.now()
 
         for name in self.container_names:
             mem_usage, cpu_usage = get_container_stats(name)
 
             # Update and maintain window for stats
             container_stats = self.stats[name]
-            container_stats['times'].append(round(current_time, 2))
+            container_stats['times'].append(current_time)
             container_stats['mem_usages'].append(round(mem_usage, 2))
             container_stats['cpu_usages'].append(round(cpu_usage, 2))
 
-            if len(container_stats['times']) > self.window_duration:
-                container_stats['times'] = container_stats['times'][
-                    -self.window_duration :
-                ]
-                container_stats['mem_usages'] = container_stats['mem_usages'][
-                    -self.window_duration :
-                ]
-                container_stats['cpu_usages'] = container_stats['cpu_usages'][
-                    -self.window_duration :
-                ]
+            container_stats['times'] = container_stats['times'][
+                -self.window_duration :
+            ]
+            container_stats['mem_usages'] = container_stats['mem_usages'][
+                -self.window_duration :
+            ]
+            container_stats['cpu_usages'] = container_stats['cpu_usages'][
+                -self.window_duration :
+            ]
+
+        self.reset_chart()
 
         for name in self.container_names:
             container_stats = self.stats[name]
@@ -155,13 +203,13 @@ def __init__(self, container_names: list[str], *args, **kwargs) -> None:
     def on_mount(self) -> None:
         """Set up the widget."""
         # Set up a periodic update, adjust the interval as needed
-        interval_time = 1
+        interval_time = CHART_TIME_INTERVAL
         self.set_interval(
             interval_time, self.update_plot
         )  # Update every second
         self.stats_plot = StatsPlot(
             container_names=self.container_names,
-            window_duration=60,
+            window_duration=CHART_WINDOW_DURATION,
             interval=interval_time,
         )
 
@@ -183,7 +231,7 @@ def render(self) -> Text:
 class StatsPlotApp(App[str]):
     """StatsPlotApp app class."""
 
-    TITLE = 'Containers Stats'
+    TITLE = 'Sugar Containers Stats'
     container_names: list[str]
 
     def __init__(self, container_names: list[str], *args, **kwargs) -> None: