Improved CLI

autoignition.py

@@ -11,6 +11,7 @@ from selenium.webdriver.support import expected_conditions as EC
 from selenium.webdriver.support.ui import WebDriverWait
 import typer
 
+from cli import cli_spinner
 import config
 from debug import debug_guard
 from utils import ensure_build_dir
@@ -110,6 +111,7 @@ def build_fuelignition():
 
 
 @debug_guard
+@cli_spinner(description="Converting json to img", total=None)
 @ensure_build_dir
 def json_to_img(
     fuelignition_json: Annotated[
@@ -170,4 +172,5 @@ def json_to_img(
 
 
 if __name__ == "__main__":
+    config.update_config("cli")
     typer.run(json_to_img)

cli.py

@@ -0,0 +1,36 @@
+from functools import wraps
+import inspect
+from typing import Callable
+
+from rich.progress import Progress, SpinnerColumn, TextColumn
+
+import config
+from utils import Singleton
+
+
+class SingletonProgress(Progress, metaclass=Singleton):
+    pass
+
+
+def cli_spinner(*spinner_args, **spinner_kwargs) -> Callable:
+    def decorator(f: Callable) -> Callable:
+        # Indent the spinner to match its nesting level
+        indent = len(inspect.stack()) - 1
+        spinner_kwargs["indent"] = f"├{"─"*indent}► "
+        
+        @wraps(f)
+        def wrapped(*func_args, **func_kwargs):
+            if not config.CLI:
+                return f(*func_args, **func_kwargs)
+            with SingletonProgress(
+                SpinnerColumn(),
+                TextColumn("{task.fields[indent]}[progress.description]{task.description}"),
+                transient=True,
+                expand=True,
+            ) as progress:
+                task_id = progress.add_task(*spinner_args, **spinner_kwargs)
+                out = f(*func_args, **func_kwargs)
+                progress.stop_task(task_id)
+                return out
+        return wrapped
+    return decorator

create_disk.py

@@ -3,11 +3,12 @@ from typing import Annotated
 
 import typer
 
+from cli import cli_spinner
 import config
 from create_img import create_img
 from debug import debug_guard
-from utils import ensure_build_dir
 from docker.types import Mount
+from utils import ensure_build_dir
 
 
 def filter_validation_response(response: str) -> str:
@@ -44,12 +45,14 @@ def validation_result() -> str:
     return response
 
 
+@cli_spinner(description="Validating ignition image", total=None)
 def validate() -> (bool, str):
     response = validation_result().decode()
     response = filter_validation_response(response)
     return (not bool(response), response)
 
 
+@cli_spinner(description="Writing ignition image to disk", total=None)
 def write_disk(disk: str) -> None:
     config.CLIENT.containers.run(
         "alpine",
@@ -60,6 +63,7 @@ def write_disk(disk: str) -> None:
 
 
 @debug_guard
+@cli_spinner(description="Creating ignition initialisation disk", total=None)
 @ensure_build_dir
 def create_ignition_disk(
     disk: Annotated[str, typer.Option(help="Path to the disk to write to", prompt=True)],
@@ -92,4 +96,5 @@ def create_ignition_disk(
 
 
 if __name__ == "__main__":
+    config.update_config("cli")
     typer.run(create_ignition_disk)

create_img.py

@@ -5,6 +5,8 @@ from typing import Annotated
 import typer
 
 from autoignition import json_to_img
+from cli import cli_spinner
+import config
 from debug import debug_guard
 from utils import ensure_build_dir
 
@@ -67,6 +69,7 @@ def apply_ignition_settings(
 
 
 @debug_guard
+@cli_spinner(description="Creating ignition image", total=None)
 @ensure_build_dir
 def create_img(
     hostname: Annotated[str, typer.Option(help="Hostname for the new node", prompt=True)],
@@ -118,4 +121,5 @@ def create_img(
 
 
 if __name__ == "__main__":
+    config.update_config("cli")
     typer.run(create_img)

utils.py

@@ -12,3 +12,11 @@ def ensure_build_dir(f: Callable) -> Callable:
         return f(*args, **kwargs)
 
     return wrapper
+
+
+class Singleton(type):
+    _instance = None
+    def __call__(cls, *args, **kwargs):
+        if cls._instance is None:
+            cls._instance = super().__call__(*args, **kwargs)
+        return cls._instance