From 330105a5a4816c818c12639a272474401a678708 Mon Sep 17 00:00:00 2001 From: vvzvlad Date: Tue, 13 Aug 2024 18:32:24 +0300 Subject: [PATCH] refactoring --- shutdown_ups.py | 99 +++++++++++++++++++++++++++++++++---------------- 1 file changed, 68 insertions(+), 31 deletions(-) diff --git a/shutdown_ups.py b/shutdown_ups.py index 6c647d7..729cfe7 100644 --- a/shutdown_ups.py +++ b/shutdown_ups.py @@ -1,25 +1,31 @@ #!/usr/bin/env python3 + import subprocess import time -import requests import socket import os -ups_ip = "10.31.41.46" -time_threshold = 6 -interval = 60 +UPS_IP = "10.31.41.46" +TIME_THRESHOLD = 6 +INTERVAL_LINE = 60 +INTERVAL_BATTERY = 5 +OID_ON_BATTERY_TIME = "1.3.6.1.4.1.318.1.1.1.2.2.3.0" +OID_STATES = "1.3.6.1.4.1.318.1.1.1.11.1.1.0" +OID_BATTERY_PERCENT = "1.3.6.1.4.1.318.1.1.1.2.2.1.0" def parse_time(time_str): + if time_str == "" or time_str is None: + return None try: time_parts = time_str.split(":")[-4:] days = int(time_parts[0]) hours = int(time_parts[1]) minutes = int(time_parts[2]) seconds = float(time_parts[3]) - total_minutes = (days * 24 * 60) + (hours * 60) + minutes + (int(seconds // 60)) - return total_minutes - except: - return False + return (days * 24 * 60) + (hours * 60) + minutes + (int(seconds // 60)) + except Exception as e: + print(f"Error parsing time: {e}") + return None def send_email(recipient, subject, body, sender): email_message = f"Subject: {subject}\nFrom: {sender}\nTo: {recipient}\n\n{body}" @@ -27,44 +33,75 @@ def send_email(recipient, subject, body, sender): process.write(email_message) process.close() - def parse_states(states_str): - if len(states_str) < 2: return "Invalid input" - second_char = states_str[1] - if second_char == '0': return "on-line" - elif second_char == '1': return "on-battery" - else: return "Invalid input" + states_str = states_str.strip('"') + if states_str == "": return "Invalid input" + if len(states_str) < 64: return "Invalid input" + if states_str[1] == '0': return "on-line" + if states_str[1] == '1': return "on-battery" + return "Invalid input" def run_command(command): - result = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + result = subprocess.run(f"{command}", shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, check=False) + data = result.stdout.strip() + if "No such file or directory" in data: + print(f"Error: {data}", flush=True) + return None return result.stdout.strip() +def snmp_get(oid): + result = run_command(f'/usr/bin/snmpget -v2c -O v -O q -c public {UPS_IP} {oid}') + if result is None or result == "": + return None + if "Timeout" in result: + print(f"UPS {UPS_IP} OID {oid} is unreachable, skipping...", flush=True) + return None + if "No Such Instance" in result: + print(f"UPS {UPS_IP} OID {oid} not found, skipping...", flush=True) + return None + if "No Such Object" in result: + print(f"UPS {UPS_IP} OID {oid} not found, skipping...", flush=True) + return None + if "No response" in result: + print(f"UPS {UPS_IP} OID {oid} not response, skipping...", flush=True) + return None + return result + def main(): + state = "on-line" while True: - oid_on_battery_time = "1.3.6.1.4.1.318.1.1.1.2.2.3.0" - result = run_command(f'/usr/bin/snmpget -v2c -O v -O q -c public {ups_ip} {oid_on_battery_time}') + if state == "on-battery": time.sleep(INTERVAL_BATTERY) + if state == "on-line": time.sleep(INTERVAL_LINE) + + result = snmp_get(OID_ON_BATTERY_TIME) + if result is None: continue time_on_battery = parse_time(result) - oid_states = "1.3.6.1.4.1.318.1.1.1.11.1.1.0" - result = run_command(f'/usr/bin/snmpget -v2c -O v -O q -c public {ups_ip} {oid_states}') - state = parse_states(result.strip('"')) + result = snmp_get(OID_STATES) + if result is None: continue + state = parse_states(result) - print(f"Time on battery: {time_on_battery} minutes, State: {state}", flush=True) + result = snmp_get(OID_BATTERY_PERCENT) + if result is None: continue + battery_percent = int(result) - shutdown_status = run_command('shutdown --show 2>&1') + print(f"UPS percent: {battery_percent}, time on battery: {time_on_battery} minutes, state: {state}", flush=True) + + shutdown_status = run_command('shutdown --show') - if (state == "on-battery") and (time_on_battery > time_threshold) and ("No scheduled shutdown" in shutdown_status): + if (state == "on-battery") and (time_on_battery > TIME_THRESHOLD) and ("No scheduled shutdown" in shutdown_status): print("UPS is failing, shutting down in 60 seconds...", flush=True) - send_email(f"{socket.gethostname()}@vvzvlad.xyz", "UPS is failing, shutting down in 60 seconds...", f"Proxmox node '{socket.gethostname()}': UPS is failing, shutting down in 60 seconds...", f"{socket.gethostname()}@vvzvlad.xyz") - run_command('sudo shutdown -h 3 "Shutting down due to UPS failure"') + send_email(f"{socket.gethostname()}@vvzvlad.xyz", "UPS is failing, shutting down in 60 seconds...", + f"Proxmox node '{socket.gethostname()}': UPS is failing, shutting down in 60 seconds...", + f"{socket.gethostname()}@vvzvlad.xyz") + run_command('shutdown -h 3 "Shutting down due to UPS failure"') + if (state == "on-line") and ("Shutdown scheduled" in shutdown_status): print("UPS is back online, cancelling shutdown...", flush=True) - run_command('sudo shutdown -c') - - if state == "on-battery": - time.sleep(5) - else: - time.sleep(interval) + run_command('shutdown -c') if __name__ == "__main__": print("Starting UPS monitoring...", flush=True)