make CLI spinner animation more robust
This commit is contained in:
parent
3ab468205f
commit
50473f4cea
@ -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)
|
||||
(?P<major>0|[1-9]\\d*)\\.
|
||||
(?P<minor>0|[1-9]\\d*)\\.
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user