Skip to content

Commit

Permalink
Small code refactor for outputting text
Browse files Browse the repository at this point in the history
  • Loading branch information
r0x0d committed Jan 6, 2025
1 parent c7d9389 commit e784d6a
Show file tree
Hide file tree
Showing 14 changed files with 266 additions and 199 deletions.
127 changes: 49 additions & 78 deletions command_line_assistant/commands/history.py
Original file line number Diff line number Diff line change
@@ -1,63 +1,25 @@
"""Module to handle the history command."""

import logging
from argparse import Namespace

from command_line_assistant.dbus.constants import HISTORY_IDENTIFIER
from command_line_assistant.dbus.exceptions import (
CorruptedHistoryError,
MissingHistoryFileError,
)
from command_line_assistant.dbus.structures import HistoryEntry
from command_line_assistant.dbus.structures import HistoryEntry, HistoryItem
from command_line_assistant.rendering.decorators.colors import ColorDecorator
from command_line_assistant.rendering.decorators.text import (
EmojiDecorator,
TextWrapDecorator,
)
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 StdoutStream
from command_line_assistant.utils.cli import BaseCLICommand, SubParsersAction

logger = logging.getLogger(__name__)


def _initialize_spinner_renderer() -> SpinnerRenderer:
"""Initialize a new text renderer class
Returns:
SpinnerRenderer: Instance of a text renderer class with decorators.
"""
spinner = SpinnerRenderer(message="Loading history", stream=StdoutStream(end=""))
spinner.update(TextWrapDecorator())

return spinner


def _initialize_qa_renderer(is_assistant: bool = False) -> TextRenderer:
"""Initialize a new text renderer class.
Args:
is_assistant (bool): Apply different decorators if it is assistant.
Returns:
TextRenderer: Instance of a text renderer class with decorators.
"""
text = TextRenderer(stream=StdoutStream(end="\n"))
foreground = "lightblue" if is_assistant else "lightgreen"
text.update(ColorDecorator(foreground=foreground))
text.update(EmojiDecorator("🤖"))
return text


def _initialize_text_renderer() -> TextRenderer:
"""Initialize a new text renderer class
Returns:
TextRenderer: Instance of a text renderer class with decorators.
"""
text = TextRenderer(stream=StdoutStream(end="\n"))
return text
from command_line_assistant.utils.renderers import (
create_error_renderer,
create_spinner_renderer,
create_text_renderer,
)


class HistoryCommand(BaseCLICommand):
Expand All @@ -80,10 +42,19 @@ def __init__(self, clear: bool, first: bool, last: bool) -> None:
self._last = last

self._proxy = HISTORY_IDENTIFIER.get_proxy()
self._user_renderer = _initialize_qa_renderer()
self._assistant_renderer = _initialize_qa_renderer(is_assistant=True)
self._text_renderer = _initialize_text_renderer()
self._spinner_renderer = _initialize_spinner_renderer()

self._spinner_renderer: SpinnerRenderer = create_spinner_renderer(
message="Loading history",
decorators=[EmojiDecorator(emoji="U+1F916")],
)
self._q_renderer: TextRenderer = create_text_renderer(
decorators=[ColorDecorator("lightgreen")]
)
self._a_renderer: TextRenderer = create_text_renderer(
decorators=[ColorDecorator("lightblue")]
)
self._text_renderer: TextRenderer = create_text_renderer()
self._error_renderer: TextRenderer = create_error_renderer()

super().__init__()

Expand All @@ -108,9 +79,7 @@ def run(self) -> int:

return 0
except (MissingHistoryFileError, CorruptedHistoryError) as e:
self._text_renderer.update(ColorDecorator(foreground="red"))
self._text_renderer.update(EmojiDecorator(emoji="U+1F641"))
self._text_renderer.render(str(e))
self._error_renderer.render(str(e))

Check warning on line 82 in command_line_assistant/commands/history.py

View check run for this annotation

Codecov / codecov/patch

