mirror of
https://github.com/samjage/metro-warden.git
synced 2026-06-06 04:00:42 +00:00
123 lines
3.2 KiB
Python
123 lines
3.2 KiB
Python
"""
|
|
Metro Warden header widget — industrial metro-map style banner.
|
|
|
|
Displays the application name, a live clock, and a pulsing status indicator
|
|
that reflects overall system health sourced from the state store.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from textual.app import ComposeResult
|
|
from textual.reactive import reactive
|
|
from textual.widget import Widget
|
|
from textual.widgets import Label, Static
|
|
|
|
|
|
class PulseIndicator(Static):
|
|
"""A small pulsing dot that animates to show the system is alive."""
|
|
|
|
DEFAULT_CSS = """
|
|
PulseIndicator {
|
|
width: 3;
|
|
height: 1;
|
|
color: $success;
|
|
}
|
|
PulseIndicator.pulse-off {
|
|
color: $panel;
|
|
}
|
|
"""
|
|
|
|
_frames = ["●", "○"]
|
|
_frame_idx: reactive[int] = reactive(0)
|
|
|
|
def on_mount(self) -> None:
|
|
self.set_interval(1.0, self._tick)
|
|
|
|
def _tick(self) -> None:
|
|
self._frame_idx = (self._frame_idx + 1) % len(self._frames)
|
|
self.update(self._frames[self._frame_idx])
|
|
if self._frame_idx == 0:
|
|
self.add_class("pulse-off")
|
|
else:
|
|
self.remove_class("pulse-off")
|
|
|
|
def render(self):
|
|
return self._frames[self._frame_idx]
|
|
|
|
|
|
class MetroHeader(Widget):
|
|
"""
|
|
Industrial metro-map style header bar.
|
|
|
|
Shows:
|
|
- Application name and tagline (left)
|
|
- Live UTC clock (centre)
|
|
- Pulse indicator + status label (right)
|
|
"""
|
|
|
|
DEFAULT_CSS = """
|
|
MetroHeader {
|
|
height: 3;
|
|
background: $surface;
|
|
border-bottom: heavy $primary;
|
|
layout: horizontal;
|
|
align: left middle;
|
|
padding: 0 1;
|
|
}
|
|
|
|
MetroHeader .header-brand {
|
|
width: 1fr;
|
|
color: $primary;
|
|
text-style: bold;
|
|
}
|
|
|
|
MetroHeader .header-clock {
|
|
width: auto;
|
|
color: $text-muted;
|
|
text-align: center;
|
|
min-width: 24;
|
|
}
|
|
|
|
MetroHeader .header-status {
|
|
width: 1fr;
|
|
text-align: right;
|
|
color: $success;
|
|
}
|
|
|
|
MetroHeader .header-sep {
|
|
color: $primary;
|
|
width: 1;
|
|
}
|
|
"""
|
|
|
|
_clock: reactive[str] = reactive("--:--:-- UTC")
|
|
_status: reactive[str] = reactive("NOMINAL")
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Static("METRO WARDEN // NOC", classes="header-brand")
|
|
yield Static(self._clock, id="header-clock", classes="header-clock")
|
|
yield Static("▶ " + self._status, id="header-status", classes="header-status")
|
|
|
|
def on_mount(self) -> None:
|
|
self.set_interval(1.0, self._update_clock)
|
|
|
|
def _update_clock(self) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
self._clock = now.strftime("%Y-%m-%d %H:%M:%S UTC")
|
|
clock_widget = self.query_one("#header-clock", Static)
|
|
clock_widget.update(self._clock)
|
|
|
|
def set_status(self, status: str, healthy: bool = True) -> None:
|
|
"""Update the status label text and colour."""
|
|
self._status = status
|
|
status_widget = self.query_one("#header-status", Static)
|
|
status_widget.update(("▶ " if healthy else "▲ ") + status)
|
|
if healthy:
|
|
status_widget.remove_class("status-warn")
|
|
else:
|
|
status_widget.add_class("status-warn")
|