summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/rich/progress.py
diff options
context:
space:
mode:
authorcyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
committercyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
commit12cf076118570eebbff08c6b3090e0d4798447a1 (patch)
tree3ba25e17e3c3a5e82316558ba3864b955919ff72 /venv/lib/python3.11/site-packages/rich/progress.py
parentc45662ff3923b34614ddcc8feb9195541166dcc5 (diff)
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/rich/progress.py')
-rw-r--r--venv/lib/python3.11/site-packages/rich/progress.py1699
1 files changed, 0 insertions, 1699 deletions
diff --git a/venv/lib/python3.11/site-packages/rich/progress.py b/venv/lib/python3.11/site-packages/rich/progress.py
deleted file mode 100644
index 8810aea..0000000
--- a/venv/lib/python3.11/site-packages/rich/progress.py
+++ /dev/null
@@ -1,1699 +0,0 @@
-import io
-import sys
-import typing
-import warnings
-from abc import ABC, abstractmethod
-from collections import deque
-from dataclasses import dataclass, field
-from datetime import timedelta
-from io import RawIOBase, UnsupportedOperation
-from math import ceil
-from mmap import mmap
-from operator import length_hint
-from os import PathLike, stat
-from threading import Event, RLock, Thread
-from types import TracebackType
-from typing import (
- Any,
- BinaryIO,
- Callable,
- ContextManager,
- Deque,
- Dict,
- Generic,
- Iterable,
- List,
- NamedTuple,
- NewType,
- Optional,
- Sequence,
- TextIO,
- Tuple,
- Type,
- TypeVar,
- Union,
-)
-
-if sys.version_info >= (3, 8):
- from typing import Literal
-else:
- from typing_extensions import Literal # pragma: no cover
-
-from . import filesize, get_console
-from .console import Console, Group, JustifyMethod, RenderableType
-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]
-
-
-_I = typing.TypeVar("_I", TextIO, BinaryIO)
-
-
-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,
- show_speed: bool = True,
-) -> 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.finished".
- 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.
- show_speed (bool, optional): Show speed if total isn't known. Defaults to True.
- 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,
- ),
- TaskProgressColumn(show_speed=show_speed),
- TimeRemainingColumn(elapsed_when_finished=True),
- )
- )
- 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 _Reader(RawIOBase, BinaryIO):
- """A reader that tracks progress while it's being read from."""
-
- def __init__(
- self,
- handle: BinaryIO,
- progress: "Progress",
- task: TaskID,
- close_handle: bool = True,
- ) -> None:
- self.handle = handle
- self.progress = progress
- self.task = task
- self.close_handle = close_handle
- self._closed = False
-
- def __enter__(self) -> "_Reader":
- self.handle.__enter__()
- return self
-
- def __exit__(
- self,
- exc_type: Optional[Type[BaseException]],
- exc_val: Optional[BaseException],
- exc_tb: Optional[TracebackType],
- ) -> None:
- self.close()
-
- def __iter__(self) -> BinaryIO:
- return self
-
- def __next__(self) -> bytes:
- line = next(self.handle)
- self.progress.advance(self.task, advance=len(line))
- return line
-
- @property
- def closed(self) -> bool:
- return self._closed
-
- def fileno(self) -> int:
- return self.handle.fileno()
-
- def isatty(self) -> bool:
- return self.handle.isatty()
-
- @property
- def mode(self) -> str:
- return self.handle.mode
-
- @property
- def name(self) -> str:
- return self.handle.name
-
- def readable(self) -> bool:
- return self.handle.readable()
-
- def seekable(self) -> bool:
- return self.handle.seekable()
-
- def writable(self) -> bool:
- return False
-
- def read(self, size: int = -1) -> bytes:
- block = self.handle.read(size)
- self.progress.advance(self.task, advance=len(block))
- return block
-
- def readinto(self, b: Union[bytearray, memoryview, mmap]): # type: ignore[no-untyped-def, override]
- n = self.handle.readinto(b) # type: ignore[attr-defined]
- self.progress.advance(self.task, advance=n)
- return n
-
- def readline(self, size: int = -1) -> bytes: # type: ignore[override]
- line = self.handle.readline(size)
- self.progress.advance(self.task, advance=len(line))
- return line
-
- def readlines(self, hint: int = -1) -> List[bytes]:
- lines = self.handle.readlines(hint)
- self.progress.advance(self.task, advance=sum(map(len, lines)))
- return lines
-
- def close(self) -> None:
- if self.close_handle:
- self.handle.close()
- self._closed = True
-
- def seek(self, offset: int, whence: int = 0) -> int:
- pos = self.handle.seek(offset, whence)
- self.progress.update(self.task, completed=pos)
- return pos
-
- def tell(self) -> int:
- return self.handle.tell()
-
- def write(self, s: Any) -> int:
- raise UnsupportedOperation("write")
-
-
-class _ReadContext(ContextManager[_I], Generic[_I]):
- """A utility class to handle a context for both a reader and a progress."""
-
- def __init__(self, progress: "Progress", reader: _I) -> None:
- self.progress = progress
- self.reader: _I = reader
-
- def __enter__(self) -> _I:
- self.progress.start()
- return self.reader.__enter__()
-
- def __exit__(
- self,
- exc_type: Optional[Type[BaseException]],
- exc_val: Optional[BaseException],
- exc_tb: Optional[TracebackType],
- ) -> None:
- self.progress.stop()
- self.reader.__exit__(exc_type, exc_val, exc_tb)
-
-
-def wrap_file(
- file: BinaryIO,
- total: int,
- *,
- description: str = "Reading...",
- 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",
- disable: bool = False,
-) -> ContextManager[BinaryIO]:
- """Read bytes from a file while tracking progress.
-
- Args:
- file (Union[str, PathLike[str], BinaryIO]): The path to the file to read, or a file-like object in binary mode.
- total (int): Total number of bytes to read.
- description (str, optional): Description of task show next to progress bar. Defaults to "Reading".
- 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.finished".
- pulse_style (StyleType, optional): Style for pulsing bars. Defaults to "bar.pulse".
- disable (bool, optional): Disable display of progress.
- Returns:
- ContextManager[BinaryIO]: A context manager yielding a progress reader.
-
- """
-
- 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,
- ),
- DownloadColumn(),
- 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,
- )
-
- reader = progress.wrap_file(file, total=total, description=description)
- return _ReadContext(progress, reader)
-
-
-@typing.overload
-def open(
- file: Union[str, "PathLike[str]", bytes],
- mode: Union[Literal["rt"], Literal["r"]],
- buffering: int = -1,
- encoding: Optional[str] = None,
- errors: Optional[str] = None,
- newline: Optional[str] = None,
- *,
- total: Optional[int] = None,
- description: str = "Reading...",
- 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",
- disable: bool = False,
-) -> ContextManager[TextIO]:
- pass
-
-
-@typing.overload
-def open(
- file: Union[str, "PathLike[str]", bytes],
- mode: Literal["rb"],
- buffering: int = -1,
- encoding: Optional[str] = None,
- errors: Optional[str] = None,
- newline: Optional[str] = None,
- *,
- total: Optional[int] = None,
- description: str = "Reading...",
- 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",
- disable: bool = False,
-) -> ContextManager[BinaryIO]:
- pass
-
-
-def open(
- file: Union[str, "PathLike[str]", bytes],
- mode: Union[Literal["rb"], Literal["rt"], Literal["r"]] = "r",
- buffering: int = -1,
- encoding: Optional[str] = None,
- errors: Optional[str] = None,
- newline: Optional[str] = None,
- *,
- total: Optional[int] = None,
- description: str = "Reading...",
- 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",
- disable: bool = False,
-) -> Union[ContextManager[BinaryIO], ContextManager[TextIO]]:
- """Read bytes from a file while tracking progress.
-
- Args:
- path (Union[str, PathLike[str], BinaryIO]): The path to the file to read, or a file-like object in binary mode.
- mode (str): The mode to use to open the file. Only supports "r", "rb" or "rt".
- buffering (int): The buffering strategy to use, see :func:`io.open`.
- encoding (str, optional): The encoding to use when reading in text mode, see :func:`io.open`.
- errors (str, optional): The error handling strategy for decoding errors, see :func:`io.open`.
- newline (str, optional): The strategy for handling newlines in text mode, see :func:`io.open`
- total: (int, optional): Total number of bytes to read. Must be provided if reading from a file handle. Default for a path is os.stat(file).st_size.
- description (str, optional): Description of task show next to progress bar. Defaults to "Reading".
- 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.finished".
- pulse_style (StyleType, optional): Style for pulsing bars. Defaults to "bar.pulse".
- disable (bool, optional): Disable display of progress.
- encoding (str, optional): The encoding to use when reading in text mode.
-
- Returns:
- ContextManager[BinaryIO]: A context manager yielding a progress reader.
-
- """
-
- 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,
- ),
- DownloadColumn(),
- 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,
- )
-
- reader = progress.open(
- file,
- mode=mode,
- buffering=buffering,
- encoding=encoding,
- errors=errors,
- newline=newline,
- total=total,
- description=description,
- )
- return _ReadContext(progress, reader) # type: ignore[return-value, type-var]
-
-
-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.finished".
- 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) if task.total is not None else None,
- 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 elapsed."""
- elapsed = task.finished_time if task.finished else task.elapsed
- if elapsed is None:
- return Text("-:--:--", style="progress.elapsed")
- delta = timedelta(seconds=max(0, int(elapsed)))
- return Text(str(delta), style="progress.elapsed")
-
-
-class TaskProgressColumn(TextColumn):
- """Show task progress as a percentage.
-
- Args:
- text_format (str, optional): Format for percentage display. Defaults to "[progress.percentage]{task.percentage:>3.0f}%".
- text_format_no_percentage (str, optional): Format if percentage is unknown. Defaults to "".
- style (StyleType, optional): Style of output. Defaults to "none".
- justify (JustifyMethod, optional): Text justification. Defaults to "left".
- markup (bool, optional): Enable markup. Defaults to True.
- highlighter (Optional[Highlighter], optional): Highlighter to apply to output. Defaults to None.
- table_column (Optional[Column], optional): Table Column to use. Defaults to None.
- show_speed (bool, optional): Show speed if total is unknown. Defaults to False.
- """
-
- def __init__(
- self,
- text_format: str = "[progress.percentage]{task.percentage:>3.0f}%",
- text_format_no_percentage: str = "",
- style: StyleType = "none",
- justify: JustifyMethod = "left",
- markup: bool = True,
- highlighter: Optional[Highlighter] = None,
- table_column: Optional[Column] = None,
- show_speed: bool = False,
- ) -> None:
- self.text_format_no_percentage = text_format_no_percentage
- self.show_speed = show_speed
- super().__init__(
- text_format=text_format,
- style=style,
- justify=justify,
- markup=markup,
- highlighter=highlighter,
- table_column=table_column,
- )
-
- @classmethod
- def render_speed(cls, speed: Optional[float]) -> Text:
- """Render the speed in iterations per second.
-
- Args:
- task (Task): A Task object.
-
- Returns:
- Text: Text object containing the task speed.
- """
- if speed is None:
- return Text("", style="progress.percentage")
- unit, suffix = filesize.pick_unit_and_suffix(
- int(speed),
- ["", "×10³", "×10⁶", "×10⁹", "×10¹²"],
- 1000,
- )
- data_speed = speed / unit
- return Text(f"{data_speed:.1f}{suffix} it/s", style="progress.percentage")
-
- def render(self, task: "Task") -> Text:
- if task.total is None and self.show_speed:
- return self.render_speed(task.finished_speed or task.speed)
- text_format = (
- self.text_format_no_percentage if task.total is None else self.text_format
- )
- _text = 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 TimeRemainingColumn(ProgressColumn):
- """Renders estimated time remaining.
-
- Args:
- compact (bool, optional): Render MM:SS when time remaining is less than an hour. Defaults to False.
- elapsed_when_finished (bool, optional): Render time elapsed when the task is finished. Defaults to False.
- """
-
- # Only refresh twice a second to prevent jitter
- max_refresh = 0.5
-
- def __init__(
- self,
- compact: bool = False,
- elapsed_when_finished: bool = False,
- table_column: Optional[Column] = None,
- ):
- self.compact = compact
- self.elapsed_when_finished = elapsed_when_finished
- super().__init__(table_column=table_column)
-
- def render(self, task: "Task") -> Text:
- """Show time remaining."""
- if self.elapsed_when_finished and task.finished:
- task_time = task.finished_time
- style = "progress.elapsed"
- else:
- task_time = task.time_remaining
- style = "progress.remaining"
-
- if task.total is None:
- return Text("", style=style)
-
- if task_time is None:
- return Text("--:--" if self.compact else "-:--:--", style=style)
-
- # Based on https://github.com/tqdm/tqdm/blob/master/tqdm/std.py
- minutes, seconds = divmod(int(task_time), 60)
- hours, minutes = divmod(minutes, 60)
-
- if self.compact and not hours:
- formatted = f"{minutes:02d}:{seconds:02d}"
- else:
- formatted = f"{hours:d}:{minutes:02d}:{seconds:02d}"
-
- return Text(formatted, style=style)
-
-
-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)) if task.total is not None else ""
- return Text(data_size, style="progress.filesize.total")
-
-
-class MofNCompleteColumn(ProgressColumn):
- """Renders completed count/total, e.g. ' 10/1000'.
-
- Best for bounded tasks with int quantities.
-
- Space pads the completed count so that progress length does not change as task progresses
- past powers of 10.
-
- Args:
- separator (str, optional): Text to separate completed and total values. Defaults to "/".
- """
-
- def __init__(self, separator: str = "/", table_column: Optional[Column] = None):
- self.separator = separator
- super().__init__(table_column=table_column)
-
- def render(self, task: "Task") -> Text:
- """Show completed/total."""
- completed = int(task.completed)
- total = int(task.total) if task.total is not None else "?"
- total_width = len(str(total))
- return Text(
- f"{completed:{total_width}d}{self.separator}{total}",
- style="progress.download",
- )
-
-
-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)
-
- unit_and_suffix_calculation_base = (
- int(task.total) if task.total is not None else completed
- )
- if self.binary_units:
- unit, suffix = filesize.pick_unit_and_suffix(
- unit_and_suffix_calculation_base,
- ["bytes", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB"],
- 1024,
- )
- else:
- unit, suffix = filesize.pick_unit_and_suffix(
- unit_and_suffix_calculation_base,
- ["bytes", "kB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB"],
- 1000,
- )
- precision = 0 if unit == 1 else 1
-
- completed_ratio = completed / unit
- completed_str = f"{completed_ratio:,.{precision}f}"
-
- if task.total is not None:
- total = int(task.total)
- total_ratio = total / unit
- total_str = f"{total_ratio:,.{precision}f}"
- else:
- total_str = "?"
-
- 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: Optional[float]
- """Optional[float]: 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=lambda: deque(maxlen=1000), 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) -> Optional[float]:
- """Optional[float]: Get the number of steps remaining, if a non-None total was set."""
- if self.total is None:
- return None
- 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 a None total was set, returns 0"""
- 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
- remaining = self.remaining
- if remaining is None:
- return None
- estimate = ceil(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 > 0, "refresh_per_second must be > 0"
- self._lock = RLock()
- self.columns = columns or self.get_default_columns()
- 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
-
- @classmethod
- def get_default_columns(cls) -> Tuple[ProgressColumn, ...]:
- """Get the default columns used for a new Progress instance:
- - a text column for the description (TextColumn)
- - the bar itself (BarColumn)
- - a text column showing completion percentage (TextColumn)
- - an estimated-time-remaining column (TimeRemainingColumn)
- If the Progress instance is created without passing a columns argument,
- the default columns defined here will be used.
-
- You can also create a Progress instance using custom columns before
- and/or after the defaults, as in this example:
-
- progress = Progress(
- SpinnerColumn(),
- *Progress.get_default_columns(),
- "Elapsed:",
- TimeElapsedColumn(),
- )
-
- This code shows the creation of a Progress display, containing
- a spinner to the left, the default columns, and a labeled elapsed
- time column.
- """
- return (
- TextColumn("[progress.description]{task.description}"),
- BarColumn(),
- TaskProgressColumn(),
- TimeRemainingColumn(),
- )
-
- @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:
- total = float(length_hint(sequence)) or None
-
- if task_id is None:
- task_id = self.add_task(description, total=total)
- else:
- self.update(task_id, total=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 wrap_file(
- self,
- file: BinaryIO,
- total: Optional[int] = None,
- *,
- task_id: Optional[TaskID] = None,
- description: str = "Reading...",
- ) -> BinaryIO:
- """Track progress file reading from a binary file.
-
- Args:
- file (BinaryIO): A file-like object opened in binary mode.
- total (int, optional): Total number of bytes to read. This must be provided unless a task with a total is also given.
- task_id (TaskID): Task to track. Default is new task.
- description (str, optional): Description of task, if new task is created.
-
- Returns:
- BinaryIO: A readable file-like object in binary mode.
-
- Raises:
- ValueError: When no total value can be extracted from the arguments or the task.
- """
- # attempt to recover the total from the task
- total_bytes: Optional[float] = None
- if total is not None:
- total_bytes = total
- elif task_id is not None:
- with self._lock:
- total_bytes = self._tasks[task_id].total
- if total_bytes is None:
- raise ValueError(
- f"unable to get the total number of bytes, please specify 'total'"
- )
-
- # update total of task or create new task
- if task_id is None:
- task_id = self.add_task(description, total=total_bytes)
- else:
- self.update(task_id, total=total_bytes)
-
- return _Reader(file, self, task_id, close_handle=False)
-
- @typing.overload
- def open(
- self,
- file: Union[str, "PathLike[str]", bytes],
- mode: Literal["rb"],
- buffering: int = -1,
- encoding: Optional[str] = None,
- errors: Optional[str] = None,
- newline: Optional[str] = None,
- *,
- total: Optional[int] = None,
- task_id: Optional[TaskID] = None,
- description: str = "Reading...",
- ) -> BinaryIO:
- pass
-
- @typing.overload
- def open(
- self,
- file: Union[str, "PathLike[str]", bytes],
- mode: Union[Literal["r"], Literal["rt"]],
- buffering: int = -1,
- encoding: Optional[str] = None,
- errors: Optional[str] = None,
- newline: Optional[str] = None,
- *,
- total: Optional[int] = None,
- task_id: Optional[TaskID] = None,
- description: str = "Reading...",
- ) -> TextIO:
- pass
-
- def open(
- self,
- file: Union[str, "PathLike[str]", bytes],
- mode: Union[Literal["rb"], Literal["rt"], Literal["r"]] = "r",
- buffering: int = -1,
- encoding: Optional[str] = None,
- errors: Optional[str] = None,
- newline: Optional[str] = None,
- *,
- total: Optional[int] = None,
- task_id: Optional[TaskID] = None,
- description: str = "Reading...",
- ) -> Union[BinaryIO, TextIO]:
- """Track progress while reading from a binary file.
-
- Args:
- path (Union[str, PathLike[str]]): The path to the file to read.
- mode (str): The mode to use to open the file. Only supports "r", "rb" or "rt".
- buffering (int): The buffering strategy to use, see :func:`io.open`.
- encoding (str, optional): The encoding to use when reading in text mode, see :func:`io.open`.
- errors (str, optional): The error handling strategy for decoding errors, see :func:`io.open`.
- newline (str, optional): The strategy for handling newlines in text mode, see :func:`io.open`.
- total (int, optional): Total number of bytes to read. If none given, os.stat(path).st_size is used.
- task_id (TaskID): Task to track. Default is new task.
- description (str, optional): Description of task, if new task is created.
-
- Returns:
- BinaryIO: A readable file-like object in binary mode.
-
- Raises:
- ValueError: When an invalid mode is given.
- """
- # normalize the mode (always rb, rt)
- _mode = "".join(sorted(mode, reverse=False))
- if _mode not in ("br", "rt", "r"):
- raise ValueError("invalid mode {!r}".format(mode))
-
- # patch buffering to provide the same behaviour as the builtin `open`
- line_buffering = buffering == 1
- if _mode == "br" and buffering == 1:
- warnings.warn(
- "line buffering (buffering=1) isn't supported in binary mode, the default buffer size will be used",
- RuntimeWarning,
- )
- buffering = -1
- elif _mode in ("rt", "r"):
- if buffering == 0:
- raise ValueError("can't have unbuffered text I/O")
- elif buffering == 1:
- buffering = -1
-
- # attempt to get the total with `os.stat`
- if total is None:
- total = stat(file).st_size
-
- # update total of task or create new task
- if task_id is None:
- task_id = self.add_task(description, total=total)
- else:
- self.update(task_id, total=total)
-
- # open the file in binary mode,
- handle = io.open(file, "rb", buffering=buffering)
- reader = _Reader(handle, self, task_id, close_handle=True)
-
- # wrap the reader in a `TextIOWrapper` if text mode
- if mode in ("r", "rt"):
- return io.TextIOWrapper(
- reader,
- encoding=encoding,
- errors=errors,
- newline=newline,
- line_buffering=line_buffering,
- )
-
- return reader
-
- 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()
- if update_completed > 0:
- _progress.append(ProgressSample(current_time, update_completed))
- if (
- task.total is not None
- and 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.
- visible (bool, optional): Enable display of the task. Defaults to True.
- description (str, optional): Change task description if not None. Defaults to None.
- **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.total is not None
- and 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: Optional[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 known.
- Set to None to render a pulsing animation. 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(),
- *Progress.get_default_columns(),
- TimeElapsedColumn(),
- console=console,
- transient=False,
- ) 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=None)
-
- 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))