You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
bazarr/libs/click/_termui_impl.py

718 lines
23 KiB

6 years ago
"""
5 years ago
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.
6 years ago
"""
import contextlib
import math
6 years ago
import os
import sys
import time
import typing as t
from gettext import gettext as _
from ._compat import _default_text_stdout
from ._compat import CYGWIN
from ._compat import get_best_encoding
from ._compat import isatty
from ._compat import open_stream
from ._compat import strip_ansi
from ._compat import term_len
from ._compat import WIN
6 years ago
from .exceptions import ClickException
from .utils import echo
6 years ago
V = t.TypeVar("V")
6 years ago
if os.name == "nt":
BEFORE_BAR = "\r"
AFTER_BAR = "\n"
6 years ago
else:
BEFORE_BAR = "\r\033[?25l"
AFTER_BAR = "\033[?25h\n"
class ProgressBar(t.Generic[V]):
def __init__(
self,
iterable: t.Optional[t.Iterable[V]],
length: t.Optional[int] = None,
fill_char: str = "#",
empty_char: str = " ",
bar_template: str = "%(bar)s",
info_sep: str = " ",
show_eta: bool = True,
show_percent: t.Optional[bool] = None,
show_pos: bool = False,
item_show_func: t.Optional[t.Callable[[t.Optional[V]], t.Optional[str]]] = None,
label: t.Optional[str] = None,
file: t.Optional[t.TextIO] = None,
color: t.Optional[bool] = None,
update_min_steps: int = 1,
width: int = 30,
) -> None:
6 years ago
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 ""
6 years ago
if file is None:
file = _default_text_stdout()
self.file = file
self.color = color
self.update_min_steps = update_min_steps
self._completed_intervals = 0
6 years ago
self.width = width
self.autowidth = width == 0
if length is None:
from operator import length_hint
length = length_hint(iterable, -1)
if length == -1:
length = None
6 years ago
if iterable is None:
if length is None:
raise TypeError("iterable or length is required")
iterable = t.cast(t.Iterable[V], range(length))
6 years ago
self.iter = iter(iterable)
self.length = length
self.pos = 0
self.avg: t.List[float] = []
6 years ago
self.start = self.last_eta = time.time()
self.eta_known = False
self.finished = False
self.max_width: t.Optional[int] = None
6 years ago
self.entered = False
self.current_item: t.Optional[V] = None
6 years ago
self.is_hidden = not isatty(self.file)
self._last_line: t.Optional[str] = None
6 years ago
def __enter__(self) -> "ProgressBar":
6 years ago
self.entered = True
self.render_progress()
return self
def __exit__(self, exc_type, exc_value, tb): # type: ignore
6 years ago
self.render_finish()
def __iter__(self) -> t.Iterator[V]:
6 years ago
if not self.entered:
raise RuntimeError("You need to use progress bars in a with block.")
6 years ago
self.render_progress()
5 years ago
return self.generator()
def __next__(self) -> V:
# 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))
6 years ago
def render_finish(self) -> None:
if self.is_hidden:
6 years ago
return
self.file.write(AFTER_BAR)
self.file.flush()
@property
def pct(self) -> float:
6 years ago
if self.finished:
return 1.0
return min(self.pos / (float(self.length or 1) or 1), 1.0)
6 years ago
@property
def time_per_iteration(self) -> float:
6 years ago
if not self.avg:
return 0.0
return sum(self.avg) / float(len(self.avg))
@property
def eta(self) -> float:
if self.length is not None and not self.finished:
6 years ago
return self.time_per_iteration * (self.length - self.pos)
return 0.0
def format_eta(self) -> str:
6 years ago
if self.eta_known:
5 years ago
t = int(self.eta)
6 years ago
seconds = t % 60
5 years ago
t //= 60
6 years ago
minutes = t % 60
5 years ago
t //= 60
6 years ago
hours = t % 24
5 years ago
t //= 24
6 years ago
if t > 0:
return f"{t}d {hours:02}:{minutes:02}:{seconds:02}"
6 years ago
else:
return f"{hours:02}:{minutes:02}:{seconds:02}"
return ""
6 years ago
def format_pos(self) -> str:
6 years ago
pos = str(self.pos)
if self.length is not None:
pos += f"/{self.length}"
6 years ago
return pos
def format_pct(self) -> str:
return f"{int(self.pct * 100): 4}%"[1:]
6 years ago
def format_bar(self) -> str:
if self.length is not None:
6 years ago
bar_length = int(self.pct * self.width)
bar = self.fill_char * bar_length
bar += self.empty_char * (self.width - bar_length)
5 years ago
elif self.finished:
bar = self.fill_char * self.width
6 years ago
else:
chars = list(self.empty_char * (self.width or 1))
5 years ago
if self.time_per_iteration != 0:
chars[
int(
(math.cos(self.pos * self.time_per_iteration) / 2.0 + 0.5)
* self.width
)
] = self.fill_char
bar = "".join(chars)
5 years ago
return bar
def format_progress_line(self) -> str:
5 years ago
show_percent = self.show_percent
info_bits = []
if self.length is not None and show_percent is None:
5 years ago
show_percent = not self.show_pos
6 years ago
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()
6 years ago
def render_progress(self) -> None:
import shutil
6 years ago
if self.is_hidden:
# Only output the label as it changes if the output is not a
# TTY. Use file=stderr if you expect to be piping stdout.
if self._last_line != self.label:
self._last_line = self.label
echo(self.label, file=self.file, color=self.color)
5 years ago
return
6 years ago
5 years ago
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, shutil.get_terminal_size().columns - clutter_length)
5 years ago
if new_width < old_width:
buf.append(BEFORE_BAR)
buf.append(" " * self.max_width) # type: ignore
5 years ago
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)
6 years ago
# Render the line only if it changed.
5 years ago
if line != self._last_line:
6 years ago
self._last_line = line
5 years ago
echo(line, file=self.file, color=self.color, nl=False)
6 years ago
self.file.flush()
def make_step(self, n_steps: int) -> None:
6 years ago
self.pos += n_steps
if self.length is not None and self.pos >= self.length:
6 years ago
self.finished = True
if (time.time() - self.last_eta) < 1.0:
return
self.last_eta = time.time()
5 years ago
# 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]
6 years ago
self.eta_known = self.length is not None
6 years ago
def update(self, n_steps: int, current_item: t.Optional[V] = None) -> 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.
.. versionchanged:: 8.0
Added the ``current_item`` optional parameter.
.. versionchanged:: 8.0
Only render when the number of steps meets the
``update_min_steps`` threshold.
"""
if current_item is not None:
self.current_item = current_item
self._completed_intervals += n_steps
if self._completed_intervals >= self.update_min_steps:
self.make_step(self._completed_intervals)
self.render_progress()
self._completed_intervals = 0
6 years ago
def finish(self) -> None:
self.eta_known = False
6 years ago
self.current_item = None
self.finished = True
def generator(self) -> t.Iterator[V]:
"""Return a generator which yields the items added to the bar
during construction, and updates the progress bar *after* the
yielded block returns.
5 years ago
"""
# 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.
5 years ago
if not self.entered:
raise RuntimeError("You need to use progress bars in a with block.")
5 years ago
6 years ago
if self.is_hidden:
yield from self.iter
5 years ago
else:
for rv in self.iter:
self.current_item = rv
# This allows show_item_func to be updated before the
# item is processed. Only trigger at the beginning of
# the update interval.
if self._completed_intervals == 0:
self.render_progress()
5 years ago
yield rv
self.update(1)
6 years ago
self.finish()
self.render_progress()
def pager(generator: t.Iterable[str], color: t.Optional[bool] = None) -> None:
6 years ago
"""Decide what method to use for paging through text."""
stdout = _default_text_stdout()
if not isatty(sys.stdin) or not isatty(stdout):
5 years ago
return _nullpager(stdout, generator, color)
pager_cmd = (os.environ.get("PAGER", None) or "").strip()
6 years ago
if pager_cmd:
if WIN:
5 years ago
return _tempfilepager(generator, pager_cmd, color)
return _pipepager(generator, pager_cmd, color)
if os.environ.get("TERM") in ("dumb", "emacs"):
5 years ago
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)
6 years ago
import tempfile
6 years ago
fd, filename = tempfile.mkstemp()
os.close(fd)
try:
if hasattr(os, "system") and os.system(f'more "{filename}"') == 0:
return _pipepager(generator, "more", color)
5 years ago
return _nullpager(stdout, generator, color)
6 years ago
finally:
os.unlink(filename)
def _pipepager(generator: t.Iterable[str], cmd: str, color: t.Optional[bool]) -> None:
6 years ago
"""Page through text by feeding it to another program. Invoking a
pager through this might support colors.
"""
import subprocess
6 years ago
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 = f"{os.environ.get('LESS', '')}{' '.join(cmd_detail[1:])}"
6 years ago
if not less_flags:
env["LESS"] = "-R"
6 years ago
color = True
elif "r" in less_flags or "R" in less_flags:
6 years ago
color = True
c = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, env=env)
stdin = t.cast(t.BinaryIO, c.stdin)
encoding = get_best_encoding(stdin)
6 years ago
try:
5 years ago
for text in generator:
if not color:
text = strip_ansi(text)
stdin.write(text.encode(encoding, "replace"))
except (OSError, KeyboardInterrupt):
6 years ago
pass
5 years ago
else:
stdin.close()
6 years ago
# 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: t.Iterable[str], cmd: str, color: t.Optional[bool]
) -> None:
6 years ago
"""Page through text by invoking a program on a temporary file."""
import tempfile
fd, filename = tempfile.mkstemp()
5 years ago
# TODO: This never terminates if the passed generator never terminates.
text = "".join(generator)
6 years ago
if not color:
text = strip_ansi(text)
encoding = get_best_encoding(sys.stdout)
with open_stream(filename, "wb")[0] as f:
6 years ago
f.write(text.encode(encoding))
try:
os.system(f'{cmd} "{filename}"')
6 years ago
finally:
os.close(fd)
6 years ago
os.unlink(filename)
def _nullpager(
stream: t.TextIO, generator: t.Iterable[str], color: t.Optional[bool]
) -> None:
6 years ago
"""Simply print unformatted text. This is the ultimate fallback."""
5 years ago
for text in generator:
if not color:
text = strip_ansi(text)
stream.write(text)
6 years ago
class Editor:
def __init__(
self,
editor: t.Optional[str] = None,
env: t.Optional[t.Mapping[str, str]] = None,
require_save: bool = True,
extension: str = ".txt",
) -> None:
6 years ago
self.editor = editor
self.env = env
self.require_save = require_save
self.extension = extension
def get_editor(self) -> str:
6 years ago
if self.editor is not None:
return self.editor
for key in "VISUAL", "EDITOR":
6 years ago
rv = os.environ.get(key)
if rv:
return rv
if WIN:
return "notepad"
for editor in "sensible-editor", "vim", "nano":
if os.system(f"which {editor} >/dev/null 2>&1") == 0:
6 years ago
return editor
return "vi"
6 years ago
def edit_file(self, filename: str) -> None:
6 years ago
import subprocess
6 years ago
editor = self.get_editor()
environ: t.Optional[t.Dict[str, str]] = None
6 years ago
if self.env:
environ = os.environ.copy()
environ.update(self.env)
6 years ago
try:
c = subprocess.Popen(f'{editor} "{filename}"', env=environ, shell=True)
6 years ago
exit_code = c.wait()
if exit_code != 0:
raise ClickException(
_("{editor}: Editing failed").format(editor=editor)
)
6 years ago
except OSError as e:
raise ClickException(
_("{editor}: Editing failed: {e}").format(editor=editor, e=e)
) from e
6 years ago
def edit(self, text: t.Optional[t.AnyStr]) -> t.Optional[t.AnyStr]:
6 years ago
import tempfile
if not text:
data = b""
elif isinstance(text, (bytes, bytearray)):
data = text
else:
if text and not text.endswith("\n"):
text += "\n"
6 years ago
if WIN:
data = text.replace("\n", "\r\n").encode("utf-8-sig")
6 years ago
else:
data = text.encode("utf-8")
6 years ago
fd, name = tempfile.mkstemp(prefix="editor-", suffix=self.extension)
f: t.BinaryIO
try:
with os.fdopen(fd, "wb") as f:
f.write(data)
# If the filesystem resolution is 1 second, like Mac OS
# 10.12 Extended, or 2 seconds, like FAT32, and the editor
# closes very fast, require_save can fail. Set the modified
# time to be 2 seconds in the past to work around this.
os.utime(name, (os.path.getatime(name), os.path.getmtime(name) - 2))
# Depending on the resolution, the exact value might not be
# recorded, so get the new recorded value.
6 years ago
timestamp = os.path.getmtime(name)
self.edit_file(name)
if self.require_save and os.path.getmtime(name) == timestamp:
6 years ago
return None
with open(name, "rb") as f:
6 years ago
rv = f.read()
if isinstance(text, (bytes, bytearray)):
return rv
return rv.decode("utf-8-sig").replace("\r\n", "\n") # type: ignore
6 years ago
finally:
os.unlink(name)
def open_url(url: str, wait: bool = False, locate: bool = False) -> int:
6 years ago
import subprocess
def _unquote_file(url: str) -> str:
from urllib.parse import unquote
if url.startswith("file://"):
url = unquote(url[7:])
6 years ago
return url
if sys.platform == "darwin":
args = ["open"]
6 years ago
if wait:
args.append("-W")
6 years ago
if locate:
args.append("-R")
6 years ago
args.append(_unquote_file(url))
null = open("/dev/null", "w")
6 years ago
try:
return subprocess.Popen(args, stderr=null).wait()
finally:
null.close()
elif WIN:
if locate:
url = _unquote_file(url.replace('"', ""))
args = f'explorer /select,"{url}"'
6 years ago
else:
url = url.replace('"', "")
wait_str = "/WAIT" if wait else ""
args = f'start {wait_str} "" "{url}"'
6 years ago
return os.system(args)
5 years ago
elif CYGWIN:
if locate:
url = os.path.dirname(_unquote_file(url).replace('"', ""))
args = f'cygstart "{url}"'
5 years ago
else:
url = url.replace('"', "")
wait_str = "-w" if wait else ""
args = f'cygstart {wait_str} "{url}"'
5 years ago
return os.system(args)
6 years ago
try:
if locate:
url = os.path.dirname(_unquote_file(url)) or "."
6 years ago
else:
url = _unquote_file(url)
c = subprocess.Popen(["xdg-open", url])
6 years ago
if wait:
return c.wait()
return 0
except OSError:
if url.startswith(("http://", "https://")) and not locate and not wait:
6 years ago
import webbrowser
6 years ago
webbrowser.open(url)
return 0
return 1
def _translate_ch_to_exc(ch: str) -> t.Optional[BaseException]:
if ch == "\x03":
6 years ago
raise KeyboardInterrupt()
if ch == "\x04" and not WIN: # Unix-like, Ctrl+D
5 years ago
raise EOFError()
if ch == "\x1a" and WIN: # Windows, Ctrl+Z
6 years ago
raise EOFError()
return None
6 years ago
if WIN:
import msvcrt
5 years ago
@contextlib.contextmanager
def raw_terminal() -> t.Iterator[int]:
yield -1
5 years ago
def getchar(echo: bool) -> str:
5 years ago
# 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`.
func: t.Callable[[], str]
6 years ago
if echo:
func = msvcrt.getwche # type: ignore
5 years ago
else:
func = msvcrt.getwch # type: ignore
5 years ago
rv = func()
if rv in ("\x00", "\xe0"):
5 years ago
# \x00 and \xe0 are control characters that indicate special key,
# see above.
rv += func()
6 years ago
_translate_ch_to_exc(rv)
return rv
6 years ago
else:
import tty
import termios
5 years ago
@contextlib.contextmanager
def raw_terminal() -> t.Iterator[int]:
f: t.Optional[t.TextIO]
fd: int
6 years ago
if not isatty(sys.stdin):
f = open("/dev/tty")
6 years ago
fd = f.fileno()
else:
fd = sys.stdin.fileno()
f = None
6 years ago
try:
old_settings = termios.tcgetattr(fd)
6 years ago
try:
tty.setraw(fd)
5 years ago
yield fd
6 years ago
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
sys.stdout.flush()
6 years ago
if f is not None:
f.close()
except termios.error:
pass
5 years ago
def getchar(echo: bool) -> str:
5 years ago
with raw_terminal() as fd:
ch = os.read(fd, 32).decode(get_best_encoding(sys.stdin), "replace")
5 years ago
if echo and isatty(sys.stdout):
sys.stdout.write(ch)
5 years ago
_translate_ch_to_exc(ch)
return ch