HitonabiandClaude Opus 5.5
904598031a
- kern/platten.py: SMART aus pvesh, smartctl und SNMP in einer Form; rot bei SMART FAILED, kritischer Warnung, aufgebrauchter Reserve, ≥ 90 % verbraucht; gelb bei ≥ 80 %, Hitze, gewachsenen Fehlerzählern (Ausgangslage je Platte, „Zur Kenntnis genommen“ setzt sie neu) - Ausführer: stündlich disks/list + disks/smart im Bericht (nur lesend) - services/homelab/nas_snmp.py: kleiner SNMPv2c-GetBulk-Leser ohne neue Abhängigkeit, QNAP hdTable/diskTable, Zugang 0600, Einstellungen „Verbindung testen und speichern“ - services/platten.py: NVMe der KI-Box über sudo smartctl (smartmontools am 26.09. nachinstalliert) - Wächter beider Rollen, Plattenfeld in beiden Cockpits, Doku Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
90 lines
3.1 KiB
Python
90 lines
3.1 KiB
Python
"""Plattengesundheit der KI-Box (Ausbauplan „Später“, Punkt 18, seit 26.09.2026): smartctl (Paket smartmontools,
|
|
seit 26.09.2026 auf der Box) je Platte aus lsblk, höchstens stündlich, bewertet mit kern/platten.py. smartctl braucht
|
|
root (sudo -n); fehlt es oder verweigert sudo, gibt es schlicht keine Werte — nie einen Absturz (Windows, Tests).
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import shutil
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
|
|
from kern import platten as kern_platten
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
CACHE_S = 3600
|
|
_lock = threading.Lock()
|
|
_cache: dict = {"zeit": 0.0, "platten": None}
|
|
|
|
|
|
def _laufen(befehl: list[str], zeitlimit: int = 30) -> tuple[int, str]:
|
|
try:
|
|
r = subprocess.run(befehl, capture_output=True, text=True, timeout=zeitlimit)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return 127, ""
|
|
return r.returncode, r.stdout
|
|
|
|
|
|
def geraete() -> list[str]:
|
|
code, aus = _laufen(["lsblk", "-J", "-d", "-o", "NAME,TYPE"], 10)
|
|
try:
|
|
return [f"/dev/{d['name']}" for d in json.loads(aus).get("blockdevices") or [] if d.get("type") == "disk"
|
|
and not str(d.get("name", "")).startswith(("zram", "loop"))] if code == 0 else []
|
|
except (ValueError, AttributeError):
|
|
return []
|
|
|
|
|
|
def lesen() -> list[dict]:
|
|
if not shutil.which("sudo"): # Windows, Tests ohne sudo: keine Werte
|
|
return []
|
|
ergebnis = []
|
|
for geraet in geraete():
|
|
_, aus = _laufen(["sudo", "-n", "smartctl", "-j", "-a", geraet])
|
|
try:
|
|
daten = json.loads(aus)
|
|
except ValueError:
|
|
continue
|
|
# smartctl setzt Bits im Exit-Code auch für „Fehler im Protokoll“ — gelesen ist gelesen.
|
|
if not isinstance(daten, dict) or not (daten.get("model_name") or daten.get("nvme_smart_health_information_log")):
|
|
continue
|
|
ergebnis.append(kern_platten.aus_smartctl(daten, geraet, "KI-Box"))
|
|
return ergebnis
|
|
|
|
|
|
def platten() -> list[dict]:
|
|
with _lock:
|
|
if _cache["platten"] is not None and time.time() - _cache["zeit"] < CACHE_S:
|
|
return _cache["platten"]
|
|
werte = lesen()
|
|
with _lock:
|
|
_cache.update(zeit=time.time(), platten=werte)
|
|
return werte
|
|
|
|
|
|
def lage() -> dict:
|
|
"""Für GET /api/platten."""
|
|
return {"platten": kern_platten.bewertet(platten())}
|
|
|
|
|
|
def hinnehmen(pid: str) -> dict:
|
|
return kern_platten.hinnehmen(pid, platten())
|
|
|
|
|
|
def pruefe_platten_gesundheit() -> list:
|
|
"""Für den Wächter (Rolle box)."""
|
|
from services.waechter import Befund, _aktion
|
|
|
|
befunde = []
|
|
for p in lage()["platten"]:
|
|
if p["stufe"] == "ok":
|
|
continue
|
|
fehler = any("→" in s for s in p["saetze"])
|
|
befunde.append(Befund(
|
|
id=f"platte:{p['id']}", stufe=p["stufe"], titel=kern_platten.befund_titel(p),
|
|
text="; ".join(p["saetze"]) + ". Wichtiges sichern; wird es schlechter, die Platte ersetzen.",
|
|
quelle="platte", aktionen=[_aktion("hinnehmen", "Zur Kenntnis genommen", platte=p["id"])] if fehler else [],
|
|
sofort=True))
|
|
return befunde
|