diff --git a/pyproject.toml b/pyproject.toml index fc000ee..f175257 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "dopt-basics" -version = "0.2.1" +version = "0.2.2" description = "basic cross-project tools for Python-based d-opt projects" authors = [ {name = "Florian Förster", email = "f.foerster@d-opt.com"}, @@ -69,7 +69,7 @@ directory = "reports/coverage" [tool.bumpversion] -current_version = "0.2.1" +current_version = "0.2.2" parse = """(?x) (?P0|[1-9]\\d*)\\. (?P0|[1-9]\\d*)\\. diff --git a/src/dopt_basics/cli.py b/src/dopt_basics/cli.py index 94c52b9..6c4be60 100644 --- a/src/dopt_basics/cli.py +++ b/src/dopt_basics/cli.py @@ -1,4 +1,5 @@ import shutil +import sys import time import typing as t from collections.abc import Callable @@ -17,41 +18,77 @@ class LoadingAnimation: loading_txt: str, ending_text: str, timeout: float = 0.1, + force_ascii: bool = False, ) -> None: self.loading_txt = loading_txt self.ending_text = ending_text self.timeout = timeout self.done: bool = False - self.steps: list[str] = ["⢿", "⣻", "⣽", "⣾", "⣷", "⣯", "⣟", "⡿"] - self._thread: Thread = Thread(target=self._animation, daemon=True) + self._isatty = sys.stdout.isatty() + self._do_animation = bool(sys.stdout.isatty()) - def __enter__(self) -> None: + self._animate_dots: list[str] = ["⢿", "⣻", "⣽", "⣾", "⣷", "⣯", "⣟", "⡿"] + self._animate_ascii: list[str] = ["|", "/", "-", "\\"] + + self.frames: list[str] + enc = (sys.stdout.encoding or "utf-8").lower() + if force_ascii: + self.frames = self._animate_ascii + else: + try: + self._animate_dots[0].encode(enc) + self.frames = self._animate_dots + except Exception: + self.frames = self._animate_ascii + + self._thread: Thread = Thread(target=self._animation) + + def __enter__(self) -> t.Self: self.start() + return self def __exit__( self, - exc_type, + exc_type: type[Exception], exc_value, tb, - ) -> None: + ) -> bool: + if exc_type is not None: + self.stop(interrupt=True) + if exc_type is KeyboardInterrupt: + print("Operation cancelled by user. (KeyboardInterrupt)", flush=True) + return True + return False + self.stop() + return False def _animation(self) -> None: - for c in cycle(self.steps): + for frame in cycle(self.frames): if self.done: break - print(f"\r{c} {self.loading_txt}", flush=True, end="") + print(f"\r{frame} {self.loading_txt}", flush=True, end="") time.sleep(self.timeout) def start(self) -> None: - self._thread.start() + if self._do_animation: + self._thread.start() + else: + print(f"\r{self.loading_txt}", end="", flush=True) - def stop(self): + def stop( + self, + interrupt: bool = False, + ) -> None: self.done = True + if self._do_animation: + self._thread.join() cols = shutil.get_terminal_size((80, 20)).columns - print("\r" + " " * cols, end="", flush=True) - print(f"\r{self.ending_text}", flush=True) + print("\r" + " " * cols, end="\r", flush=True) + if interrupt: + return + print(f"{self.ending_text}", flush=True) def default_loading(func: Callable[P, T]) -> Callable[P, T]: @@ -63,6 +100,6 @@ def default_loading(func: Callable[P, T]) -> Callable[P, T]: with LoadingAnimation(loading_txt, ending_text): res = func(*args, **kwargs) - return res + return res # type: ignore return wrapper