command_line_assistant/commands/history.py#L82

Added line #L82 was not covered by tests
return 1

def _retrieve_all_conversations(self) -> None:
Expand All @@ -119,54 +88,56 @@ def _retrieve_all_conversations(self) -> None:
response = self._proxy.GetHistory()
history = HistoryEntry.from_structure(response)

if not history.entries:
self._text_renderer.render("No history found.")
return

for entry in history.entries:
self._user_renderer.render(f"Query: {entry.query}")
self._assistant_renderer.render(f"Answer: {entry.response}")
self._text_renderer.render(f"Time: {entry.timestamp}")
self._text_renderer.render("-" * 50) # Separator between conversations
# Display the conversation
self._show_history(history.entries)

def _retrieve_first_conversation(self) -> None:
"""Retrieve the first conversation in the conversation cache."""
logger.info("Getting first conversation from history.")
self._text_renderer.render("Getting first conversation from history.")
response = self._proxy.GetFirstConversation()
history = HistoryEntry.from_structure(response)

if not history.entries:
self._text_renderer.render("No history found.")
return

entry = history.entries[0]
self._user_renderer.render(f"Query: {entry.query}")
self._assistant_renderer.render(f"Answer: {entry.response}")
self._text_renderer.render(f"Time: {entry.timestamp}")
# Display the conversation
self._show_history(history.entries)

def _retrieve_last_conversation(self):
"""Retrieve the last conversation in the conversation cache."""
logger.info("Getting last conversation from history.")
self._text_renderer.render("Getting last conversation from history.")
response = self._proxy.GetLastConversation()

# Handle and display the response
history = HistoryEntry.from_structure(response)

if not history.entries:
self._text_renderer.render("No history found.")
return

# Display the conversation
entry = history.entries[-1]
self._user_renderer.render(f"Query: {entry.query}")
self._assistant_renderer.render(f"Answer: {entry.response}")
self._text_renderer.render(f"Time: {entry.timestamp}")
self._show_history(history.entries)

def _clear_history(self) -> None:
"""Clear the user history"""
self._text_renderer.render("Cleaning the history.")
self._proxy.ClearHistory()

def _show_history(self, entries: list[HistoryItem]) -> None:
"""Internal method to show the history in a standarized way
Args:
entries (list[HistoryItem]): The list of entries in the history
"""
if not entries:
self._text_renderer.render("No history found.")
return

is_separator_needed = len(entries) > 1
for entry in entries:
self._q_renderer.render(f"Query: {entry.query}")
self._a_renderer.render(f"Answer: {entry.response}")

timestamp = f"Time: {entry.timestamp}"
self._text_renderer.render(timestamp)

if is_separator_needed:
# Separator between conversations
self._text_renderer.render("-" * len(timestamp))


