From 05aaec19a96e64f4c85a2c663e19e5cc8007e318 Mon Sep 17 00:00:00 2001 From: giovannino Date: Tue, 21 Jul 2026 02:07:37 +0200 Subject: [PATCH] Add server_dash.py --- server_dash.py | 302 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 302 insertions(+) create mode 100644 server_dash.py diff --git a/server_dash.py b/server_dash.py new file mode 100644 index 0000000..cb380b6 --- /dev/null +++ b/server_dash.py @@ -0,0 +1,302 @@ +#!/usr/bin/env python3 +# Created by John A. Hoeven with the ethical assistance of Claude AI +# --------------------------------------------------------------------------- +# bigboy_dash.py +# /server-monitor-dash/server_dash.py +# Version: v0.0.1 | Status: DEVELOPMENT +# --------------------------------------------------------------------------- +# Purpose: Case-mounted touchscreen monitoring dashboard for servers — CPU, +# GPU, memory telemetry rendered "supercar dash" style. Textual +# built-ins only (Sparkline, Digits, ProgressBar) 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 touchscreen +# 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: Starter skeleton — dashboard tab is functional; Request Assistance +# and Switch to X tabs are stubs, wired for later completion. Not yet a +# CE OS-compliant script (no test gates / cleanup trap) — this is a +# prototype to iterate the look and feel on, not a production deployment. +# --------------------------------------------------------------------------- + +from __future__ import annotations + +import subprocess +import time +from dataclasses import dataclass + +from textual.app import App, ComposeResult +from textual.containers import Horizontal, Vertical +from textual.screen import Screen +from textual.widgets import ( + Digits, + Footer, + Header, + Label, + ProgressBar, + Sparkline, + TabbedContent, + TabPane, +) + +IDLE_TIMEOUT_SECONDS = 15 * 60 +REFRESH_INTERVAL_SECONDS = 1.0 +SPARKLINE_HISTORY_LENGTH = 60 + +# --- Supercar-dash colour thresholds ----------------------------------------- +# Green -> amber -> red, same instinct as a tachometer redline. +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} + + +# --- Dashboard screen --------------------------------------------------------- + +class DashboardScreen(Screen): + """Main telemetry view — the 'dash' itself.""" + + CSS = """ + Screen { + background: #0a0a0a; + } + .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._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") + 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 _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) + + # --- Memory --- + mem_pct = read_mem_used_pct() + self.query_one("#mem-bar", ProgressBar).update(progress=mem_pct) + + # --- 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._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 + + +# --- Idle blank screen -------------------------------------------------------- + +class BlankScreen(Screen): + """Solid dark screen shown after idle timeout — protects the 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") + + +# --- Stub tabs — Request Assistance / Switch to X ----------------------------- + +class RequestAssistanceTab(Screen): + """STUB: will send a scripted email via smtplib on button press.""" + + def compose(self) -> ComposeResult: + yield Label("Request Assistance — not yet wired up. [STUB]") + + +class SwitchToXTab(Screen): + """STUB: will call the sudoers-scoped vt-switch-tuning wrapper + (chvt to the tuning VT) on button press.""" + + def compose(self) -> ComposeResult: + yield Label("Switch to X (GPU tuning) — not yet wired up. [STUB]") + + +# --- App shell with tabs + idle/blank handling --------------------------------- + +class BigBoyDashApp(App): + """ + Purpose: BigBoy case-display dashboard. + Note: idle/blank + tab structure only — DashboardScreen carries the real + telemetry logic; the other two tabs are placeholders per project scope. + """ + + def compose(self) -> ComposeResult: + with TabbedContent(): + with TabPane("Dashboard"): + yield DashboardScreen() + with TabPane("Request Assistance"): + yield RequestAssistanceTab() + with TabPane("Switch to X"): + yield SwitchToXTab() + + def on_mount(self) -> None: + self._reset_idle_timer() + + 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 on_click(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: + BigBoyDashApp().run() + + +if __name__ == "__main__": + main() \ No newline at end of file