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
|
#!/usr/bin/env python
"""
More complex demonstration of what's possible with the progress bar.
"""
import random
import threading
import time
from prompt_toolkit import HTML
from prompt_toolkit.shortcuts import ProgressBar
def main():
with ProgressBar(
title=HTML("<b>Example of many parallel tasks.</b>"),
bottom_toolbar=HTML("<b>[Control-L]</b> clear <b>[Control-C]</b> abort"),
) as pb:
def run_task(label, total, sleep_time):
"""Complete a normal run."""
for i in pb(range(total), label=label):
time.sleep(sleep_time)
def stop_task(label, total, sleep_time):
"""Stop at some random index.
Breaking out of iteration at some stop index mimics how progress
bars behave in cases where errors are raised.
"""
stop_i = random.randrange(total)
bar = pb(range(total), label=label)
for i in bar:
if stop_i == i:
bar.label = f"{label} BREAK"
break
time.sleep(sleep_time)
threads = []
for i in range(160):
label = "Task %i" % i
total = random.randrange(50, 200)
sleep_time = random.randrange(5, 20) / 100.0
threads.append(
threading.Thread(
target=random.choice((run_task, stop_task)),
args=(label, total, sleep_time),
)
)
for t in threads:
t.daemon = True
t.start()
# Wait for the threads to finish. We use a timeout for the join() call,
# because on Windows, join cannot be interrupted by Control-C or any other
# signal.
for t in threads:
while t.is_alive():
t.join(timeout=0.5)
if __name__ == "__main__":
main()
|
