venv/lib/python3.11/site-packages/pip/_vendor/rich/live.py

   1 import sys
   2 from threading import Event, RLock, Thread
   3 from types import TracebackType
   4 from typing import IO, Any, Callable, List, Optional, TextIO, Type, cast
   5
   6 from . import get_console
   7 from .console import Console, ConsoleRenderable, RenderableType, RenderHook
   8 from .control import Control
   9 from .file_proxy import FileProxy
  10 from .jupyter import JupyterMixin
  11 from .live_render import LiveRender, VerticalOverflowMethod
  12 from .screen import Screen
  13 from .text import Text
  14
  15
  16 class _RefreshThread(Thread):
  17     """A thread that calls refresh() at regular intervals."""
  18
  19     def __init__(self, live: "Live", refresh_per_second: float) -> None:
  20         self.live = live
  21         self.refresh_per_second = refresh_per_second
  22         self.done = Event()
  23         super().__init__(daemon=True)
  24
  25     def stop(self) -> None:
  26         self.done.set()
  27
  28     def run(self) -> None:
  29         while not self.done.wait(1 / self.refresh_per_second):
  30             with self.live._lock:
  31                 if not self.done.is_set():
  32                     self.live.refresh()
  33
  34
  35 class Live(JupyterMixin, RenderHook):
  36     """Renders an auto-updating live display of any given renderable.
  37
  38     Args:
  39         renderable (RenderableType, optional): The renderable to live display. Defaults to displaying nothing.
  40         console (Console, optional): Optional Console instance. Default will an internal Console instance writing to stdout.
  41         screen (bool, optional): Enable alternate screen mode. Defaults to False.
  42         auto_refresh (bool, optional): Enable auto refresh. If disabled, you will need to call `refresh()` or `update()` with refresh flag. Defaults to True
  43         refresh_per_second (float, optional): Number of times per second to refresh the live display. Defaults to 4.
  44         transient (bool, optional): Clear the renderable on exit (has no effect when screen=True). Defaults to False.
  45         redirect_stdout (bool, optional): Enable redirection of stdout, so ``print`` may be used. Defaults to True.
  46         redirect_stderr (bool, optional): Enable redirection of stderr. Defaults to True.
  47         vertical_overflow (VerticalOverflowMethod, optional): How to handle renderable when it is too tall for the console. Defaults to "ellipsis".
  48         get_renderable (Callable[[], RenderableType], optional): Optional callable to get renderable. Defaults to None.
  49     """
  50
  51     def __init__(
  52         self,
  53         renderable: Optional[RenderableType] = None,
  54         *,
  55         console: Optional[Console] = None,
  56         screen: bool = False,
  57         auto_refresh: bool = True,
  58         refresh_per_second: float = 4,
  59         transient: bool = False,
  60         redirect_stdout: bool = True,
  61         redirect_stderr: bool = True,
  62         vertical_overflow: VerticalOverflowMethod = "ellipsis",
  63         get_renderable: Optional[Callable[[], RenderableType]] = None,
  64     ) -> None:
  65         assert refresh_per_second > 0, "refresh_per_second must be > 0"
  66         self._renderable = renderable
  67         self.console = console if console is not None else get_console()
  68         self._screen = screen
  69         self._alt_screen = False
  70
  71         self._redirect_stdout = redirect_stdout
  72         self._redirect_stderr = redirect_stderr
  73         self._restore_stdout: Optional[IO[str]] = None
  74         self._restore_stderr: Optional[IO[str]] = None
  75
  76         self._lock = RLock()
  77         self.ipy_widget: Optional[Any] = None
  78         self.auto_refresh = auto_refresh
  79         self._started: bool = False
  80         self.transient = True if screen else transient
  81
  82         self._refresh_thread: Optional[_RefreshThread] = None
  83         self.refresh_per_second = refresh_per_second
  84
  85         self.vertical_overflow = vertical_overflow
  86         self._get_renderable = get_renderable
  87         self._live_render = LiveRender(
  88             self.get_renderable(), vertical_overflow=vertical_overflow
  89         )
  90
  91     @property
  92     def is_started(self) -> bool:
  93         """Check if live display has been started."""
  94         return self._started
  95
  96     def get_renderable(self) -> RenderableType:
  97         renderable = (
  98             self._get_renderable()
  99             if self._get_renderable is not None
 100             else self._renderable
 101         )
 102         return renderable or ""
 103
 104     def start(self, refresh: bool = False) -> None:
 105         """Start live rendering display.
 106
 107         Args:
 108             refresh (bool, optional): Also refresh. Defaults to False.
 109         """
 110         with self._lock:
 111             if self._started:
 112                 return
 113             self.console.set_live(self)
 114             self._started = True
 115             if self._screen:
 116                 self._alt_screen = self.console.set_alt_screen(True)
 117             self.console.show_cursor(False)
 118             self._enable_redirect_io()
 119             self.console.push_render_hook(self)
 120             if refresh:
 121                 try:
 122                     self.refresh()
 123                 except Exception:
 124                     # If refresh fails, we want to stop the redirection of sys.stderr,
 125                     # so the error stacktrace is properly displayed in the terminal.
 126                     # (or, if the code that calls Rich captures the exception and wants to display something,
 127                     # let this be displayed in the terminal).
 128                     self.stop()
 129                     raise
 130             if self.auto_refresh:
 131                 self._refresh_thread = _RefreshThread(self, self.refresh_per_second)
 132                 self._refresh_thread.start()
 133
 134     def stop(self) -> None:
 135         """Stop live rendering display."""
 136         with self._lock:
 137             if not self._started:
 138                 return
 139             self.console.clear_live()
 140             self._started = False
 141
 142             if self.auto_refresh and self._refresh_thread is not None:
 143                 self._refresh_thread.stop()
 144                 self._refresh_thread = None
 145             # allow it to fully render on the last even if overflow
 146             self.vertical_overflow = "visible"
 147             with self.console:
 148                 try:
 149                     if not self._alt_screen and not self.console.is_jupyter:
 150                         self.refresh()
 151                 finally:
 152                     self._disable_redirect_io()
 153                     self.console.pop_render_hook()
 154                     if not self._alt_screen and self.console.is_terminal:
 155                         self.console.line()
 156                     self.console.show_cursor(True)
 157                     if self._alt_screen:
 158                         self.console.set_alt_screen(False)
 159
 160                     if self.transient and not self._alt_screen:
 161                         self.console.control(self._live_render.restore_cursor())
 162                     if self.ipy_widget is not None and self.transient:
 163                         self.ipy_widget.close()  # pragma: no cover
 164
 165     def __enter__(self) -> "Live":
 166         self.start(refresh=self._renderable is not None)
 167         return self
 168
 169     def __exit__(
 170         self,
 171         exc_type: Optional[Type[BaseException]],
 172         exc_val: Optional[BaseException],
 173         exc_tb: Optional[TracebackType],
 174     ) -> None:
 175         self.stop()
 176
 177     def _enable_redirect_io(self) -> None:
 178         """Enable redirecting of stdout / stderr."""
 179         if self.console.is_terminal or self.console.is_jupyter:
 180             if self._redirect_stdout and not isinstance(sys.stdout, FileProxy):
 181                 self._restore_stdout = sys.stdout
 182                 sys.stdout = cast("TextIO", FileProxy(self.console, sys.stdout))
 183             if self._redirect_stderr and not isinstance(sys.stderr, FileProxy):
 184                 self._restore_stderr = sys.stderr
 185                 sys.stderr = cast("TextIO", FileProxy(self.console, sys.stderr))
 186
 187     def _disable_redirect_io(self) -> None:
 188         """Disable redirecting of stdout / stderr."""
 189         if self._restore_stdout:
 190             sys.stdout = cast("TextIO", self._restore_stdout)
 191             self._restore_stdout = None
 192         if self._restore_stderr:
 193             sys.stderr = cast("TextIO", self._restore_stderr)
 194             self._restore_stderr = None
 195
 196     @property
 197     def renderable(self) -> RenderableType:
 198         """Get the renderable that is being displayed
 199
 200         Returns:
 201             RenderableType: Displayed renderable.
 202         """
 203         renderable = self.get_renderable()
 204         return Screen(renderable) if self._alt_screen else renderable
 205
 206     def update(self, renderable: RenderableType, *, refresh: bool = False) -> None:
 207         """Update the renderable that is being displayed
 208
 209         Args:
 210             renderable (RenderableType): New renderable to use.
 211             refresh (bool, optional): Refresh the display. Defaults to False.
 212         """
 213         if isinstance(renderable, str):
 214             renderable = self.console.render_str(renderable)
 215         with self._lock:
 216             self._renderable = renderable
 217             if refresh:
 218                 self.refresh()
 219
 220     def refresh(self) -> None:
 221         """Update the display of the Live Render."""
 222         with self._lock:
 223             self._live_render.set_renderable(self.renderable)
 224             if self.console.is_jupyter:  # pragma: no cover
 225                 try:
 226                     from IPython.display import display
 227                     from ipywidgets import Output
 228                 except ImportError:
 229                     import warnings
 230
 231                     warnings.warn('install "ipywidgets" for Jupyter support')
 232                 else:
 233                     if self.ipy_widget is None:
 234                         self.ipy_widget = Output()
 235                         display(self.ipy_widget)
 236
 237                     with self.ipy_widget:
 238                         self.ipy_widget.clear_output(wait=True)
 239                         self.console.print(self._live_render.renderable)
 240             elif self.console.is_terminal and not self.console.is_dumb_terminal:
 241                 with self.console:
 242                     self.console.print(Control())
 243             elif (
 244                 not self._started and not self.transient
 245             ):  # if it is finished allow files or dumb-terminals to see final result
 246                 with self.console:
 247                     self.console.print(Control())
 248
 249     def process_renderables(
 250         self, renderables: List[ConsoleRenderable]
 251     ) -> List[ConsoleRenderable]:
 252         """Process renderables to restore cursor and display progress."""
 253         self._live_render.vertical_overflow = self.vertical_overflow
 254         if self.console.is_interactive:
 255             # lock needs acquiring as user can modify live_render renderable at any time unlike in Progress.
 256             with self._lock:
 257                 reset = (
 258                     Control.home()
 259                     if self._alt_screen
 260                     else self._live_render.position_cursor()
 261                 )
 262                 renderables = [reset, *renderables, self._live_render]
 263         elif (
 264             not self._started and not self.transient
 265         ):  # if it is finished render the final output for files or dumb_terminals
 266             renderables = [*renderables, self._live_render]
 267
 268         return renderables
 269
 270
 271 if __name__ == "__main__":  # pragma: no cover
 272     import random
 273     import time
 274     from itertools import cycle
 275     from typing import Dict, List, Tuple
 276
 277     from .align import Align
 278     from .console import Console
 279     from .live import Live as Live
 280     from .panel import Panel
 281     from .rule import Rule
 282     from .syntax import Syntax
 283     from .table import Table
 284
 285     console = Console()
 286
 287     syntax = Syntax(
 288         '''def loop_last(values: Iterable[T]) -> Iterable[Tuple[bool, T]]:
 289     """Iterate and generate a tuple with a flag for last value."""
 290     iter_values = iter(values)
 291     try:
 292         previous_value = next(iter_values)
 293     except StopIteration:
 294         return
 295     for value in iter_values:
 296         yield False, previous_value
 297         previous_value = value
 298     yield True, previous_value''',
 299         "python",
 300         line_numbers=True,
 301     )
 302
 303     table = Table("foo", "bar", "baz")
 304     table.add_row("1", "2", "3")
 305
 306     progress_renderables = [
 307         "You can make the terminal shorter and taller to see the live table hide"
 308         "Text may be printed while the progress bars are rendering.",
 309         Panel("In fact, [i]any[/i] renderable will work"),
 310         "Such as [magenta]tables[/]...",
 311         table,
 312         "Pretty printed structures...",
 313         {"type": "example", "text": "Pretty printed"},
 314         "Syntax...",
 315         syntax,
 316         Rule("Give it a try!"),
 317     ]
 318
 319     examples = cycle(progress_renderables)
 320
 321     exchanges = [
 322         "SGD",
 323         "MYR",
 324         "EUR",
 325         "USD",
 326         "AUD",
 327         "JPY",
 328         "CNH",
 329         "HKD",
 330         "CAD",
 331         "INR",
 332         "DKK",
 333         "GBP",
 334         "RUB",
 335         "NZD",
 336         "MXN",
 337         "IDR",
 338         "TWD",
 339         "THB",
 340         "VND",
 341     ]
 342     with Live(console=console) as live_table:
 343         exchange_rate_dict: Dict[Tuple[str, str], float] = {}
 344
 345         for index in range(100):
 346             select_exchange = exchanges[index % len(exchanges)]
 347
 348             for exchange in exchanges:
 349                 if exchange == select_exchange:
 350                     continue
 351                 time.sleep(0.4)
 352                 if random.randint(0, 10) < 1:
 353                     console.log(next(examples))
 354                 exchange_rate_dict[(select_exchange, exchange)] = 200 / (
 355                     (random.random() * 320) + 1
 356                 )
 357                 if len(exchange_rate_dict) > len(exchanges) - 1:
 358                     exchange_rate_dict.pop(list(exchange_rate_dict.keys())[0])
 359                 table = Table(title="Exchange Rates")
 360
 361                 table.add_column("Source Currency")
 362                 table.add_column("Destination Currency")
 363                 table.add_column("Exchange Rate")
 364
 365                 for ((source, dest), exchange_rate) in exchange_rate_dict.items():
 366                     table.add_row(
 367                         source,
 368                         dest,
 369                         Text(
 370                             f"{exchange_rate:.4f}",
 371                             style="red" if exchange_rate < 1.0 else "green",
 372                         ),
 373                     )
 374
 375                 live_table.update(Align.center(table))