"""Plattengesundheit (Ausbauplan „Später“, Punkt 18, seit 26.09.2026): SMART-Werte vereinheitlichen und bewerten. Quellen: der Proxmox-Host (Ausführer: pvesh disks/list und disks/smart), die KI-Box (smartctl -j) und das NAS (SNMP, services/homelab/nas_snmp.py). Einheitliche Form je Platte: {"id", "name", "wo", "typ", "groesse", "gesund": True/False/None, "verbraucht": % oder None, "reserve": % oder None, "reserve_grenze": % oder None, "warnung": 0 oder Bits, "temperatur": °C oder None, "stunden": h oder None, "fehler": {"medien", "umgelagert", "ausstehend", "unkorrigierbar", "kabel": Zahl}} Gemeldet wird, was schlecht ist oder schlechter wird (User-Entscheid „wie empfohlen“ 25.09.2026): rot die Platte hält sich selbst für krank (SMART FAILED, kritische Warnung, Reserve unter der Grenze) oder ist zu 90 % verbraucht gelb zu 80 % verbraucht, zu heiß, oder ein Fehlerzähler ist seit dem letzten „Zur Kenntnis genommen“ gewachsen Fehler, die schon beim ersten Blick da waren, gelten als Ausgangslage (basis) — sonst meldete eine ältere, aber stabile Platte für immer. Die Ausgangslage liegt je Instanz in /platten-basis.json. """ import json import re from pathlib import Path from kern.einstellungen import einstellungen VERBRAUCHT_GELB, VERBRAUCHT_ROT = 80, 90 HEISS = {"nvme": 70, "ssd": 60, "hdd": 55} FEHLER_NAME = {"medien": "Medienfehler", "umgelagert": "umgelagerte Sektoren", "ausstehend": "wartende Sektoren", "unkorrigierbar": "unkorrigierbare Sektoren", "kabel": "Übertragungsfehler (Kabel)"} # SATA-Attribute (Nummer → Fehlerart) und die Temperatur. ATA_FEHLER = {5: "umgelagert", 197: "ausstehend", 198: "unkorrigierbar", 199: "kabel"} ATA_TEMPERATUR = (194, 190) def _zahl(text: object) -> int | None: m = re.search(r"-?\d[\d,.]*", str(text or "")) if not m: return None try: return int(m.group(0).replace(",", "").replace(".", "")) except ValueError: return None def aus_nvme_text(text: str) -> dict: """Die NVMe-Werte aus dem Text von smartctl (so liefert ihn pvesh disks/smart).""" werte: dict[str, str] = {} for zeile in (text or "").splitlines(): if ":" in zeile: k, v = zeile.split(":", 1) werte.setdefault(k.strip(), v.strip()) warnung = werte.get("Critical Warning") return { "warnung": int(warnung, 16) if warnung and re.fullmatch(r"0x[0-9a-fA-F]+", warnung) else _zahl(warnung) or 0, "temperatur": _zahl(werte.get("Temperature")), "reserve": _zahl(werte.get("Available Spare")), "reserve_grenze": _zahl(werte.get("Available Spare Threshold")), "verbraucht": _zahl(werte.get("Percentage Used")), "stunden": _zahl(werte.get("Power On Hours")), "fehler": {"medien": _zahl(werte.get("Media and Data Integrity Errors")) or 0}, } def aus_ata_attributen(attribute: list) -> dict: """SATA: Fehlerzähler und Temperatur aus den Attributen (pvesh disks/smart „attributes“).""" fehler: dict[str, int] = {} temperatur = None for a in attribute or []: try: nr = int(a.get("id")) except (TypeError, ValueError): continue roh = _zahl(str(a.get("raw") or "").split()[0] if str(a.get("raw") or "").split() else None) if nr in ATA_FEHLER and roh is not None: fehler[ATA_FEHLER[nr]] = roh if nr in ATA_TEMPERATUR and temperatur is None and roh is not None: temperatur = roh return {"fehler": fehler, "temperatur": temperatur} def aus_proxmox(disk: dict, smart: dict | None) -> dict: """Eine Platte des Proxmox-Hosts aus pvesh disks/list (+ disks/smart).""" typ = str(disk.get("type") or "").lower() or "unbekannt" health = str((smart or {}).get("health") or disk.get("health") or "").upper() platte = {"id": f"pve:{disk.get('serial') or disk.get('devpath')}", "name": disk.get("model") or disk.get("devpath"), "wo": f"Proxmox-Host {disk.get('devpath')}", "typ": typ, "groesse": disk.get("size"), "gesund": True if health in ("PASSED", "OK") else False if health == "FAILED" else None, "verbraucht": None, "reserve": None, "reserve_grenze": None, "warnung": 0, "temperatur": None, "stunden": None, "fehler": {}} wearout = disk.get("wearout") if isinstance(wearout, (int, float)) and 0 <= wearout <= 100: platte["verbraucht"] = 100 - int(wearout) # Proxmox zeigt, was noch übrig ist if smart: if (smart.get("type") == "text" or typ == "nvme") and smart.get("text"): werte = aus_nvme_text(smart["text"]) platte.update({k: v for k, v in werte.items() if v is not None and k != "fehler"}) platte["fehler"] = werte["fehler"] elif smart.get("attributes"): platte.update(aus_ata_attributen(smart["attributes"])) return platte def aus_smartctl(d: dict, geraet: str, wo: str) -> dict: """Eine Platte aus `smartctl -j -a` (KI-Box).""" n = d.get("nvme_smart_health_information_log") or {} status = (d.get("smart_status") or {}).get("passed") typ = "nvme" if n else ("ssd" if (d.get("rotation_rate") == 0) else "hdd") platte = {"id": f"{wo}:{d.get('serial_number') or geraet}", "name": d.get("model_name") or geraet, "wo": f"{wo} {geraet}", "typ": typ, "groesse": (d.get("user_capacity") or {}).get("bytes"), "gesund": status if isinstance(status, bool) else None, "verbraucht": n.get("percentage_used"), "reserve": n.get("available_spare"), "reserve_grenze": n.get("available_spare_threshold"), "warnung": int(n.get("critical_warning") or 0), "temperatur": (d.get("temperature") or {}).get("current"), "stunden": (d.get("power_on_time") or {}).get("hours") or n.get("power_on_hours"), "fehler": {}} if n: platte["fehler"] = {"medien": int(n.get("media_errors") or 0)} else: tabelle = [{"id": a.get("id"), "raw": str((a.get("raw") or {}).get("value", ""))} for a in (d.get("ata_smart_attributes") or {}).get("table") or []] platte.update(aus_ata_attributen(tabelle)) return platte def bewerten(p: dict, basis: dict | None) -> tuple[str, list[str]]: """(Stufe, Sätze). basis = die Fehlerzähler, die als Ausgangslage gelten.""" stufe, saetze = "ok", [] rang = {"ok": 0, "gelb": 1, "rot": 2} def schlimmer(neu: str, satz: str) -> None: nonlocal stufe saetze.append(satz) if rang[neu] > rang[stufe]: stufe = neu if p.get("gesund") is False: schlimmer("rot", "meldet sich selbst als defekt (SMART)") if p.get("smart_warnung"): schlimmer("gelb", "SMART meldet eine Warnung") if p.get("warnung"): schlimmer("rot", f"meldet eine kritische Warnung (0x{int(p['warnung']):02x})") reserve, grenze = p.get("reserve"), p.get("reserve_grenze") if isinstance(reserve, int) and isinstance(grenze, int) and grenze > 0 and reserve <= grenze: schlimmer("rot", f"Ersatzblöcke fast aufgebraucht ({reserve} %)") v = p.get("verbraucht") if isinstance(v, (int, float)) and v >= VERBRAUCHT_ROT: schlimmer("rot", f"zu {int(v)} % verbraucht — Ersatz einplanen") elif isinstance(v, (int, float)) and v >= VERBRAUCHT_GELB: schlimmer("gelb", f"zu {int(v)} % verbraucht") t, heiss = p.get("temperatur"), HEISS.get(p.get("typ") or "", 60) if isinstance(t, (int, float)) and t >= heiss: schlimmer("gelb", f"{int(t)} °C warm (ab {heiss} °C zu heiß)") for art, n in (p.get("fehler") or {}).items(): alt = (basis or {}).get(art, n) if isinstance(n, int) and isinstance(alt, int) and n > alt: schlimmer("gelb", f"{FEHLER_NAME.get(art, art)}: {alt} → {n}") return stufe, saetze # --- Ausgangslage je Instanz ------------------------------------------------------------------------------------- def _pfad() -> Path: return einstellungen().daten_dir / "platten-basis.json" def basis_lesen() -> dict[str, dict]: try: daten = json.loads(_pfad().read_text(encoding="utf-8")) except (OSError, ValueError): return {} return daten if isinstance(daten, dict) else {} def _basis_schreiben(daten: dict) -> None: pfad = _pfad() pfad.parent.mkdir(parents=True, exist_ok=True) tmp = pfad.with_suffix(".tmp") tmp.write_text(json.dumps(daten, ensure_ascii=False, indent=1), encoding="utf-8") tmp.replace(pfad) def bewertet(platten: list[dict]) -> list[dict]: """Die Platten mit stufe und saetze. Neue Platten bekommen ihre Ausgangslage beim ersten Blick.""" basis = basis_lesen() neu = {p["id"]: dict(p.get("fehler") or {}) for p in platten if p.get("id") and p["id"] not in basis} if neu: _basis_schreiben({**basis, **neu}) basis = {**basis, **neu} ergebnis = [] for p in platten: stufe, saetze = bewerten(p, basis.get(p.get("id"))) ergebnis.append({**p, "stufe": stufe, "saetze": saetze}) return ergebnis def hinnehmen(pid: str, platten: list[dict]) -> dict: """„Zur Kenntnis genommen“: die heutigen Fehlerzähler werden die neue Ausgangslage.""" p = next((x for x in platten if x.get("id") == pid), None) if p is None: return {"ok": False, "detail": "Diese Platte ist gerade nicht zu sehen."} _basis_schreiben({**basis_lesen(), pid: dict(p.get("fehler") or {})}) return {"ok": True, "text": f"Zur Kenntnis genommen: {p.get('name')}. Gemeldet wird erst wieder, wenn die Fehler " "weiter wachsen."} def befund_titel(p: dict) -> str: return f"Platte {p.get('name')} ({p.get('wo')}): {p['saetze'][0]}" if p.get("saetze") else str(p.get("name"))