From 12cf076118570eebbff08c6b3090e0d4798447a1 Mon Sep 17 00:00:00 2001 From: cyfraeviolae Date: Wed, 3 Apr 2024 03:17:55 -0400 Subject: no venv --- venv/lib/python3.11/site-packages/rich/live.py | 375 ------------------------- 1 file changed, 375 deletions(-) delete mode 100644 venv/lib/python3.11/site-packages/rich/live.py (limited to 'venv/lib/python3.11/site-packages/rich/live.py') 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)) -- cgit v1.2.3