mirrors/rich
1
0
Fork 0
mirror of https://github.com/Textualize/rich.git synced 2025-12-23 07:08:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

live refactor

Browse source
This commit is contained in:
Will McGugan 2021-02-06 11:49:54 +00:00
parent 0a54346bda
commit 122b8131bb
24 changed files with 412 additions and 299 deletions
Download patch file Download diff file Expand all files Collapse all files

12
CHANGELOG.md
View file

@ -5,6 +5,18 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [9.11.0] - Unreleased
### Fixed
- Fixed error message for tracebacks with broken `__str__` https://github.com/willmcgugan/rich/issues/980
- Fixed markup edge case https://github.com/willmcgugan/rich/issues/987
### Added
- Added cheeky sponsorship request to test card
- Added `quiet` argument to Console constructor
## [9.10.0] - 2021-01-27
### Changed

7
docs/source/console.rst
View file

@ -306,6 +306,13 @@ If Rich detects that it is not writing to a terminal it will strip control codes
Letting Rich auto-detect terminals is useful as it will write plain text when you pipe output to a file or other application.
Interactive mode
~~~~~~~~~~~~~~~~
Rich will remove animations such as progress bars and status indicators when not writing to a terminal as you probably don't want to write these out to a text file (for example). You can override this behavior by setting the ``force_interactive`` argument on the constructor. Set it to True to enable animations or False to disable them.
.. note::
Some CI systems support ANSI color and style but not anything that moves the cursor or selectively refreshes parts of the terminal. For these you might want to set ``force_terminal`` to ``True`` and ``force_interactve`` to ``False``.
Environment variables
---------------------

19
docs/source/live.rst
View file

@ -3,9 +3,9 @@
Live Display
============
Rich can display continiuously updated information for any renderable.
Progress bars and status indicators use a *live* display to animate parts of the terminal. You can build custom live displays with the :class:`~rich.live.Live` class.
To see some live display examples, try this from the command line::
For a demonstration of a live display, running the following command:
python -m rich.live
@ -21,7 +21,7 @@ The basic usage can be split into two use cases.
1. Same Renderable
~~~~~~~~~~~~~~~~~~
When keeping the same renderable, you simply pass the :class:`~rich.console.RenderableType` you would like to see updating and provide
When keeping the same renderable, pass the :class:`~rich.console.RenderableType` you would like to see updating and provide
a ``refresh_per_second`` parameter. The Live :class:`~rich.live.Live` will automatically update the console at the provided refresh rate.
@ -47,8 +47,8 @@ a ``refresh_per_second`` parameter. The Live :class:`~rich.live.Live` will autom
2. New Renderable
~~~~~~~~~~~~~~~~~
You can also provide constant new renderable to :class:`~rich.live.Live` using the :meth:`~rich.live.Live.update` function. This allows you to
completely change what is rendered live.
You can also provide a new renderable to :class:`~rich.live.Live` using the :meth:`~rich.live.Live.update` function. This allows you to
completely change the live display.
**Example**::
@ -163,6 +163,15 @@ Redirecting stdout / stderr
To avoid breaking the live display visuals, Rich will redirect ``stdout`` and ``stderr`` so that you can use the builtin ``print`` statement.
This feature is enabled by default, but you can disable by setting ``redirect_stdout`` or ``redirect_stderr`` to ``False``.
Nesting Lives
-------------
Note that only a single live context may be active at any one time. The following will raise a :class:`~rich.errors.LiveError` because status also uses Live::
with Live(table, console=console):
with console.status("working"): # Will not work
do_work()
Examples
--------

6
docs/source/progress.rst
View file

@ -162,7 +162,13 @@ If the :class:`~rich.progress.Progress` class doesn't offer exactly what you nee
def get_renderables(self):
yield Panel(self.make_tasks_table(self.tasks))
Multiple Progress
-----------------
You can't have different columns per task with a single Progress instance. However, you can have as many Progress instance as you like in a :ref:`live`. See `live_progress.py <https://github.com/willmcgugan/rich/blob/master/examples/live_progress.py>`_ for an example of using mutiple Progress instances.
Example
-------
See `downloader.py <https://github.com/willmcgugan/rich/blob/master/examples/downloader.py>`_ for a realistic application of a progress display. This script can download multiple concurrent files with a progress bar, transfer speed and file size.

45
examples/live_progress.py Normal file
View file

@ -0,0 +1,45 @@
"""
Demonstrates the use of multiple Progress instances in a single Live display.
"""
from time import sleep
from rich.live import Live
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, BarColumn, TextColumn
from rich.table import Table
job_progress = Progress(
"{task.description}",
SpinnerColumn(),
BarColumn(),
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
)
job1 = job_progress.add_task("[green]Cooking")
job2 = job_progress.add_task("[magenta]Baking", total=200)
job3 = job_progress.add_task("[cyan]Mixing", total=400)
total = sum(task.total for task in job_progress.tasks)
overall_progress = Progress()
overall_task = overall_progress.add_task("All Jobs", total=int(total))
progress_table = Table.grid()
progress_table.add_row(
Panel.fit(
overall_progress, title="Overall Progress", border_style="green", padding=(2, 2)
),
Panel.fit(job_progress, title="[b]Jobs", border_style="red", padding=(1, 2)),
)
with Live(progress_table, refresh_per_second=10):
while not overall_progress.finished:
sleep(0.1)
for job in job_progress.tasks:
if not job.finished:
job_progress.advance(job.id)
completed = sum(task.completed for task in job_progress.tasks)
overall_progress.update(overall_task, completed=completed)

