"""Terminal progress bar spinners."""
import atexit
import logging
import random
import sys
from typing import IO, Any, Sequence
__all__ = ["Spinner", "SpinnerHandler"]
SPINNER_ARC: Sequence[str] = [
"◜",
"◠",
"◝",
"◞",
"◡",
"◟",
]
SPINNER_ARROW: Sequence[str] = [
"←",
"↖",
"↑",
"↗",
"→",
"↘",
"↓",
"↙",
]
SPINNER_CIRCLE: Sequence[str] = [
"•",
"◦",
"●",
"○",
"◎",
"◉",
"⦿",
]
SPINNER_SQUARE: Sequence[str] = [
"◢",
"◣",
"◤",
"◥",
]
SPINNER_MOON: Sequence[str] = [
"🌑 ",
"🌒 ",
"🌓 ",
"🌔 ",
"🌕 ",
"🌖 ",
"🌗 ",
"🌘 ",
]
SPINNERS: Sequence[Sequence[str]] = [
SPINNER_ARC,
SPINNER_ARROW,
SPINNER_CIRCLE,
SPINNER_SQUARE,
SPINNER_MOON,
]
ACTIVE_SPINNER: Sequence[str] = random.choice(SPINNERS) # nosec B311
[docs]class Spinner:
"""Progress bar spinner."""
bell = "\b"
sprites: Sequence[str] = ACTIVE_SPINNER
cursor_hide: str = "\x1b[?25l"
cursor_show: str = "\x1b[?25h"
hide_cursor: bool = True
stopped: bool = False
def __init__(self, file: IO = sys.stderr) -> None:
self.file: IO = file
self.width: int = 0
self.count = 0
self.stopped = False
[docs] def update(self) -> None:
"""Draw spinner, single iteration."""
if not self.stopped:
if not self.count:
self.begin()
i = self.count % len(self.sprites)
self.count += 1
self.write(self.sprites[i])
[docs] def stop(self) -> None:
"""Stop spinner from being emitted."""
self.stopped = True
[docs] def reset(self) -> None:
"""Reset state or allow restart."""
self.stopped = False
self.count = 0
[docs] def write(self, s: str) -> None:
"""Write spinner character to terminal."""
if self.file.isatty():
self._print(f"{self.bell * self.width}{s.ljust(self.width)}")
self.width = max(self.width, len(s))
def _print(self, s: str) -> None:
print(s, end="", file=self.file)
self.file.flush()
[docs] def begin(self) -> None:
"""Prepare terminal for spinner starting."""
atexit.register(type(self)._finish, self.file, at_exit=True)
self._print(self.cursor_hide)
[docs] def finish(self) -> None:
"""Finish spinner and reset terminal."""
print(f"{self.bell * (self.width + 1)}", end="", file=self.file)
self._finish(self.file)
self.stop()
@classmethod
def _finish(cls, file: IO, *, at_exit: bool = False) -> None:
print(cls.cursor_show, end="", file=file)
file.flush()
[docs]class SpinnerHandler(logging.Handler):
"""A logger handler that iterates our progress spinner for each log."""
spinner: Spinner
# For every logging call we advance the terminal spinner (-\/-)
def __init__(self, spinner: Spinner, **kwargs: Any) -> None:
self.spinner = spinner
super().__init__(**kwargs)
[docs] def emit(self, _record: logging.LogRecord) -> None:
"""Emit the next spinner character."""
# the spinner is only in effect with WARN level and below.
if self.spinner and not self.spinner.stopped:
self.spinner.update()
