otk/ui: introduce some sort of ui

src/otk/command.py

@@ -6,6 +6,7 @@
 
 from . import __version__
 from .document import Omnifest
+from . import ui
 
 log = logging.getLogger(__name__)
 
@@ -28,6 +29,8 @@ def run(argv: List[str]) -> int:
         print(f"otk {__version__}")
         return 0
 
+    ui.motd()
+
     if arguments.command == "compile":
         return compile(arguments)
     if arguments.command == "validate":
@@ -47,6 +50,8 @@ def _process(arguments: argparse.Namespace, dry_run: bool) -> int:
     else:
         path = pathlib.Path(arguments.input)
 
+    ui.print(f"Compiling {path}")
+
     # First pass of resolving the otk file is "shallow", it will not run
     # externals and not resolve anything under otk.target.*
     #
@@ -68,12 +73,16 @@ def _process(arguments: argparse.Namespace, dry_run: bool) -> int:
         log.fatal("requested target %r does not exist in INPUT", target_requested)
         return 1
 
+    ui.print(f"Selected the {target_requested} target")
+
     # Now do the real resolve that takes the target into account. It needs
     # a full run so that resolving includes works correctly.
     warn_duplicated_defs = any(arg in getattr(arguments, "warn", [])
                                for arg in ["duplicate-definition", "all"])
     doc = Omnifest(path, target=target_requested, warn_duplicated_defs=warn_duplicated_defs)
 
+    ui.print("Done")
+
     # and then output by writing to the output
     if not dry_run:
         dst.write(doc.as_target_string())

src/otk/external.py

@@ -9,10 +9,13 @@
 import os
 from typing import Any
 
+from . import ui
+
 from .constant import PREFIX_EXTERNAL
 from .error import ExternalFailedError
 from .traversal import State
 
+
 log = logging.getLogger(__name__)
 
 
@@ -26,7 +29,9 @@ def call(state: State, directive: str, tree: Any) -> Any:
         }
     )
 
-    process = subprocess.run([exe], input=data, encoding="utf8", capture_output=True, check=False)
+    with ui.spinner(f"Executing external {os.path.basename(exe)}"):
+        process = subprocess.run([exe], input=data, encoding="utf8", capture_output=True, check=False)
+
     if process.returncode != 0:
         msg = f"call {exe} {directive!r} failed: stdout={process.stdout!r}, stderr={process.stderr!r}"
         log.error(msg)

src/otk/transform.py

@@ -223,7 +223,7 @@ def process_include(ctx: Context, state: State, path: pathlib.Path) -> dict:
     if not path.is_absolute():
         cur_path = state.path.parent
         path = (cur_path / pathlib.Path(path)).resolve()
-    log.info("resolving %s", path)
+    log.debug("resolving %s", path)
     try:
         with path.open(encoding="utf8") as fp:
             data = yaml.load(fp, Loader=SafeUniqueKeyLoader)

src/otk/ui.py

@@ -0,0 +1,67 @@
+import sys
+import time
+import threading
+
+from typing import Optional, Type
+from types import TracebackType
+
+from . import __version__
+
+
+class _Spinner:
+    """Render a spinner with an optional prompt to stderr, if stderr isn't a
+    tty this only prints the prompt."""
+
+    active: bool
+    thread: threading.Thread
+
+    def __init__(self, prompt: str) -> None:
+        self.prompt = prompt
+
+    def _loop(self):
+        sys.stderr.write(self.prompt)
+        sys.stderr.flush()
+
+        # No spinner business if we aren't printing to tty's.
+        if sys.stderr.isatty():
+            while self.active:
+                sys.stderr.write(".")
+                sys.stderr.flush()
+                time.sleep(0.1)
+
+        # Only write a newline if the prompt was actually there
+        if self.prompt:
+            sys.stderr.write("\n")
+            sys.stderr.flush()
+
+    def __enter__(self, prompt: str = "") -> None:
+        self.active = True
+        self.thread = threading.Thread(target=self._loop)
+        self.thread.start()
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> bool:
+        self.active = False
+        self.thread.join()
+
+        return True
+
+
+def spinner(prompt: str) -> _Spinner:
+    return _Spinner(prompt)
+
+
+def print(text: str) -> None:  # pylint: disable=redefined-builtin
+    """Write a line of text to stderr if it's attached to a tty."""
+    if sys.stderr.isatty():
+        sys.stderr.write(text + "\n")
+        sys.stderr.flush()
+
+
+def motd() -> None:
+    # Note, this is the local print function
+    print(f"otk {__version__} -- https://www.osbuild.org/")

test/test_integration.py

@@ -60,7 +60,7 @@ def test_otk_define_empty(tmp_path, caplog):
 
 
 def test_verbose_logs_processed_files(tmp_path, caplog):
-    caplog.set_level(logging.INFO)
+    caplog.set_level(logging.DEBUG)
 
     test_otk = tmp_path / "foo.yaml"
     test_otk.write_text(TEST_OTK)