--disable-startup-ping and a cleaner main.py start process

This commit is contained in:
2025-06-07 15:10:50 -04:00
parent b828a84310
commit 32109b7acd
4 changed files with 45 additions and 27 deletions

View File

@ -1,12 +1,17 @@
import argparse import argparse
import cpu
def Interface(): def Interface():
parser = argparse.ArgumentParser() parser = argparse.ArgumentParser()
parser.add_argument("server_address", help="The ntfy server address.") parser.add_argument("server_address", help="The ntfy server address.")
parser.add_argument("--disable-uptime-notifys", action="store_true", help="Disable uptime notifications.") parser.add_argument("--disable-uptime-notifys", action="store_true", help="Disable uptime notifications.")
parser.add_argument("--disable-startup-ping", action="store_true", help="Disable the start up ping.")
parser.add_argument("--disable-cpu-temp", action="store_true", help="Disable notifications for CPU tempature.") parser.add_argument("--disable-cpu-temp", action="store_true", help="Disable notifications for CPU tempature.")
parser.add_argument("--cpu-temp-warning", type=int, default=70, help="CPU tempature for the warning alert. default = 70") parser.add_argument("--cpu-temp-warning", type=int, default=70, help="CPU tempature for the warning alert. default = 70")
parser.add_argument("--update-rate", type=int, default=1, help="How often updates happen in seconds. default = 1") parser.add_argument("--update-rate", type=int, default=1, help="How often updates happen in seconds. default = 1")
parser.add_argument("--cpu-temp-warning-message", default=cpu.Tempature.cpu_temp_warning_message, help="The notification message if the CPU is at a high tempature. (message) [TEMP] C")
return parser.parse_args() return parser.parse_args()

View File

@ -8,7 +8,7 @@ from ntfy import Ntfy
_time_now = time.time() _time_now = time.time()
last_cpu_check_warning: float = _time_now last_cpu_check_warning: float = _time_now
last_check_debounce: int = 120 # Seconds cpu_temp_queue_check: int = 120 # Seconds
class Tempature: class Tempature:
cpu_temp_warning_message: str = "🌡️ CPU is at a high tempature." cpu_temp_warning_message: str = "🌡️ CPU is at a high tempature."
@ -26,12 +26,12 @@ class Tempature:
return float(match.group(1)) return float(match.group(1))
return None return None
def __check_time(self, last_check: float) -> bool: def __queue_time(self, last_check: float) -> bool:
return (time.time() - last_check) > last_check_debounce * 1000 return (time.time() - last_check) > cpu_temp_queue_check * 1000
def ntfy_check(self): def ntfy_check(self):
cpu_temp = self.get() cpu_temp = self.get()
if cpu_temp: if cpu_temp:
cpu_temp = math.floor(cpu_temp) cpu_temp = math.floor(cpu_temp)
if cpu_temp >= self.cpu_warning_temp and self.__check_time(last_cpu_check_warning): if cpu_temp >= self.cpu_warning_temp and self.__queue_time(last_cpu_check_warning):
self.ntfy.send(f"{Tempature.cpu_temp_warning_message} {cpu_temp} C") self.ntfy.send(f"{Tempature.cpu_temp_warning_message} {cpu_temp} C")

View File

@ -8,12 +8,6 @@ from datetime import datetime
from typing import TypedDict from typing import TypedDict
from ntfy import Ntfy from ntfy import Ntfy
class Config(TypedDict):
cpu_temp_check_disabled: bool
cpu_warning_temp: int
update_interval: int
ntfy_server_url: str
def start_prompt(server_url: str) -> str: def start_prompt(server_url: str) -> str:
return f"""{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} return f"""{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
Listening and sending notifications to: \033[32m{server_url}\033[0m. Listening and sending notifications to: \033[32m{server_url}\033[0m.
@ -23,24 +17,43 @@ Source code available at:
<https://git.rhpidfyre.io/rhpidfyre/proxmox-ntfy> <https://git.rhpidfyre.io/rhpidfyre/proxmox-ntfy>
------""" ------"""
def start(config: Config): class Config(TypedDict):
ntfy = Ntfy(config["ntfy_server_url"]) cpu_temp_warning_message: str
ntfy_cpu_temp_monitor = cpu.Tempature(ntfy, config["cpu_warning_temp"]) cpu_temp_check_disabled: bool
startup_ping_disabled: bool
cpu_warning_temp: int
update_interval: int
ntfy_server_url: str
print(config) class Init:
print(start_prompt(config["ntfy_server_url"])) def __init__(self, config: Config):
self.config = config
self.ntfy = Ntfy(config["ntfy_server_url"])
self.monitor_cpu_temp = cpu.Tempature(self.ntfy, config["cpu_warning_temp"])
while True: cpu.Tempature.cpu_temp_warning_message = config["cpu_temp_warning_message"]
if not config["cpu_temp_check_disabled"]:
ntfy_cpu_temp_monitor.ntfy_check() def __listen(self):
time.sleep(config["update_interval"]) while True:
if not self.config["cpu_temp_check_disabled"]:
self.monitor_cpu_temp.ntfy_check()
time.sleep(self.config["update_interval"])
def start(self):
print(f"{self.config}\n" + start_prompt(self.config["ntfy_server_url"]))
if not self.config["startup_ping_disabled"]:
self.ntfy.send("started")
self.__listen()
if __name__ == "__main__": if __name__ == "__main__":
if package.installed("lm-sensors"): if package.installed("lm-sensors"):
cli_args = cli.Interface() cli_args = cli.Interface()
start({ Init({
"cpu_temp_check_disabled": cli_args.disable_cpu_temp, "cpu_temp_warning_message": cli_args.cpu_temp_warning_message,
"cpu_warning_temp": cli_args.cpu_temp_warning, "cpu_temp_check_disabled": cli_args.disable_cpu_temp,
"update_interval": cli_args.update_rate, "startup_ping_disabled": cli_args.disable_startup_ping,
"ntfy_server_url": cli_args.server_address, "cpu_warning_temp": cli_args.cpu_temp_warning,
}) "update_interval": cli_args.update_rate,
"ntfy_server_url": cli_args.server_address,
}).start()

View File

@ -11,8 +11,8 @@ class Ntfy:
self.server = server self.server = server
def send(self, message: str): def send(self, message: str):
print_t(message) print_t("Ntfy OUT: " + message)
try: try:
requests.post(self.server, data=message) requests.post(self.server, data=message)
except Exception as err: except Exception as err:
print_t(f"Ntfy failed. \033[31m{err}\033[0m") print_t(f"\033[31m{err}\033[0m")