make CLI spinner animation more robust
This commit is contained in:
parent
3ab468205f
commit
18e0a8ecea
@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "dopt-basics"
|
name = "dopt-basics"
|
||||||
version = "0.2.1"
|
version = "0.2.2"
|
||||||
description = "basic cross-project tools for Python-based d-opt projects"
|
description = "basic cross-project tools for Python-based d-opt projects"
|
||||||
authors = [
|
authors = [
|
||||||
{name = "Florian Förster", email = "f.foerster@d-opt.com"},
|
{name = "Florian Förster", email = "f.foerster@d-opt.com"},
|
||||||
@ -69,7 +69,7 @@ directory = "reports/coverage"
|
|||||||
|
|
||||||
|
|
||||||
[tool.bumpversion]
|
[tool.bumpversion]
|
||||||
current_version = "0.2.1"
|
current_version = "0.2.2"
|
||||||
parse = """(?x)
|
parse = """(?x)
|
||||||
(?P<major>0|[1-9]\\d*)\\.
|
(?P<major>0|[1-9]\\d*)\\.
|
||||||
(?P<minor>0|[1-9]\\d*)\\.
|
(?P<minor>0|[1-9]\\d*)\\.
|
||||||
|
|||||||
@ -1,4 +1,5 @@
|
|||||||
import shutil
|
import shutil
|
||||||
|
import sys
|
||||||
import time
|
import time
|
||||||
import typing as t
|
import typing as t
|
||||||
from collections.abc import Callable
|
from collections.abc import Callable
|
||||||
@ -17,41 +18,77 @@ class LoadingAnimation:
|
|||||||
loading_txt: str,
|
loading_txt: str,
|
||||||
ending_text: str,
|
ending_text: str,
|
||||||
timeout: float = 0.1,
|
timeout: float = 0.1,
|
||||||
|
force_ascii: bool = False,
|
||||||
) -> None:
|
) -> None:
|
||||||
self.loading_txt = loading_txt
|
self.loading_txt = loading_txt
|
||||||
self.ending_text = ending_text
|
self.ending_text = ending_text
|
||||||
self.timeout = timeout
|
self.timeout = timeout
|
||||||
self.done: bool = False
|
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()
|
self.start()
|
||||||
|
return self
|
||||||
|
|
||||||
def __exit__(
|
def __exit__(
|
||||||
self,
|
self,
|
||||||
exc_type,
|
exc_type: type[Exception],
|
||||||
exc_value,
|
exc_value,
|
||||||
tb,
|
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()
|
self.stop()
|
||||||
|
return False
|
||||||
|
|
||||||
def _animation(self) -> None:
|
def _animation(self) -> None:
|
||||||
for c in cycle(self.steps):
|
for frame in cycle(self.frames):
|
||||||
if self.done:
|
if self.done:
|
||||||
break
|
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)
|
time.sleep(self.timeout)
|
||||||
|
|
||||||
def start(self) -> None:
|
def start(self) -> None:
|
||||||
|
if self._do_animation:
|
||||||
self._thread.start()
|
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
|
self.done = True
|
||||||
|
if self._do_animation:
|
||||||
|
self._thread.join()
|
||||||
cols = shutil.get_terminal_size((80, 20)).columns
|
cols = shutil.get_terminal_size((80, 20)).columns
|
||||||
print("\r" + " " * cols, end="", flush=True)
|
print("\r" + " " * cols, end="\r", flush=True)
|
||||||
print(f"\r{self.ending_text}", flush=True)
|
if interrupt:
|
||||||
|
return
|
||||||
|
print(f"{self.ending_text}", flush=True)
|
||||||
|
|
||||||
|
|
||||||
def default_loading(func: Callable[P, T]) -> Callable[P, T]:
|
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):
|
with LoadingAnimation(loading_txt, ending_text):
|
||||||
res = func(*args, **kwargs)
|
res = func(*args, **kwargs)
|
||||||
|
|
||||||
return res
|
return res # type: ignore
|
||||||
|
|
||||||
return wrapper
|
return wrapper
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user