diff --git a/alerting/alerts.py b/alerting/alerts.py index f387dc4..e403c5a 100644 --- a/alerting/alerts.py +++ b/alerting/alerts.py @@ -19,6 +19,7 @@ class AlertType(StrEnum): LOGIN = "LOGIN" # TODO SMART = "SMART" # TODO RAID = "RAID" # TODO + UPS = "UPS" class Severity(Enum): diff --git a/misc/checks.py b/misc/checks.py index 6efefd2..e41c2ba 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,101 @@ async def vuln_check() -> list[alerts.Alert]: ) alert_list.append(alert) return alert_list + + +async def ups_check() -> list[alerts.Alert]: + sensor = await sensors.Sensors.get_ups() + + if not sensor: + return + + 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, + ) + ) + + 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: + 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( + 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 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( + alert_type=alerts.AlertType.UPS, + message=f"Overvoltage detected: trimming voltage to nominal.", + severity=alerts.Severity.INFO, + ) + ) + elif IS_TESTING or status == UPSStatus.UPS_BOOST: + alert_list.append( + alerts.Alert( + alert_type=alerts.AlertType.UPS, + message=f"Undervoltage detected: boosting voltage to nominal.", + 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/enums.py b/misc/enums.py new file mode 100644 index 0000000..2ea1fc1 --- /dev/null +++ b/misc/enums.py @@ -0,0 +1,21 @@ +from enum import StrEnum + + +class UPSStatus(StrEnum): + """https://networkupstools.org/docs/developer-guide.chunked/new-drivers.html#_status_data""" + + ON_LINE = "OL" + ON_BATTERY = "OB" + BATTERY_LOW = "LB" + BATTERY_HIGH = "HB" + BATTERY_REPLACE = "RB" + BATTERY_CHARGING = "CHRG" + BATTERY_DISCHARGING = "DISCHRG" + UPS_BYPASS = "BYPASS" + """Battery and connected devices are not protected from power outage!""" + UPS_OFFLINE = "OFF" + UPS_OVERLOAD = "OVER" + UPS_CALIBRATION = "CAL" + UPS_TRIM = "TRIM" + UPS_BOOST = "BOOST" + UPS_FSD = "FSD" diff --git a/misc/sensors.py b/misc/sensors.py index b1b3f1b..ed554cb 100644 --- a/misc/sensors.py +++ b/misc/sensors.py @@ -1,7 +1,11 @@ +import subprocess from dataclasses import dataclass from psutil import cpu_percent, sensors_temperatures, virtual_memory +from .enums import UPSStatus +from alerting import alerts + @dataclass class TemperatureSensor: @@ -27,6 +31,15 @@ class RamSensor: critical_avail: int = 2 * 1024**3 +@dataclass +class UPSSensor: + ups_status: list[UPSStatus] = None + battery_charge_percentage: int = None + battery_warning_percentage: int = 20 + battery_critical_percentage: int = 10 + battery_runtime: int = 1000 + + class Sensors: @staticmethod def get_temperatures() -> dict[str, list[TemperatureSensor]]: @@ -96,10 +109,40 @@ class Sensors: @staticmethod def get_ram() -> RamSensor: ram = virtual_memory() - return RamSensor(current_avail=ram.available, - current_avail_percentage=ram.percent) + return RamSensor(current_avail=ram.available, current_avail_percentage=ram.percent) + @staticmethod + async def get_ups() -> None | UPSSensor: + 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 + )) + return None -if __name__ == "__main__": - for i in Sensors.get_temperatures(): - print(i) + sensor_data = UPSSensor() + + for line in raw_data.stdout.splitlines(): + sensor, value = line.split(": ")[:2] + match sensor: + 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 + 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 + case "battery.runtime": + sensor_data.battery_runtime = int(value) + case "ups.status": + sensor_data.ups_status = [UPSStatus(status) for status in value.split()] + case _: + ... + + return sensor_data 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), )