create LiveProgressReporter

This commit is contained in:
Sina Atalay 2024-02-05 19:41:43 +01:00
parent 382914eee9
commit 0b09a5095b
2 changed files with 41 additions and 73 deletions

View File

@ -5,26 +5,15 @@ from typing import Annotated
import typer import typer
from rich.prompt import Prompt from rich.prompt import Prompt
import ruamel.yaml import ruamel.yaml
from rich.console import Console, Group
from rich.panel import Panel
from rich.live import Live
from rich.progress import (
BarColumn,
Progress,
ProgressColumn,
TextColumn,
Task,
)
from rich.text import Text
from .user_communicator import handle_exceptions, welcome, LiveProgress from . import user_communicator as uc
from . import data_models as dm from . import data_models as dm
from . import renderer as r from . import renderer as r
app = typer.Typer( app = typer.Typer(
callback=welcome(), # callback=uc.print_rendercv_graphics(),
help="RenderCV - A LateX CV generator from YAML", help="RenderCV - A LateX CV generator from YAML",
rich_markup_mode=( # see https://typer.tiangolo.com/tutorial/commands/help/#rich-markdown rich_markup_mode=( # see https://typer.tiangolo.com/tutorial/commands/help/#rich-markdown
"markdown" "markdown"
@ -33,7 +22,7 @@ app = typer.Typer(
@app.command(help="Render a YAML input file") @app.command(help="Render a YAML input file")
@handle_exceptions @uc.handle_exceptions
def render( def render(
input_file_path: Annotated[ input_file_path: Annotated[
pathlib.Path, pathlib.Path,
@ -46,50 +35,18 @@ def render(
input_file (str): Name of the YAML input file input_file (str): Name of the YAML input file
""" """
output_directory = input_file_path.parent / "rendercv_output" output_directory = input_file_path.parent / "rendercv_output"
with uc.LiveProgressReporter(number_of_steps=3) as progress:
number_of_steps = 3
class TimeElapsedColumn(ProgressColumn):
"""Renders time elapsed."""
def render(self, task: "Task") -> Text:
"""Show time elapsed."""
elapsed = task.finished_time if task.finished else task.elapsed
if elapsed is None:
return Text("--.-", style="progress.elapsed")
delta = f"{elapsed:.1f} s"
return Text(str(delta), style="progress.elapsed")
step_progress = Progress(TimeElapsedColumn(), TextColumn("{task.description}"))
# overall progress bar
overall_progress = Progress(
TimeElapsedColumn(),
BarColumn(),
TextColumn("{task.description}"),
)
# group of progress bars;
# some are always visible, others will disappear when progress is complete
group = Group(
Panel(Group(step_progress)),
overall_progress,
)
overall_task_id = overall_progress.add_task("", total=number_of_steps)
overall_progress.update(
overall_task_id,
description=f"[bold #AAAAAA](0 out of 3 steps finished)",
)
with LiveProgress(
step_progress, overall_progress, group, overall_task_id
) as progress:
progress.start_a_step("Reading the input file") progress.start_a_step("Reading the input file")
data_model = dm.read_input_file(input_file_path) data_model = dm.read_input_file(input_file_path)
progress.finish_the_current_step() progress.finish_the_current_step()
progress.start_a_step("Generating the LaTeX file")
latex_file_path = r.generate_latex_file(data_model, output_directory) latex_file_path = r.generate_latex_file(data_model, output_directory)
progress.finish_the_current_step()
progress.start_a_step("Converting the LaTeX file to PDF")
r.latex_to_pdf(latex_file_path) r.latex_to_pdf(latex_file_path)
progress.finish_the_current_step()
@app.command(help="Generate a YAML input file to get started.") @app.command(help="Generate a YAML input file to get started.")

View File

@ -19,7 +19,7 @@ import pydantic
error_console = Console(stderr=True) error_console = Console(stderr=True)
def welcome(): def print_rendercv_graphics():
"""Print a welcome message to the terminal.""" """Print a welcome message to the terminal."""
print("Welcome to [bold blue]RenderCV[/bold blue]!") print("Welcome to [bold blue]RenderCV[/bold blue]!")
print("Documentation: [link=https://sinaatalay.github.io/rendercv/]") print("Documentation: [link=https://sinaatalay.github.io/rendercv/]")
@ -91,35 +91,43 @@ def handle_exceptions(function: Callable) -> Callable:
return wrapper return wrapper
class LiveProgressReporter(Live):
def __init__(self, number_of_steps: int):
class TimeElapsedColumn(ProgressColumn): class TimeElapsedColumn(ProgressColumn):
"""Renders time elapsed."""
def render(self, task: "Task") -> Text: def render(self, task: "Task") -> Text:
"""Show time elapsed."""
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 Text("--.-", style="progress.elapsed")
delta = f"{elapsed:.1f} s" delta = f"{elapsed:.1f} s"
return Text(str(delta), style="progress.elapsed") return Text(str(delta), style="progress.elapsed")
self.step_progress = Progress(
TimeElapsedColumn(), TextColumn("{task.description}")
)
class LiveProgress(Live): self.overall_progress = Progress(
def __init__(self, step_progress, overall_progress, group, overall_task_id): TimeElapsedColumn(),
super().__init__() BarColumn(),
self.step_progress = step_progress TextColumn("{task.description}"),
self.overall_progress = overall_progress )
self.group = group
self.overall_task_id = overall_task_id self.group = Group(
self.number_of_tasks = 3 Panel(Group(self.step_progress)),
self.overall_progress,
)
self.overall_task_id = self.overall_progress.add_task("", total=number_of_steps)
self.number_of_steps = number_of_steps
self.current_step = 0
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_tasks} steps finished)" f"[bold #AAAAAA](0 out of {self.number_of_steps} steps finished)"
), ),
) )
super().__init__(self.group)
def __enter__(self) -> "LiveProgress": def __enter__(self) -> "LiveProgressReporter":
self.start(refresh=self._renderable is not None) self.start(refresh=self._renderable is not None)
return self return self
@ -135,6 +143,9 @@ class LiveProgress(Live):
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.overall_progress.update(self.overall_task_id, advance=1)
self.current_step += 1
if self.current_step == self.number_of_steps:
self.end()
def end(self): def end(self):
self.overall_progress.update( self.overall_progress.update(