144 lines
No EOL
3.7 KiB
Python
144 lines
No EOL
3.7 KiB
Python
import sys
|
|
import threading
|
|
import time
|
|
|
|
RESET = "\033[0m"
|
|
BOLD = "\033[1m"
|
|
DIM = "\033[2m"
|
|
|
|
RED = "\033[0;31m"
|
|
GREEN = "\033[0;32m"
|
|
YELLOW = "\033[0;33m"
|
|
BLUE = "\033[0;34m"
|
|
CYAN = "\033[0;36m"
|
|
WHITE = "\033[0;37m"
|
|
|
|
BRED = "\033[1;31m"
|
|
BGREEN = "\033[1;32m"
|
|
BYELLOW = "\033[1;33m"
|
|
BBLUE = "\033[1;34m"
|
|
BCYAN = "\033[1;36m"
|
|
BWHITE = "\033[1;37m"
|
|
|
|
def label_ok(msg: str):
|
|
print(f" {BGREEN}[ OK ]{RESET} {msg}")
|
|
|
|
def label_warn(msg: str):
|
|
print(f" {BYELLOW}[ WARN ]{RESET} {msg}")
|
|
|
|
def label_error(msg: str):
|
|
print(f" {BRED}[ ERROR]{RESET} {msg}")
|
|
|
|
def label_info(msg: str):
|
|
print(f" {BCYAN}[ INFO ]{RESET} {msg}")
|
|
|
|
def label_skip(msg: str):
|
|
print(f" {DIM}[ SKIP ]{RESET} {msg}")
|
|
|
|
def label_connected(msg: str):
|
|
print(f" {BGREEN}[CONNECTED ]{RESET} {msg}")
|
|
|
|
def label_disconnect(msg: str):
|
|
print(f" {BRED}[DISCONNECTED]{RESET} {msg}")
|
|
|
|
def label_step(msg: str, current: int, total: int):
|
|
print(f" {BBLUE}[ {current:<2}/{total:<2} ]{RESET} {msg}")
|
|
|
|
def print_header(title: str, version: str = ""):
|
|
width = 60
|
|
line = "─" * width
|
|
ver = f"{DIM}{version}{RESET}" if version else ""
|
|
print()
|
|
print(f"{BCYAN}{line}{RESET}")
|
|
print(f"{BCYAN} {'':<56}{RESET}")
|
|
print(f"{BCYAN} {BWHITE}{title} {ver}{'':<40}{BCYAN} {RESET}")
|
|
print(f"{BCYAN} {'':<56}{RESET}")
|
|
print(f"{BCYAN}{line}{RESET}")
|
|
print()
|
|
|
|
def print_section(label: str):
|
|
print()
|
|
print(f"{DIM}{CYAN}┌─{RESET} {BWHITE}{label}{RESET}")
|
|
|
|
def print_divider():
|
|
print(f"{DIM}{'─' * 60}{RESET}")
|
|
|
|
def progress_bar(current: int, total: int, label: str = "", width: int = 40):
|
|
filled = int(current * width / total)
|
|
empty = width - filled
|
|
bar = "#" * filled + "·" * empty
|
|
pct = int(current * 100 / total)
|
|
sys.stdout.write(
|
|
f"\r {DIM}[{RESET}{BGREEN}{bar}{RESET}{DIM}]{RESET}"
|
|
f" {BWHITE}{pct:3d}%{RESET} {DIM}{label}{RESET} "
|
|
)
|
|
sys.stdout.flush()
|
|
|
|
def progress_simulate(steps: int = 20, label: str = "loading", delay: float = 0.04):
|
|
for i in range(1, steps + 1):
|
|
progress_bar(i, steps, label)
|
|
time.sleep(delay)
|
|
print()
|
|
|
|
class Spinner:
|
|
FRAMES = ("─", "\\", "│", "/")
|
|
|
|
def __init__(self, msg: str = "Processing..."):
|
|
self.msg = msg
|
|
self._stop_event = threading.Event()
|
|
self._thread: threading.Thread | None = None
|
|
|
|
def start(self):
|
|
self._stop_event.clear()
|
|
self._thread = threading.Thread(target=self._spin, daemon=True)
|
|
self._thread.start()
|
|
return self
|
|
|
|
def stop(self, result_msg: str = ""):
|
|
self._stop_event.set()
|
|
if self._thread:
|
|
self._thread.join()
|
|
sys.stdout.write(f"\r{' ' * 60}\r")
|
|
sys.stdout.flush()
|
|
if result_msg:
|
|
label_ok(result_msg)
|
|
|
|
def _spin(self):
|
|
i = 0
|
|
while not self._stop_event.is_set():
|
|
frame = self.FRAMES[i % len(self.FRAMES)]
|
|
sys.stdout.write(
|
|
f"\r {BCYAN}{frame}{RESET} {DIM}{self.msg}{RESET} "
|
|
)
|
|
sys.stdout.flush()
|
|
i += 1
|
|
time.sleep(0.1)
|
|
|
|
def __enter__(self):
|
|
self.start()
|
|
return self
|
|
|
|
def __exit__(self, *_):
|
|
self.stop()
|
|
|
|
|
|
def wait_with_spinner(msg: str, fn, *args, result_msg: str = "", **kwargs):
|
|
|
|
result_box = [None, None]
|
|
|
|
def _run():
|
|
try:
|
|
result_box[0] = fn(*args, **kwargs)
|
|
except Exception as e:
|
|
result_box[1] = e
|
|
|
|
s = Spinner(msg)
|
|
s.start()
|
|
t = threading.Thread(target=_run, daemon=True)
|
|
t.start()
|
|
t.join()
|
|
s.stop(result_msg)
|
|
|
|
if result_box[1] is not None:
|
|
raise result_box[1]
|
|
return result_box[0] |