UPS check and runtime monitoring

misc/checks.py

@@ -1,5 +1,8 @@
+from datetime import timedelta
+
 from alerting import alerts
 from misc import sensors, vuln
+from misc.enums import UPSStatus
 
 IS_TESTING = False
 
@@ -92,3 +95,54 @@ async def vuln_check() -> list[alerts.Alert]:
             )
             alert_list.append(alert)
     return alert_list
+
+
+async def ups_check() -> list[alerts.Alert]:
+    sensor = sensors.Sensors.get_ups()
+    alert_list = []
+
+    if IS_TESTING or sensor.battery_charge_percentage < sensor.battery_critical_percentage:
+        alert_list.append(alerts.Alert(
+            alert_type=alerts.AlertType.UPS,
+            message=f"Battery is under {sensor.battery_critical_percentage}%\n{sensor.battery_charge_percentage}% ({timedelta(seconds=sensor.battery_runtime)} remaining.",
+            severity=alerts.Severity.CRITICAL
+        ))
+    elif IS_TESTING or sensor.battery_charge_percentage < sensor.battery_warning_percentage:
+        alert_list.append(alerts.Alert(
+            alert_type=alerts.AlertType.UPS,
+            message=f"Battery is under {sensor.battery_warning_percentage}%\n{sensor.battery_charge_percentage}% ({timedelta(seconds=sensor.battery_runtime)} remaining.",
+            severity=alerts.Severity.WARNING
+        ))
+    elif IS_TESTING or sensor.ups_status == UPSStatus.ON_BATTERY:
+        alert_list.append(alerts.Alert(
+            alert_type=alerts.AlertType.UPS,
+            message=f"UPS is on battery.\n{sensor.battery_charge_percentage}% ({timedelta(seconds=sensor.battery_runtime)}) remaining.",
+            severity=alerts.Severity.INFO
+        ))
+
+    elif IS_TESTING or sensor.ups_status == UPSStatus.BATTERY_REPLACE:
+        alert_list.append(alerts.Alert(
+            alert_type=alerts.AlertType.UPS,
+            message=f"UPS battery needs to be replaced ASAP!",
+            severity=alerts.Severity.CRITICAL
+        ))
+    elif IS_TESTING or sensor.ups_status == UPSStatus.UPS_OVERLOAD:
+        alert_list.append(alerts.Alert(
+            alert_type=alerts.AlertType.UPS,
+            message=f"UPS is overloaded!",
+            severity=alerts.Severity.CRITICAL
+        ))
+    elif IS_TESTING or sensor.ups_status == UPSStatus.UPS_BYPASS:
+        alert_list.append(alerts.Alert(
+            alert_type=alerts.AlertType.UPS,
+            message=f"BYPASS MODE: Battery and connected devices are not protected from power outage!",
+            severity=alerts.Severity.WARNING
+        ))
+    elif IS_TESTING or sensor.ups_status == UPSStatus.UPS_CALIBRATION:
+        alert_list.append(alerts.Alert(
+            alert_type=alerts.AlertType.UPS,
+            message=f"UPS is currently performing runtime calibration.",
+            severity=alerts.Severity.INFO
+        ))
+
+    return alert_list

misc/sensors.py

@@ -36,6 +36,7 @@ class UPSSensor:
     battery_charge_percentage: int = None
     battery_warning_percentage: int = 20
     battery_critical_percentage: int = 10
+    battery_runtime: int = 1000
 
 
 class Sensors:
@@ -121,9 +122,15 @@ class Sensors:
                 case "battery.charge":
                     sensor_data.battery_charge_percentage = int(value)
                 case "battery.charge.low":
-                    sensor_data.battery_critical_percentage = int(value)
+                    # ? in case we need to evaluate critical% from sensor
+                    # sensor_data.battery_critical_percentage = int(value)
+                    sensor_data.battery_critical_percentage = 25
                 case "battery.charge.warning":
-                    sensor_data.battery_warning_percentage = int(value)
+                    # ? in case we need to evaluate warning% from sensor
+                    # sensor_data.battery_warning_percentage = int(value)
+                    sensor_data.battery_warning_percentage = 50
+                case "battery.runtime":
+                    sensor_data.battery_runtime = int(value)
                 case "ups.status":
                     sensor_data.ups_status = UPSStatus(value)
                 case _:

service.py

@@ -47,6 +47,7 @@ async def main():
     checkers = (
         checker(checks.temp_check, 5 * MINUTE, client),
         checker(checks.cpu_check, 5 * MINUTE, client),
+        checker(checks.ups_check, 5 * MINUTE, client),
         checker(checks.ram_check, 1 * MINUTE, client),
         checker(checks.vuln_check, 1 * DAY, client),
     )