2
pyproject.toml
View file

@ -2,7 +2,7 @@
name = "rich"
homepage = "https://github.com/willmcgugan/rich"
documentation = "https://rich.readthedocs.io/en/latest/"
version = "9.10.0"
version = "9.11.0"
description = "Render rich text, tables, progress bars, syntax highlighting, markdown and more to the terminal"
authors = ["Will McGugan <willmcgugan@gmail.com>"]
license = "MIT"

39
rich/__main__.py
View file

@ -227,3 +227,42 @@ if __name__ == "__main__": # pragma: no cover
print(line)
print(f"rendered in {taken}ms")
from rich.panel import Panel
console = Console()
sponsor_message = Text.from_markup(
"""\
[green]Github Sponsor:[/green] [u blue][link=https://github.com/sponsors/willmcgugan]https://github.com/sponsors/willmcgugan[/][/]
[green]Buy me a coffee:[/green] [u blue][link=https://ko-fi.com/willmcgugan]https://ko-fi.com/willmcgugan[/link][/]
[green]Twitter:[/green] [u blue][link=https://twitter.com/willmcgugan]https://twitter.com/willmcgugan[/link][/]
[green]Blog:[/green] [u blue][link=https://www.willmcgugan.com]https://www.willmcgugan.com[/link][/]"""
)
intro_message = Text.from_markup(
"""\
Developing Rich and responding to issues can take a lot of my time
Consider supporting my work via Github Sponsors (ask your company / Organization), or buy me a coffee to say thanks.
- Will McGugan"""
)
message = Table.grid(padding=2)
message.add_row(intro_message, sponsor_message)
console.print(
Panel.fit(
message,
box=box.ROUNDED,
padding=(1, 2),
title="[b red]Thanks for trying out Rich!",
border_style="bright_blue",
width=122,
),
justify="center",
)

38
rich/console.py
View file

