-
Notifications
You must be signed in to change notification settings - Fork 8
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
27 changed files
with
1,084 additions
and
227 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,16 @@ | ||
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.""" | ||
decorated_text = self._apply_decorators(text) | ||
self._stream.execute(decorated_text) |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,47 @@ | ||
import sys | ||
|
||
from command_line_assistant.rendering.base import ( | ||
OutputStreamWritter, | ||
) | ||
|
||
|
||
class StderrStream(OutputStreamWritter): | ||
"""Decorator for outputting text to stderr""" | ||
|
||
def __init__(self, end: str = "\n") -> None: | ||
""" | ||
Initialize the stderr decorator. | ||
Args: | ||
end: The string to append after the text (defaults to newline) | ||
""" | ||
super().__init__(stream=sys.stderr, end=end) | ||
|
||
def write(self, text: str) -> None: | ||
"""Write the text to stderr""" | ||
self._stream.write(text + self._end) | ||
|
||
def flush(self) -> None: | ||
"""Flush stderr""" | ||
self._stream.flush() | ||
|
||
|
||
class StdoutStream(OutputStreamWritter): | ||
"""Decorator for outputting text to stdout""" | ||
|
||
def __init__(self, end: str = "\n") -> None: | ||
""" | ||
Initialize the stdout decorator. | ||
Args: | ||
end: The string to append after the text (defaults to newline) | ||
""" | ||
super().__init__(stream=sys.stdout, end=end) | ||
|
||
def write(self, text: str) -> None: | ||
"""Write the text to stdout""" | ||
self._stream.write(text + self._end) | ||
|
||
def flush(self) -> None: | ||
"""Flush stdout""" | ||
self._stream.flush() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.