Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/humanfriendly/terminal/spinners.py @ 0:26e78fe6e8c4 draft
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author | shellac |
---|---|
date | Sat, 02 May 2020 07:14:21 -0400 |
parents | |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.7/site-packages/humanfriendly/terminal/spinners.py Sat May 02 07:14:21 2020 -0400 @@ -0,0 +1,310 @@ +# Human friendly input/output in Python. +# +# Author: Peter Odding <peter@peterodding.com> +# Last Change: March 1, 2020 +# URL: https://humanfriendly.readthedocs.io + +""" +Support for spinners that represent progress on interactive terminals. + +The :class:`Spinner` class shows a "spinner" on the terminal to let the user +know that something is happening during long running operations that would +otherwise be silent (leaving the user to wonder what they're waiting for). +Below are some visual examples that should illustrate the point. + +**Simple spinners:** + + Here's a screen capture that shows the simplest form of spinner: + + .. image:: images/spinner-basic.gif + :alt: Animated screen capture of a simple spinner. + + The following code was used to create the spinner above: + + .. code-block:: python + + import itertools + import time + from humanfriendly import Spinner + + with Spinner(label="Downloading") as spinner: + for i in itertools.count(): + # Do something useful here. + time.sleep(0.1) + # Advance the spinner. + spinner.step() + +**Spinners that show elapsed time:** + + Here's a spinner that shows the elapsed time since it started: + + .. image:: images/spinner-with-timer.gif + :alt: Animated screen capture of a spinner showing elapsed time. + + The following code was used to create the spinner above: + + .. code-block:: python + + import itertools + import time + from humanfriendly import Spinner, Timer + + with Spinner(label="Downloading", timer=Timer()) as spinner: + for i in itertools.count(): + # Do something useful here. + time.sleep(0.1) + # Advance the spinner. + spinner.step() + +**Spinners that show progress:** + + Here's a spinner that shows a progress percentage: + + .. image:: images/spinner-with-progress.gif + :alt: Animated screen capture of spinner showing progress. + + The following code was used to create the spinner above: + + .. code-block:: python + + import itertools + import random + import time + from humanfriendly import Spinner, Timer + + with Spinner(label="Downloading", total=100) as spinner: + progress = 0 + while progress < 100: + # Do something useful here. + time.sleep(0.1) + # Advance the spinner. + spinner.step(progress) + # Determine the new progress value. + progress += random.random() * 5 + +If you want to provide user feedback during a long running operation but it's +not practical to periodically call the :func:`~Spinner.step()` method consider +using :class:`AutomaticSpinner` instead. + +As you may already have noticed in the examples above, :class:`Spinner` objects +can be used as context managers to automatically call :func:`Spinner.clear()` +when the spinner ends. +""" + +# Standard library modules. +import multiprocessing +import sys +import time + +# Modules included in our package. +from humanfriendly import Timer +from humanfriendly.deprecation import deprecated_args +from humanfriendly.terminal import ANSI_ERASE_LINE + +# Public identifiers that require documentation. +__all__ = ("AutomaticSpinner", "GLYPHS", "MINIMUM_INTERVAL", "Spinner") + +GLYPHS = ["-", "\\", "|", "/"] +"""A list of strings with characters that together form a crude animation :-).""" + +MINIMUM_INTERVAL = 0.2 +"""Spinners are redrawn with a frequency no higher than this number (a floating point number of seconds).""" + + +class Spinner(object): + + """Show a spinner on the terminal as a simple means of feedback to the user.""" + + @deprecated_args('label', 'total', 'stream', 'interactive', 'timer') + def __init__(self, **options): + """ + Initialize a :class:`Spinner` object. + + :param label: + + The label for the spinner (a string or :data:`None`, defaults to + :data:`None`). + + :param total: + + The expected number of steps (an integer or :data:`None`). If this is + provided the spinner will show a progress percentage. + + :param stream: + + The output stream to show the spinner on (a file-like object, + defaults to :data:`sys.stderr`). + + :param interactive: + + :data:`True` to enable rendering of the spinner, :data:`False` to + disable (defaults to the result of ``stream.isatty()``). + + :param timer: + + A :class:`.Timer` object (optional). If this is given the spinner + will show the elapsed time according to the timer. + + :param interval: + + The spinner will be updated at most once every this many seconds + (a floating point number, defaults to :data:`MINIMUM_INTERVAL`). + + :param glyphs: + + A list of strings with single characters that are drawn in the same + place in succession to implement a simple animated effect (defaults + to :data:`GLYPHS`). + """ + # Store initializer arguments. + self.interactive = options.get('interactive') + self.interval = options.get('interval', MINIMUM_INTERVAL) + self.label = options.get('label') + self.states = options.get('glyphs', GLYPHS) + self.stream = options.get('stream', sys.stderr) + self.timer = options.get('timer') + self.total = options.get('total') + # Define instance variables. + self.counter = 0 + self.last_update = 0 + # Try to automatically discover whether the stream is connected to + # a terminal, but don't fail if no isatty() method is available. + if self.interactive is None: + try: + self.interactive = self.stream.isatty() + except Exception: + self.interactive = False + + def step(self, progress=0, label=None): + """ + Advance the spinner by one step and redraw it. + + :param progress: The number of the current step, relative to the total + given to the :class:`Spinner` constructor (an integer, + optional). If not provided the spinner will not show + progress. + :param label: The label to use while redrawing (a string, optional). If + not provided the label given to the :class:`Spinner` + constructor is used instead. + + This method advances the spinner by one step without starting a new + line, causing an animated effect which is very simple but much nicer + than waiting for a prompt which is completely silent for a long time. + + .. note:: This method uses time based rate limiting to avoid redrawing + the spinner too frequently. If you know you're dealing with + code that will call :func:`step()` at a high frequency, + consider using :func:`sleep()` to avoid creating the + equivalent of a busy loop that's rate limiting the spinner + 99% of the time. + """ + if self.interactive: + time_now = time.time() + if time_now - self.last_update >= self.interval: + self.last_update = time_now + state = self.states[self.counter % len(self.states)] + label = label or self.label + if not label: + raise Exception("No label set for spinner!") + elif self.total and progress: + label = "%s: %.2f%%" % (label, progress / (self.total / 100.0)) + elif self.timer and self.timer.elapsed_time > 2: + label = "%s (%s)" % (label, self.timer.rounded) + self.stream.write("%s %s %s ..\r" % (ANSI_ERASE_LINE, state, label)) + self.counter += 1 + + def sleep(self): + """ + Sleep for a short period before redrawing the spinner. + + This method is useful when you know you're dealing with code that will + call :func:`step()` at a high frequency. It will sleep for the interval + with which the spinner is redrawn (less than a second). This avoids + creating the equivalent of a busy loop that's rate limiting the + spinner 99% of the time. + + This method doesn't redraw the spinner, you still have to call + :func:`step()` in order to do that. + """ + time.sleep(MINIMUM_INTERVAL) + + def clear(self): + """ + Clear the spinner. + + The next line which is shown on the standard output or error stream + after calling this method will overwrite the line that used to show the + spinner. + """ + if self.interactive: + self.stream.write(ANSI_ERASE_LINE) + + def __enter__(self): + """ + Enable the use of spinners as context managers. + + :returns: The :class:`Spinner` object. + """ + return self + + def __exit__(self, exc_type=None, exc_value=None, traceback=None): + """Clear the spinner when leaving the context.""" + self.clear() + + +class AutomaticSpinner(object): + + """ + Show a spinner on the terminal that automatically starts animating. + + This class shows a spinner on the terminal (just like :class:`Spinner` + does) that automatically starts animating. This class should be used as a + context manager using the :keyword:`with` statement. The animation + continues for as long as the context is active. + + :class:`AutomaticSpinner` provides an alternative to :class:`Spinner` + for situations where it is not practical for the caller to periodically + call :func:`~Spinner.step()` to advance the animation, e.g. because + you're performing a blocking call and don't fancy implementing threading or + subprocess handling just to provide some user feedback. + + This works using the :mod:`multiprocessing` module by spawning a + subprocess to render the spinner while the main process is busy doing + something more useful. By using the :keyword:`with` statement you're + guaranteed that the subprocess is properly terminated at the appropriate + time. + """ + + def __init__(self, label, show_time=True): + """ + Initialize an automatic spinner. + + :param label: The label for the spinner (a string). + :param show_time: If this is :data:`True` (the default) then the spinner + shows elapsed time. + """ + self.label = label + self.show_time = show_time + self.shutdown_event = multiprocessing.Event() + self.subprocess = multiprocessing.Process(target=self._target) + + def __enter__(self): + """Enable the use of automatic spinners as context managers.""" + self.subprocess.start() + + def __exit__(self, exc_type=None, exc_value=None, traceback=None): + """Enable the use of automatic spinners as context managers.""" + self.shutdown_event.set() + self.subprocess.join() + + def _target(self): + try: + timer = Timer() if self.show_time else None + with Spinner(label=self.label, timer=timer) as spinner: + while not self.shutdown_event.is_set(): + spinner.step() + spinner.sleep() + except KeyboardInterrupt: + # Swallow Control-C signals without producing a nasty traceback that + # won't make any sense to the average user. + pass