Merge branch 'upsmon' into 'main'

move checks to upsmon, where possible

See merge request lego/lego-monitoring!4

misc/checks.py

@@ -123,36 +123,12 @@ async def ups_check() -> list[alerts.Alert]:
         )
 
     for status in sensor.ups_status:
-        if IS_TESTING or status == UPSStatus.BATTERY_REPLACE:
+        if IS_TESTING or status == UPSStatus.UPS_OVERLOAD:
-            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 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 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 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,
-                )
-            )
         elif IS_TESTING or status == UPSStatus.ON_BATTERY:
             alert_list.append(
                 alerts.Alert(
@@ -161,12 +137,6 @@ async def ups_check() -> list[alerts.Alert]:
                     severity=alerts.Severity.INFO,
                 )
             )
-        elif IS_TESTING or status == UPSStatus.UPS_OFFLINE:
-            alert_list.append(
-                alerts.Alert(
-                    alert_type=alerts.AlertType.UPS, message=f"UPS seems to be offline.", severity=alerts.Severity.INFO
-                )
-            )
         elif IS_TESTING or status == UPSStatus.UPS_TRIM:
             alert_list.append(
                 alerts.Alert(
@@ -183,13 +153,5 @@ async def ups_check() -> list[alerts.Alert]:
                     severity=alerts.Severity.INFO,
                 )
             )
-        elif IS_TESTING or status == UPSStatus.UPS_FSD:
-            alert_list.append(
-                alerts.Alert(
-                    alert_type=alerts.AlertType.UPS,
-                    message=f"Shutdown imminent!",
-                    severity=alerts.Severity.CRITICAL,
-                )
-            )
 
     return alert_list

misc/sensors.py

@@ -3,9 +3,10 @@ from dataclasses import dataclass
 
 from psutil import cpu_percent, sensors_temperatures, virtual_memory
 
-from .enums import UPSStatus
 from alerting import alerts
 
+from .enums import UPSStatus
+
 
 @dataclass
 class TemperatureSensor:
@@ -116,11 +117,13 @@ class Sensors:
         try:
             raw_data = subprocess.run(["upsc", "cp1300"], stdout=subprocess.PIPE, encoding="utf-8")
         except FileNotFoundError:
-            await alerts.send_alert(alerts.Alert(
+            await alerts.send_alert(
-                alert_type=alerts.AlertType.ERROR,
+                alerts.Alert(
-                message="upsc is not installed!",
+                    alert_type=alerts.AlertType.ERROR,
-                severity=alerts.Severity.CRITICAL
+                    message="upsc is not installed!",
-            ))
+                    severity=alerts.Severity.CRITICAL,
+                )
+            )
             return None
 
         sensor_data = UPSSensor()
@@ -131,13 +134,9 @@ class Sensors:
                 case "battery.charge":
                     sensor_data.battery_charge_percentage = int(value)
                 case "battery.charge.low":
-                    # ? in case we need to evaluate critical% from sensor
+                    sensor_data.battery_critical_percentage = int(value)
-                    # sensor_data.battery_critical_percentage = int(value)
-                    sensor_data.battery_critical_percentage = 25
                 case "battery.charge.warning":
-                    # ? in case we need to evaluate warning% from sensor
+                    sensor_data.battery_warning_percentage = int(value)
-                    # 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":

send_ups_alert.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+import asyncio
+from os import environ
+from sys import argv
+
+from alerting import alerts
+
+type_priority_map = {
+    "ONLINE": alerts.Severity.INFO,  # UPS is back online
+    "ONBATT": alerts.Severity.WARNING,  # UPS is on battery
+    "LOWBATT": alerts.Severity.CRITICAL,  # UPS is on battery and has a low battery (is critical)
+    "FSD": alerts.Severity.CRITICAL,  # UPS is being shutdown by the primary (FSD = "Forced Shutdown")
+    "COMMOK": alerts.Severity.INFO,  # Communications established with the UPS
+    "COMMBAD": alerts.Severity.WARNING,  # Communications lost to the UPS
+    "SHUTDOWN": alerts.Severity.CRITICAL,  # The system is being shutdown
+    "REPLBATT": alerts.Severity.WARNING,  # The UPS battery is bad and needs to be replaced
+    "NOCOMM": alerts.Severity.WARNING,  # A UPS is unavailable (can’t be contacted for monitoring)
+    "NOPARENT": alerts.Severity.CRITICAL,  # upsmon parent process died - shutdown impossible
+    "CAL": alerts.Severity.INFO,  # UPS calibration in progress
+    "NOTCAL": alerts.Severity.INFO,  # UPS calibration finished
+    "OFF": alerts.Severity.CRITICAL,  # UPS administratively OFF or asleep
+    "NOTOFF": alerts.Severity.INFO,  # UPS no longer administratively OFF or asleep
+    "BYPASS": alerts.Severity.WARNING,  # UPS on bypass (powered, not protecting)
+    "NOTBYPASS": alerts.Severity.INFO,  # UPS no longer on bypass
+    None: alerts.Severity.CRITICAL,  # unknown alert type
+}
+
+
+async def main():
+    """Sends an alert about a UPS event. Meant to be used from upsmon with NOTIFYCMD"""
+    if len(argv) != 2:
+        raise Exception("provide exactly one argument: alert message")
+    message = argv[1]
+    typestr = environ.get("NOTIFYTYPE", None)
+    severity = type_priority_map[typestr]
+    alert = alerts.Alert(alert_type=alerts.AlertType.UPS, message=message, severity=severity)
+    await alerts.send_alert(alert)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())