Add server_dash.py
This commit is contained in:
parent
afac54c7bb
commit
05aaec19a9
1 changed files with 302 additions and 0 deletions
302
server_dash.py
Normal file
302
server_dash.py
Normal file
|
|
@ -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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue