server-monitor-dash/server_dash.py

#!/usr/bin/env python3
# Created by John A. Hoeven with the ethical assistance of Claude AI
# ---------------------------------------------------------------------------
# server_dash.py
# /server-monitor-dash/server_dash.py
# Version: v0.0.2 | Status: DEVELOPMENT
# ---------------------------------------------------------------------------
# Purpose:  Case-mounted touchscreen monitoring dashboard for Cervello
#           Elettrico servers — CPU, GPU, memory telemetry rendered
#           "supercar dash" style. Textual built-ins only (Digits,
#           ProgressBar, Sparkline) plus stdlib /proc parsing for CPU/mem
#           and a subprocess call to nvidia-smi for GPU, reusing the same
#           parsing approach as the thermal logging script.
# Target:   Cervello Elettrico servers (AlmaLinux 10), 5" case-mounted
#           resistive touchscreen (800x480, e.g. Waveshare 5" HDMI LCD)
# Entry:    python3 server_dash.py  (run under the monitoring user's
#           autologin session, no X required)
# Depends:  textual (pip install textual --break-system-packages, or pipx)
# ---------------------------------------------------------------------------
# STATUS: v0.0.2 — replaces the v0.0.1 tabbed layout with a scroll-section
# layout navigated via a fixed touch-bar sidebar (up/down), designed for
# resistive touch and for use by hands of any shape or digit count — no
# gesture assumes a specific hand anatomy, and touch zones are deliberately
# oversized with a forgiving hit-margin beyond their visible bar (see
# TouchZone below). Fixes from the v0.0.1 review: removed the unused
# `time` import, renamed the app class away from the BigBoy-specific name,
# and applied UTIL_THRESHOLDS consistently across all gauges (previously
# only the GPU temp readout was colour-coded).
#
# Still not yet a CE OS-compliant script (no test gates / cleanup trap) —
# this remains a prototype for iterating look, feel, and navigation, not
# a production deployment.
# ---------------------------------------------------------------------------

from __future__ import annotations

import subprocess
from dataclasses import dataclass

from textual.app import App, ComposeResult
from textual.containers import Container, Horizontal, Vertical, VerticalScroll
from textual.screen import Screen
from textual.widgets import Digits, Footer, Header, Label, ProgressBar, Sparkline, Static

REFRESH_INTERVAL_SECONDS = 1.0
SPARKLINE_HISTORY_LENGTH = 60
IDLE_TIMEOUT_SECONDS = 15 * 60

SECTION_IDS = ["sec-dashboard", "sec-assistance", "sec-switch-x"]

# --- Supercar-dash colour thresholds -----------------------------------------
# Green -> amber -> red, same instinct as a tachometer redline. Values are
# based on empirically observed thermal-error thresholds on constrained
# hardware in this project, not arbitrary numbers.
TEMP_THRESHOLDS = {"amber": 65, "red": 75}   # degrees C
UTIL_THRESHOLDS = {"amber": 60, "red": 90}   # percent


def threshold_colour(value: float, thresholds: dict[str, float]) -> str:
    """Return a Textual colour name for a gauge value against its thresholds."""
    if value >= thresholds["red"]:
        return "red"
    if value >= thresholds["amber"]:
        return "yellow"
    return "green"


# --- CPU stats, pure stdlib, /proc/stat -------------------------------------

@dataclass
class CpuSample:
    total: int
    idle: int


def read_cpu_sample(core_index: int | None = None) -> CpuSample:
    """
    Read one instantaneous sample from /proc/stat.
    core_index=None reads the aggregate 'cpu' line; an int reads 'cpuN'.
    """
    target = "cpu" if core_index is None else f"cpu{core_index}"
    with open("/proc/stat") as f:
        for line in f:
            fields = line.split()
            if fields[0] == target:
                nums = [int(x) for x in fields[1:]]
                idle = nums[3]
                total = sum(nums)
                return CpuSample(total=total, idle=idle)
    raise RuntimeError(f"Could not find {target} in /proc/stat")


def cpu_utilization_pct(prev: CpuSample, curr: CpuSample) -> float:
    """Utilization % between two /proc/stat samples of the same core."""
    total_delta = curr.total - prev.total
    idle_delta = curr.idle - prev.idle
    if total_delta <= 0:
        return 0.0
    return 100.0 * (1.0 - (idle_delta / total_delta))


def count_cpu_cores() -> int:
    with open("/proc/stat") as f:
        return sum(1 for line in f if line.startswith("cpu") and line[3].isdigit())


def read_mem_used_pct() -> float:
    """Memory used % from /proc/meminfo (MemTotal vs MemAvailable)."""
    values: dict[str, int] = {}
    with open("/proc/meminfo") as f:
        for line in f:
            key, rest = line.split(":", 1)
            values[key] = int(rest.strip().split()[0])  # kB
    total = values["MemTotal"]
    available = values.get("MemAvailable", values["MemFree"])
    if total <= 0:
        return 0.0
    return 100.0 * (1.0 - (available / total))


# --- GPU stats via nvidia-smi (same parsing approach as thermal logger) ----

def read_gpu_stats() -> dict[str, float]:
    """
    Query nvidia-smi for temp/util/power/fan. Returns zeros on any failure
    rather than raising — a monitoring dashboard should degrade gracefully,
    not crash, if the GPU query hiccups for one cycle.
    """
    try:
        out = subprocess.run(
            [
                "nvidia-smi",
                "--query-gpu=temperature.gpu,utilization.gpu,power.draw,fan.speed",
                "--format=csv,noheader,nounits",
            ],
            capture_output=True,
            text=True,
            timeout=3,
            check=True,
        ).stdout.strip()
        temp, util, power, fan = (float(x.strip()) for x in out.split(","))
        return {"temp": temp, "util": util, "power": power, "fan": fan}
    except Exception:
        return {"temp": 0.0, "util": 0.0, "power": 0.0, "fan": 0.0}


# --- Touch bar: fixed sidebar, oversized forgiving touch zones --------------

class TouchZone(Container):
    """
    A clickable zone WIDER than its visible coloured bar. The visible bar
    (an inner Static, TOUCHBAR_VISIBLE_WIDTH cells) is what a person sees;
    the zone's actual clickable region (TOUCHBAR_CLICK_WIDTH cells) extends
    further right, so a touch that lands just past the visible edge still
    registers. Deliberately generous and forgiving — sized to "whatever's
    closest to the index finger on the hands a person has," not to a
    specific hand shape or digit count, and not requiring a precise
    pointer. See TouchZone CSS below for actual dimensions.
    """

    def __init__(self, label: str, direction: str, **kwargs) -> None:
        super().__init__(**kwargs)
        self.direction = direction
        self._label = label

    def compose(self) -> ComposeResult:
        yield Static(self._label, classes="visible-bar")

    def on_click(self) -> None:
        self.app.handle_zone_tap(self.direction)  # type: ignore[attr-defined]


# --- Dashboard section content ------------------------------------------------

class DashboardSection(Container):
    """Live telemetry section — the 'dash' itself. Composed inline into the
    scroll stack, not a Screen (Screen is reserved for BlankScreen, which
    genuinely uses the app's push/pop screen stack)."""

    def __init__(self) -> None:
        super().__init__(id=SECTION_IDS[0], classes="section")
        self._core_count = count_cpu_cores()
        self._prev_core_samples = [read_cpu_sample(i) for i in range(self._core_count)]
        self._gpu_temp_history: list[float] = []
        self._gpu_util_history: list[float] = []

    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
        with Vertical():
            with Horizontal(id="gpu-panel"):
                yield Label("GPU TEMP", classes="gauge-label")
                yield Digits("--", id="gpu-temp-digits", classes="big-digits")
                yield Label("GPU UTIL", classes="gauge-label")
                yield ProgressBar(id="gpu-util-bar", total=100)
                yield Sparkline([], id="gpu-temp-spark")
                yield Sparkline([], id="gpu-util-spark")
            with Horizontal(id="cpu-panel"):
                yield Label("CPU CORES", classes="gauge-label")
                for i in range(self._core_count):
                    yield ProgressBar(id=f"cpu-core-{i}", total=100)
            with Horizontal(id="mem-panel"):
                yield Label("MEMORY", classes="gauge-label")
                yield ProgressBar(id="mem-bar", total=100)
        yield Footer()

    def on_mount(self) -> None:
        self.set_interval(REFRESH_INTERVAL_SECONDS, self._refresh_telemetry)

    def _colour_bar(self, bar: ProgressBar, pct: float, thresholds: dict[str, float]) -> None:
        """Apply threshold colour to a ProgressBar's inner Bar widget."""
        bar.query_one("Bar").styles.color = threshold_colour(pct, thresholds)

    def _refresh_telemetry(self) -> None:
        # --- CPU per-core ---
        for i in range(self._core_count):
            curr = read_cpu_sample(i)
            pct = cpu_utilization_pct(self._prev_core_samples[i], curr)
            self._prev_core_samples[i] = curr
            bar = self.query_one(f"#cpu-core-{i}", ProgressBar)
            bar.update(progress=pct)
            self._colour_bar(bar, pct, UTIL_THRESHOLDS)

        # --- Memory ---
        mem_pct = read_mem_used_pct()
        mem_bar = self.query_one("#mem-bar", ProgressBar)
        mem_bar.update(progress=mem_pct)
        self._colour_bar(mem_bar, mem_pct, UTIL_THRESHOLDS)

        # --- GPU ---
        gpu = read_gpu_stats()

        temp_digits = self.query_one("#gpu-temp-digits", Digits)
        temp_digits.update(f"{gpu['temp']:.0f}°")
        temp_digits.styles.color = threshold_colour(gpu["temp"], TEMP_THRESHOLDS)

        util_bar = self.query_one("#gpu-util-bar", ProgressBar)
        util_bar.update(progress=gpu["util"])
        self._colour_bar(util_bar, gpu["util"], UTIL_THRESHOLDS)

        self._gpu_temp_history.append(gpu["temp"])
        self._gpu_temp_history = self._gpu_temp_history[-SPARKLINE_HISTORY_LENGTH:]
        self.query_one("#gpu-temp-spark", Sparkline).data = self._gpu_temp_history

        self._gpu_util_history.append(gpu["util"])
        self._gpu_util_history = self._gpu_util_history[-SPARKLINE_HISTORY_LENGTH:]
        self.query_one("#gpu-util-spark", Sparkline).data = self._gpu_util_history


# --- Stub sections — Request Assistance / Switch to X -----------------------

class RequestAssistanceSection(Container):
    """STUB: will send a scripted email via smtplib on button press,
    notifying the responsible admin that this host requests assistance."""

    def __init__(self) -> None:
        super().__init__(id=SECTION_IDS[1], classes="section")

    def compose(self) -> ComposeResult:
        yield Label("Request Assistance — not yet wired up. [STUB]")


class SwitchToXSection(Container):
    """STUB: will call a narrowly-scoped, sudoers-gated wrapper script that
    switches to a dedicated VT and drops to a standard login prompt
    (password required) before X starts, for GPU tuning tools that need
    it. See the CE security-in-deploy context doc for why this is gated
    behind a real login, not a direct launch from an unauthenticated
    monitoring session."""

    def __init__(self) -> None:
        super().__init__(id=SECTION_IDS[2], classes="section")

    def compose(self) -> ComposeResult:
        yield Label("Switch to X (GPU tuning) — not yet wired up. [STUB]")


# --- Idle blank screen --------------------------------------------------------

class BlankScreen(Screen):
    """Solid dark screen shown after idle timeout — protects the physical
    panel from years of always-on burn-in, consistent with the fleet's
    decade-longevity philosophy applied to this display specifically."""

    CSS = "Screen { background: #000000; }"

    def compose(self) -> ComposeResult:
        yield Label("", id="blank-spacer")


# --- App shell: touch-bar + scroll-sections + idle/blank ----------------------

class ServerMonitorDashApp(App):
    """
    Case-display dashboard for Cervello Elettrico servers.

    Layout: a fixed-width touch-bar sidebar (up/down navigation) beside a
    VerticalScroll containing one min-height:100vh section per "page"
    (Dashboard, Request Assistance, Switch to X). Navigating is
    button-triggered only — no reliance on raw touch-swipe fidelity, which
    was left unverified on this specific touchscreen hardware. Navigation
    index is tracked explicitly rather than inferred from scroll pixel
    offset (inferring from offset proved fragile in testing).
    """

    CSS = """
    Screen { background: #0a0a0a; }

    #touchbar {
        width: 20;
        height: 1fr;
    }
    TouchZone {
        height: 1fr;
        background: #000000;
    }
    .visible-bar {
        width: 12;
        height: 100%;
        background: #1a1a1a;
        border-right: heavy #444444;
        content-align: center middle;
        text-style: bold;
    }

    .section { min-height: 100vh; }

    .gauge-label { color: #888888; text-style: bold; }
    .big-digits { text-style: bold; }
    #gpu-panel, #cpu-panel, #mem-panel {
        border: heavy #333333;
        padding: 1 2;
        margin: 1;
    }
    """

    def __init__(self) -> None:
        super().__init__()
        self._current_index = 0

    def compose(self) -> ComposeResult:
        with Horizontal():
            with Vertical(id="touchbar"):
                yield TouchZone("^\nUP", "up", id="zone-up")
                yield TouchZone("v\nDOWN", "down", id="zone-down")
            with VerticalScroll(id="scroller"):
                yield DashboardSection()
                yield RequestAssistanceSection()
                yield SwitchToXSection()

    def on_mount(self) -> None:
        self._reset_idle_timer()

    # --- Navigation ---

    def handle_zone_tap(self, direction: str) -> None:
        self._wake_if_blanked()
        self._reset_idle_timer()

        if direction == "down":
            self._current_index = min(self._current_index + 1, len(SECTION_IDS) - 1)
        else:
            self._current_index = max(self._current_index - 1, 0)

        target = self.query_one(f"#{SECTION_IDS[self._current_index]}")
        self.query_one("#scroller", VerticalScroll).scroll_to_widget(target, animate=True)

    # --- Idle / blank handling ---

    def _reset_idle_timer(self) -> None:
        if hasattr(self, "_idle_timer"):
            self._idle_timer.stop()
        self._idle_timer = self.set_timer(IDLE_TIMEOUT_SECONDS, self._blank_screen)

    def on_key(self) -> None:
        self._wake_if_blanked()
        self._reset_idle_timer()

    def _blank_screen(self) -> None:
        self.push_screen(BlankScreen())

    def _wake_if_blanked(self) -> None:
        if isinstance(self.screen, BlankScreen):
            self.pop_screen()


def main() -> None:
    ServerMonitorDashApp().run()


if __name__ == "__main__":
    main()