diff --git a/misc/checks.py b/misc/checks.py index e41c2ba..aa00920 100644 --- a/misc/checks.py +++ b/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: - 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: + if 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 diff --git a/misc/sensors.py b/misc/sensors.py index ed554cb..f5082cc 100644 --- a/misc/sensors.py +++ b/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( - alert_type=alerts.AlertType.ERROR, - message="upsc is not installed!", - severity=alerts.Severity.CRITICAL - )) + await alerts.send_alert( + alerts.Alert( + alert_type=alerts.AlertType.ERROR, + 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 = 25 + sensor_data.battery_critical_percentage = int(value) 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 = 50 + sensor_data.battery_warning_percentage = int(value) case "battery.runtime": sensor_data.battery_runtime = int(value) case "ups.status": diff --git a/send_ups_alert.py b/send_ups_alert.py new file mode 100644 index 0000000..9c7b3d1 --- /dev/null +++ b/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())