From 3a02adde569c1a222b11f47583ae45725a4045d4 Mon Sep 17 00:00:00 2001 From: saqriphnix Date: Sat, 17 Aug 2024 00:00:16 +0300 Subject: [PATCH] UPS check and runtime monitoring --- misc/checks.py | 54 +++++++++++++++++++++++++++++++++++++++++++++++++ misc/sensors.py | 11 ++++++++-- service.py | 1 + 3 files changed, 64 insertions(+), 2 deletions(-) diff --git a/misc/checks.py b/misc/checks.py index 6efefd2..baee3bd 100644 --- a/misc/checks.py +++ b/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 diff --git a/misc/sensors.py b/misc/sensors.py index f136847..6169d44 100644 --- a/misc/sensors.py +++ b/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 _: diff --git a/service.py b/service.py index 59d8e60..fb63030 100755 --- a/service.py +++ b/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), )