from contextlib import contextmanager
from enum import Enum, auto as auto_enum
from logging import basicConfig, Handler, LogRecord, INFO, WARNING
from time import monotonic
from typing import List, Dict, Generator, Protocol
from rich.text import Text
from textual.app import ComposeResult
from textual.notifications import SeverityLevel
from textual.reactive import reactive, Reactive
from textual.timer import Timer
from textual.widget import Widget
from textual.widgets import Label, Static
class Notifier(Protocol):
def notify(self, message: str, *, title: str, severity: SeverityLevel, timeout: float) -> None:
...
class NotificationHandler(Handler):
def __init__(self, handler: Notifier, timeout: float = 10.0) -> None:
self._handler: Notifier = handler
self._timeout: float = timeout
super().__init__()
@classmethod
def _translate_level(cls, level: int) -> SeverityLevel:
if level <= INFO:
return "information"
elif level <= WARNING:
return "warning"
else:
return "error"
def emit(self, record: LogRecord) -> None:
self._handler.notify(message=self.format(record), title=record.name,
severity=self._translate_level(record.levelno), timeout=self._timeout)
@classmethod
def configure_logging(cls, handler: Notifier) -> None:
basicConfig(level=INFO, format="%(message)s", handlers=[NotificationHandler(handler)])
class BusyIndicator(Static):
busy: Reactive[bool] = reactive(False)
def __init__(self) -> None:
self._frames: List[Text] = [Text(_) for _ in "⠒⠐⠰⠴⠤⠄⠆⠖"]
self._placeholder: Text = Text("⠶")
super().__init__(self._placeholder)
self._interval: float = 0.1
self._timer: Timer = self.set_interval(self._interval / 2.0, self._update, pause=True)
def _update(self) -> None:
if self.busy: # spurious leftover call otherwise
self.renderable = self._frames[round(monotonic() / self._interval) % len(self._frames)]
self.refresh()
def watch_busy(self, busy: bool) -> None:
if busy:
self._timer.resume()
else:
self._timer.pause()
self.renderable = self._placeholder
self.refresh()
class StatusIcon(Static):
DEFAULT_CSS = """
StatusIcon.success {
color: $success;
}
StatusIcon.warning {
color: $warning;
}
StatusIcon.error {
color: $error;
}
"""
class Status(Enum):
unknown = auto_enum()
idle = auto_enum()
dirty = auto_enum()
busy = auto_enum()
error = auto_enum()
status: Reactive[Status] = reactive(Status.unknown)
def __init__(self) -> None:
self._statuses: Dict[StatusIcon.Status, Text] = {
StatusIcon.Status.unknown: Text(" "),
StatusIcon.Status.idle: Text("✓"),
StatusIcon.Status.dirty: Text("✱"),
StatusIcon.Status.busy: Text("⇅"),
StatusIcon.Status.error: Text("✗"),
}
self._status_classes: Dict[StatusIcon.Status, str] = {
StatusIcon.Status.unknown: "warning",
StatusIcon.Status.idle: "success",
StatusIcon.Status.dirty: "warning",
StatusIcon.Status.busy: "warning",
StatusIcon.Status.error: "error",
}
super().__init__(classes=self._status_classes[StatusIcon.Status.unknown])
self.renderable = self._statuses[StatusIcon.Status.unknown]
def watch_status(self, status: Status) -> None:
self.renderable = self._statuses[status]
self.classes = self._status_classes[status]
self.refresh()
class StatusBar(Widget):
DEFAULT_CSS = """
StatusBar {
dock: top;
width: 100%;
background: $foreground 5%;
color: $text;
height: 1;
layout: grid;
grid-size: 3 1;
grid-columns: 1 1fr 1;
grid-rows: 1;
grid-gutter: 0 1;
padding: 0 1;
}
StatusBar Label {
content-align: center middle;
width: 100%;
}
"""
busy: Reactive[int] = reactive(0)
sync: Reactive[bool] = reactive(False)
active: Reactive[bool] = reactive(False)
dirty: Reactive[bool] = reactive(False)
def __init__(self, title: str) -> None:
super().__init__()
self._busy_indicator: BusyIndicator = BusyIndicator()
self._title: Label = Label(title, markup=False)
self._status_icon: StatusIcon = StatusIcon()
def compose(self) -> ComposeResult:
yield self._busy_indicator
yield self._title
yield self._status_icon
def _update(self) -> None:
self._busy_indicator.busy = self.busy > 0 or self.sync
if self.sync:
self._status_icon.status = StatusIcon.Status.busy
elif self.dirty:
self._status_icon.status = StatusIcon.Status.dirty
elif self.active:
self._status_icon.status = StatusIcon.Status.idle
else:
self._status_icon.status = StatusIcon.Status.unknown
def watch_dirty(self) -> None:
self._update()
def watch_sync(self) -> None:
self._update()
def watch_busy(self) -> None:
self._update()
def watch_active(self) -> None:
self._update()
@contextmanager
def sync_ctx(self) -> Generator[None, None, None]:
try:
self.sync = True
yield
finally:
self.sync = False
@contextmanager
def busy_ctx(self) -> Generator[None, None, None]:
try:
self.busy += 1
yield
finally:
self.busy -= 1