@ -49,6 +49,7 @@ from .theme import Theme, ThemeStack
if TYPE_CHECKING:
from ._windows import WindowsConsoleFeatures
from .live import Live
from .status import Status
WINDOWS = platform.system() == "Windows"
@ -395,10 +396,12 @@ class Console:
either ``"standard"``, ``"256"`` or ``"truecolor"``. Leave as ``"auto"`` to autodetect.
force_terminal (Optional[bool], optional): Enable/disable terminal control codes, or None to auto-detect terminal. Defaults to None.
force_jupyter (Optional[bool], optional): Enable/disable Jupyter rendering, or None to auto-detect Jupyter. Defaults to None.
force_interactive (Optional[bool], optional): Enable/disable interactive mode, or None to auto detect. Defaults to None.
soft_wrap (Optional[bool], optional): Set soft wrap default on print method. Defaults to False.
theme (Theme, optional): An optional style theme object, or ``None`` for default theme.
stderr (bool, optional): Use stderr rather than stdout if ``file`` is not specified. Defaults to False.
file (IO, optional): A file object where the console should write to. Defaults to stdout.
quiet (bool, Optional): Boolean to suppress all output. Defaults to False.
width (int, optional): The width of the terminal. Leave as default to auto-detect width.
height (int, optional): The height of the terminal. Leave as default to auto-detect height.
style (StyleType, optional): Style to apply to all output, or None for no style. Defaults to None.
@ -428,10 +431,12 @@ class Console:
] = "auto",
force_terminal: bool = None,
force_jupyter: bool = None,
force_interactive: bool = None,
soft_wrap: bool = False,
theme: Theme = None,
stderr: bool = False,
file: IO[str] = None,
quiet: bool = False,
width: int = None,
height: int = None,
style: StyleType = None,
@ -475,6 +480,7 @@ class Console:
self._color_system: Optional[ColorSystem]
self._force_terminal = force_terminal
self._file = file
self.quiet = quiet
self.stderr = stderr
if color_system is None:
@ -498,6 +504,11 @@ class Console:
self.no_color = (
no_color if no_color is not None else "NO_COLOR" in self._environ
)
self.is_interactive = (
(self.is_terminal and not self.is_dumb_terminal)
if force_interactive is None
else force_interactive
)
self._record_buffer_lock = threading.RLock()
self._thread_locals = ConsoleThreadLocals(
@ -505,6 +516,7 @@ class Console:
)
self._record_buffer: List[Segment] = []
self._render_hooks: List[RenderHook] = []
self._live: Optional["Live"] = None
def __repr__(self) -> str:
return f"<console width={self.width} {str(self._color_system)}>"
@ -573,6 +585,25 @@ class Console:
self._buffer_index -= 1
self._check_buffer()
def set_live(self, live: "Live") -> None:
"""Set Live instance. Used by Live context manager.
Args:
live (Live): Live instance using this Console.
Raises:
errors.LiveError: If this Console has a Live context currently active.
"""
with self._lock:
if self._live is not None:
raise errors.LiveError("Only one live display may be active at once")
self._live = live
def clear_live(self) -> None:
"""Clear the Live instance."""
with self._lock:
self._live = None
def push_render_hook(self, hook: RenderHook) -> None:
"""Add a new render hook to the stack.
@ -1007,7 +1038,9 @@ class Console:
except errors.StyleSyntaxError as error:
if default is not None:
return self.get_style(default)
raise errors.MissingStyle(f"Failed to get style {name!r}; {error}")
raise errors.MissingStyle(
f"Failed to get style {name!r}; {error}"
) from None
def _collect_renderables(
self,
@ -1356,6 +1389,9 @@ class Console:
def _check_buffer(self) -> None:
"""Check if the buffer may be rendered."""
if self.quiet:
del self._buffer[:]
return
with self._lock:
if self._buffer_index == 0:
if self.is_jupyter: # pragma: no cover

4
rich/errors.py
View file

@ -24,3 +24,7 @@ class NotRenderableError(ConsoleError):
class MarkupError(ConsoleError):
"""Markup was badly formatted."""
class LiveError(ConsoleError):
"""Error related to Live display."""

163
rich/live.py
View file

@ -1,29 +1,15 @@
import sys
from threading import Event, RLock, Thread
from typing import IO, Any, List, Optional
from typing_extensions import Literal
from typing import IO, Any, Callable, List, Optional
from . import get_console
from ._loop import loop_last
from .console import (
Console,
ConsoleOptions,
ConsoleRenderable,
RenderableType,
RenderHook,
RenderResult,
)
from .console import Console, ConsoleRenderable, RenderableType, RenderHook
from .control import Control
from .file_proxy import FileProxy
from .jupyter import JupyterMixin
from .live_render import LiveRender
from .segment import Segment
from .style import Style
from .live_render import LiveRender, VerticalOverflowMethod
from .text import Text
VerticalOverflowMethod = Literal["crop", "ellipsis", "visible"]
class _RefreshThread(Thread):
"""A thread that calls refresh() at regular intervals."""
@ -32,7 +18,7 @@ class _RefreshThread(Thread):
self.live = live
self.refresh_per_second = refresh_per_second
self.done = Event()
super().__init__()
super().__init__(daemon=True)
def stop(self) -> None:
self.done.set()
@ -44,54 +30,11 @@ class _RefreshThread(Thread):
self.live.refresh()
class _LiveRender(LiveRender):
def __init__(self, live: "Live", renderable: RenderableType) -> None:
self._live = live
self.renderable = renderable
self._shape: Optional[Tuple[int, int]] = None
def __rich_console__(
self, console: Console, options: ConsoleOptions
) -> RenderResult:
with self._live._lock:
lines = console.render_lines(self.renderable, options, pad=False)
shape = Segment.get_shape(lines)
_, height = shape
if height > console.size.height:
if self._live.vertical_overflow == "crop":
lines = lines[: console.size.height]
shape = Segment.get_shape(lines)
elif self._live.vertical_overflow == "ellipsis":
lines = lines[: (console.size.height - 1)]
lines.append(
list(
console.render(
Text(
"...",
overflow="crop",
justify="center",
end="",
style="live.ellipsis",
)
)
)
)
shape = Segment.get_shape(lines)
self._shape = shape
for last, line in loop_last(lines):
yield from line
if not last:
yield Segment.line()
class Live(JupyterMixin, RenderHook):
"""Renders an auto-updating live display of any given renderable.
Args:
renderable (RenderableType, optional): [The renderable to live display. Defaults to displaying nothing.
renderable (RenderableType, optional): The renderable to live display. Defaults to displaying nothing.
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()` or `update()` with refresh flag. Defaults to True
refresh_per_second (float, optional): Number of times per second to refresh the live display. Defaults to 1.
@ -99,11 +42,12 @@ class Live(JupyterMixin, RenderHook):
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.
vertical_overflow (VerticalOverflowMethod, optional): How to handle renderable when it is too tall for the console. Defaults to "ellipsis".
get_renderable (Callable[[], RenderableType], optional): Optional callable to get renderable. Defaults to None.
"""
def __init__(
self,
renderable: RenderableType = "",
renderable: RenderableType = None,
*,
console: Console = None,
auto_refresh: bool = True,
@ -112,10 +56,11 @@ class Live(JupyterMixin, RenderHook):
redirect_stdout: bool = True,
redirect_stderr: bool = True,
vertical_overflow: VerticalOverflowMethod = "ellipsis",
get_renderable: Callable[[], RenderableType] = None,
) -> None:
assert refresh_per_second > 0, "refresh_per_second must be > 0"
self._renderable = renderable
self.console = console if console is not None else get_console()
self._live_render = _LiveRender(self, renderable)
self._redirect_stdout = redirect_stdout
self._redirect_stderr = redirect_stderr
@ -132,19 +77,36 @@ class Live(JupyterMixin, RenderHook):
self.refresh_per_second = refresh_per_second
self.vertical_overflow = vertical_overflow
self._get_renderable = get_renderable
self._live_render = LiveRender(
self.get_renderable(), vertical_overflow=vertical_overflow
)
# cant store just clear_control as the live_render shape is lazily computed on render
def start(self) -> None:
"""Start live rendering display."""
def get_renderable(self) -> RenderableType:
renderable = (
self._get_renderable()
if self._get_renderable is not None
else self._renderable
)
return renderable or ""
def start(self, refresh=False) -> None:
"""Start live rendering display.
Args:
refresh (bool, optional): Also refresh. Defaults to False.
"""
with self._lock:
if self._started:
return
self.console.set_live(self)
self._started = True
self.console.show_cursor(False)
self._enable_redirect_io()
self.console.push_render_hook(self)
self._started = True
if refresh:
self.refresh()
if self.auto_refresh:
self._refresh_thread = _RefreshThread(self, self.refresh_per_second)
self._refresh_thread.start()
@ -154,6 +116,7 @@ class Live(JupyterMixin, RenderHook):
with self._lock:
if not self._started:
return
self.console.clear_live()
self._started = False
try:
if self.auto_refresh and self._refresh_thread is not None:
@ -165,22 +128,22 @@ class Live(JupyterMixin, RenderHook):
if self.console.is_terminal:
self.console.line()
finally:
self.console.show_cursor(True)
self._disable_redirect_io()
self.console.pop_render_hook()
self.console.show_cursor(True)
if self.transient:
self.console.control(self._live_render.restore_cursor())
if self.ipy_widget is not None: # pragma: no cover
if self.transient:
self.ipy_widget.close()
else:
# jupyter last refresh must occur after console pop render hook
# i am not sure why this is needed
self.refresh()
if self.auto_refresh and self._refresh_thread is not None:
if self._refresh_thread is not None:
self._refresh_thread.join()
self._refresh_thread = None
if self.transient:
self.console.control(self._live_render.restore_cursor())
if self.ipy_widget is not None: # pragma: no cover
if self.transient:
self.ipy_widget.close()
else:
# jupyter last refresh must occur after console pop render hook
# i am not sure why this is needed
self.refresh()
def __enter__(self) -> "Live":
self.start()
@ -192,13 +155,22 @@ class Live(JupyterMixin, RenderHook):
def _enable_redirect_io(self):
"""Enable redirecting of stdout / stderr."""
if self.console.is_terminal:
if self._redirect_stdout:
if self._redirect_stdout and not isinstance(sys.stdout, FileProxy): # type: ignore
self._restore_stdout = sys.stdout
sys.stdout = FileProxy(self.console, sys.stdout)
if self._redirect_stderr:
if self._redirect_stderr and not isinstance(sys.stderr, FileProxy): # type: ignore
self._restore_stderr = sys.stderr
sys.stderr = FileProxy(self.console, sys.stderr)
def _disable_redirect_io(self):
"""Disable redirecting of stdout / stderr."""
if self._restore_stdout:
sys.stdout = self._restore_stdout
self._restore_stdout = None
if self._restore_stderr:
sys.stderr = self._restore_stderr
self._restore_stderr = None
@property
def renderable(self) -> RenderableType:
"""Get the renderable that is being displayed
@ -206,8 +178,7 @@ class Live(JupyterMixin, RenderHook):
Returns:
RenderableType: Displayed renderable.
"""
with self._lock:
return self._live_render.renderable
return self.get_renderable()
def update(self, renderable: RenderableType, *, refresh: bool = False) -> None:
"""Update the renderable that is being displayed
@ -217,12 +188,13 @@ class Live(JupyterMixin, RenderHook):
refresh (bool, optional): Refresh the display. Defaults to False.
"""
with self._lock:
self._live_render.set_renderable(renderable)
self._renderable = renderable
if refresh:
self.refresh()
def refresh(self) -> None:
"""Update the display of the Live Render."""
self._live_render.set_renderable(self.renderable)
if self.console.is_jupyter: # pragma: no cover
try:
from IPython.display import display
@ -249,21 +221,13 @@ class Live(JupyterMixin, RenderHook):
with self.console:
self.console.print(Control(""))
def _disable_redirect_io(self):
"""Disable redirecting of stdout / stderr."""
if self._restore_stdout:
sys.stdout = self._restore_stdout
self._restore_stdout = None
if self._restore_stderr:
sys.stderr = self._restore_stderr
self._restore_stderr = None
def process_renderables(
self, renderables: List[ConsoleRenderable]
) -> List[ConsoleRenderable]:
"""Process renderables to restore cursor and display progress."""
if self.console.is_terminal:
# lock needs acquiring as user can modify live_render renerable at any time unlike in Progress.
self._live_render.vertical_overflow = self.vertical_overflow
if self.console.is_interactive:
# lock needs acquiring as user can modify live_render renderable at any time unlike in Progress.
with self._lock:
# determine the control command needed to clear previous rendering
renderables = [
@ -285,6 +249,7 @@ if __name__ == "__main__": # pragma: no cover
from itertools import cycle
from typing import Dict, List, Tuple
from .align import Align
from .console import Console
from .live import Live
from .panel import Panel
@ -372,9 +337,9 @@ if __name__ == "__main__": # pragma: no cover
table.add_column("Destination Currency")
table.add_column("Exchange Rate")
for ((soure, dest), exchange_rate) in exchange_rate_dict.items():
for ((source, dest), exchange_rate) in exchange_rate_dict.items():
table.add_row(
soure,
source,
dest,
Text(
f"{exchange_rate:.4f}",
@ -382,4 +347,4 @@ if __name__ == "__main__": # pragma: no cover
),
)
live_table.update(table)
live_table.update(Align.center(table))

48
rich/live_render.py
View file

@ -1,11 +1,14 @@
from typing import Optional, Tuple
from typing import Literal, Optional, Tuple
from .console import Console, ConsoleOptions, RenderableType, RenderResult
from .control import Control
from .segment import Segment
from .text import Text
from .style import StyleType
from ._loop import loop_last
VerticalOverflowMethod = Literal["crop", "ellipsis", "visible"]
class LiveRender:
"""Creates a renderable that may be updated.
@ -15,9 +18,15 @@ class LiveRender:
style (StyleType, optional): An optional style to apply to the renderable. Defaults to "".
"""
def __init__(self, renderable: RenderableType, style: StyleType = "") -> None:
def __init__(
self,
renderable: RenderableType,
style: StyleType = "",
vertical_overflow: VerticalOverflowMethod = "ellipsis",
) -> None:
self.renderable = renderable
self.style = style
self.vertical_overflow = vertical_overflow
self._shape: Optional[Tuple[int, int]] = None
def set_renderable(self, renderable: RenderableType) -> None:
@ -53,23 +62,30 @@ class LiveRender:
def __rich_console__(
self, console: Console, options: ConsoleOptions
) -> RenderResult:
_Segment = Segment
style = console.get_style(self.style)
lines = console.render_lines(self.renderable, options, style=style, pad=False)
_Segment = Segment
shape = _Segment.get_shape(lines)
if self._shape is None:
self._shape = shape
else:
width1, height1 = shape
width2, height2 = self._shape
self._shape = (
max(width1, min(options.max_width, width2)),
max(height1, height2),
)
width, height = self._shape
lines = _Segment.set_shape(lines, width, height)
_, height = shape
if height > console.size.height:
if self.vertical_overflow == "crop":
lines = lines[: console.size.height]
shape = _Segment.get_shape(lines)
elif self.vertical_overflow == "ellipsis":
lines = lines[: (console.size.height - 1)]
overflow_text = Text(
"...",
overflow="crop",
justify="center",
end="",
style="live.ellipsis",
)
lines.append(list(console.render(overflow_text)))
shape = _Segment.get_shape(lines)
self._shape = shape
for last, line in loop_last(lines):
yield from _Segment.make_control(line)
yield from line
if not last:
yield _Segment.line(is_control=True)
yield _Segment.line()

200
rich/progress.py
View file

@ -7,7 +7,6 @@ from datetime import timedelta
from math import ceil
from threading import Event, RLock, Thread
from typing import (
IO,
Any,
Callable,
Deque,
@ -26,17 +25,13 @@ from typing import (
from . import filesize, get_console
from .console import (
Console,
ConsoleRenderable,
JustifyMethod,
RenderableType,
RenderGroup,
RenderHook,
)
from .control import Control
from .file_proxy import FileProxy
from .highlighter import Highlighter
from .jupyter import JupyterMixin
from .live_render import LiveRender
from .highlighter import Highlighter
from .live import Live
from .progress_bar import ProgressBar
from .spinner import Spinner
from .style import StyleType
@ -93,7 +88,7 @@ def track(
console: Optional[Console] = None,
transient: bool = False,
get_time: Callable[[], float] = None,
refresh_per_second: float = None,
refresh_per_second: float = 10,
style: StyleType = "bar.back",
complete_style: StyleType = "bar.complete",
finished_style: StyleType = "bar.finished",
@ -110,7 +105,7 @@ def track(
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 (Optional[float], optional): Number of times per second to refresh the progress information, or None to use default. Defaults to None.
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".
@ -143,7 +138,7 @@ def track(
console=console,
transient=transient,
get_time=get_time,
refresh_per_second=refresh_per_second,
refresh_per_second=refresh_per_second or 10,
disable=disable,
)
@ -246,9 +241,11 @@ class SpinnerColumn(ProgressColumn):
self.spinner = Spinner(spinner_name, style=spinner_style, speed=speed)
def render(self, task: "Task") -> Text:
if task.finished:
return self.finished_text
text = self.spinner.render(task.get_time())
text = (
self.finished_text
if task.finished
else self.spinner.render(task.get_time())
)
return text
@ -465,6 +462,9 @@ class Task:
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() # type: ignore
@ -507,17 +507,18 @@ class Task:
"""Optional[float]: Get the estimated speed in steps per second."""
if self.start_time is None:
return None
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
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]:
@ -536,24 +537,7 @@ class Task:
self.finished_time = None
class _RefreshThread(Thread):
"""A thread that calls refresh() on the Process object at regular intervals."""
def __init__(self, progress: "Progress", refresh_per_second: float = 10) -> None:
self.progress = progress
self.refresh_per_second = refresh_per_second
self.done = Event()
super().__init__()
def stop(self) -> None:
self.done.set()
def run(self) -> None:
while not self.done.wait(1.0 / self.refresh_per_second):
self.progress.refresh()
class Progress(JupyterMixin, RenderHook):
class Progress(JupyterMixin):
"""Renders an auto-updating progress bar(s).
Args:
@ -573,7 +557,7 @@ class Progress(JupyterMixin, RenderHook):
*columns: Union[str, ProgressColumn],
console: Console = None,
auto_refresh: bool = True,
refresh_per_second: float = None,
refresh_per_second: float = 10,
speed_estimate_period: float = 30.0,
transient: bool = False,
redirect_stdout: bool = True,
@ -583,7 +567,7 @@ class Progress(JupyterMixin, RenderHook):
) -> None:
assert (
refresh_per_second is None or refresh_per_second > 0
), "refresh_per_second must be > 0"
), "refresh_per_second must be > 0" # type: ignore
self._lock = RLock()
self.columns = columns or (
TextColumn("[progress.description]{task.description}"),
@ -591,25 +575,27 @@ class Progress(JupyterMixin, RenderHook):
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
TimeRemainingColumn(),
)
self.console = console or get_console()
self.auto_refresh = auto_refresh and not self.console.is_jupyter
self.refresh_per_second = refresh_per_second or 10
self.speed_estimate_period = speed_estimate_period
self.transient = transient
self._redirect_stdout = redirect_stdout
self._redirect_stderr = redirect_stderr
self.get_time = get_time or self.console.get_time
self.disable = disable
self._tasks: Dict[TaskID, Task] = {}
self._live_render = LiveRender(self.get_renderable())
self._task_index: TaskID = TaskID(0)
self._refresh_thread: Optional[_RefreshThread] = None
self._started = False
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
self._restore_stdout: Optional[IO[str]] = None
self._restore_stderr: Optional[IO[str]] = None
self.ipy_widget: Optional[Any] = None
@property
def console(self) -> Console:
return self.live.console
@property
def tasks(self) -> List[Task]:
@ -631,63 +617,13 @@ class Progress(JupyterMixin, RenderHook):
return True
return all(task.finished for task in self._tasks.values())
def _enable_redirect_io(self):
"""Enable redirecting of stdout / stderr."""
if self.console.is_terminal:
if self._redirect_stdout:
self._restore_stdout = sys.stdout
sys.stdout = FileProxy(self.console, sys.stdout)
if self._redirect_stderr:
self._restore_stderr = sys.stderr
sys.stderr = FileProxy(self.console, sys.stderr)
def _disable_redirect_io(self):
"""Disable redirecting of stdout / stderr."""
if self._restore_stdout:
sys.stdout = self._restore_stdout
self._restore_stdout = None
if self._restore_stderr:
sys.stderr = self._restore_stderr
self._restore_stderr = None
def start(self) -> None:
"""Start the progress display."""
with self._lock:
if self._started:
return
self._started = True
self.console.show_cursor(False)
self._enable_redirect_io()
self.console.push_render_hook(self)
self.refresh()
if self.auto_refresh:
self._refresh_thread = _RefreshThread(self, self.refresh_per_second)
self._refresh_thread.start()
self.live.start(refresh=True)
def stop(self) -> None:
"""Stop the progress display."""
with self._lock:
if not self._started:
return
self._started = False
try:
if self.auto_refresh and self._refresh_thread is not None:
self._refresh_thread.stop()
self.refresh()
if self.console.is_terminal:
self.console.line()
finally:
self.console.show_cursor(True)
self._disable_redirect_io()
self.console.pop_render_hook()
if self._refresh_thread is not None:
self._refresh_thread.join()
self._refresh_thread = None
if self.transient:
self.console.control(self._live_render.restore_cursor())
if self.ipy_widget is not None and self.transient: # pragma: no cover
self.ipy_widget.clear_output()
self.ipy_widget.close()
self.live.stop()
def __enter__(self) -> "Progress":
self.start()
@ -731,7 +667,7 @@ class Progress(JupyterMixin, RenderHook):
else:
self.update(task_id, total=task_total)
if self.auto_refresh:
if self.live.auto_refresh:
with _TrackThread(self, task_id, update_period) as track_thread:
for value in sequence:
yield value
@ -897,29 +833,7 @@ class Progress(JupyterMixin, RenderHook):
def refresh(self) -> None:
"""Refresh (render) the progress information."""
if not self.disable:
if self.console.is_jupyter: # pragma: no cover
try:
from IPython.display import display
from ipywidgets import Output
except ImportError:
import warnings
warnings.warn('install "ipywidgets" for Jupyter support')
else:
with self._lock:
if self.ipy_widget is None:
self.ipy_widget = Output()
display(self.ipy_widget)
with self.ipy_widget:
self.ipy_widget.clear_output(wait=True)
self.console.print(self.get_renderable())
elif self.console.is_terminal and not self.console.is_dumb_terminal:
with self._lock:
self._live_render.set_renderable(self.get_renderable())
with self.console:
self.console.print(Control(""))
self.live.refresh()
def get_renderable(self) -> RenderableType:
"""Get a renderable for the progress display."""
@ -960,6 +874,9 @@ class Progress(JupyterMixin, RenderHook):
table.add_row(*row)
return table
def __rich__(self) -> RenderableType:
return self.get_renderable()
def add_task(
self,
description: str,
@ -992,6 +909,7 @@ class Progress(JupyterMixin, RenderHook):
visible=visible,
fields=fields,
_get_time=self.get_time,
_lock=self._lock,
)
self._tasks[self._task_index] = task
if start:
@ -1012,18 +930,6 @@ class Progress(JupyterMixin, RenderHook):
with self._lock:
del self._tasks[task_id]
def process_renderables(
self, renderables: List[ConsoleRenderable]
) -> List[ConsoleRenderable]:
"""Process renderables to restore cursor and display progress."""
if self.console.is_terminal:
renderables = [
self._live_render.position_cursor(),
*renderables,
self._live_render,
]
return renderables
if __name__ == "__main__": # pragma: no coverage
@ -1094,6 +1000,6 @@ if __name__ == "__main__": # pragma: no coverage
if random.randint(0, 100) < 1:
progress.log(next(examples))
except:
console.save_html("progress.html")
print("wrote progress.html")
# console.save_html("progress.html")
# print("wrote progress.html")
raise

6
rich/status.py
View file

@ -1,13 +1,14 @@
from typing import Optional
from .console import Console, RenderableType
from .jupyter import JupyterMixin
from .live import Live
from .spinner import Spinner
from .style import StyleType
from .table import Table
class Status:
class Status(JupyterMixin):
"""Displays a status indicator with a 'spinner' animation.
Args:
@ -93,6 +94,9 @@ class Status:
"""Stop the spinner animation."""
self._live.stop()
def __rich__(self) -> RenderableType:
return self.renderable
def __enter__(self) -> "Status":
self.start()
return self

4
rich/syntax.py
View file

@ -2,7 +2,7 @@ import os.path
import platform
import textwrap
from abc import ABC, abstractmethod
from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Type, Union
from typing import Any, Dict, Iterable, Optional, Set, Tuple, Type, Union
from pygments.lexers import get_lexer_by_name, guess_lexer_for_filename
from pygments.style import Style as PygmentsStyle
@ -22,7 +22,7 @@ from pygments.token import (
from pygments.util import ClassNotFound
from ._loop import loop_first
from .color import Color, blend_rgb, parse_rgb_hex
from .color import Color, blend_rgb
from .console import Console, ConsoleOptions, JustifyMethod, RenderResult, Segment
from .jupyter import JupyterMixin
from .measure import Measurement

6
rich/text.py
View file

@ -397,14 +397,14 @@ class Text(JupyterMixin):
Returns:
Style: A Style instance.
"""
# TODO: This is a little inefficient, it is only used by full justify
if offset < 0:
offset = len(self) + offset
get_style = console.get_style
style = get_style(self.style).copy()
for start, end, span_style in self._spans:
if offset >= start and offset < end:
style += get_style(span_style)
if end > offset >= start:
style += get_style(span_style, default="")
return style
def highlight_regex(

14
rich/traceback.py
View file

@ -6,7 +6,7 @@ import sys
from dataclasses import dataclass, field
from traceback import walk_tb
from types import TracebackType
from typing import Callable, Dict, Iterable, List, Optional, Type
from typing import Any, Callable, Dict, Iterable, List, Optional, Type
from pygments.lexers import guess_lexer_for_filename
from pygments.token import Comment, Keyword, Name, Number, Operator, String
@ -258,10 +258,17 @@ class Traceback:
from rich import _IMPORT_CWD
def safe_str(_object: Any) -> str:
"""Don't allow exceptions from __str__ to propegate."""
try:
return str(_object)
except Exception:
return "<exception str() failed>"
while True:
stack = Stack(
exc_type=str(exc_type.__name__),
exc_value=str(exc_value),
exc_type=safe_str(exc_type.__name__),
exc_value=safe_str(exc_value),
is_cause=is_cause,
)
@ -386,6 +393,7 @@ class Traceback:
highlighter(stack.syntax_error.msg),
)
else:
print(stack.exc_value)
yield Text.assemble(
(f"{stack.exc_type}: ", "traceback.exc_type"),
highlighter(stack.exc_value),

14
tests/test_console.py
View file

@ -478,3 +478,17 @@ def test_no_color():
result = console.file.getvalue()
print(repr(result))
assert result == expected
def test_quiet():
console = Console(file=io.StringIO(), quiet=True)
console.print("Hello, World!")
assert console.file.getvalue() == ""
def test_no_nested_live():
console = Console()
with pytest.raises(errors.LiveError):
with console.status("foo"):
with console.status("bar"):
pass

1
tests/test_live.py
View file

@ -45,6 +45,7 @@ def test_growing_display() -> None:
display += f"Step {step}\n"
live.update(display, refresh=True)
output = console.end_capture()
print(repr(output))
assert (
output
== "\x1b[?25lStep 0\n\r\x1b[2K\x1b[1A\x1b[2KStep 0\nStep 1\n\r\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2KStep 0\nStep 1\nStep 2\n\r\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2KStep 0\nStep 1\nStep 2\nStep 3\n\r\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2KStep 0\nStep 1\nStep 2\nStep 3\nStep 4\n\r\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2KStep 0\nStep 1\nStep 2\nStep 3\nStep 4\nStep 5\n\r\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2KStep 0\nStep 1\nStep 2\nStep 3\nStep 4\nStep 5\nStep 6\n\r\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2KStep 0\nStep 1\nStep 2\nStep 3\nStep 4\nStep 5\nStep 6\nStep 7\n\r\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2KStep 0\nStep 1\nStep 2\nStep 3\nStep 4\nStep 5\nStep 6\nStep 7\nStep 8\n\r\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2KStep 0\nStep 1\nStep 2\nStep 3\nStep 4\nStep 5\nStep 6\nStep 7\nStep 8\nStep 9\n\r\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2K\x1b[1A\x1b[2KStep 0\nStep 1\nStep 2\nStep 3\nStep 4\nStep 5\nStep 6\nStep 7\nStep 8\nStep 9\n\n\x1b[?25h"

4
tests/test_live_render.py
View file

@ -37,7 +37,7 @@ def test_rich_console(live_render):
encoding="utf-8",
)
rich_console = live_render.__rich_console__(Console(), options)
assert [Segment.control("my string", Style.parse("none"))] == list(rich_console)
assert [Segment("my string", Style.parse("none"))] == list(rich_console)
live_render.style = "red"
rich_console = live_render.__rich_console__(Console(), options)
assert [Segment.control("my string", Style.parse("red"))] == list(rich_console)
assert [Segment("my string", Style.parse("red"))] == list(rich_console)

39
tests/test_progress.py
View file

@ -14,6 +14,8 @@ from rich.progress import (
TotalFileSizeColumn,
DownloadColumn,
TransferSpeedColumn,
RenderableColumn,
SpinnerColumn,
Progress,
Task,
TextColumn,
@ -22,7 +24,6 @@ from rich.progress import (
track,
_TrackThread,
TaskID,
_RefreshThread,
)
from rich.text import Text
@ -80,6 +81,22 @@ def test_time_remaining_column():
assert str(text) == "0:01:00"
def test_renderable_column():
column = RenderableColumn("foo")
task = Task(1, "test", 100, 20, _get_time=lambda: 1.0)
assert column.render(task) == "foo"
def test_spinner_column():
column = SpinnerColumn()
column.set_spinner("dots2")
task = Task(1, "test", 100, 20, _get_time=lambda: 1.0)
result = column.render(task)
print(repr(result))
expected = "⡿"
assert str(result) == expected
def test_download_progress_uses_decimal_units() -> None:
column = DownloadColumn()
@ -171,7 +188,8 @@ def test_expand_bar() -> None:
pass
expected = "\x1b[?25l\x1b[38;5;237m━━━━━━━━━━\x1b[0m\r\x1b[2K\x1b[38;5;237m━━━━━━━━━━\x1b[0m\n\x1b[?25h"
render_result = console.file.getvalue()
print(repr(render_result))
print("RESULT\n", repr(render_result))
print("EXPECTED\n", repr(expected))
assert render_result == expected
@ -331,23 +349,6 @@ def test_progress_create() -> None:
assert progress.task_ids == []
def test_refresh_thread() -> None:
class MockProgress:
def __init__(self):
self.count = 0
def refresh(self):
self.count += 1
progress = MockProgress()
thread = _RefreshThread(progress, 100)
assert thread.progress == progress
thread.start()
sleep(0.2)
thread.stop()
assert progress.count >= 1
def test_track_thread() -> None:
progress = Progress()
task_id = progress.add_task("foo")

8
tests/test_segment.py
View file

@ -107,3 +107,11 @@ def test_remove_color():
Segment("foo", Style(bold=True)),
Segment("bar", None),
]
def test_make_control():
segments = [Segment("foo"), Segment("bar")]
assert Segment.make_control(segments) == [
Segment.control("foo"),
Segment.control("bar"),
]

10
tests/test_status.py
View file

@ -19,3 +19,13 @@ def test_status():
# TODO: Testing output is tricky with threads
with status:
sleep(0.2)
def test_renderable():
console = Console(
color_system=None, width=80, legacy_windows=False, get_time=lambda: 0.0
)
status = Status("foo", console=console)
console.begin_capture()
console.print(status)
assert console.end_capture() == "⠋ foo\n"

7
tests/test_text.py
View file

@ -668,3 +668,10 @@ def test_slice():
with pytest.raises(TypeError):
text[::-1]
def test_wrap_invalid_style():
# https://github.com/willmcgugan/rich/issues/987
console = Console(width=100, color_system="truecolor")
a = "[#######.................] xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx [#######.................]"
console.print(a, justify="full")

15
tests/test_traceback.py
View file

@ -166,6 +166,21 @@ def test_filename_not_a_file():
assert "string" in exception_text
def test_broken_str():
class BrokenStr(Exception):
def __str__(self):
1 / 0
console = Console(width=100, file=io.StringIO())
try:
raise BrokenStr()
except Exception:
console.print_exception()
result = console.file.getvalue()
print(result)
assert "<exception str() failed>" in result
if __name__ == "__main__": # pragma: no cover
expected = render(get_exception())