Improve rendering libary (#65)

command_line_assistant/commands/query.py

@@ -1,13 +1,70 @@
 from argparse import Namespace
 
+from dasbus.error import DBusError
+
 from command_line_assistant.dbus.constants import SERVICE_IDENTIFIER
 from command_line_assistant.dbus.definitions import MessageInput, MessageOutput
+from command_line_assistant.rendering.decorators.colors import ColorDecorator
+from command_line_assistant.rendering.decorators.text import (
+    EmojiDecorator,
+    TextWrapDecorator,
+    WriteOnceDecorator,
+)
+from command_line_assistant.rendering.renders.spinner import SpinnerRenderer
+from command_line_assistant.rendering.renders.text import TextRenderer
+from command_line_assistant.rendering.stream import StderrStream, StdoutStream
 from command_line_assistant.utils.cli import BaseCLICommand, SubParsersAction
 
+LEGAL_NOTICE = (
+    "RHEL Lightspeed Command Line Assistant can answer questions related to RHEL."
+    " Do not include personal or business sensitive information in your input."
+    "Interactions with RHEL Lightspeed may be reviewed and used to improve our "
+    "products and service."
+)
+ALWAYS_LEGAL_MESSAGE = (
+    "Always check AI/LLM-generated responses for accuracy prior to use."
+)
+
+
+def _initialize_spinner_renderer() -> SpinnerRenderer:
+    spinner = SpinnerRenderer(
+        message="Requesting knowledge from AI", stream=StdoutStream(end="")
+    )
+
+    spinner.update(EmojiDecorator(emoji="U+1F916"))  # Robot emoji
+    spinner.update(TextWrapDecorator())
+
+    return spinner
+
+
+def _initialize_text_renderer() -> TextRenderer:
+    text = TextRenderer(stream=StdoutStream(end="\n"))
+    text.update(ColorDecorator(foreground="green"))  # Robot emoji
+    text.update(TextWrapDecorator())
+
+    return text
+
+
+def _initialize_legal_renderer(write_once: bool = False) -> TextRenderer:
+    text = TextRenderer(stream=StderrStream())
+    text.update(ColorDecorator(foreground="lightyellow"))
+    text.update(TextWrapDecorator())
+
+    if write_once:
+        text.update(WriteOnceDecorator(state_filename="legal"))
+
+    return text
+
 
 class QueryCommand(BaseCLICommand):
     def __init__(self, query_string: str) -> None:
         self._query = query_string
+
+        self._spinner_renderer: SpinnerRenderer = _initialize_spinner_renderer()
+        self._text_renderer: TextRenderer = _initialize_text_renderer()
+        self._legal_renderer: TextRenderer = _initialize_legal_renderer(write_once=True)
+        self._warning_renderer: TextRenderer = _initialize_legal_renderer()
+
         super().__init__()
 
     def run(self) -> None:
@@ -16,13 +73,21 @@ def run(self) -> None:
         input_query = MessageInput()
         input_query.message = self._query
 
-        print("Requesting knowledge from the AI :robot:")
-        proxy.ProcessQuery(MessageInput.to_structure(input_query))
-
-        output = MessageOutput.from_structure(proxy.RetrieveAnswer).message
+        output = "Nothing to see here..."
+        try:
+            with self._spinner_renderer:
+                proxy.ProcessQuery(MessageInput.to_structure(input_query))
+                output = MessageOutput.from_structure(proxy.RetrieveAnswer).message
 
-        if output:
-            print("\n", output)
+            self._legal_renderer.render(LEGAL_NOTICE)
+            self._text_renderer.render(output)
+            self._warning_renderer.render(ALWAYS_LEGAL_MESSAGE)
+        except DBusError:
+            self._text_renderer.update(ColorDecorator(foreground="red"))
+            self._text_renderer.update(EmojiDecorator(emoji="U+1F641"))
+            self._text_renderer.render(
+                "Uh oh... Something went wrong. Try again later."
+            )
 
 
 def register_subcommand(parser: SubParsersAction) -> None:

command_line_assistant/config/__init__.py

@@ -10,7 +10,7 @@
     LoggingSchema,
     OutputSchema,
 )
