Compare commits
No commits in common. "main" and "v0.2.0" have entirely different histories.
@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "dopt-basics"
|
name = "dopt-basics"
|
||||||
version = "0.2.4"
|
version = "0.2.0"
|
||||||
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.4"
|
current_version = "0.2.0"
|
||||||
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,108 +0,0 @@
|
|||||||
import shutil
|
|
||||||
import sys
|
|
||||||
import time
|
|
||||||
import typing as t
|
|
||||||
from collections.abc import Callable
|
|
||||||
from functools import wraps
|
|
||||||
from itertools import cycle
|
|
||||||
from threading import Thread
|
|
||||||
|
|
||||||
P = t.ParamSpec("P")
|
|
||||||
T = t.TypeVar("T")
|
|
||||||
|
|
||||||
|
|
||||||
# based on: https://stackoverflow.com/questions/22029562/python-how-to-make-simple-animated-loading-while-process-is-running
|
|
||||||
class LoadingAnimation:
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
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.keyboard_interrupt: bool = False
|
|
||||||
|
|
||||||
self._isatty = sys.stdout.isatty()
|
|
||||||
self._do_animation = bool(sys.stdout.isatty())
|
|
||||||
|
|
||||||
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: type[Exception],
|
|
||||||
exc_value,
|
|
||||||
tb,
|
|
||||||
) -> None:
|
|
||||||
if exc_type is not None:
|
|
||||||
self.stop(interrupt=True)
|
|
||||||
if exc_type is KeyboardInterrupt:
|
|
||||||
self.keyboard_interrupt = True
|
|
||||||
print("Operation cancelled by user. (KeyboardInterrupt)", flush=True)
|
|
||||||
return
|
|
||||||
|
|
||||||
self.stop()
|
|
||||||
|
|
||||||
def _animation(self) -> None:
|
|
||||||
for frame in cycle(self.frames):
|
|
||||||
if self.done:
|
|
||||||
break
|
|
||||||
print(f"\r{frame} {self.loading_txt}", flush=True, end="")
|
|
||||||
time.sleep(self.timeout)
|
|
||||||
|
|
||||||
def start(self) -> None:
|
|
||||||
if self._do_animation:
|
|
||||||
self._thread.start()
|
|
||||||
else:
|
|
||||||
print(f"\r{self.loading_txt}", end="", flush=True)
|
|
||||||
|
|
||||||
def stop(
|
|
||||||
self,
|
|
||||||
interrupt: bool = False,
|
|
||||||
) -> None:
|
|
||||||
if self.done:
|
|
||||||
return
|
|
||||||
|
|
||||||
self.done = True
|
|
||||||
if self._do_animation:
|
|
||||||
self._thread.join()
|
|
||||||
cols = shutil.get_terminal_size((80, 20)).columns
|
|
||||||
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]:
|
|
||||||
@wraps(func)
|
|
||||||
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
|
|
||||||
loading_txt: str = "Performing operation..."
|
|
||||||
ending_text: str = "Operation successful"
|
|
||||||
|
|
||||||
with LoadingAnimation(loading_txt, ending_text):
|
|
||||||
res = func(*args, **kwargs)
|
|
||||||
|
|
||||||
return res
|
|
||||||
|
|
||||||
return wrapper
|
|
||||||
Loading…
x
Reference in New Issue
Block a user