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
|
"""
Inserting `tqdm` as a "pipe" in a chain of coroutines.
Not to be confused with `asyncio.coroutine`.
"""
from functools import wraps
from tqdm.auto import tqdm
def autonext(func):
@wraps(func)
def inner(*args, **kwargs):
res = func(*args, **kwargs)
next(res)
return res
return inner
@autonext
def tqdm_pipe(target, **tqdm_kwargs):
"""
Coroutine chain pipe `send()`ing to `target`.
This:
>>> r = receiver()
>>> p = producer(r)
>>> next(r)
>>> next(p)
Becomes:
>>> r = receiver()
>>> t = tqdm.pipe(r)
>>> p = producer(t)
>>> next(r)
>>> next(p)
"""
with tqdm(**tqdm_kwargs) as pbar:
while True:
obj = (yield)
target.send(obj)
pbar.update()
def source(target):
for i in ["foo", "bar", "baz", "pythonista", "python", "py"]:
target.send(i)
target.close()
@autonext
def grep(pattern, target):
while True:
line = (yield)
if pattern in line:
target.send(line)
@autonext
def sink():
while True:
line = (yield)
tqdm.write(line)
if __name__ == "__main__":
source(
tqdm_pipe(
grep('python',
sink())))
|
