diff --git a/README.md b/README.md index b4fb7d4..00ac24c 100644 --- a/README.md +++ b/README.md @@ -821,6 +821,20 @@ weekly dashboard at `/ups/week.html` shows one-week history for every collected value. The pages link to each other, refresh every 60 seconds, and open a full-size PNG when a graph is clicked. +## UPS notifications + +The collector also sends an ntfy notification when: + +- The UPS status changes +- Battery charge crosses below 50% +- Battery charge crosses below 5% +- Estimated battery runtime crosses below 5 minutes + +Alert state is stored in `/var/lib/rrd/bx950mi-alert-state.json`, allowing +changes to be detected across cron runs. The first run establishes a baseline +without sending notifications. Threshold notifications can fire again after a +value recovers above the threshold and later drops below it. + --- # 19. Testing the Python program diff --git a/ups2rrd.py b/ups2rrd.py index fd60387..2369116 100644 --- a/ups2rrd.py +++ b/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()