with alerts

ups2rrd.py

@@ -34,11 +34,13 @@ NUT UPS name:
 """
 
 import os
+import json
 import sys
 import time
 
 import rrdtool
 from PyNUTClient import PyNUT
+from ntfy_notify import send_notification
 
 
 # ---------------------------------------------------------------------------
@@ -49,6 +51,7 @@ UPS_NAME = "BX950MI"
 
 RRD_DIR = "/var/lib/rrd"
 RRD_FILE = os.path.join(RRD_DIR, "bx950mi.rrd")
+ALERT_STATE_FILE = os.path.join(RRD_DIR, "bx950mi-alert-state.json")
 
 GRAPH_DIR = "/var/www/html/ups"
 HTML_FILE = os.path.join(GRAPH_DIR, "index.html")
@@ -400,6 +403,116 @@ def update_rrd():
         f"nominal={realpower_nominal}",
     )
 
+    return {
+        "status": status,
+        "battery_charge": battery_charge,
+        "battery_runtime": battery_runtime,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Notifications
+# ---------------------------------------------------------------------------
+
+def load_alert_state():
+    """Load values saved by the previous collector run."""
+
+    try:
+        with open(ALERT_STATE_FILE, encoding="utf-8") as state_file:
+            return json.load(state_file)
+    except FileNotFoundError:
+        return {}
+    except (OSError, ValueError) as exc:
+        print(f"Could not read alert state: {exc}", file=sys.stderr)
+        return {}
+
+
+def save_alert_state(state):
+    """Atomically save values for the next collector run."""
+
+    temporary_file = f"{ALERT_STATE_FILE}.tmp"
+
+    try:
+        with open(temporary_file, "w", encoding="utf-8") as state_file:
+            json.dump(state, state_file, indent=2, sort_keys=True)
+            state_file.write("\n")
+        os.replace(temporary_file, ALERT_STATE_FILE)
+    except OSError as exc:
+        print(f"Could not save alert state: {exc}", file=sys.stderr)
+
+
+def notify(message):
+    """Send an alert without stopping UPS data collection on failure."""
+
+    try:
+        send_notification(message)
+        print(f"Notification sent: {message}")
+    except Exception as exc:
+        print(f"Notification failed: {exc}", file=sys.stderr)
+
+
+def crossed_below(previous, current, threshold):
+    """Return whether a numeric value crossed below a threshold."""
+
+    return (
+        previous is not None
+        and current is not None
+        and previous >= threshold
+        and current < threshold
+    )
+
+
+def process_alerts(current):
+    """Notify on status changes and configured threshold crossings."""
+
+    previous = load_alert_state()
+    previous_status = previous.get("status")
+    current_status = current.get("status")
+    previous_charge = previous.get("battery_charge")
+    current_charge = current.get("battery_charge")
+    previous_runtime = previous.get("battery_runtime")
+    current_runtime = current.get("battery_runtime")
+
+    # An empty state is the initial baseline, not an alert condition.
+    if previous:
+        if (
+            previous_status
+            and current_status
+            and current_status != previous_status
+        ):
+            notify(
+                f"UPS status changed: {previous_status} -> {current_status}"
+            )
+
+        if crossed_below(previous_charge, current_charge, 50):
+            notify(
+                f"UPS battery charge is below 50%: {current_charge:.1f}%"
+            )
+
+        if crossed_below(previous_charge, current_charge, 5):
+            notify(
+                "UPS battery charge is critically low "
+                f"(below 5%): {current_charge:.1f}%"
+            )
+
+        if crossed_below(previous_runtime, current_runtime, 5 * 60):
+            notify(
+                "UPS battery runtime is below 5 minutes: "
+                f"{current_runtime / 60:.1f} minutes"
+            )
+
+    # Keep the last valid reading when NUT temporarily omits a value.
+    state = {
+        "status": current_status or previous_status,
+        "battery_charge": (
+            current_charge if current_charge is not None else previous_charge
+        ),
+        "battery_runtime": (
+            current_runtime if current_runtime is not None else previous_runtime
+        ),
+    }
+    save_alert_state(state)
+
 
 # ---------------------------------------------------------------------------
 # Graph: UPS status
@@ -1109,7 +1222,9 @@ def create_dashboards():
 def main():
     create_rrd()
 
-    update_rrd()
+    current_values = update_rrd()
+
+    process_alerts(current_values)
 
     create_graphs()