battery-charge alerts using a rolling mean of the last 10 valid measurements.

ups2rrd.py

@@ -70,6 +70,7 @@ SHUTDOWN_KNOWN_HOSTS = os.environ.get(
 SHUTDOWN_COMMAND = "sudo /usr/sbin/shutdown -h now"
 REMOTE_SHUTDOWN_RUNTIME_SECONDS = 5 * 60
 LOCAL_SHUTDOWN_RUNTIME_SECONDS = 200
+BATTERY_CHARGE_MEAN_SAMPLES = 10
 
 GRAPH_DIR = "/var/www/html/ups"
 HTML_FILE = os.path.join(GRAPH_DIR, "index.html")
@@ -480,6 +481,33 @@ def crossed_below(previous, current, threshold):
     )
 
 
+def battery_charge_history(previous, current_charge):
+    """Return up to the last 10 valid battery-charge measurements."""
+
+    history = previous.get("battery_charge_history", [])
+    if not isinstance(history, list):
+        history = []
+    history = [
+        value
+        for value in history
+        if isinstance(value, (int, float)) and not isinstance(value, bool)
+    ]
+
+    if current_charge is not None:
+        history.append(current_charge)
+
+    return history[-BATTERY_CHARGE_MEAN_SAMPLES:]
+
+
+def complete_charge_mean(history):
+    """Return the mean only when a complete charge window is available."""
+
+    if len(history) < BATTERY_CHARGE_MEAN_SAMPLES:
+        return None
+
+    return sum(history) / len(history)
+
+
 def shutdown_remote_server(
     host=SHUTDOWN_HOST,
     user=SHUTDOWN_USER,
@@ -544,6 +572,10 @@ def process_alerts(current):
     current_status = current.get("status")
     previous_charge = previous.get("battery_charge")
     current_charge = current.get("battery_charge")
+    previous_charge_history = battery_charge_history(previous, None)
+    charge_history = battery_charge_history(previous, current_charge)
+    previous_charge_mean = complete_charge_mean(previous_charge_history)
+    current_charge_mean = complete_charge_mean(charge_history)
     previous_runtime = previous.get("battery_runtime")
     current_runtime = current.get("battery_runtime")
     shutdown_sent = previous.get("remote_shutdown_sent", False)
@@ -560,15 +592,22 @@ def process_alerts(current):
                 f"UPS status changed: {previous_status} -> {current_status}"
             )
 
-        if crossed_below(previous_charge, current_charge, 50):
+        if current_charge_mean is not None and (
+            previous_charge_mean is None
+            or crossed_below(previous_charge_mean, current_charge_mean, 50)
+        ) and current_charge_mean < 50:
             notify(
-                f"UPS battery charge is below 50%: {current_charge:.1f}%"
+                "UPS mean battery charge over the last 10 measurements "
+                f"is below 50%: {current_charge_mean:.1f}%"
             )
 
-        if crossed_below(previous_charge, current_charge, 5):
+        if current_charge_mean is not None and (
+            previous_charge_mean is None
+            or crossed_below(previous_charge_mean, current_charge_mean, 5)
+        ) and current_charge_mean < 5:
             notify(
-                "UPS battery charge is critically low "
-                f"(below 5%): {current_charge:.1f}%"
+                "UPS mean battery charge over the last 10 measurements "
+                f"is critically low (below 5%): {current_charge_mean:.1f}%"
             )
 
         if crossed_below(previous_runtime, current_runtime, 5 * 60):
@@ -633,6 +672,7 @@ def process_alerts(current):
         "battery_charge": (
             current_charge if current_charge is not None else previous_charge
         ),
+        "battery_charge_history": charge_history,
         "battery_runtime": (
             current_runtime if current_runtime is not None else previous_runtime
         ),