python (3.11.7)

Browse
Build Log
Usage
       1  import functools
       2  from typing import Callable, Generator, Iterable, Iterator, Optional, Tuple
       3  
       4  from pip._vendor.rich.progress import (
       5      BarColumn,
       6      DownloadColumn,
       7      FileSizeColumn,
       8      Progress,
       9      ProgressColumn,
      10      SpinnerColumn,
      11      TextColumn,
      12      TimeElapsedColumn,
      13      TimeRemainingColumn,
      14      TransferSpeedColumn,
      15  )
      16  
      17  from pip._internal.utils.logging import get_indentation
      18  
      19  DownloadProgressRenderer = Callable[[Iterable[bytes]], Iterator[bytes]]
      20  
      21  
      22  def _rich_progress_bar(
      23      iterable: Iterable[bytes],
      24      *,
      25      bar_type: str,
      26      size: int,
      27  ) -> Generator[bytes, None, None]:
      28      assert bar_type == "on", "This should only be used in the default mode."
      29  
      30      if not size:
      31          total = float("inf")
      32          columns: Tuple[ProgressColumn, ...] = (
      33              TextColumn("[progress.description]{task.description}"),
      34              SpinnerColumn("line", speed=1.5),
      35              FileSizeColumn(),
      36              TransferSpeedColumn(),
      37              TimeElapsedColumn(),
      38          )
      39      else:
      40          total = size
      41          columns = (
      42              TextColumn("[progress.description]{task.description}"),
      43              BarColumn(),
      44              DownloadColumn(),
      45              TransferSpeedColumn(),
      46              TextColumn("eta"),
      47              TimeRemainingColumn(),
      48          )
      49  
      50      progress = Progress(*columns, refresh_per_second=30)
      51      task_id = progress.add_task(" " * (get_indentation() + 2), total=total)
      52      with progress:
      53          for chunk in iterable:
      54              yield chunk
      55              progress.update(task_id, advance=len(chunk))
      56  
      57  
      58  def get_download_progress_renderer(
      59      *, bar_type: str, size: Optional[int] = None
      60  ) -> DownloadProgressRenderer:
      61      """Get an object that can be used to render the download progress.
      62  
      63      Returns a callable, that takes an iterable to "wrap".
      64      """
      65      if bar_type == "on":
      66          return functools.partial(_rich_progress_bar, bar_type=bar_type, size=size)
      67      else:
      68          return iter  # no-op, when passed an iterator