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