summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/rich/live.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/live.py
parentc45662ff3923b34614ddcc8feb9195541166dcc5 (diff)
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/rich/live.py')
-rw-r--r--venv/lib/python3.11/site-packages/rich/live.py375
1 files changed, 0 insertions, 375 deletions
diff --git a/venv/lib/python3.11/site-packages/rich/live.py b/venv/lib/python3.11/site-packages/rich/live.py
deleted file mode 100644
index f0529a7..0000000
--- a/venv/lib/python3.11/site-packages/rich/live.py
+++ /dev/null
@@ -1,375 +0,0 @@
-import sys
-from threading import Event, RLock, Thread
-from types import TracebackType
-from typing import IO, Any, Callable, List, Optional, TextIO, Type, cast
-
-from . import get_console
-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, VerticalOverflowMethod
-from .screen import Screen
-from .text import Text
-
-
-class _RefreshThread(Thread):
- """A thread that calls refresh() at regular intervals."""
-
- def __init__(self, live: "Live", refresh_per_second: float) -> None:
- self.live = live
- self.refresh_per_second = refresh_per_second
- self.done = Event()
- super().__init__(daemon=True)
-
- def stop(self) -> None:
- self.done.set()
-
- def run(self) -> None:
- while not self.done.wait(1 / self.refresh_per_second):
- with self.live._lock:
- if not self.done.is_set():
- self.live.refresh()
-
-
-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.
- console (Console, optional): Optional Console instance. Default will an internal Console instance writing to stdout.
- screen (bool, optional): Enable alternate screen mode. Defaults to False.
- 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 4.
- transient (bool, optional): Clear the renderable on exit (has no effect when screen=True). 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.
- 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: Optional[RenderableType] = None,
- *,
- console: Optional[Console] = None,
- screen: bool = False,
- auto_refresh: bool = True,
- refresh_per_second: float = 4,
- transient: bool = False,
- redirect_stdout: bool = True,
- redirect_stderr: bool = True,
- vertical_overflow: VerticalOverflowMethod = "ellipsis",
- get_renderable: Optional[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._screen = screen
- self._alt_screen = False
-
- self._redirect_stdout = redirect_stdout
- self._redirect_stderr = redirect_stderr
- self._restore_stdout: Optional[IO[str]] = None
- self._restore_stderr: Optional[IO[str]] = None
-
- self._lock = RLock()
- self.ipy_widget: Optional[Any] = None
- self.auto_refresh = auto_refresh
- self._started: bool = False
- self.transient = True if screen else transient
-
- self._refresh_thread: Optional[_RefreshThread] = None
- 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
- )
-
- @property
- def is_started(self) -> bool:
- """Check if live display has been started."""
- return self._started
-
- 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: bool = 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
- if self._screen:
- self._alt_screen = self.console.set_alt_screen(True)
- self.console.show_cursor(False)
- self._enable_redirect_io()
- self.console.push_render_hook(self)
- if refresh:
- try:
- self.refresh()
- except Exception:
- # If refresh fails, we want to stop the redirection of sys.stderr,
- # so the error stacktrace is properly displayed in the terminal.
- # (or, if the code that calls Rich captures the exception and wants to display something,
- # let this be displayed in the terminal).
- self.stop()
- raise
- if self.auto_refresh:
- self._refresh_thread = _RefreshThread(self, self.refresh_per_second)
- self._refresh_thread.start()
-
- def stop(self) -> None:
- """Stop live rendering display."""
- with self._lock:
- if not self._started:
- return
- self.console.clear_live()
- self._started = False
-
- if self.auto_refresh and self._refresh_thread is not None:
- self._refresh_thread.stop()
- self._refresh_thread = None
- # allow it to fully render on the last even if overflow
- self.vertical_overflow = "visible"
- with self.console:
- try:
- if not self._alt_screen and not self.console.is_jupyter:
- self.refresh()
- finally:
- self._disable_redirect_io()
- self.console.pop_render_hook()
- if not self._alt_screen and self.console.is_terminal:
- self.console.line()
- self.console.show_cursor(True)
- if self._alt_screen:
- self.console.set_alt_screen(False)
-
- if self.transient and not self._alt_screen:
- self.console.control(self._live_render.restore_cursor())
- if self.ipy_widget is not None and self.transient:
- self.ipy_widget.close() # pragma: no cover
-
- def __enter__(self) -> "Live":
- self.start(refresh=self._renderable is not None)
- return self
-
- def __exit__(
- self,
- exc_type: Optional[Type[BaseException]],
- exc_val: Optional[BaseException],
- exc_tb: Optional[TracebackType],
- ) -> None:
- self.stop()
-
- def _enable_redirect_io(self) -> None:
- """Enable redirecting of stdout / stderr."""
- if self.console.is_terminal or self.console.is_jupyter:
- if self._redirect_stdout and not isinstance(sys.stdout, FileProxy):
- self._restore_stdout = sys.stdout
- sys.stdout = cast("TextIO", FileProxy(self.console, sys.stdout))
- if self._redirect_stderr and not isinstance(sys.stderr, FileProxy):
- self._restore_stderr = sys.stderr
- sys.stderr = cast("TextIO", FileProxy(self.console, sys.stderr))
-
- def _disable_redirect_io(self) -> None:
- """Disable redirecting of stdout / stderr."""
- if self._restore_stdout:
- sys.stdout = cast("TextIO", self._restore_stdout)
- self._restore_stdout = None
- if self._restore_stderr:
- sys.stderr = cast("TextIO", self._restore_stderr)
- self._restore_stderr = None
-
- @property
- def renderable(self) -> RenderableType:
- """Get the renderable that is being displayed
-
- Returns:
- RenderableType: Displayed renderable.
- """
- renderable = self.get_renderable()
- return Screen(renderable) if self._alt_screen else renderable
-
- def update(self, renderable: RenderableType, *, refresh: bool = False) -> None:
- """Update the renderable that is being displayed
-
- Args:
- renderable (RenderableType): New renderable to use.
- refresh (bool, optional): Refresh the display. Defaults to False.
- """
- if isinstance(renderable, str):
- renderable = self.console.render_str(renderable)
- with self._lock:
- self._renderable = renderable
- if refresh:
- self.refresh()
-
- def refresh(self) -> None:
- """Update the display of the Live Render."""
- with self._lock:
- self._live_render.set_renderable(self.renderable)
- 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:
- 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._live_render.renderable)
- elif self.console.is_terminal and not self.console.is_dumb_terminal:
- with self.console:
- self.console.print(Control())
- elif (
- not self._started and not self.transient
- ): # if it is finished allow files or dumb-terminals to see final result
- with self.console:
- self.console.print(Control())
-
- def process_renderables(
- self, renderables: List[ConsoleRenderable]
- ) -> List[ConsoleRenderable]:
- """Process renderables to restore cursor and display 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:
- reset = (
- Control.home()
- if self._alt_screen
- else self._live_render.position_cursor()
- )
- renderables = [reset, *renderables, self._live_render]
- elif (
- not self._started and not self.transient
- ): # if it is finished render the final output for files or dumb_terminals
- renderables = [*renderables, self._live_render]
-
- return renderables
-
-
-if __name__ == "__main__": # pragma: no cover
- import random
- import time
- from itertools import cycle
- from typing import Dict, List, Tuple
-
- from .align import Align
- from .console import Console
- from .live import Live as Live
- from .panel import Panel
- from .rule import Rule
- from .syntax import Syntax
- from .table import Table
-
- console = Console()
-
- 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 = [
- "You can make the terminal shorter and taller to see the live table hide"
- "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!"),
- ]
-
- examples = cycle(progress_renderables)
-
- exchanges = [
- "SGD",
- "MYR",
- "EUR",
- "USD",
- "AUD",
- "JPY",
- "CNH",
- "HKD",
- "CAD",
- "INR",
- "DKK",
- "GBP",
- "RUB",
- "NZD",
- "MXN",
- "IDR",
- "TWD",
- "THB",
- "VND",
- ]
- with Live(console=console) as live_table:
- exchange_rate_dict: Dict[Tuple[str, str], float] = {}
-
- for index in range(100):
- select_exchange = exchanges[index % len(exchanges)]
-
- for exchange in exchanges:
- if exchange == select_exchange:
- continue
- time.sleep(0.4)
- if random.randint(0, 10) < 1:
- console.log(next(examples))
- exchange_rate_dict[(select_exchange, exchange)] = 200 / (
- (random.random() * 320) + 1
- )
- if len(exchange_rate_dict) > len(exchanges) - 1:
- exchange_rate_dict.pop(list(exchange_rate_dict.keys())[0])
- table = Table(title="Exchange Rates")
-
- table.add_column("Source Currency")
- table.add_column("Destination Currency")
- table.add_column("Exchange Rate")
-
- for (source, dest), exchange_rate in exchange_rate_dict.items():
- table.add_row(
- source,
- dest,
- Text(
- f"{exchange_rate:.4f}",
- style="red" if exchange_rate < 1.0 else "green",
- ),
- )
-
- live_table.update(Align.center(table))