Mercurial > repos > rliterman > csp2

comparison CSP2/CSP2_env/env-d9b9114564458d9d-741b3de822f2aaca6c6caa4325c4afce/lib/python3.8/site-packages/tqdm/rich.py @ 68:5028fdace37b

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
planemo upload commit 2e9511a184a1ca667c7be0c6321a36dc4e3d116d
author jpayne
date Tue, 18 Mar 2025 16:23:26 -0400
parents
children
comparison
equal deleted inserted replaced
67:0e9998148a16 68:5028fdace37b
1 """
2 `rich.progress` decorator for iterators.
3
4 Usage:
5 >>> from tqdm.rich import trange, tqdm
6 >>> for i in trange(10):
7 ... ...
8 """
9 from warnings import warn
10
11 from rich.progress import (
12 BarColumn, Progress, ProgressColumn, Text, TimeElapsedColumn, TimeRemainingColumn, filesize)
13
14 from .std import TqdmExperimentalWarning
15 from .std import tqdm as std_tqdm
16
17 __author__ = {"github.com/": ["casperdcl"]}
18 __all__ = ['tqdm_rich', 'trrange', 'tqdm', 'trange']
19
20
21 class FractionColumn(ProgressColumn):
22 """Renders completed/total, e.g. '0.5/2.3 G'."""
23 def __init__(self, unit_scale=False, unit_divisor=1000):
24 self.unit_scale = unit_scale
25 self.unit_divisor = unit_divisor
26 super().__init__()
27
28 def render(self, task):
29 """Calculate common unit for completed and total."""
30 completed = int(task.completed)
31 total = int(task.total)
32 if self.unit_scale:
33 unit, suffix = filesize.pick_unit_and_suffix(
34 total,
35 ["", "K", "M", "G", "T", "P", "E", "Z", "Y"],
36 self.unit_divisor,
37 )
38 else:
39 unit, suffix = filesize.pick_unit_and_suffix(total, [""], 1)
40 precision = 0 if unit == 1 else 1
41 return Text(
42 f"{completed/unit:,.{precision}f}/{total/unit:,.{precision}f} {suffix}",
43 style="progress.download")
44
45
46 class RateColumn(ProgressColumn):
47 """Renders human readable transfer speed."""
48 def __init__(self, unit="", unit_scale=False, unit_divisor=1000):
49 self.unit = unit
50 self.unit_scale = unit_scale
51 self.unit_divisor = unit_divisor
52 super().__init__()
53
54 def render(self, task):
55 """Show data transfer speed."""
56 speed = task.speed
57 if speed is None:
58 return Text(f"? {self.unit}/s", style="progress.data.speed")
59 if self.unit_scale:
60 unit, suffix = filesize.pick_unit_and_suffix(
61 speed,
62 ["", "K", "M", "G", "T", "P", "E", "Z", "Y"],
63 self.unit_divisor,
64 )
65 else:
66 unit, suffix = filesize.pick_unit_and_suffix(speed, [""], 1)
67 precision = 0 if unit == 1 else 1
68 return Text(f"{speed/unit:,.{precision}f} {suffix}{self.unit}/s",
69 style="progress.data.speed")
70
71
72 class tqdm_rich(std_tqdm): # pragma: no cover
73 """Experimental rich.progress GUI version of tqdm!"""
74 # TODO: @classmethod: write()?
75 def __init__(self, *args, **kwargs):
76 """
77 This class accepts the following parameters *in addition* to
78 the parameters accepted by `tqdm`.
79
80 Parameters
81 ----------
82 progress : tuple, optional
83 arguments for `rich.progress.Progress()`.
84 options : dict, optional
85 keyword arguments for `rich.progress.Progress()`.
86 """
87 kwargs = kwargs.copy()
88 kwargs['gui'] = True
89 # convert disable = None to False
90 kwargs['disable'] = bool(kwargs.get('disable', False))
91 progress = kwargs.pop('progress', None)
92 options = kwargs.pop('options', {}).copy()
93 super().__init__(*args, **kwargs)
94
95 if self.disable:
96 return
97
98 warn("rich is experimental/alpha", TqdmExperimentalWarning, stacklevel=2)
99 d = self.format_dict
100 if progress is None:
101 progress = (
102 "[progress.description]{task.description}"
103 "[progress.percentage]{task.percentage:>4.0f}%",
104 BarColumn(bar_width=None),
105 FractionColumn(
106 unit_scale=d['unit_scale'], unit_divisor=d['unit_divisor']),
107 "[", TimeElapsedColumn(), "<", TimeRemainingColumn(),
108 ",", RateColumn(unit=d['unit'], unit_scale=d['unit_scale'],
109 unit_divisor=d['unit_divisor']), "]"
110 )
111 options.setdefault('transient', not self.leave)
112 self._prog = Progress(*progress, **options)
113 self._prog.__enter__()
114 self._task_id = self._prog.add_task(self.desc or "", **d)
115
116 def close(self):
117 if self.disable:
118 return
119 self.display() # print 100%, vis #1306
120 super().close()
121 self._prog.__exit__(None, None, None)
122
123 def clear(self, *_, **__):
124 pass
125
126 def display(self, *_, **__):
127 if not hasattr(self, '_prog'):
128 return
129 self._prog.update(self._task_id, completed=self.n, description=self.desc)
130
131 def reset(self, total=None):
132 """
133 Resets to 0 iterations for repeated use.
134
135 Parameters
136 ----------
137 total : int or float, optional. Total to use for the new bar.
138 """
139 if hasattr(self, '_prog'):
140 self._prog.reset(total=total)
141 super().reset(total=total)
142
143
144 def trrange(*args, **kwargs):
145 """Shortcut for `tqdm.rich.tqdm(range(*args), **kwargs)`."""
146 return tqdm_rich(range(*args), **kwargs)
147
148
149 # Aliases
150 tqdm = tqdm_rich
151 trange = trrange