from __future__ import annotations
from checks.base import Check
from state import make_key
class NutCheck(Check):
def run(self) -> None:
for ups in self.config.nut.ups:
self.log.info("--- USV: %s ---", ups)
self._check_ups(ups)
def _check_ups(self, ups_name: str) -> None:
key_conn = make_key("nut_conn", ups_name)
key_ob = make_key("nut_ob", ups_name)
key_rb = make_key("nut_rb", ups_name)
key_batt = make_key("nut_batt", ups_name)
key_load = make_key("nut_load", ups_name)
key_runtime = make_key("nut_rt", ups_name)
raw = self._run(["upsc", ups_name], timeout=self.config.nut.timeout)
if not raw or raw.strip() == "":
if self.alerter.should_alert(key_conn, "WARNING"):
self.alerter.send_alert(
"WARNING",
f"USV nicht erreichbar: {ups_name}",
f"Keine Verbindung zu NUT-USV {ups_name}!\n\n"
f"Läuft der Dienst? → systemctl status nut-server\n"
f"upsc {ups_name}",
key_conn,
)
return
else:
if self.alerter.clear_alert(key_conn):
self.alerter.send_alert(
"INFO",
f"USV wieder erreichbar: {ups_name}",
f"NUT-USV {ups_name} ist wieder erreichbar ✅",
)
# Werte parsen
values: dict[str, str] = {}
for line in raw.splitlines():
if ": " in line:
k, _, v = line.partition(": ")
values[k.strip()] = v.strip()
status = values.get("ups.status", "")
batt_charge = _to_int(values.get("battery.charge", "100"))
batt_runtime = _to_int(values.get("battery.runtime", "9999"))
ups_load = _to_int(values.get("ups.load", "0"))
# ── On Battery ───────────────────────────────────────
if "OB" in status:
runtime_min = batt_runtime // 60
lb_hint = "\n⚠️ LOW BATTERY – Shutdown droht!" if "LB" in status else ""
if self.alerter.should_alert(key_ob, "CRITICAL"):
self.alerter.send_alert(
"CRITICAL",
f"USV {ups_name}: STROMAUSFALL!",
f"Netzstrom ausgefallen! USV läuft auf Batterie.{lb_hint}\n\n"
f"🔋 Akku: {batt_charge}%\n"
f"⏱️ Restlaufzeit: {runtime_min} min\n"
f"⚡ Last: {ups_load}%\n\n"
f"upsc {ups_name}",
key_ob,
)
# Restlaufzeit kritisch
if batt_runtime < self.config.nut.runtime_crit_sec:
if self.alerter.should_alert(key_runtime, "CRITICAL"):
self.alerter.send_alert(
"CRITICAL",
f"USV {ups_name}: Restlaufzeit kritisch!",
f"USV {ups_name}: Nur noch {runtime_min} min Restlaufzeit!\n\n"
f"System sofort herunterfahren:\n"
f"upsmon -c fsd",
key_runtime,
)
else:
if self.alerter.clear_alert(key_ob):
self.alerter.send_alert(
"INFO",
f"USV {ups_name}: Netzstrom wiederhergestellt ✅",
f"Netzstrom wieder vorhanden. USV {ups_name} "
f"zurück im Normalbetrieb.",
)
self.alerter.clear_alert(key_runtime)
# ── Replace Battery ───────────────────────────────────
if "RB" in status:
if self.alerter.should_alert(key_rb, "WARNING"):
self.alerter.send_alert(
"WARNING",
f"USV {ups_name}: Batterie ersetzen!",
f"USV {ups_name} meldet: Batterie muss ersetzt werden!\n\n"
f"Aktueller Ladestand: {batt_charge}%\n"
f"upsc {ups_name}",
key_rb,
)
else:
if self.alerter.clear_alert(key_rb):
self.alerter.send_alert(
"INFO",
f"USV {ups_name}: Batterie-Status OK ✅",
f"USV {ups_name}: Replace-Battery Flag wurde aufgehoben.",
)
# ── Akkustand (nur im Netzbetrieb) ────────────────────
if "OB" not in status:
cfg = self.config.nut
if batt_charge <= cfg.batt_crit_pct:
if self.alerter.should_alert(key_batt, "CRITICAL"):
self.alerter.send_alert(
"CRITICAL",
f"USV {ups_name}: Akkustand kritisch!",
f"USV {ups_name}: Akkustand {batt_charge}% "
f"≤ {cfg.batt_crit_pct}%!\n\nAkku lädt nicht korrekt?",
key_batt,
)
elif batt_charge <= cfg.batt_warn_pct:
if self.alerter.should_alert(key_batt, "WARNING"):
self.alerter.send_alert(
"WARNING",
f"USV {ups_name}: Akkustand niedrig",
f"USV {ups_name}: Akkustand {batt_charge}% "
f"≤ {cfg.batt_warn_pct}%",
key_batt,
)
else:
if self.alerter.clear_alert(key_batt):
self.alerter.send_alert(
"INFO",
f"USV {ups_name}: Akkustand OK ✅",
f"USV {ups_name}: Akkustand wieder bei {batt_charge}% ✅",
)
# ── Last ──────────────────────────────────────────────
if ups_load >= self.config.nut.load_warn_pct:
if self.alerter.should_alert(key_load, "WARNING"):
self.alerter.send_alert(
"WARNING",
f"USV {ups_name}: Hohe Last",
f"USV {ups_name}: Last {ups_load}% "
f"≥ {self.config.nut.load_warn_pct}%",
key_load,
)
else:
if self.alerter.clear_alert(key_load):
self.alerter.send_alert(
"INFO",
f"USV {ups_name}: Last normal ✅",
f"USV {ups_name}: Last wieder unter "
f"{self.config.nut.load_warn_pct}% ({ups_load}%) ✅",
)
self.log.info(
"USV %s: status=%s batt=%d%% runtime=%ds load=%d%%",
ups_name, status, batt_charge, batt_runtime, ups_load,
)
def _to_int(val: str) -> int:
"""Parse int from NUT value, stripping decimals and units."""
try:
return int(float(val.split()[0]))
except (ValueError, IndexError):
return 0