-from command_line_assistant.utils.environment import get_xdg_path
+from command_line_assistant.utils.environment import get_xdg_config_path
 
 # tomllib is available in the stdlib after Python3.11. Before that, we import
 # from tomli.
@@ -54,7 +54,7 @@ def load_config_file() -> Config:
     """Read configuration file."""
 
     config_dict = {}
-    config_file_path = Path(get_xdg_path(), *CONFIG_FILE_DEFINITION)
+    config_file_path = Path(get_xdg_config_path(), *CONFIG_FILE_DEFINITION)
 
     try:
         print(f"Loading configuration file from {config_file_path}")

command_line_assistant/rendering/base.py

@@ -0,0 +1,67 @@
+from abc import ABC, abstractmethod
+from typing import TextIO
+
+
+class RenderDecorator(ABC):
+    """Abstract base class for render decorators"""
+
+    @abstractmethod
+    def decorate(self, text: str) -> str:
+        pass
+
+
+class OutputStreamWritter(ABC):
+    """Abstract base class for output stream decorators"""
+
+    def __init__(self, stream: TextIO, end: str = "\n") -> None:
+        """
+        Initialize the output stream decorator.
+
+        Args:
+            stream: The output stream to use
+            end: The string to append after the text (defaults to newline)
+        """
+        self._stream = stream
+        self._end = end
+
+    @abstractmethod
+    def write(self, text: str) -> None:
+        """Write the text to the output stream"""
+        pass
+
+    @abstractmethod
+    def flush(self) -> None:
+        """Flush the output stream"""
+        pass
+
+    def execute(self, text: str) -> None:
+        """
+        Write the text to the output stream and return the original text for chaining.
+        """
+        if text:
+            self.write(text)
+            self.flush()
+
+
+class BaseRenderer(ABC):
+    """Base class for renderers providing common functionality."""
+
+    def __init__(self, stream: OutputStreamWritter) -> None:
+        self._stream = stream
+        self._decorators: dict[type, RenderDecorator] = {}
+
+    def update(self, decorator: RenderDecorator) -> None:
+        """Update or add a decorator of the same type."""
+        self._decorators[type(decorator)] = decorator
+
+    def _apply_decorators(self, text: str) -> str:
+        """Apply all decorators to the text."""
+        decorated_text = text
+        for decorator in self._decorators.values():
+            decorated_text = decorator.decorate(decorated_text)
+        return decorated_text
+
+    @abstractmethod
+    def render(self, text: str) -> None:
+        """Render the text with all decorators applied."""
+        pass

command_line_assistant/rendering/decorators/colors.py

@@ -3,7 +3,7 @@
 
 from colorama import Back, Fore, Style
 
-from command_line_assistant.rendering.decorators.base import RenderDecorator
+from command_line_assistant.rendering.base import RenderDecorator
 
 
 class ColorDecorator(RenderDecorator):

command_line_assistant/rendering/decorators/style.py

@@ -2,7 +2,7 @@
 
 from colorama import Style
 
-from command_line_assistant.rendering.decorators.base import RenderDecorator
+from command_line_assistant.rendering.base import RenderDecorator
 
 
 class StyleDecorator(RenderDecorator):

command_line_assistant/rendering/decorators/text.py

@@ -1,8 +1,10 @@
 import shutil
 import textwrap
+from pathlib import Path
 from typing import Optional, Union
 
-from command_line_assistant.rendering.decorators.base import RenderDecorator
+from command_line_assistant.rendering.base import RenderDecorator
+from command_line_assistant.utils.environment import get_xdg_state_path
 
 
 class EmojiDecorator(RenderDecorator):
@@ -40,3 +42,43 @@ def decorate(self, text: str) -> str:
             initial_indent=self._indent,
             subsequent_indent=self._indent,
         )
+
+
+class WriteOnceDecorator(RenderDecorator):
+    """Decorator that ensures content is written only once by checking a state file.
+
+    The state file is created under $XDG_STATE_HOME/command-line-assistant/legal/
+    """
+
+    def __init__(self, state_filename: str = "written") -> None:
+        """Initialize the write once decorator.
+
+        Args:
+            state_filename: Name of the state file to create/check
+        """
+        self._state_dir = Path(get_xdg_state_path(), "command-line-assistant")
+        self._state_file = self._state_dir / state_filename
+
+    def _should_write(self) -> bool:
+        """Check if content should be written by verifying state file existence."""
+        if self._state_file.exists():
+            return False
+
+        if not self._state_dir.exists():
+            # Create directory if it doesn't exist
+            self._state_dir.mkdir(parents=True)
+
+        # Write state file
+        self._state_file.write_text("1")
+        return True
+
+    def decorate(self, text: str) -> str:
+        """Write the text only if it hasn't been written before.
+
+        Args:
+            text: The text to potentially write
+
+        Returns:
+            The text if it should be written, None otherwise
+        """
+        return text if self._should_write() else ""

command_line_assistant/rendering/renders/spinner.py

@@ -0,0 +1,72 @@
+import itertools
+import threading
+import time
+from dataclasses import dataclass
+from typing import Iterator, Optional
+
+from command_line_assistant.rendering.base import BaseRenderer, OutputStreamWritter
+from command_line_assistant.rendering.stream import StdoutStream
+
+
+@dataclass
+class Frames:
+    default: Iterator[str] = itertools.cycle(["⠋", "⠙", "⠸", "⠴", "⠦", "⠇"])
+    dash: Iterator[str] = itertools.cycle(["-", "\\", "|", "/"])
+    circular: Iterator[str] = itertools.cycle(["◐", "◓", "◑", "◒"])
+    dots: Iterator[str] = itertools.cycle([".  ", ".. ", "...", " ..", "  .", "   "])
+    arrows: Iterator[str] = itertools.cycle(["←", "↖", "↑", "↗", "→", "↘", "↓", "↙"])
+    moving: Iterator[str] = itertools.cycle(
+        ["[   ]", "[=  ]", "[== ]", "[===]", "[ ==]", "[  =]", "[   ]"]
+    )
+
+
+class SpinnerRenderer(BaseRenderer):
+    def __init__(
+        self,
+        message: str,
+        stream: Optional[OutputStreamWritter] = None,
+        frames: Iterator[str] = Frames.default,
+        delay: float = 0.1,
+        clear_message: bool = False,
+    ) -> None:
+        super().__init__(stream or StdoutStream())
+        self._message = message
+        self._frames = frames
+        self._delay = delay
+        self._clear_message = clear_message
+        self._done = threading.Event()
+        self._spinner_thread: Optional[threading.Thread] = None
+
+    def render(self, text: str) -> None:
+        """Render text with all decorators applied."""
+        decorated_text = self._apply_decorators(text)
+        self._stream.execute(decorated_text)
+
+    def _animation(self) -> None:
+        while not self._done.is_set():
+            frame = next(self._frames)
+            message = self._apply_decorators(f"{frame} {self._message}")
+            self._stream.execute(f"\r{message}")
+            time.sleep(self._delay)
+
+    def start(self) -> None:
+        """Start the spinner animation"""
+        self._done.clear()
+        self._spinner_thread = threading.Thread(target=self._animation)
+        self._spinner_thread.start()
+
+    def stop(self) -> None:
+        """Stop the spinner animation"""
+        if self._spinner_thread:
+            self._done.set()
+            self._spinner_thread.join()
+            self._stream.execute("\n")
+            if self._clear_message:
+                self._stream.execute(f"\r{' ' * (len(self._message) + 2)}\r")
+
+    def __enter__(self) -> "SpinnerRenderer":
+        self.start()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        self.stop()

command_line_assistant/rendering/renders/text.py

@@ -0,0 +1,18 @@
+import shutil
+from typing import Optional
+
+from command_line_assistant.rendering.base import BaseRenderer, OutputStreamWritter
+from command_line_assistant.rendering.stream import StdoutStream
+
+
+class TextRenderer(BaseRenderer):
+    def __init__(self, stream: Optional[OutputStreamWritter] = None) -> None:
+        super().__init__(stream or StdoutStream())
+        self.terminal_width = shutil.get_terminal_size().columns
+
+    def render(self, text: str) -> None:
+        """Render text with all decorators applied."""
+        lines = text.splitlines()
+        for line in lines:
+            decorated_text = self._apply_decorators(line)
+            self._stream.execute(decorated_text)