_termui_impl.py

# -*- coding: utf-8 -*-
"""
click._termui_impl
~~~~~~~~~~~~~~~~~~

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.

:copyright: © 2014 by the Pallets team.
:license: BSD, see LICENSE.rst for more details.
"""

import os
import sys
import time
import math
import contextlib
from ._compat import _default_text_stdout, range_type, isatty, \
     open_stream, strip_ansi, term_len, get_best_encoding, WIN, int_types, \
     CYGWIN, which
from .utils import echo
from .exceptions import ClickException


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:
                days = t
                return '%dd %02d:%02d:%02d' % (days, hours, minutes, seconds)
            else:
                return '%02d:%02d:%02d' % (hours, minutes, seconds)
        return ''

    def format_pos(self):
        pos = str(self.pos)
        if self.length_known:
            pos += '/%s' % self.length
        return pos

    def format_pct(self):
        return ('% 4d%%' % 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, current_item=None):
        """Update the progress bar by advancing a specified number of
        steps, and optionally set the ``current_item`` for this new
        position.

        :param n_steps: Number of steps to advance.
        :param current_item: Optional item to set as ``current_item``
            for the updated position.

        .. versionadded:: 8.0
            Added the ``current_item`` optional parameter.
        """
        self.make_step(n_steps)
        if current_item is not None:
            self.current_item = current_item
        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 _pipetempfilepager(generator, 'more', color)
    if which('less') is not None:
        return _pipepager(generator, 'less', color)

    import tempfile
    fd, filename = tempfile.mkstemp()
    os.close(fd)
    try:
        if which('more') is not None:
            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 = 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 subprocess
    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:
        subprocess.call([cmd, filename])
    except OSError:
        # Command not found
        pass
    finally:
        os.unlink(filename)


def _pipetempfilepager(generator, cmd, color):
    """
    Page through text by invoking a program with a temporary file redirected
    as input.
    """
    import subprocess
    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:
        with open_stream(filename, 'rb')[0] as f:
            subprocess.call([cmd], stdin=f)
    except OSError:
        # Command not found
        pass
    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 which(editor) is not None:
                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('%s "%s"' % (editor, filename),
                                 env=environ, shell=True)
            exit_code = c.wait()
            if exit_code != 0:
                raise ClickException('%s: Editing failed!' % editor)
        except OSError as e:
            raise ClickException('%s: Editing failed: %s' % (editor, e))

    def edit(self, text):
        import tempfile

        text = text or ''
        binary_data = type(text) in [bytes, bytearray]

        if not binary_data and text and not text.endswith('\n'):
            text += '\n'

        fd, name = tempfile.mkstemp(prefix='editor-', suffix=self.extension)
        try:
            if not binary_data:
                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()
            if binary_data:
                return rv
            else:
                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,%s' % _unquote_file(url)]
        else:
            args = ['start']
            if wait:
                args.append('/WAIT')
            args.append('')
            args.append(url)
        try:
            return subprocess.call(args)
        except OSError:
            # Command not found
            return 127
    elif CYGWIN:
        if locate:
            url = _unquote_file(url)
            args = ['cygstart', os.path.dirname(url)]
        else:
            args = ['cygstart']
            if wait:
                args.append('-w')
            args.append(url)
        try:
            return subprocess.call(args)
        except OSError:
            # Command not found
            return 127

    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