diff env/lib/python3.9/site-packages/click/_termui_impl.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/env/lib/python3.9/site-packages/click/_termui_impl.py	Mon Mar 22 18:12:50 2021 +0000
@@ -0,0 +1,657 @@
+# -*- coding: utf-8 -*-
+"""
+This module contains implementations for the termui module. To keep the
+import time of Click down, some infrequently used functionality is
+placed in this module and only imported as needed.
+"""
+import contextlib
+import math
+import os
+import sys
+import time
+
+from ._compat import _default_text_stdout
+from ._compat import CYGWIN
+from ._compat import get_best_encoding
+from ._compat import int_types
+from ._compat import isatty
+from ._compat import open_stream
+from ._compat import range_type
+from ._compat import strip_ansi
+from ._compat import term_len
+from ._compat import WIN
+from .exceptions import ClickException
+from .utils import echo
+
+if os.name == "nt":
+    BEFORE_BAR = "\r"
+    AFTER_BAR = "\n"
+else:
+    BEFORE_BAR = "\r\033[?25l"
+    AFTER_BAR = "\033[?25h\n"
+
+
+def _length_hint(obj):
+    """Returns the length hint of an object."""
+    try:
+        return len(obj)
+    except (AttributeError, TypeError):
+        try:
+            get_hint = type(obj).__length_hint__
+        except AttributeError:
+            return None
+        try:
+            hint = get_hint(obj)
+        except TypeError:
+            return None
+        if hint is NotImplemented or not isinstance(hint, int_types) or hint < 0:
+            return None
+        return hint
+
+
+class ProgressBar(object):
+    def __init__(
+        self,
+        iterable,
+        length=None,
+        fill_char="#",
+        empty_char=" ",
+        bar_template="%(bar)s",
+        info_sep="  ",
+        show_eta=True,
+        show_percent=None,
+        show_pos=False,
+        item_show_func=None,
+        label=None,
+        file=None,
+        color=None,
+        width=30,
+    ):
+        self.fill_char = fill_char
+        self.empty_char = empty_char
+        self.bar_template = bar_template
+        self.info_sep = info_sep
+        self.show_eta = show_eta
+        self.show_percent = show_percent
+        self.show_pos = show_pos
+        self.item_show_func = item_show_func
+        self.label = label or ""
+        if file is None:
+            file = _default_text_stdout()
+        self.file = file
+        self.color = color
+        self.width = width
+        self.autowidth = width == 0
+
+        if length is None:
+            length = _length_hint(iterable)
+        if iterable is None:
+            if length is None:
+                raise TypeError("iterable or length is required")
+            iterable = range_type(length)
+        self.iter = iter(iterable)
+        self.length = length
+        self.length_known = length is not None
+        self.pos = 0
+        self.avg = []
+        self.start = self.last_eta = time.time()
+        self.eta_known = False
+        self.finished = False
+        self.max_width = None
+        self.entered = False
+        self.current_item = None
+        self.is_hidden = not isatty(self.file)
+        self._last_line = None
+        self.short_limit = 0.5
+
+    def __enter__(self):
+        self.entered = True
+        self.render_progress()
+        return self
+
+    def __exit__(self, exc_type, exc_value, tb):
+        self.render_finish()
+
+    def __iter__(self):
+        if not self.entered:
+            raise RuntimeError("You need to use progress bars in a with block.")
+        self.render_progress()
+        return self.generator()
+
+    def __next__(self):
+        # Iteration is defined in terms of a generator function,
+        # returned by iter(self); use that to define next(). This works
+        # because `self.iter` is an iterable consumed by that generator,
+        # so it is re-entry safe. Calling `next(self.generator())`
+        # twice works and does "what you want".
+        return next(iter(self))
+
+    # Python 2 compat
+    next = __next__
+
+    def is_fast(self):
+        return time.time() - self.start <= self.short_limit
+
+    def render_finish(self):
+        if self.is_hidden or self.is_fast():
+            return
+        self.file.write(AFTER_BAR)
+        self.file.flush()
+
+    @property
+    def pct(self):
+        if self.finished:
+            return 1.0
+        return min(self.pos / (float(self.length) or 1), 1.0)
+
+    @property
+    def time_per_iteration(self):
+        if not self.avg:
+            return 0.0
+        return sum(self.avg) / float(len(self.avg))
+
+    @property
+    def eta(self):
+        if self.length_known and not self.finished:
+            return self.time_per_iteration * (self.length - self.pos)
+        return 0.0
+
+    def format_eta(self):
+        if self.eta_known:
+            t = int(self.eta)
+            seconds = t % 60
+            t //= 60
+            minutes = t % 60
+            t //= 60
+            hours = t % 24
+            t //= 24
+            if t > 0:
+                return "{}d {:02}:{:02}:{:02}".format(t, hours, minutes, seconds)
+            else:
+                return "{:02}:{:02}:{:02}".format(hours, minutes, seconds)
+        return ""
+
+    def format_pos(self):
+        pos = str(self.pos)
+        if self.length_known:
+            pos += "/{}".format(self.length)
+        return pos
+
+    def format_pct(self):
+        return "{: 4}%".format(int(self.pct * 100))[1:]
+
+    def format_bar(self):
+        if self.length_known:
+            bar_length = int(self.pct * self.width)
+            bar = self.fill_char * bar_length
+            bar += self.empty_char * (self.width - bar_length)
+        elif self.finished:
+            bar = self.fill_char * self.width
+        else:
+            bar = list(self.empty_char * (self.width or 1))
+            if self.time_per_iteration != 0:
+                bar[
+                    int(
+                        (math.cos(self.pos * self.time_per_iteration) / 2.0 + 0.5)
+                        * self.width
+                    )
+                ] = self.fill_char
+            bar = "".join(bar)
+        return bar
+
+    def format_progress_line(self):
+        show_percent = self.show_percent
+
+        info_bits = []
+        if self.length_known and show_percent is None:
+            show_percent = not self.show_pos
+
+        if self.show_pos:
+            info_bits.append(self.format_pos())
+        if show_percent:
+            info_bits.append(self.format_pct())
+        if self.show_eta and self.eta_known and not self.finished:
+            info_bits.append(self.format_eta())
+        if self.item_show_func is not None:
+            item_info = self.item_show_func(self.current_item)
+            if item_info is not None:
+                info_bits.append(item_info)
+
+        return (
+            self.bar_template
+            % {
+                "label": self.label,
+                "bar": self.format_bar(),
+                "info": self.info_sep.join(info_bits),
+            }
+        ).rstrip()
+
+    def render_progress(self):
+        from .termui import get_terminal_size
+
+        if self.is_hidden:
+            return
+
+        buf = []
+        # Update width in case the terminal has been resized
+        if self.autowidth:
+            old_width = self.width
+            self.width = 0
+            clutter_length = term_len(self.format_progress_line())
+            new_width = max(0, get_terminal_size()[0] - clutter_length)
+            if new_width < old_width:
+                buf.append(BEFORE_BAR)
+                buf.append(" " * self.max_width)
+                self.max_width = new_width
+            self.width = new_width
+
+        clear_width = self.width
+        if self.max_width is not None:
+            clear_width = self.max_width
+
+        buf.append(BEFORE_BAR)
+        line = self.format_progress_line()
+        line_len = term_len(line)
+        if self.max_width is None or self.max_width < line_len:
+            self.max_width = line_len
+
+        buf.append(line)
+        buf.append(" " * (clear_width - line_len))
+        line = "".join(buf)
+        # Render the line only if it changed.
+
+        if line != self._last_line and not self.is_fast():
+            self._last_line = line
+            echo(line, file=self.file, color=self.color, nl=False)
+            self.file.flush()
+
+    def make_step(self, n_steps):
+        self.pos += n_steps
+        if self.length_known and self.pos >= self.length:
+            self.finished = True
+
+        if (time.time() - self.last_eta) < 1.0:
+            return
+
+        self.last_eta = time.time()
+
+        # self.avg is a rolling list of length <= 7 of steps where steps are
+        # defined as time elapsed divided by the total progress through
+        # self.length.
+        if self.pos:
+            step = (time.time() - self.start) / self.pos
+        else:
+            step = time.time() - self.start
+
+        self.avg = self.avg[-6:] + [step]
+
+        self.eta_known = self.length_known
+
+    def update(self, n_steps):
+        self.make_step(n_steps)
+        self.render_progress()
+
+    def finish(self):
+        self.eta_known = 0
+        self.current_item = None
+        self.finished = True
+
+    def generator(self):
+        """Return a generator which yields the items added to the bar
+        during construction, and updates the progress bar *after* the
+        yielded block returns.
+        """
+        # WARNING: the iterator interface for `ProgressBar` relies on
+        # this and only works because this is a simple generator which
+        # doesn't create or manage additional state. If this function
+        # changes, the impact should be evaluated both against
+        # `iter(bar)` and `next(bar)`. `next()` in particular may call
+        # `self.generator()` repeatedly, and this must remain safe in
+        # order for that interface to work.
+        if not self.entered:
+            raise RuntimeError("You need to use progress bars in a with block.")
+
+        if self.is_hidden:
+            for rv in self.iter:
+                yield rv
+        else:
+            for rv in self.iter:
+                self.current_item = rv
+                yield rv
+                self.update(1)
+            self.finish()
+            self.render_progress()
+
+
+def pager(generator, color=None):
+    """Decide what method to use for paging through text."""
+    stdout = _default_text_stdout()
+    if not isatty(sys.stdin) or not isatty(stdout):
+        return _nullpager(stdout, generator, color)
+    pager_cmd = (os.environ.get("PAGER", None) or "").strip()
+    if pager_cmd:
+        if WIN:
+            return _tempfilepager(generator, pager_cmd, color)
+        return _pipepager(generator, pager_cmd, color)
+    if os.environ.get("TERM") in ("dumb", "emacs"):
+        return _nullpager(stdout, generator, color)
+    if WIN or sys.platform.startswith("os2"):
+        return _tempfilepager(generator, "more <", color)
+    if hasattr(os, "system") and os.system("(less) 2>/dev/null") == 0:
+        return _pipepager(generator, "less", color)
+
+    import tempfile
+
+    fd, filename = tempfile.mkstemp()
+    os.close(fd)
+    try:
+        if hasattr(os, "system") and os.system('more "{}"'.format(filename)) == 0:
+            return _pipepager(generator, "more", color)
+        return _nullpager(stdout, generator, color)
+    finally:
+        os.unlink(filename)
+
+
+def _pipepager(generator, cmd, color):
+    """Page through text by feeding it to another program.  Invoking a
+    pager through this might support colors.
+    """
+    import subprocess
+
+    env = dict(os.environ)
+
+    # If we're piping to less we might support colors under the
+    # condition that
+    cmd_detail = cmd.rsplit("/", 1)[-1].split()
+    if color is None and cmd_detail[0] == "less":
+        less_flags = "{}{}".format(os.environ.get("LESS", ""), " ".join(cmd_detail[1:]))
+        if not less_flags:
+            env["LESS"] = "-R"
+            color = True
+        elif "r" in less_flags or "R" in less_flags:
+            color = True
+
+    c = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, env=env)
+    encoding = get_best_encoding(c.stdin)
+    try:
+        for text in generator:
+            if not color:
+                text = strip_ansi(text)
+
+            c.stdin.write(text.encode(encoding, "replace"))
+    except (IOError, KeyboardInterrupt):
+        pass
+    else:
+        c.stdin.close()
+
+    # Less doesn't respect ^C, but catches it for its own UI purposes (aborting
+    # search or other commands inside less).
+    #
+    # That means when the user hits ^C, the parent process (click) terminates,
+    # but less is still alive, paging the output and messing up the terminal.
+    #
+    # If the user wants to make the pager exit on ^C, they should set
+    # `LESS='-K'`. It's not our decision to make.
+    while True:
+        try:
+            c.wait()
+        except KeyboardInterrupt:
+            pass
+        else:
+            break
+
+
+def _tempfilepager(generator, cmd, color):
+    """Page through text by invoking a program on a temporary file."""
+    import tempfile
+
+    filename = tempfile.mktemp()
+    # TODO: This never terminates if the passed generator never terminates.
+    text = "".join(generator)
+    if not color:
+        text = strip_ansi(text)
+    encoding = get_best_encoding(sys.stdout)
+    with open_stream(filename, "wb")[0] as f:
+        f.write(text.encode(encoding))
+    try:
+        os.system('{} "{}"'.format(cmd, filename))
+    finally:
+        os.unlink(filename)
+
+
+def _nullpager(stream, generator, color):
+    """Simply print unformatted text.  This is the ultimate fallback."""
+    for text in generator:
+        if not color:
+            text = strip_ansi(text)
+        stream.write(text)
+
+
+class Editor(object):
+    def __init__(self, editor=None, env=None, require_save=True, extension=".txt"):
+        self.editor = editor
+        self.env = env
+        self.require_save = require_save
+        self.extension = extension
+
+    def get_editor(self):
+        if self.editor is not None:
+            return self.editor
+        for key in "VISUAL", "EDITOR":
+            rv = os.environ.get(key)
+            if rv:
+                return rv
+        if WIN:
+            return "notepad"
+        for editor in "sensible-editor", "vim", "nano":
+            if os.system("which {} >/dev/null 2>&1".format(editor)) == 0:
+                return editor
+        return "vi"
+
+    def edit_file(self, filename):
+        import subprocess
+
+        editor = self.get_editor()
+        if self.env:
+            environ = os.environ.copy()
+            environ.update(self.env)
+        else:
+            environ = None
+        try:
+            c = subprocess.Popen(
+                '{} "{}"'.format(editor, filename), env=environ, shell=True,
+            )
+            exit_code = c.wait()
+            if exit_code != 0:
+                raise ClickException("{}: Editing failed!".format(editor))
+        except OSError as e:
+            raise ClickException("{}: Editing failed: {}".format(editor, e))
+
+    def edit(self, text):
+        import tempfile
+
+        text = text or ""
+        if text and not text.endswith("\n"):
+            text += "\n"
+
+        fd, name = tempfile.mkstemp(prefix="editor-", suffix=self.extension)
+        try:
+            if WIN:
+                encoding = "utf-8-sig"
+                text = text.replace("\n", "\r\n")
+            else:
+                encoding = "utf-8"
+            text = text.encode(encoding)
+
+            f = os.fdopen(fd, "wb")
+            f.write(text)
+            f.close()
+            timestamp = os.path.getmtime(name)
+
+            self.edit_file(name)
+
+            if self.require_save and os.path.getmtime(name) == timestamp:
+                return None
+
+            f = open(name, "rb")
+            try:
+                rv = f.read()
+            finally:
+                f.close()
+            return rv.decode("utf-8-sig").replace("\r\n", "\n")
+        finally:
+            os.unlink(name)
+
+
+def open_url(url, wait=False, locate=False):
+    import subprocess
+
+    def _unquote_file(url):
+        try:
+            import urllib
+        except ImportError:
+            import urllib
+        if url.startswith("file://"):
+            url = urllib.unquote(url[7:])
+        return url
+
+    if sys.platform == "darwin":
+        args = ["open"]
+        if wait:
+            args.append("-W")
+        if locate:
+            args.append("-R")
+        args.append(_unquote_file(url))
+        null = open("/dev/null", "w")
+        try:
+            return subprocess.Popen(args, stderr=null).wait()
+        finally:
+            null.close()
+    elif WIN:
+        if locate:
+            url = _unquote_file(url)
+            args = 'explorer /select,"{}"'.format(_unquote_file(url.replace('"', "")))
+        else:
+            args = 'start {} "" "{}"'.format(
+                "/WAIT" if wait else "", url.replace('"', "")
+            )
+        return os.system(args)
+    elif CYGWIN:
+        if locate:
+            url = _unquote_file(url)
+            args = 'cygstart "{}"'.format(os.path.dirname(url).replace('"', ""))
+        else:
+            args = 'cygstart {} "{}"'.format("-w" if wait else "", url.replace('"', ""))
+        return os.system(args)
+
+    try:
+        if locate:
+            url = os.path.dirname(_unquote_file(url)) or "."
+        else:
+            url = _unquote_file(url)
+        c = subprocess.Popen(["xdg-open", url])
+        if wait:
+            return c.wait()
+        return 0
+    except OSError:
+        if url.startswith(("http://", "https://")) and not locate and not wait:
+            import webbrowser
+
+            webbrowser.open(url)
+            return 0
+        return 1
+
+
+def _translate_ch_to_exc(ch):
+    if ch == u"\x03":
+        raise KeyboardInterrupt()
+    if ch == u"\x04" and not WIN:  # Unix-like, Ctrl+D
+        raise EOFError()
+    if ch == u"\x1a" and WIN:  # Windows, Ctrl+Z
+        raise EOFError()
+
+
+if WIN:
+    import msvcrt
+
+    @contextlib.contextmanager
+    def raw_terminal():
+        yield
+
+    def getchar(echo):
+        # The function `getch` will return a bytes object corresponding to
+        # the pressed character. Since Windows 10 build 1803, it will also
+        # return \x00 when called a second time after pressing a regular key.
+        #
+        # `getwch` does not share this probably-bugged behavior. Moreover, it
+        # returns a Unicode object by default, which is what we want.
+        #
+        # Either of these functions will return \x00 or \xe0 to indicate
+        # a special key, and you need to call the same function again to get
+        # the "rest" of the code. The fun part is that \u00e0 is
+        # "latin small letter a with grave", so if you type that on a French
+        # keyboard, you _also_ get a \xe0.
+        # E.g., consider the Up arrow. This returns \xe0 and then \x48. The
+        # resulting Unicode string reads as "a with grave" + "capital H".
+        # This is indistinguishable from when the user actually types
+        # "a with grave" and then "capital H".
+        #
+        # When \xe0 is returned, we assume it's part of a special-key sequence
+        # and call `getwch` again, but that means that when the user types
+        # the \u00e0 character, `getchar` doesn't return until a second
+        # character is typed.
+        # The alternative is returning immediately, but that would mess up
+        # cross-platform handling of arrow keys and others that start with
+        # \xe0. Another option is using `getch`, but then we can't reliably
+        # read non-ASCII characters, because return values of `getch` are
+        # limited to the current 8-bit codepage.
+        #
+        # Anyway, Click doesn't claim to do this Right(tm), and using `getwch`
+        # is doing the right thing in more situations than with `getch`.
+        if echo:
+            func = msvcrt.getwche
+        else:
+            func = msvcrt.getwch
+
+        rv = func()
+        if rv in (u"\x00", u"\xe0"):
+            # \x00 and \xe0 are control characters that indicate special key,
+            # see above.
+            rv += func()
+        _translate_ch_to_exc(rv)
+        return rv
+
+
+else:
+    import tty
+    import termios
+
+    @contextlib.contextmanager
+    def raw_terminal():
+        if not isatty(sys.stdin):
+            f = open("/dev/tty")
+            fd = f.fileno()
+        else:
+            fd = sys.stdin.fileno()
+            f = None
+        try:
+            old_settings = termios.tcgetattr(fd)
+            try:
+                tty.setraw(fd)
+                yield fd
+            finally:
+                termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
+                sys.stdout.flush()
+                if f is not None:
+                    f.close()
+        except termios.error:
+            pass
+
+    def getchar(echo):
+        with raw_terminal() as fd:
+            ch = os.read(fd, 32)
+            ch = ch.decode(get_best_encoding(sys.stdin), "replace")
+            if echo and isatty(sys.stdout):
+                sys.stdout.write(ch)
+            _translate_ch_to_exc(ch)
+            return ch