| from abc import ABC, abstractmethod |
| from collections import deque |
| from collections.abc import Sized |
| from dataclasses import dataclass, field |
| from datetime import timedelta |
| from math import ceil |
| from threading import Event, RLock, Thread |
| from types import TracebackType |
| from typing import ( |
| Any, |
| Callable, |
| Deque, |
| Dict, |
| Iterable, |
| List, |
| NamedTuple, |
| NewType, |
| Optional, |
| Sequence, |
| Tuple, |
| Type, |
| TypeVar, |
| Union, |
| ) |
| |
| from . import filesize, get_console |
| from .console import Console, JustifyMethod, RenderableType, Group |
| from .highlighter import Highlighter |
| from .jupyter import JupyterMixin |
| from .live import Live |
| from .progress_bar import ProgressBar |
| from .spinner import Spinner |
| from .style import StyleType |
| from .table import Column, Table |
| from .text import Text, TextType |
| |
| TaskID = NewType("TaskID", int) |
| |
| ProgressType = TypeVar("ProgressType") |
| |
| GetTimeCallable = Callable[[], float] |
| |
| |
| class _TrackThread(Thread): |
| """A thread to periodically update progress.""" |
| |
| def __init__(self, progress: "Progress", task_id: "TaskID", update_period: float): |
| self.progress = progress |
| self.task_id = task_id |
| self.update_period = update_period |
| self.done = Event() |
| |
| self.completed = 0 |
| super().__init__() |
| |
| def run(self) -> None: |
| task_id = self.task_id |
| advance = self.progress.advance |
| update_period = self.update_period |
| last_completed = 0 |
| wait = self.done.wait |
| while not wait(update_period): |
| completed = self.completed |
| if last_completed != completed: |
| advance(task_id, completed - last_completed) |
| last_completed = completed |
| |
| self.progress.update(self.task_id, completed=self.completed, refresh=True) |
| |
| def __enter__(self) -> "_TrackThread": |
| self.start() |
| return self |
| |
| def __exit__( |
| self, |
| exc_type: Optional[Type[BaseException]], |
| exc_val: Optional[BaseException], |
| exc_tb: Optional[TracebackType], |
| ) -> None: |
| self.done.set() |
| self.join() |
| |
| |
| def track( |
| sequence: Union[Sequence[ProgressType], Iterable[ProgressType]], |
| description: str = "Working...", |
| total: Optional[float] = None, |
| auto_refresh: bool = True, |
| console: Optional[Console] = None, |
| transient: bool = False, |
| get_time: Optional[Callable[[], float]] = None, |
| refresh_per_second: float = 10, |
| style: StyleType = "bar.back", |
| complete_style: StyleType = "bar.complete", |
| finished_style: StyleType = "bar.finished", |
| pulse_style: StyleType = "bar.pulse", |
| update_period: float = 0.1, |
| disable: bool = False, |
| ) -> Iterable[ProgressType]: |
| """Track progress by iterating over a sequence. |
| |
| Args: |
| sequence (Iterable[ProgressType]): A sequence (must support "len") you wish to iterate over. |
| description (str, optional): Description of task show next to progress bar. Defaults to "Working". |
| total: (float, optional): Total number of steps. Default is len(sequence). |
| auto_refresh (bool, optional): Automatic refresh, disable to force a refresh after each iteration. Default is True. |
| transient: (bool, optional): Clear the progress on exit. Defaults to False. |
| console (Console, optional): Console to write to. Default creates internal Console instance. |
| refresh_per_second (float): Number of times per second to refresh the progress information. Defaults to 10. |
| style (StyleType, optional): Style for the bar background. Defaults to "bar.back". |
| complete_style (StyleType, optional): Style for the completed bar. Defaults to "bar.complete". |
| finished_style (StyleType, optional): Style for a finished bar. Defaults to "bar.done". |
| pulse_style (StyleType, optional): Style for pulsing bars. Defaults to "bar.pulse". |
| update_period (float, optional): Minimum time (in seconds) between calls to update(). Defaults to 0.1. |
| disable (bool, optional): Disable display of progress. |
| Returns: |
| Iterable[ProgressType]: An iterable of the values in the sequence. |
| |
| """ |
| |
| columns: List["ProgressColumn"] = ( |
| [TextColumn("[progress.description]{task.description}")] if description else [] |
| ) |
| columns.extend( |
| ( |
| BarColumn( |
| style=style, |
| complete_style=complete_style, |
| finished_style=finished_style, |
| pulse_style=pulse_style, |
| ), |
| TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), |
| TimeRemainingColumn(), |
| ) |
| ) |
| progress = Progress( |
| *columns, |
| auto_refresh=auto_refresh, |
| console=console, |
| transient=transient, |
| get_time=get_time, |
| refresh_per_second=refresh_per_second or 10, |
| disable=disable, |
| ) |
| |
| with progress: |
| yield from progress.track( |
| sequence, total=total, description=description, update_period=update_period |
| ) |
| |
| |
| class ProgressColumn(ABC): |
| """Base class for a widget to use in progress display.""" |
| |
| max_refresh: Optional[float] = None |
| |
| def __init__(self, table_column: Optional[Column] = None) -> None: |
| self._table_column = table_column |
| self._renderable_cache: Dict[TaskID, Tuple[float, RenderableType]] = {} |
| self._update_time: Optional[float] = None |
| |
| def get_table_column(self) -> Column: |
| """Get a table column, used to build tasks table.""" |
| return self._table_column or Column() |
| |
| def __call__(self, task: "Task") -> RenderableType: |
| """Called by the Progress object to return a renderable for the given task. |
| |
| Args: |
| task (Task): An object containing information regarding the task. |
| |
| Returns: |
| RenderableType: Anything renderable (including str). |
| """ |
| current_time = task.get_time() |
| if self.max_refresh is not None and not task.completed: |
| try: |
| timestamp, renderable = self._renderable_cache[task.id] |
| except KeyError: |
| pass |
| else: |
| if timestamp + self.max_refresh > current_time: |
| return renderable |
| |
| renderable = self.render(task) |
| self._renderable_cache[task.id] = (current_time, renderable) |
| return renderable |
| |
| @abstractmethod |
| def render(self, task: "Task") -> RenderableType: |
| """Should return a renderable object.""" |
| |
| |
| class RenderableColumn(ProgressColumn): |
| """A column to insert an arbitrary column. |
| |
| Args: |
| renderable (RenderableType, optional): Any renderable. Defaults to empty string. |
| """ |
| |
| def __init__( |
| self, renderable: RenderableType = "", *, table_column: Optional[Column] = None |
| ): |
| self.renderable = renderable |
| super().__init__(table_column=table_column) |
| |
| def render(self, task: "Task") -> RenderableType: |
| return self.renderable |
| |
| |
| class SpinnerColumn(ProgressColumn): |
| """A column with a 'spinner' animation. |
| |
| Args: |
| spinner_name (str, optional): Name of spinner animation. Defaults to "dots". |
| style (StyleType, optional): Style of spinner. Defaults to "progress.spinner". |
| speed (float, optional): Speed factor of spinner. Defaults to 1.0. |
| finished_text (TextType, optional): Text used when task is finished. Defaults to " ". |
| """ |
| |
| def __init__( |
| self, |
| spinner_name: str = "dots", |
| style: Optional[StyleType] = "progress.spinner", |
| speed: float = 1.0, |
| finished_text: TextType = " ", |
| table_column: Optional[Column] = None, |
| ): |
| self.spinner = Spinner(spinner_name, style=style, speed=speed) |
| self.finished_text = ( |
| Text.from_markup(finished_text) |
| if isinstance(finished_text, str) |
| else finished_text |
| ) |
| super().__init__(table_column=table_column) |
| |
| def set_spinner( |
| self, |
| spinner_name: str, |
| spinner_style: Optional[StyleType] = "progress.spinner", |
| speed: float = 1.0, |
| ) -> None: |
| """Set a new spinner. |
| |
| Args: |
| spinner_name (str): Spinner name, see python -m rich.spinner. |
| spinner_style (Optional[StyleType], optional): Spinner style. Defaults to "progress.spinner". |
| speed (float, optional): Speed factor of spinner. Defaults to 1.0. |
| """ |
| self.spinner = Spinner(spinner_name, style=spinner_style, speed=speed) |
| |
| def render(self, task: "Task") -> RenderableType: |
| text = ( |
| self.finished_text |
| if task.finished |
| else self.spinner.render(task.get_time()) |
| ) |
| return text |
| |
| |
| class TextColumn(ProgressColumn): |
| """A column containing text.""" |
| |
| def __init__( |
| self, |
| text_format: str, |
| style: StyleType = "none", |
| justify: JustifyMethod = "left", |
| markup: bool = True, |
| highlighter: Optional[Highlighter] = None, |
| table_column: Optional[Column] = None, |
| ) -> None: |
| self.text_format = text_format |
| self.justify: JustifyMethod = justify |
| self.style = style |
| self.markup = markup |
| self.highlighter = highlighter |
| super().__init__(table_column=table_column or Column(no_wrap=True)) |
| |
| def render(self, task: "Task") -> Text: |
| _text = self.text_format.format(task=task) |
| if self.markup: |
| text = Text.from_markup(_text, style=self.style, justify=self.justify) |
| else: |
| text = Text(_text, style=self.style, justify=self.justify) |
| if self.highlighter: |
| self.highlighter.highlight(text) |
| return text |
| |
| |
| class BarColumn(ProgressColumn): |
| """Renders a visual progress bar. |
| |
| Args: |
| bar_width (Optional[int], optional): Width of bar or None for full width. Defaults to 40. |
| style (StyleType, optional): Style for the bar background. Defaults to "bar.back". |
| complete_style (StyleType, optional): Style for the completed bar. Defaults to "bar.complete". |
| finished_style (StyleType, optional): Style for a finished bar. Defaults to "bar.done". |
| pulse_style (StyleType, optional): Style for pulsing bars. Defaults to "bar.pulse". |
| """ |
| |
| def __init__( |
| self, |
| bar_width: Optional[int] = 40, |
| style: StyleType = "bar.back", |
| complete_style: StyleType = "bar.complete", |
| finished_style: StyleType = "bar.finished", |
| pulse_style: StyleType = "bar.pulse", |
| table_column: Optional[Column] = None, |
| ) -> None: |
| self.bar_width = bar_width |
| self.style = style |
| self.complete_style = complete_style |
| self.finished_style = finished_style |
| self.pulse_style = pulse_style |
| super().__init__(table_column=table_column) |
| |
| def render(self, task: "Task") -> ProgressBar: |
| """Gets a progress bar widget for a task.""" |
| return ProgressBar( |
| total=max(0, task.total), |
| completed=max(0, task.completed), |
| width=None if self.bar_width is None else max(1, self.bar_width), |
| pulse=not task.started, |
| animation_time=task.get_time(), |
| style=self.style, |
| complete_style=self.complete_style, |
| finished_style=self.finished_style, |
| pulse_style=self.pulse_style, |
| ) |
| |
| |
| class TimeElapsedColumn(ProgressColumn): |
| """Renders time elapsed.""" |
| |
| def render(self, task: "Task") -> Text: |
| """Show time remaining.""" |
| elapsed = task.finished_time if task.finished else task.elapsed |
| if elapsed is None: |
| return Text("-:--:--", style="progress.elapsed") |
| delta = timedelta(seconds=int(elapsed)) |
| return Text(str(delta), style="progress.elapsed") |
| |
| |
| class TimeRemainingColumn(ProgressColumn): |
| """Renders estimated time remaining.""" |
| |
| # Only refresh twice a second to prevent jitter |
| max_refresh = 0.5 |
| |
| def render(self, task: "Task") -> Text: |
| """Show time remaining.""" |
| remaining = task.time_remaining |
| if remaining is None: |
| return Text("-:--:--", style="progress.remaining") |
| remaining_delta = timedelta(seconds=int(remaining)) |
| return Text(str(remaining_delta), style="progress.remaining") |
| |
| |
| class FileSizeColumn(ProgressColumn): |
| """Renders completed filesize.""" |
| |
| def render(self, task: "Task") -> Text: |
| """Show data completed.""" |
| data_size = filesize.decimal(int(task.completed)) |
| return Text(data_size, style="progress.filesize") |
| |
| |
| class TotalFileSizeColumn(ProgressColumn): |
| """Renders total filesize.""" |
| |
| def render(self, task: "Task") -> Text: |
| """Show data completed.""" |
| data_size = filesize.decimal(int(task.total)) |
| return Text(data_size, style="progress.filesize.total") |
| |
| |
| class DownloadColumn(ProgressColumn): |
| """Renders file size downloaded and total, e.g. '0.5/2.3 GB'. |
| |
| Args: |
| binary_units (bool, optional): Use binary units, KiB, MiB etc. Defaults to False. |
| """ |
| |
| def __init__( |
| self, binary_units: bool = False, table_column: Optional[Column] = None |
| ) -> None: |
| self.binary_units = binary_units |
| super().__init__(table_column=table_column) |
| |
| def render(self, task: "Task") -> Text: |
| """Calculate common unit for completed and total.""" |
| completed = int(task.completed) |
| total = int(task.total) |
| if self.binary_units: |
| unit, suffix = filesize.pick_unit_and_suffix( |
| total, |
| ["bytes", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB"], |
| 1024, |
| ) |
| else: |
| unit, suffix = filesize.pick_unit_and_suffix( |
| total, ["bytes", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB"], 1000 |
| ) |
| completed_ratio = completed / unit |
| total_ratio = total / unit |
| precision = 0 if unit == 1 else 1 |
| completed_str = f"{completed_ratio:,.{precision}f}" |
| total_str = f"{total_ratio:,.{precision}f}" |
| download_status = f"{completed_str}/{total_str} {suffix}" |
| download_text = Text(download_status, style="progress.download") |
| return download_text |
| |
| |
| class TransferSpeedColumn(ProgressColumn): |
| """Renders human readable transfer speed.""" |
| |
| def render(self, task: "Task") -> Text: |
| """Show data transfer speed.""" |
| speed = task.finished_speed or task.speed |
| if speed is None: |
| return Text("?", style="progress.data.speed") |
| data_speed = filesize.decimal(int(speed)) |
| return Text(f"{data_speed}/s", style="progress.data.speed") |
| |
| |
| class ProgressSample(NamedTuple): |
| """Sample of progress for a given time.""" |
| |
| timestamp: float |
| """Timestamp of sample.""" |
| completed: float |
| """Number of steps completed.""" |
| |
| |
| @dataclass |
| class Task: |
| """Information regarding a progress task. |
| |
| This object should be considered read-only outside of the :class:`~Progress` class. |
| |
| """ |
| |
| id: TaskID |
| """Task ID associated with this task (used in Progress methods).""" |
| |
| description: str |
| """str: Description of the task.""" |
| |
| total: float |
| """str: Total number of steps in this task.""" |
| |
| completed: float |
| """float: Number of steps completed""" |
| |
| _get_time: GetTimeCallable |
| """Callable to get the current time.""" |
| |
| finished_time: Optional[float] = None |
| """float: Time task was finished.""" |
| |
| visible: bool = True |
| """bool: Indicates if this task is visible in the progress display.""" |
| |
| fields: Dict[str, Any] = field(default_factory=dict) |
| """dict: Arbitrary fields passed in via Progress.update.""" |
| |
| start_time: Optional[float] = field(default=None, init=False, repr=False) |
| """Optional[float]: Time this task was started, or None if not started.""" |
| |
| stop_time: Optional[float] = field(default=None, init=False, repr=False) |
| """Optional[float]: Time this task was stopped, or None if not stopped.""" |
| |
| finished_speed: Optional[float] = None |
| """Optional[float]: The last speed for a finished task.""" |
| |
| _progress: Deque[ProgressSample] = field( |
| default_factory=deque, init=False, repr=False |
| ) |
| |
| _lock: RLock = field(repr=False, default_factory=RLock) |
| """Thread lock.""" |
| |
| def get_time(self) -> float: |
| """float: Get the current time, in seconds.""" |
| return self._get_time() |
| |
| @property |
| def started(self) -> bool: |
| """bool: Check if the task as started.""" |
| return self.start_time is not None |
| |
| @property |
| def remaining(self) -> float: |
| """float: Get the number of steps remaining.""" |
| return self.total - self.completed |
| |
| @property |
| def elapsed(self) -> Optional[float]: |
| """Optional[float]: Time elapsed since task was started, or ``None`` if the task hasn't started.""" |
| if self.start_time is None: |
| return None |
| if self.stop_time is not None: |
| return self.stop_time - self.start_time |
| return self.get_time() - self.start_time |
| |
| @property |
| def finished(self) -> bool: |
| """Check if the task has finished.""" |
| return self.finished_time is not None |
| |
| @property |
| def percentage(self) -> float: |
| """float: Get progress of task as a percentage.""" |
| if not self.total: |
| return 0.0 |
| completed = (self.completed / self.total) * 100.0 |
| completed = min(100.0, max(0.0, completed)) |
| return completed |
| |
| @property |
| def speed(self) -> Optional[float]: |
| """Optional[float]: Get the estimated speed in steps per second.""" |
| if self.start_time is None: |
| return None |
| with self._lock: |
| progress = self._progress |
| if not progress: |
| return None |
| total_time = progress[-1].timestamp - progress[0].timestamp |
| if total_time == 0: |
| return None |
| iter_progress = iter(progress) |
| next(iter_progress) |
| total_completed = sum(sample.completed for sample in iter_progress) |
| speed = total_completed / total_time |
| return speed |
| |
| @property |
| def time_remaining(self) -> Optional[float]: |
| """Optional[float]: Get estimated time to completion, or ``None`` if no data.""" |
| if self.finished: |
| return 0.0 |
| speed = self.speed |
| if not speed: |
| return None |
| estimate = ceil(self.remaining / speed) |
| return estimate |
| |
| def _reset(self) -> None: |
| """Reset progress.""" |
| self._progress.clear() |
| self.finished_time = None |
| self.finished_speed = None |
| |
| |
| class Progress(JupyterMixin): |
| """Renders an auto-updating progress bar(s). |
| |
| Args: |
| console (Console, optional): Optional Console instance. Default will an internal Console instance writing to stdout. |
| auto_refresh (bool, optional): Enable auto refresh. If disabled, you will need to call `refresh()`. |
| refresh_per_second (Optional[float], optional): Number of times per second to refresh the progress information or None to use default (10). Defaults to None. |
| speed_estimate_period: (float, optional): Period (in seconds) used to calculate the speed estimate. Defaults to 30. |
| transient: (bool, optional): Clear the progress on exit. Defaults to False. |
| redirect_stdout: (bool, optional): Enable redirection of stdout, so ``print`` may be used. Defaults to True. |
| redirect_stderr: (bool, optional): Enable redirection of stderr. Defaults to True. |
| get_time: (Callable, optional): A callable that gets the current time, or None to use Console.get_time. Defaults to None. |
| disable (bool, optional): Disable progress display. Defaults to False |
| expand (bool, optional): Expand tasks table to fit width. Defaults to False. |
| """ |
| |
| def __init__( |
| self, |
| *columns: Union[str, ProgressColumn], |
| console: Optional[Console] = None, |
| auto_refresh: bool = True, |
| refresh_per_second: float = 10, |
| speed_estimate_period: float = 30.0, |
| transient: bool = False, |
| redirect_stdout: bool = True, |
| redirect_stderr: bool = True, |
| get_time: Optional[GetTimeCallable] = None, |
| disable: bool = False, |
| expand: bool = False, |
| ) -> None: |
| assert ( |
| refresh_per_second is None or refresh_per_second > 0 |
| ), "refresh_per_second must be > 0" |
| self._lock = RLock() |
| self.columns = columns or ( |
| TextColumn("[progress.description]{task.description}"), |
| BarColumn(), |
| TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), |
| TimeRemainingColumn(), |
| ) |
| self.speed_estimate_period = speed_estimate_period |
| |
| self.disable = disable |
| self.expand = expand |
| self._tasks: Dict[TaskID, Task] = {} |
| self._task_index: TaskID = TaskID(0) |
| self.live = Live( |
| console=console or get_console(), |
| auto_refresh=auto_refresh, |
| refresh_per_second=refresh_per_second, |
| transient=transient, |
| redirect_stdout=redirect_stdout, |
| redirect_stderr=redirect_stderr, |
| get_renderable=self.get_renderable, |
| ) |
| self.get_time = get_time or self.console.get_time |
| self.print = self.console.print |
| self.log = self.console.log |
| |
| @property |
| def console(self) -> Console: |
| return self.live.console |
| |
| @property |
| def tasks(self) -> List[Task]: |
| """Get a list of Task instances.""" |
| with self._lock: |
| return list(self._tasks.values()) |
| |
| @property |
| def task_ids(self) -> List[TaskID]: |
| """A list of task IDs.""" |
| with self._lock: |
| return list(self._tasks.keys()) |
| |
| @property |
| def finished(self) -> bool: |
| """Check if all tasks have been completed.""" |
| with self._lock: |
| if not self._tasks: |
| return True |
| return all(task.finished for task in self._tasks.values()) |
| |
| def start(self) -> None: |
| """Start the progress display.""" |
| if not self.disable: |
| self.live.start(refresh=True) |
| |
| def stop(self) -> None: |
| """Stop the progress display.""" |
| self.live.stop() |
| if not self.console.is_interactive: |
| self.console.print() |
| |
| def __enter__(self) -> "Progress": |
| self.start() |
| return self |
| |
| def __exit__( |
| self, |
| exc_type: Optional[Type[BaseException]], |
| exc_val: Optional[BaseException], |
| exc_tb: Optional[TracebackType], |
| ) -> None: |
| self.stop() |
| |
| def track( |
| self, |
| sequence: Union[Iterable[ProgressType], Sequence[ProgressType]], |
| total: Optional[float] = None, |
| task_id: Optional[TaskID] = None, |
| description: str = "Working...", |
| update_period: float = 0.1, |
| ) -> Iterable[ProgressType]: |
| """Track progress by iterating over a sequence. |
| |
| Args: |
| sequence (Sequence[ProgressType]): A sequence of values you want to iterate over and track progress. |
| total: (float, optional): Total number of steps. Default is len(sequence). |
| task_id: (TaskID): Task to track. Default is new task. |
| description: (str, optional): Description of task, if new task is created. |
| update_period (float, optional): Minimum time (in seconds) between calls to update(). Defaults to 0.1. |
| |
| Returns: |
| Iterable[ProgressType]: An iterable of values taken from the provided sequence. |
| """ |
| |
| if total is None: |
| if isinstance(sequence, Sized): |
| task_total = float(len(sequence)) |
| else: |
| raise ValueError( |
| f"unable to get size of {sequence!r}, please specify 'total'" |
| ) |
| else: |
| task_total = total |
| |
| if task_id is None: |
| task_id = self.add_task(description, total=task_total) |
| else: |
| self.update(task_id, total=task_total) |
| |
| if self.live.auto_refresh: |
| with _TrackThread(self, task_id, update_period) as track_thread: |
| for value in sequence: |
| yield value |
| track_thread.completed += 1 |
| else: |
| advance = self.advance |
| refresh = self.refresh |
| for value in sequence: |
| yield value |
| advance(task_id, 1) |
| refresh() |
| |
| def start_task(self, task_id: TaskID) -> None: |
| """Start a task. |
| |
| Starts a task (used when calculating elapsed time). You may need to call this manually, |
| if you called ``add_task`` with ``start=False``. |
| |
| Args: |
| task_id (TaskID): ID of task. |
| """ |
| with self._lock: |
| task = self._tasks[task_id] |
| if task.start_time is None: |
| task.start_time = self.get_time() |
| |
| def stop_task(self, task_id: TaskID) -> None: |
| """Stop a task. |
| |
| This will freeze the elapsed time on the task. |
| |
| Args: |
| task_id (TaskID): ID of task. |
| """ |
| with self._lock: |
| task = self._tasks[task_id] |
| current_time = self.get_time() |
| if task.start_time is None: |
| task.start_time = current_time |
| task.stop_time = current_time |
| |
| def update( |
| self, |
| task_id: TaskID, |
| *, |
| total: Optional[float] = None, |
| completed: Optional[float] = None, |
| advance: Optional[float] = None, |
| description: Optional[str] = None, |
| visible: Optional[bool] = None, |
| refresh: bool = False, |
| **fields: Any, |
| ) -> None: |
| """Update information associated with a task. |
| |
| Args: |
| task_id (TaskID): Task id (returned by add_task). |
| total (float, optional): Updates task.total if not None. |
| completed (float, optional): Updates task.completed if not None. |
| advance (float, optional): Add a value to task.completed if not None. |
| description (str, optional): Change task description if not None. |
| visible (bool, optional): Set visible flag if not None. |
| refresh (bool): Force a refresh of progress information. Default is False. |
| **fields (Any): Additional data fields required for rendering. |
| """ |
| with self._lock: |
| task = self._tasks[task_id] |
| completed_start = task.completed |
| |
| if total is not None and total != task.total: |
| task.total = total |
| task._reset() |
| if advance is not None: |
| task.completed += advance |
| if completed is not None: |
| task.completed = completed |
| if description is not None: |
| task.description = description |
| if visible is not None: |
| task.visible = visible |
| task.fields.update(fields) |
| update_completed = task.completed - completed_start |
| |
| current_time = self.get_time() |
| old_sample_time = current_time - self.speed_estimate_period |
| _progress = task._progress |
| |
| popleft = _progress.popleft |
| while _progress and _progress[0].timestamp < old_sample_time: |
| popleft() |
| while len(_progress) > 1000: |
| popleft() |
| if update_completed > 0: |
| _progress.append(ProgressSample(current_time, update_completed)) |
| if task.completed >= task.total and task.finished_time is None: |
| task.finished_time = task.elapsed |
| |
| if refresh: |
| self.refresh() |
| |
| def reset( |
| self, |
| task_id: TaskID, |
| *, |
| start: bool = True, |
| total: Optional[float] = None, |
| completed: int = 0, |
| visible: Optional[bool] = None, |
| description: Optional[str] = None, |
| **fields: Any, |
| ) -> None: |
| """Reset a task so completed is 0 and the clock is reset. |
| |
| Args: |
| task_id (TaskID): ID of task. |
| start (bool, optional): Start the task after reset. Defaults to True. |
| total (float, optional): New total steps in task, or None to use current total. Defaults to None. |
| completed (int, optional): Number of steps completed. Defaults to 0. |
| **fields (str): Additional data fields required for rendering. |
| """ |
| current_time = self.get_time() |
| with self._lock: |
| task = self._tasks[task_id] |
| task._reset() |
| task.start_time = current_time if start else None |
| if total is not None: |
| task.total = total |
| task.completed = completed |
| if visible is not None: |
| task.visible = visible |
| if fields: |
| task.fields = fields |
| if description is not None: |
| task.description = description |
| task.finished_time = None |
| self.refresh() |
| |
| def advance(self, task_id: TaskID, advance: float = 1) -> None: |
| """Advance task by a number of steps. |
| |
| Args: |
| task_id (TaskID): ID of task. |
| advance (float): Number of steps to advance. Default is 1. |
| """ |
| current_time = self.get_time() |
| with self._lock: |
| task = self._tasks[task_id] |
| completed_start = task.completed |
| task.completed += advance |
| update_completed = task.completed - completed_start |
| old_sample_time = current_time - self.speed_estimate_period |
| _progress = task._progress |
| |
| popleft = _progress.popleft |
| while _progress and _progress[0].timestamp < old_sample_time: |
| popleft() |
| while len(_progress) > 1000: |
| popleft() |
| _progress.append(ProgressSample(current_time, update_completed)) |
| if task.completed >= task.total and task.finished_time is None: |
| task.finished_time = task.elapsed |
| task.finished_speed = task.speed |
| |
| def refresh(self) -> None: |
| """Refresh (render) the progress information.""" |
| if not self.disable and self.live.is_started: |
| self.live.refresh() |
| |
| def get_renderable(self) -> RenderableType: |
| """Get a renderable for the progress display.""" |
| renderable = Group(*self.get_renderables()) |
| return renderable |
| |
| def get_renderables(self) -> Iterable[RenderableType]: |
| """Get a number of renderables for the progress display.""" |
| table = self.make_tasks_table(self.tasks) |
| yield table |
| |
| def make_tasks_table(self, tasks: Iterable[Task]) -> Table: |
| """Get a table to render the Progress display. |
| |
| Args: |
| tasks (Iterable[Task]): An iterable of Task instances, one per row of the table. |
| |
| Returns: |
| Table: A table instance. |
| """ |
| table_columns = ( |
| ( |
| Column(no_wrap=True) |
| if isinstance(_column, str) |
| else _column.get_table_column().copy() |
| ) |
| for _column in self.columns |
| ) |
| table = Table.grid(*table_columns, padding=(0, 1), expand=self.expand) |
| |
| for task in tasks: |
| if task.visible: |
| table.add_row( |
| *( |
| ( |
| column.format(task=task) |
| if isinstance(column, str) |
| else column(task) |
| ) |
| for column in self.columns |
| ) |
| ) |
| return table |
| |
| def __rich__(self) -> RenderableType: |
| """Makes the Progress class itself renderable.""" |
| with self._lock: |
| return self.get_renderable() |
| |
| def add_task( |
| self, |
| description: str, |
| start: bool = True, |
| total: float = 100.0, |
| completed: int = 0, |
| visible: bool = True, |
| **fields: Any, |
| ) -> TaskID: |
| """Add a new 'task' to the Progress display. |
| |
| Args: |
| description (str): A description of the task. |
| start (bool, optional): Start the task immediately (to calculate elapsed time). If set to False, |
| you will need to call `start` manually. Defaults to True. |
| total (float, optional): Number of total steps in the progress if know. Defaults to 100. |
| completed (int, optional): Number of steps completed so far.. Defaults to 0. |
| visible (bool, optional): Enable display of the task. Defaults to True. |
| **fields (str): Additional data fields required for rendering. |
| |
| Returns: |
| TaskID: An ID you can use when calling `update`. |
| """ |
| with self._lock: |
| task = Task( |
| self._task_index, |
| description, |
| total, |
| completed, |
| visible=visible, |
| fields=fields, |
| _get_time=self.get_time, |
| _lock=self._lock, |
| ) |
| self._tasks[self._task_index] = task |
| if start: |
| self.start_task(self._task_index) |
| new_task_index = self._task_index |
| self._task_index = TaskID(int(self._task_index) + 1) |
| self.refresh() |
| return new_task_index |
| |
| def remove_task(self, task_id: TaskID) -> None: |
| """Delete a task if it exists. |
| |
| Args: |
| task_id (TaskID): A task ID. |
| |
| """ |
| with self._lock: |
| del self._tasks[task_id] |
| |
| |
| if __name__ == "__main__": # pragma: no coverage |
| |
| import random |
| import time |
| |
| from .panel import Panel |
| from .rule import Rule |
| from .syntax import Syntax |
| from .table import Table |
| |
| syntax = Syntax( |
| '''def loop_last(values: Iterable[T]) -> Iterable[Tuple[bool, T]]: |
| """Iterate and generate a tuple with a flag for last value.""" |
| iter_values = iter(values) |
| try: |
| previous_value = next(iter_values) |
| except StopIteration: |
| return |
| for value in iter_values: |
| yield False, previous_value |
| previous_value = value |
| yield True, previous_value''', |
| "python", |
| line_numbers=True, |
| ) |
| |
| table = Table("foo", "bar", "baz") |
| table.add_row("1", "2", "3") |
| |
| progress_renderables = [ |
| "Text may be printed while the progress bars are rendering.", |
| Panel("In fact, [i]any[/i] renderable will work"), |
| "Such as [magenta]tables[/]...", |
| table, |
| "Pretty printed structures...", |
| {"type": "example", "text": "Pretty printed"}, |
| "Syntax...", |
| syntax, |
| Rule("Give it a try!"), |
| ] |
| |
| from itertools import cycle |
| |
| examples = cycle(progress_renderables) |
| |
| console = Console(record=True) |
| |
| with Progress( |
| SpinnerColumn(), |
| TextColumn("[progress.description]{task.description}"), |
| BarColumn(), |
| TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), |
| TimeRemainingColumn(), |
| TimeElapsedColumn(), |
| console=console, |
| transient=True, |
| ) as progress: |
| |
| task1 = progress.add_task("[red]Downloading", total=1000) |
| task2 = progress.add_task("[green]Processing", total=1000) |
| task3 = progress.add_task("[yellow]Thinking", total=1000, start=False) |
| |
| while not progress.finished: |
| progress.update(task1, advance=0.5) |
| progress.update(task2, advance=0.3) |
| time.sleep(0.01) |
| if random.randint(0, 100) < 1: |
| progress.log(next(examples)) |