def register_subcommand(parser: SubParsersAction):
"""
Expand Down
82 changes: 24 additions & 58 deletions command_line_assistant/commands/query.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,13 +13,16 @@
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
from command_line_assistant.utils.renderers import (
create_error_renderer,
create_spinner_renderer,
create_text_renderer,
)

LEGAL_NOTICE = (
"RHEL Lightspeed Command Line Assistant can answer questions related to RHEL."
Expand All @@ -34,54 +37,6 @@
#: Always good to have legal message.


def _initialize_spinner_renderer() -> SpinnerRenderer:
"""Initialize a new spinner renderer class
Returns:
SpinnerRenderer: Instance of a spinner renderer class with decorators.
"""
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:
"""Initialize a new text renderer class
Returns:
TextRenderer: Instance of a text renderer class with decorators.
"""
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:
"""Initialize a new text renderer class
Args:
write_once (bool): If it should add the `py:WriteOnceDecorator` or not.
Returns:
SpinnerRenderer: Instance of a text renderer class with decorators.
"""
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):
"""Class that represents the query command."""

Expand All @@ -95,10 +50,23 @@ def __init__(self, query_string: str, stdin: Optional[str]) -> None:
self._query = query_string
self._stdin = stdin

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()
self._spinner_renderer: SpinnerRenderer = create_spinner_renderer(
message="Requesting knowledge from AI",
decorators=[EmojiDecorator(emoji="U+1F916")],
)
self._text_renderer: TextRenderer = create_text_renderer(
decorators=[ColorDecorator(foreground="green")]
)
self._legal_renderer: TextRenderer = create_text_renderer(
decorators=[
ColorDecorator(foreground="lightyellow"),
WriteOnceDecorator(state_filename="legal"),
]
)
self._warning_renderer: TextRenderer = create_text_renderer(
decorators=[ColorDecorator(foreground="lightyellow")]
)
self._error_renderer: TextRenderer = create_error_renderer()

super().__init__()

Expand Down Expand Up @@ -130,9 +98,7 @@ def run(self) -> int:
MissingHistoryFileError,
CorruptedHistoryError,
) as e:
self._text_renderer.update(ColorDecorator(foreground="red"))
self._text_renderer.update(EmojiDecorator(emoji="U+1F641"))
self._text_renderer.render(str(e))
self._error_renderer.render(str(e))
return 1

self._legal_renderer.render(LEGAL_NOTICE)
Expand All @@ -143,7 +109,7 @@ def run(self) -> int:

def register_subcommand(parser: SubParsersAction) -> None:
"""
Register this command to argparse so it's available for the root parser.
Register this command to argparse so it's available for the root parserself._.
Args:
parser (SubParsersAction): Root parser to register command-specific arguments
Expand Down
10 changes: 3 additions & 7 deletions command_line_assistant/initialize.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,14 +4,12 @@
from argparse import ArgumentParser, Namespace

from command_line_assistant.commands import history, query, record
from command_line_assistant.rendering.decorators.colors import ColorDecorator
from command_line_assistant.rendering.decorators.text import EmojiDecorator
from command_line_assistant.rendering.renders.text import TextRenderer
from command_line_assistant.utils.cli import (
add_default_command,
create_argument_parser,
read_stdin,
)
from command_line_assistant.utils.renderers import create_error_renderer


def register_subcommands() -> ArgumentParser:
Expand Down Expand Up @@ -45,10 +43,8 @@ def initialize() -> int:
stdin = read_stdin()
except UnicodeDecodeError:
# Usually happens when the user try to cat a binary file and redirect that to us.
text_renderer = TextRenderer()
text_renderer.update(ColorDecorator(foreground="red"))
text_renderer.update(EmojiDecorator(emoji="U+1F641"))
text_renderer.render(
error_renderer = create_error_renderer()
error_renderer.render(
"The stdin provided could not be decoded. Please, make sure it is in textual format."
)
return 1
Expand Down
10 changes: 7 additions & 3 deletions command_line_assistant/rendering/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,14 +74,18 @@ def __init__(self, stream: BaseStream) -> None:
self._stream = stream
self._decorators: dict[type, BaseDecorator] = {}

def update(self, decorator: BaseDecorator) -> None:
def update(self, decorators: list[BaseDecorator]) -> None:
"""Update or add a decorator of the same type.
Args:
decorator (RenderDecorator): An instance of a rendering decorator to be applied.
decorator (list[BaseDecorator]): An instance of a rendering
decorator to be applied.
"""
if not isinstance(decorators, list):
raise TypeError(f"decorators must be a list, not '{type(decorators)}'")

Check warning on line 85 in command_line_assistant/rendering/base.py

View check run for this annotation

Codecov / codecov/patch

command_line_assistant/rendering/base.py#L85

Added line #L85 was not covered by tests

self._decorators[type(decorator)] = decorator
for decorator in decorators:
self._decorators[type(decorator)] = decorator

def _apply_decorators(self, text: str) -> str:
"""Apply all decorators to the text.
Expand Down
Loading

0 comments on commit e784d6a

Please sign in to comment.