_termui_impl.py 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717
  1. """
  2. This module contains implementations for the termui module. To keep the
  3. import time of Click down, some infrequently used functionality is
  4. placed in this module and only imported as needed.
  5. """
  6. import contextlib
  7. import math
  8. import os
  9. import sys
  10. import time
  11. import typing as t
  12. from gettext import gettext as _
  13. from ._compat import _default_text_stdout
  14. from ._compat import CYGWIN
  15. from ._compat import get_best_encoding
  16. from ._compat import isatty
  17. from ._compat import open_stream
  18. from ._compat import strip_ansi
  19. from ._compat import term_len
  20. from ._compat import WIN
  21. from .exceptions import ClickException
  22. from .utils import echo
  23. V = t.TypeVar("V")
  24. if os.name == "nt":
  25. BEFORE_BAR = "\r"
  26. AFTER_BAR = "\n"
  27. else:
  28. BEFORE_BAR = "\r\033[?25l"
  29. AFTER_BAR = "\033[?25h\n"
  30. class ProgressBar(t.Generic[V]):
  31. def __init__(
  32. self,
  33. iterable: t.Optional[t.Iterable[V]],
  34. length: t.Optional[int] = None,
  35. fill_char: str = "#",
  36. empty_char: str = " ",
  37. bar_template: str = "%(bar)s",
  38. info_sep: str = " ",
  39. show_eta: bool = True,
  40. show_percent: t.Optional[bool] = None,
  41. show_pos: bool = False,
  42. item_show_func: t.Optional[t.Callable[[t.Optional[V]], t.Optional[str]]] = None,
  43. label: t.Optional[str] = None,
  44. file: t.Optional[t.TextIO] = None,
  45. color: t.Optional[bool] = None,
  46. update_min_steps: int = 1,
  47. width: int = 30,
  48. ) -> None:
  49. self.fill_char = fill_char
  50. self.empty_char = empty_char
  51. self.bar_template = bar_template
  52. self.info_sep = info_sep
  53. self.show_eta = show_eta
  54. self.show_percent = show_percent
  55. self.show_pos = show_pos
  56. self.item_show_func = item_show_func
  57. self.label = label or ""
  58. if file is None:
  59. file = _default_text_stdout()
  60. self.file = file
  61. self.color = color
  62. self.update_min_steps = update_min_steps
  63. self._completed_intervals = 0
  64. self.width = width
  65. self.autowidth = width == 0
  66. if length is None:
  67. from operator import length_hint
  68. length = length_hint(iterable, -1)
  69. if length == -1:
  70. length = None
  71. if iterable is None:
  72. if length is None:
  73. raise TypeError("iterable or length is required")
  74. iterable = t.cast(t.Iterable[V], range(length))
  75. self.iter = iter(iterable)
  76. self.length = length
  77. self.pos = 0
  78. self.avg: t.List[float] = []
  79. self.start = self.last_eta = time.time()
  80. self.eta_known = False
  81. self.finished = False
  82. self.max_width: t.Optional[int] = None
  83. self.entered = False
  84. self.current_item: t.Optional[V] = None
  85. self.is_hidden = not isatty(self.file)
  86. self._last_line: t.Optional[str] = None
  87. def __enter__(self) -> "ProgressBar":
  88. self.entered = True
  89. self.render_progress()
  90. return self
  91. def __exit__(self, exc_type, exc_value, tb): # type: ignore
  92. self.render_finish()
  93. def __iter__(self) -> t.Iterator[V]:
  94. if not self.entered:
  95. raise RuntimeError("You need to use progress bars in a with block.")
  96. self.render_progress()
  97. return self.generator()
  98. def __next__(self) -> V:
  99. # Iteration is defined in terms of a generator function,
  100. # returned by iter(self); use that to define next(). This works
  101. # because `self.iter` is an iterable consumed by that generator,
  102. # so it is re-entry safe. Calling `next(self.generator())`
  103. # twice works and does "what you want".
  104. return next(iter(self))
  105. def render_finish(self) -> None:
  106. if self.is_hidden:
  107. return
  108. self.file.write(AFTER_BAR)
  109. self.file.flush()
  110. @property
  111. def pct(self) -> float:
  112. if self.finished:
  113. return 1.0
  114. return min(self.pos / (float(self.length or 1) or 1), 1.0)
  115. @property
  116. def time_per_iteration(self) -> float:
  117. if not self.avg:
  118. return 0.0
  119. return sum(self.avg) / float(len(self.avg))
  120. @property
  121. def eta(self) -> float:
  122. if self.length is not None and not self.finished:
  123. return self.time_per_iteration * (self.length - self.pos)
  124. return 0.0
  125. def format_eta(self) -> str:
  126. if self.eta_known:
  127. t = int(self.eta)
  128. seconds = t % 60
  129. t //= 60
  130. minutes = t % 60
  131. t //= 60
  132. hours = t % 24
  133. t //= 24
  134. if t > 0:
  135. return f"{t}d {hours:02}:{minutes:02}:{seconds:02}"
  136. else:
  137. return f"{hours:02}:{minutes:02}:{seconds:02}"
  138. return ""
  139. def format_pos(self) -> str:
  140. pos = str(self.pos)
  141. if self.length is not None:
  142. pos += f"/{self.length}"
  143. return pos
  144. def format_pct(self) -> str:
  145. return f"{int(self.pct * 100): 4}%"[1:]
  146. def format_bar(self) -> str:
  147. if self.length is not None:
  148. bar_length = int(self.pct * self.width)
  149. bar = self.fill_char * bar_length
  150. bar += self.empty_char * (self.width - bar_length)
  151. elif self.finished:
  152. bar = self.fill_char * self.width
  153. else:
  154. chars = list(self.empty_char * (self.width or 1))
  155. if self.time_per_iteration != 0:
  156. chars[
  157. int(
  158. (math.cos(self.pos * self.time_per_iteration) / 2.0 + 0.5)
  159. * self.width
  160. )
  161. ] = self.fill_char
  162. bar = "".join(chars)
  163. return bar
  164. def format_progress_line(self) -> str:
  165. show_percent = self.show_percent
  166. info_bits = []
  167. if self.length is not None and show_percent is None:
  168. show_percent = not self.show_pos
  169. if self.show_pos:
  170. info_bits.append(self.format_pos())
  171. if show_percent:
  172. info_bits.append(self.format_pct())
  173. if self.show_eta and self.eta_known and not self.finished:
  174. info_bits.append(self.format_eta())
  175. if self.item_show_func is not None:
  176. item_info = self.item_show_func(self.current_item)
  177. if item_info is not None:
  178. info_bits.append(item_info)
  179. return (
  180. self.bar_template
  181. % {
  182. "label": self.label,
  183. "bar": self.format_bar(),
  184. "info": self.info_sep.join(info_bits),
  185. }
  186. ).rstrip()
  187. def render_progress(self) -> None:
  188. import shutil
  189. if self.is_hidden:
  190. # Only output the label as it changes if the output is not a
  191. # TTY. Use file=stderr if you expect to be piping stdout.
  192. if self._last_line != self.label:
  193. self._last_line = self.label
  194. echo(self.label, file=self.file, color=self.color)
  195. return
  196. buf = []
  197. # Update width in case the terminal has been resized
  198. if self.autowidth:
  199. old_width = self.width
  200. self.width = 0
  201. clutter_length = term_len(self.format_progress_line())
  202. new_width = max(0, shutil.get_terminal_size().columns - clutter_length)
  203. if new_width < old_width:
  204. buf.append(BEFORE_BAR)
  205. buf.append(" " * self.max_width) # type: ignore
  206. self.max_width = new_width
  207. self.width = new_width
  208. clear_width = self.width
  209. if self.max_width is not None:
  210. clear_width = self.max_width
  211. buf.append(BEFORE_BAR)
  212. line = self.format_progress_line()
  213. line_len = term_len(line)
  214. if self.max_width is None or self.max_width < line_len:
  215. self.max_width = line_len
  216. buf.append(line)
  217. buf.append(" " * (clear_width - line_len))
  218. line = "".join(buf)
  219. # Render the line only if it changed.
  220. if line != self._last_line:
  221. self._last_line = line
  222. echo(line, file=self.file, color=self.color, nl=False)
  223. self.file.flush()
  224. def make_step(self, n_steps: int) -> None:
  225. self.pos += n_steps
  226. if self.length is not None and self.pos >= self.length:
  227. self.finished = True
  228. if (time.time() - self.last_eta) < 1.0:
  229. return
  230. self.last_eta = time.time()
  231. # self.avg is a rolling list of length <= 7 of steps where steps are
  232. # defined as time elapsed divided by the total progress through
  233. # self.length.
  234. if self.pos:
  235. step = (time.time() - self.start) / self.pos
  236. else:
  237. step = time.time() - self.start
  238. self.avg = self.avg[-6:] + [step]
  239. self.eta_known = self.length is not None
  240. def update(self, n_steps: int, current_item: t.Optional[V] = None) -> None:
  241. """Update the progress bar by advancing a specified number of
  242. steps, and optionally set the ``current_item`` for this new
  243. position.
  244. :param n_steps: Number of steps to advance.
  245. :param current_item: Optional item to set as ``current_item``
  246. for the updated position.
  247. .. versionchanged:: 8.0
  248. Added the ``current_item`` optional parameter.
  249. .. versionchanged:: 8.0
  250. Only render when the number of steps meets the
  251. ``update_min_steps`` threshold.
  252. """
  253. if current_item is not None:
  254. self.current_item = current_item
  255. self._completed_intervals += n_steps
  256. if self._completed_intervals >= self.update_min_steps:
  257. self.make_step(self._completed_intervals)
  258. self.render_progress()
  259. self._completed_intervals = 0
  260. def finish(self) -> None:
  261. self.eta_known = False
  262. self.current_item = None
  263. self.finished = True
  264. def generator(self) -> t.Iterator[V]:
  265. """Return a generator which yields the items added to the bar
  266. during construction, and updates the progress bar *after* the
  267. yielded block returns.
  268. """
  269. # WARNING: the iterator interface for `ProgressBar` relies on
  270. # this and only works because this is a simple generator which
  271. # doesn't create or manage additional state. If this function
  272. # changes, the impact should be evaluated both against
  273. # `iter(bar)` and `next(bar)`. `next()` in particular may call
  274. # `self.generator()` repeatedly, and this must remain safe in
  275. # order for that interface to work.
  276. if not self.entered:
  277. raise RuntimeError("You need to use progress bars in a with block.")
  278. if self.is_hidden:
  279. yield from self.iter
  280. else:
  281. for rv in self.iter:
  282. self.current_item = rv
  283. # This allows show_item_func to be updated before the
  284. # item is processed. Only trigger at the beginning of
  285. # the update interval.
  286. if self._completed_intervals == 0:
  287. self.render_progress()
  288. yield rv
  289. self.update(1)
  290. self.finish()
  291. self.render_progress()
  292. def pager(generator: t.Iterable[str], color: t.Optional[bool] = None) -> None:
  293. """Decide what method to use for paging through text."""
  294. stdout = _default_text_stdout()
  295. if not isatty(sys.stdin) or not isatty(stdout):
  296. return _nullpager(stdout, generator, color)
  297. pager_cmd = (os.environ.get("PAGER", None) or "").strip()
  298. if pager_cmd:
  299. if WIN:
  300. return _tempfilepager(generator, pager_cmd, color)
  301. return _pipepager(generator, pager_cmd, color)
  302. if os.environ.get("TERM") in ("dumb", "emacs"):
  303. return _nullpager(stdout, generator, color)
  304. if WIN or sys.platform.startswith("os2"):
  305. return _tempfilepager(generator, "more <", color)
  306. if hasattr(os, "system") and os.system("(less) 2>/dev/null") == 0:
  307. return _pipepager(generator, "less", color)
  308. import tempfile
  309. fd, filename = tempfile.mkstemp()
  310. os.close(fd)
  311. try:
  312. if hasattr(os, "system") and os.system(f'more "{filename}"') == 0:
  313. return _pipepager(generator, "more", color)
  314. return _nullpager(stdout, generator, color)
  315. finally:
  316. os.unlink(filename)
  317. def _pipepager(generator: t.Iterable[str], cmd: str, color: t.Optional[bool]) -> None:
  318. """Page through text by feeding it to another program. Invoking a
  319. pager through this might support colors.
  320. """
  321. import subprocess
  322. env = dict(os.environ)
  323. # If we're piping to less we might support colors under the
  324. # condition that
  325. cmd_detail = cmd.rsplit("/", 1)[-1].split()
  326. if color is None and cmd_detail[0] == "less":
  327. less_flags = f"{os.environ.get('LESS', '')}{' '.join(cmd_detail[1:])}"
  328. if not less_flags:
  329. env["LESS"] = "-R"
  330. color = True
  331. elif "r" in less_flags or "R" in less_flags:
  332. color = True
  333. c = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, env=env)
  334. stdin = t.cast(t.BinaryIO, c.stdin)
  335. encoding = get_best_encoding(stdin)
  336. try:
  337. for text in generator:
  338. if not color:
  339. text = strip_ansi(text)
  340. stdin.write(text.encode(encoding, "replace"))
  341. except (OSError, KeyboardInterrupt):
  342. pass
  343. else:
  344. stdin.close()
  345. # Less doesn't respect ^C, but catches it for its own UI purposes (aborting
  346. # search or other commands inside less).
  347. #
  348. # That means when the user hits ^C, the parent process (click) terminates,
  349. # but less is still alive, paging the output and messing up the terminal.
  350. #
  351. # If the user wants to make the pager exit on ^C, they should set
  352. # `LESS='-K'`. It's not our decision to make.
  353. while True:
  354. try:
  355. c.wait()
  356. except KeyboardInterrupt:
  357. pass
  358. else:
  359. break
  360. def _tempfilepager(
  361. generator: t.Iterable[str], cmd: str, color: t.Optional[bool]
  362. ) -> None:
  363. """Page through text by invoking a program on a temporary file."""
  364. import tempfile
  365. fd, filename = tempfile.mkstemp()
  366. # TODO: This never terminates if the passed generator never terminates.
  367. text = "".join(generator)
  368. if not color:
  369. text = strip_ansi(text)
  370. encoding = get_best_encoding(sys.stdout)
  371. with open_stream(filename, "wb")[0] as f:
  372. f.write(text.encode(encoding))
  373. try:
  374. os.system(f'{cmd} "{filename}"')
  375. finally:
  376. os.close(fd)
  377. os.unlink(filename)
  378. def _nullpager(
  379. stream: t.TextIO, generator: t.Iterable[str], color: t.Optional[bool]
  380. ) -> None:
  381. """Simply print unformatted text. This is the ultimate fallback."""
  382. for text in generator:
  383. if not color:
  384. text = strip_ansi(text)
  385. stream.write(text)
  386. class Editor:
  387. def __init__(
  388. self,
  389. editor: t.Optional[str] = None,
  390. env: t.Optional[t.Mapping[str, str]] = None,
  391. require_save: bool = True,
  392. extension: str = ".txt",
  393. ) -> None:
  394. self.editor = editor
  395. self.env = env
  396. self.require_save = require_save
  397. self.extension = extension
  398. def get_editor(self) -> str:
  399. if self.editor is not None:
  400. return self.editor
  401. for key in "VISUAL", "EDITOR":
  402. rv = os.environ.get(key)
  403. if rv:
  404. return rv
  405. if WIN:
  406. return "notepad"
  407. for editor in "sensible-editor", "vim", "nano":
  408. if os.system(f"which {editor} >/dev/null 2>&1") == 0:
  409. return editor
  410. return "vi"
  411. def edit_file(self, filename: str) -> None:
  412. import subprocess
  413. editor = self.get_editor()
  414. environ: t.Optional[t.Dict[str, str]] = None
  415. if self.env:
  416. environ = os.environ.copy()
  417. environ.update(self.env)
  418. try:
  419. c = subprocess.Popen(f'{editor} "{filename}"', env=environ, shell=True)
  420. exit_code = c.wait()
  421. if exit_code != 0:
  422. raise ClickException(
  423. _("{editor}: Editing failed").format(editor=editor)
  424. )
  425. except OSError as e:
  426. raise ClickException(
  427. _("{editor}: Editing failed: {e}").format(editor=editor, e=e)
  428. ) from e
  429. def edit(self, text: t.Optional[t.AnyStr]) -> t.Optional[t.AnyStr]:
  430. import tempfile
  431. if not text:
  432. data = b""
  433. elif isinstance(text, (bytes, bytearray)):
  434. data = text
  435. else:
  436. if text and not text.endswith("\n"):
  437. text += "\n"
  438. if WIN:
  439. data = text.replace("\n", "\r\n").encode("utf-8-sig")
  440. else:
  441. data = text.encode("utf-8")
  442. fd, name = tempfile.mkstemp(prefix="editor-", suffix=self.extension)
  443. f: t.BinaryIO
  444. try:
  445. with os.fdopen(fd, "wb") as f:
  446. f.write(data)
  447. # If the filesystem resolution is 1 second, like Mac OS
  448. # 10.12 Extended, or 2 seconds, like FAT32, and the editor
  449. # closes very fast, require_save can fail. Set the modified
  450. # time to be 2 seconds in the past to work around this.
  451. os.utime(name, (os.path.getatime(name), os.path.getmtime(name) - 2))
  452. # Depending on the resolution, the exact value might not be
  453. # recorded, so get the new recorded value.
  454. timestamp = os.path.getmtime(name)
  455. self.edit_file(name)
  456. if self.require_save and os.path.getmtime(name) == timestamp:
  457. return None
  458. with open(name, "rb") as f:
  459. rv = f.read()
  460. if isinstance(text, (bytes, bytearray)):
  461. return rv
  462. return rv.decode("utf-8-sig").replace("\r\n", "\n") # type: ignore
  463. finally:
  464. os.unlink(name)
  465. def open_url(url: str, wait: bool = False, locate: bool = False) -> int:
  466. import subprocess
  467. def _unquote_file(url: str) -> str:
  468. from urllib.parse import unquote
  469. if url.startswith("file://"):
  470. url = unquote(url[7:])
  471. return url
  472. if sys.platform == "darwin":
  473. args = ["open"]
  474. if wait:
  475. args.append("-W")
  476. if locate:
  477. args.append("-R")
  478. args.append(_unquote_file(url))
  479. null = open("/dev/null", "w")
  480. try:
  481. return subprocess.Popen(args, stderr=null).wait()
  482. finally:
  483. null.close()
  484. elif WIN:
  485. if locate:
  486. url = _unquote_file(url.replace('"', ""))
  487. args = f'explorer /select,"{url}"'
  488. else:
  489. url = url.replace('"', "")
  490. wait_str = "/WAIT" if wait else ""
  491. args = f'start {wait_str} "" "{url}"'
  492. return os.system(args)
  493. elif CYGWIN:
  494. if locate:
  495. url = os.path.dirname(_unquote_file(url).replace('"', ""))
  496. args = f'cygstart "{url}"'
  497. else:
  498. url = url.replace('"', "")
  499. wait_str = "-w" if wait else ""
  500. args = f'cygstart {wait_str} "{url}"'
  501. return os.system(args)
  502. try:
  503. if locate:
  504. url = os.path.dirname(_unquote_file(url)) or "."
  505. else:
  506. url = _unquote_file(url)
  507. c = subprocess.Popen(["xdg-open", url])
  508. if wait:
  509. return c.wait()
  510. return 0
  511. except OSError:
  512. if url.startswith(("http://", "https://")) and not locate and not wait:
  513. import webbrowser
  514. webbrowser.open(url)
  515. return 0
  516. return 1
  517. def _translate_ch_to_exc(ch: str) -> t.Optional[BaseException]:
  518. if ch == "\x03":
  519. raise KeyboardInterrupt()
  520. if ch == "\x04" and not WIN: # Unix-like, Ctrl+D
  521. raise EOFError()
  522. if ch == "\x1a" and WIN: # Windows, Ctrl+Z
  523. raise EOFError()
  524. return None
  525. if WIN:
  526. import msvcrt
  527. @contextlib.contextmanager
  528. def raw_terminal() -> t.Iterator[int]:
  529. yield -1
  530. def getchar(echo: bool) -> str:
  531. # The function `getch` will return a bytes object corresponding to
  532. # the pressed character. Since Windows 10 build 1803, it will also
  533. # return \x00 when called a second time after pressing a regular key.
  534. #
  535. # `getwch` does not share this probably-bugged behavior. Moreover, it
  536. # returns a Unicode object by default, which is what we want.
  537. #
  538. # Either of these functions will return \x00 or \xe0 to indicate
  539. # a special key, and you need to call the same function again to get
  540. # the "rest" of the code. The fun part is that \u00e0 is
  541. # "latin small letter a with grave", so if you type that on a French
  542. # keyboard, you _also_ get a \xe0.
  543. # E.g., consider the Up arrow. This returns \xe0 and then \x48. The
  544. # resulting Unicode string reads as "a with grave" + "capital H".
  545. # This is indistinguishable from when the user actually types
  546. # "a with grave" and then "capital H".
  547. #
  548. # When \xe0 is returned, we assume it's part of a special-key sequence
  549. # and call `getwch` again, but that means that when the user types
  550. # the \u00e0 character, `getchar` doesn't return until a second
  551. # character is typed.
  552. # The alternative is returning immediately, but that would mess up
  553. # cross-platform handling of arrow keys and others that start with
  554. # \xe0. Another option is using `getch`, but then we can't reliably
  555. # read non-ASCII characters, because return values of `getch` are
  556. # limited to the current 8-bit codepage.
  557. #
  558. # Anyway, Click doesn't claim to do this Right(tm), and using `getwch`
  559. # is doing the right thing in more situations than with `getch`.
  560. func: t.Callable[[], str]
  561. if echo:
  562. func = msvcrt.getwche # type: ignore
  563. else:
  564. func = msvcrt.getwch # type: ignore
  565. rv = func()
  566. if rv in ("\x00", "\xe0"):
  567. # \x00 and \xe0 are control characters that indicate special key,
  568. # see above.
  569. rv += func()
  570. _translate_ch_to_exc(rv)
  571. return rv
  572. else:
  573. import tty
  574. import termios
  575. @contextlib.contextmanager
  576. def raw_terminal() -> t.Iterator[int]:
  577. f: t.Optional[t.TextIO]
  578. fd: int
  579. if not isatty(sys.stdin):
  580. f = open("/dev/tty")
  581. fd = f.fileno()
  582. else:
  583. fd = sys.stdin.fileno()
  584. f = None
  585. try:
  586. old_settings = termios.tcgetattr(fd)
  587. try:
  588. tty.setraw(fd)
  589. yield fd
  590. finally:
  591. termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
  592. sys.stdout.flush()
  593. if f is not None:
  594. f.close()
  595. except termios.error:
  596. pass
  597. def getchar(echo: bool) -> str:
  598. with raw_terminal() as fd:
  599. ch = os.read(fd, 32).decode(get_best_encoding(sys.stdin), "replace")
  600. if echo and isatty(sys.stdout):
  601. sys.stdout.write(ch)
  602. _translate_ch_to_exc(ch)
  603. return ch