Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/humanfriendly/terminal/spinners.py @ 2:6af9afd405e9 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 0a63dd5f4d38a1f6944587f52a8cd79874177fc1"
author shellac
date Thu, 14 May 2020 14:56:58 -0400
parents 26e78fe6e8c4
children
comparison
equal deleted inserted replaced
1:75ca89e9b81c 2:6af9afd405e9
1 # Human friendly input/output in Python.
2 #
3 # Author: Peter Odding <peter@peterodding.com>
4 # Last Change: March 1, 2020
5 # URL: https://humanfriendly.readthedocs.io
6
7 """
8 Support for spinners that represent progress on interactive terminals.
9
10 The :class:`Spinner` class shows a "spinner" on the terminal to let the user
11 know that something is happening during long running operations that would
12 otherwise be silent (leaving the user to wonder what they're waiting for).
13 Below are some visual examples that should illustrate the point.
14
15 **Simple spinners:**
16
17 Here's a screen capture that shows the simplest form of spinner:
18
19 .. image:: images/spinner-basic.gif
20 :alt: Animated screen capture of a simple spinner.
21
22 The following code was used to create the spinner above:
23
24 .. code-block:: python
25
26 import itertools
27 import time
28 from humanfriendly import Spinner
29
30 with Spinner(label="Downloading") as spinner:
31 for i in itertools.count():
32 # Do something useful here.
33 time.sleep(0.1)
34 # Advance the spinner.
35 spinner.step()
36
37 **Spinners that show elapsed time:**
38
39 Here's a spinner that shows the elapsed time since it started:
40
41 .. image:: images/spinner-with-timer.gif
42 :alt: Animated screen capture of a spinner showing elapsed time.
43
44 The following code was used to create the spinner above:
45
46 .. code-block:: python
47
48 import itertools
49 import time
50 from humanfriendly import Spinner, Timer
51
52 with Spinner(label="Downloading", timer=Timer()) as spinner:
53 for i in itertools.count():
54 # Do something useful here.
55 time.sleep(0.1)
56 # Advance the spinner.
57 spinner.step()
58
59 **Spinners that show progress:**
60
61 Here's a spinner that shows a progress percentage:
62
63 .. image:: images/spinner-with-progress.gif
64 :alt: Animated screen capture of spinner showing progress.
65
66 The following code was used to create the spinner above:
67
68 .. code-block:: python
69
70 import itertools
71 import random
72 import time
73 from humanfriendly import Spinner, Timer
74
75 with Spinner(label="Downloading", total=100) as spinner:
76 progress = 0
77 while progress < 100:
78 # Do something useful here.
79 time.sleep(0.1)
80 # Advance the spinner.
81 spinner.step(progress)
82 # Determine the new progress value.
83 progress += random.random() * 5
84
85 If you want to provide user feedback during a long running operation but it's
86 not practical to periodically call the :func:`~Spinner.step()` method consider
87 using :class:`AutomaticSpinner` instead.
88
89 As you may already have noticed in the examples above, :class:`Spinner` objects
90 can be used as context managers to automatically call :func:`Spinner.clear()`
91 when the spinner ends.
92 """
93
94 # Standard library modules.
95 import multiprocessing
96 import sys
97 import time
98
99 # Modules included in our package.
100 from humanfriendly import Timer
101 from humanfriendly.deprecation import deprecated_args
102 from humanfriendly.terminal import ANSI_ERASE_LINE
103
104 # Public identifiers that require documentation.
105 __all__ = ("AutomaticSpinner", "GLYPHS", "MINIMUM_INTERVAL", "Spinner")
106
107 GLYPHS = ["-", "\\", "|", "/"]
108 """A list of strings with characters that together form a crude animation :-)."""
109
110 MINIMUM_INTERVAL = 0.2
111 """Spinners are redrawn with a frequency no higher than this number (a floating point number of seconds)."""
112
113
114 class Spinner(object):
115
116 """Show a spinner on the terminal as a simple means of feedback to the user."""
117
118 @deprecated_args('label', 'total', 'stream', 'interactive', 'timer')
119 def __init__(self, **options):
120 """
121 Initialize a :class:`Spinner` object.
122
123 :param label:
124
125 The label for the spinner (a string or :data:`None`, defaults to
126 :data:`None`).
127
128 :param total:
129
130 The expected number of steps (an integer or :data:`None`). If this is
131 provided the spinner will show a progress percentage.
132
133 :param stream:
134
135 The output stream to show the spinner on (a file-like object,
136 defaults to :data:`sys.stderr`).
137
138 :param interactive:
139
140 :data:`True` to enable rendering of the spinner, :data:`False` to
141 disable (defaults to the result of ``stream.isatty()``).
142
143 :param timer:
144
145 A :class:`.Timer` object (optional). If this is given the spinner
146 will show the elapsed time according to the timer.
147
148 :param interval:
149
150 The spinner will be updated at most once every this many seconds
151 (a floating point number, defaults to :data:`MINIMUM_INTERVAL`).
152
153 :param glyphs:
154
155 A list of strings with single characters that are drawn in the same
156 place in succession to implement a simple animated effect (defaults
157 to :data:`GLYPHS`).
158 """
159 # Store initializer arguments.
160 self.interactive = options.get('interactive')
161 self.interval = options.get('interval', MINIMUM_INTERVAL)
162 self.label = options.get('label')
163 self.states = options.get('glyphs', GLYPHS)
164 self.stream = options.get('stream', sys.stderr)
165 self.timer = options.get('timer')
166 self.total = options.get('total')
167 # Define instance variables.
168 self.counter = 0
169 self.last_update = 0
170 # Try to automatically discover whether the stream is connected to
171 # a terminal, but don't fail if no isatty() method is available.
172 if self.interactive is None:
173 try:
174 self.interactive = self.stream.isatty()
175 except Exception:
176 self.interactive = False
177
178 def step(self, progress=0, label=None):
179 """
180 Advance the spinner by one step and redraw it.
181
182 :param progress: The number of the current step, relative to the total
183 given to the :class:`Spinner` constructor (an integer,
184 optional). If not provided the spinner will not show
185 progress.
186 :param label: The label to use while redrawing (a string, optional). If
187 not provided the label given to the :class:`Spinner`
188 constructor is used instead.
189
190 This method advances the spinner by one step without starting a new
191 line, causing an animated effect which is very simple but much nicer
192 than waiting for a prompt which is completely silent for a long time.
193
194 .. note:: This method uses time based rate limiting to avoid redrawing
195 the spinner too frequently. If you know you're dealing with
196 code that will call :func:`step()` at a high frequency,
197 consider using :func:`sleep()` to avoid creating the
198 equivalent of a busy loop that's rate limiting the spinner
199 99% of the time.
200 """
201 if self.interactive:
202 time_now = time.time()
203 if time_now - self.last_update >= self.interval:
204 self.last_update = time_now
205 state = self.states[self.counter % len(self.states)]
206 label = label or self.label
207 if not label:
208 raise Exception("No label set for spinner!")
209 elif self.total and progress:
210 label = "%s: %.2f%%" % (label, progress / (self.total / 100.0))
211 elif self.timer and self.timer.elapsed_time > 2:
212 label = "%s (%s)" % (label, self.timer.rounded)
213 self.stream.write("%s %s %s ..\r" % (ANSI_ERASE_LINE, state, label))
214 self.counter += 1
215
216 def sleep(self):
217 """
218 Sleep for a short period before redrawing the spinner.
219
220 This method is useful when you know you're dealing with code that will
221 call :func:`step()` at a high frequency. It will sleep for the interval
222 with which the spinner is redrawn (less than a second). This avoids
223 creating the equivalent of a busy loop that's rate limiting the
224 spinner 99% of the time.
225
226 This method doesn't redraw the spinner, you still have to call
227 :func:`step()` in order to do that.
228 """
229 time.sleep(MINIMUM_INTERVAL)
230
231 def clear(self):
232 """
233 Clear the spinner.
234
235 The next line which is shown on the standard output or error stream
236 after calling this method will overwrite the line that used to show the
237 spinner.
238 """
239 if self.interactive:
240 self.stream.write(ANSI_ERASE_LINE)
241
242 def __enter__(self):
243 """
244 Enable the use of spinners as context managers.
245
246 :returns: The :class:`Spinner` object.
247 """
248 return self
249
250 def __exit__(self, exc_type=None, exc_value=None, traceback=None):
251 """Clear the spinner when leaving the context."""
252 self.clear()
253
254
255 class AutomaticSpinner(object):
256
257 """
258 Show a spinner on the terminal that automatically starts animating.
259
260 This class shows a spinner on the terminal (just like :class:`Spinner`
261 does) that automatically starts animating. This class should be used as a
262 context manager using the :keyword:`with` statement. The animation
263 continues for as long as the context is active.
264
265 :class:`AutomaticSpinner` provides an alternative to :class:`Spinner`
266 for situations where it is not practical for the caller to periodically
267 call :func:`~Spinner.step()` to advance the animation, e.g. because
268 you're performing a blocking call and don't fancy implementing threading or
269 subprocess handling just to provide some user feedback.
270
271 This works using the :mod:`multiprocessing` module by spawning a
272 subprocess to render the spinner while the main process is busy doing
273 something more useful. By using the :keyword:`with` statement you're
274 guaranteed that the subprocess is properly terminated at the appropriate
275 time.
276 """
277
278 def __init__(self, label, show_time=True):
279 """
280 Initialize an automatic spinner.
281
282 :param label: The label for the spinner (a string).
283 :param show_time: If this is :data:`True` (the default) then the spinner
284 shows elapsed time.
285 """
286 self.label = label
287 self.show_time = show_time
288 self.shutdown_event = multiprocessing.Event()
289 self.subprocess = multiprocessing.Process(target=self._target)
290
291 def __enter__(self):
292 """Enable the use of automatic spinners as context managers."""
293 self.subprocess.start()
294
295 def __exit__(self, exc_type=None, exc_value=None, traceback=None):
296 """Enable the use of automatic spinners as context managers."""
297 self.shutdown_event.set()
298 self.subprocess.join()
299
300 def _target(self):
301 try:
302 timer = Timer() if self.show_time else None
303 with Spinner(label=self.label, timer=timer) as spinner:
304 while not self.shutdown_event.is_set():
305 spinner.step()
306 spinner.sleep()
307 except KeyboardInterrupt:
308 # Swallow Control-C signals without producing a nasty traceback that
309 # won't make any sense to the average user.
310 pass