finalize user_communicator

This commit is contained in:
Sina Atalay 2024-02-06 17:36:12 +01:00
parent 0b09a5095b
commit 6eac49f38e
1 changed files with 61 additions and 38 deletions

View File

@ -1,45 +1,49 @@
import time import time
from typing import Callable from typing import Callable
from rich import print import rich
from rich.console import Console, Group import rich.console
from rich.panel import Panel import rich.panel
from rich.live import Live import rich.live
from rich.progress import ( import rich.table
BarColumn, import rich.text
Progress, import rich.progress
ProgressColumn,
TextColumn,
Task,
)
from rich.text import Text
import pydantic
error_console = Console(stderr=True)
def print_rendercv_graphics(): console = rich.console.Console()
def welcome():
"""Print a welcome message to the terminal.""" """Print a welcome message to the terminal."""
print("Welcome to [bold blue]RenderCV[/bold blue]!") table = rich.table.Table(
print("Documentation: [link=https://sinaatalay.github.io/rendercv/]") title="\nWelcome to [bold]Render[dodger_blue3]CV[/dodger_blue3][/bold]!",
print("Source code: [link=https://github.com/sinaatalay/rendercv/]") title_justify="center",
print("Report bugs: [link=https://github.com/sinaatalay/rendercv/issues/]") )
table.add_column("Title", style="magenta")
table.add_column("Link", style="cyan", justify="right", no_wrap=True)
table.add_row("Documentation", "https://sinaatalay.github.io/rendercv/")
table.add_row("Source code", "https://github.com/sinaatalay/rendercv/")
table.add_row("Bug reports", "https://github.com/sinaatalay/rendercv/issues/")
table.add_row("Feature requests", "https://github.com/sinaatalay/rendercv/issues/")
console.print(table, justify="center")
def warning(text): def warning(text):
"""Print a warning message to the terminal.""" """Print a warning message to the terminal."""
print(f"[bold yellow]{text}[/bold yellow]") console.print(f"[bold yellow]{text}[/bold yellow]")
def error(text): def error(text):
"""Print an error message to the terminal.""" """Print an error message to the terminal."""
error_console.print(f"[bold red]{text}[/bold red]") console.print(f"[bold red]{text}[/bold red]")
def information(text): def information(text):
"""Print an information message to the terminal.""" """Print an information message to the terminal."""
print(f"[bold blue]{text}[/bold blue]") console.print(f"[bold green]{text}")
def time_the_event_below(event_name: str) -> Callable: def time_the_event_below(event_name: str) -> Callable:
@ -91,28 +95,35 @@ def handle_exceptions(function: Callable) -> Callable:
return wrapper return wrapper
class LiveProgressReporter(Live): class LiveProgressReporter(rich.live.Live):
"""This class is a wrapper around `rich.live.Live` that provides the live progress
reporting functionality.
Args:
number_of_steps (int): The number of steps to be finished.
"""
def __init__(self, number_of_steps: int): def __init__(self, number_of_steps: int):
class TimeElapsedColumn(ProgressColumn): class TimeElapsedColumn(rich.progress.ProgressColumn):
def render(self, task: "Task") -> Text: def render(self, task: "rich.progress.Task") -> rich.text.Text:
elapsed = task.finished_time if task.finished else task.elapsed elapsed = task.finished_time if task.finished else task.elapsed
if elapsed is None: if elapsed is None:
return Text("--.-", style="progress.elapsed") return rich.text.Text("--.-", style="progress.elapsed")
delta = f"{elapsed:.1f} s" delta = f"{elapsed:.1f} s"
return Text(str(delta), style="progress.elapsed") return rich.text.Text(str(delta), style="progress.elapsed")
self.step_progress = Progress( self.step_progress = rich.progress.Progress(
TimeElapsedColumn(), TextColumn("{task.description}") TimeElapsedColumn(), rich.progress.TextColumn("{task.description}")
) )
self.overall_progress = Progress( self.overall_progress = rich.progress.Progress(
TimeElapsedColumn(), TimeElapsedColumn(),
BarColumn(), rich.progress.BarColumn(),
TextColumn("{task.description}"), rich.progress.TextColumn("{task.description}"),
) )
self.group = Group( self.group = rich.console.Group(
Panel(Group(self.step_progress)), rich.panel.Panel(rich.console.Group(self.step_progress)),
self.overall_progress, self.overall_progress,
) )
@ -122,32 +133,44 @@ class LiveProgressReporter(Live):
self.overall_progress.update( self.overall_progress.update(
self.overall_task_id, self.overall_task_id,
description=( description=(
f"[bold #AAAAAA](0 out of {self.number_of_steps} steps finished)" f"[bold #AAAAAA]({self.current_step} out of"
f" {self.number_of_steps} steps finished)"
), ),
) )
super().__init__(self.group) super().__init__(self.group)
def __enter__(self) -> "LiveProgressReporter": def __enter__(self) -> "LiveProgressReporter":
"""Overwrite the `__enter__` method for the correct return type."""
self.start(refresh=self._renderable is not None) self.start(refresh=self._renderable is not None)
return self return self
def start_a_step(self, step_name: str): def start_a_step(self, step_name: str):
"""Start a step and update the progress bars."""
self.current_step_name = step_name self.current_step_name = step_name
self.current_step_id = self.step_progress.add_task( self.current_step_id = self.step_progress.add_task(
f"{self.current_step_name} has started." f"{self.current_step_name} has started."
) )
def finish_the_current_step(self): def finish_the_current_step(self):
"""Finish the current step and update the progress bars."""
self.step_progress.stop_task(self.current_step_id) self.step_progress.stop_task(self.current_step_id)
self.step_progress.update( self.step_progress.update(
self.current_step_id, description=f"{self.current_step_name} has finished." self.current_step_id, description=f"{self.current_step_name} has finished."
) )
self.overall_progress.update(self.overall_task_id, advance=1)
self.current_step += 1 self.current_step += 1
self.overall_progress.update(
self.overall_task_id,
description=(
f"[bold #AAAAAA]({self.current_step} out of"
f" {self.number_of_steps} steps finished)"
),
advance=1,
)
if self.current_step == self.number_of_steps: if self.current_step == self.number_of_steps:
self.end() self.end()
def end(self): def end(self):
"""End the live progress reporting."""
self.overall_progress.update( self.overall_progress.update(
self.overall_task_id, self.overall_task_id,
description="[bold green]Your CV is rendered!", description="[bold green]Your CV is rendered!",