| # -*- 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 |
