From 7e59d19b46127398bb5fb7ec856b94802c18b661 Mon Sep 17 00:00:00 2001 From: rhpidfyre Date: Sun, 8 Jun 2025 19:42:37 -0400 Subject: [PATCH] Ntfy headers --- src/cli.py | 17 ++++++++++------- src/cpu.py | 16 ++++++++-------- src/main.py | 21 ++++++++++++--------- src/ntfy.py | 23 +++++++++++++++++++++-- 4 files changed, 51 insertions(+), 26 deletions(-) diff --git a/src/cli.py b/src/cli.py index 18b0069..95c6c66 100644 --- a/src/cli.py +++ b/src/cli.py @@ -3,20 +3,23 @@ import argparse import cpu def Interface(): - parser = argparse.ArgumentParser(description="Proxmox monitoring tool for phone notifications using ntfy.sh") - parser.add_argument("server_address", help="The ntfy server address.") + parser = argparse.ArgumentParser( + prog="Proxmox monitoring", + description="Proxmox monitoring tool for phone notifications using ntfy.sh") + parser.add_argument("server_address_no_topic", help="The ntfy server address.") - parser.add_argument("--topic", default="proxmox", help="The ntfy topic name that notifications will be sent to. Default = proxmox") + parser.add_argument("-t", "--topic", default="proxmox", help="The ntfy topic name that notifications will be sent to. Default = proxmox") + parser.add_argument("--update-rate", type=int, default=1, help="How often updates happen in seconds. default = 1") 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-startup-notify", action="store_true", help="Disable the start up notify.") parser.add_argument("--disable-cpu-temp", action="store_true", help="Disable notifications for CPU tempature.") parser.add_argument("--disable-ntfy-logs", action="store_true", help="Disable logging ntfy activity to the output.") 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-message", default=cpu.Tempature.cpu_temp_warning_message, help="The notification message if the CPU is at a high tempature. (message) [TEMP] C") + parser.add_argument("--cpu-temp-warning-timeout", type=int, default=cpu.Tempature.timeout_check, help=f"Timeout in seconds until another CPU tempature related notification can be pushed. default = {cpu.Tempature.timeout_check}") + parser.add_argument("--cpu-temp-warning-message", default=cpu.Tempature.warning_message, help="The notification message if the CPU is at a high tempature. (message) [TEMP] C") - parser.add_argument("--update-rate", type=int, default=1, help="How often updates happen in seconds. default = 1") - parser.add_argument("--startup-ping-message", default="🖥️ Ntfy proxmox monitoring started.", help="The notification message when the program is started.") + parser.add_argument("--startup-notify-message", default="🖥️ Ntfy proxmox monitoring started.", help="The notification message when the program is started.") return parser.parse_args() \ No newline at end of file diff --git a/src/cpu.py b/src/cpu.py index e5d8e7f..2407d9c 100644 --- a/src/cpu.py +++ b/src/cpu.py @@ -10,11 +10,11 @@ from ntfy import Ntfy last_cpu_check_warning: float = time.time() class Tempature: - cpu_temp_warning_message: str = "🌡️ CPU is at a high tempature." - cpu_temp_queue_check: int = 120 # Seconds + warning_message: str = "🌡️ CPU is at a high tempature." + timeout_check: int = 180 # Seconds - def __init__(self, ntfy_instance: Ntfy, cpu_warning_temp: int): - self.cpu_warning_temp = cpu_warning_temp + def __init__(self, ntfy_instance: Ntfy, warning_temp: int): + self.warning_temp = warning_temp self.ntfy = ntfy_instance def get(self) -> Optional[float]: @@ -26,14 +26,14 @@ class Tempature: return float(match.group(1)) return None - def __queue_time(self, last_check: float) -> bool: - return (time.time() - last_check) > Tempature.cpu_temp_queue_check * 1000 + def __timeout_expired(self, last_check: float) -> bool: + return (time.time() - last_check) > Tempature.timeout_check * 1000 def ntfy_check(self): cpu_temp = self.get() if cpu_temp: cpu_temp = math.floor(cpu_temp) - 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") + if cpu_temp >= self.warning_temp and self.__timeout_expired(last_cpu_check_warning): + self.ntfy.send(f"{Tempature.warning_message} {cpu_temp} C") else: print_t("\033[31mCannot get a feasible tempature value for the CPU. (lm-sensors)\033[0m") diff --git a/src/main.py b/src/main.py index 51fba96..c208d84 100644 --- a/src/main.py +++ b/src/main.py @@ -13,15 +13,16 @@ def start_prompt(server_url: str) -> str: Listening and sending notifications to: \033[32m{server_url}\033[0m. Source code available at: - +* ------""" class Config(TypedDict): + cpu_temp_warning_timeout: int cpu_temp_warning_message: str cpu_temp_check_disabled: bool - startup_ping_disabled: bool - startup_ping_message: str + startup_notify_disabled: bool + startup_notify_message: str ntfy_logs_disabled: bool cpu_warning_temp: int update_interval: int @@ -33,7 +34,8 @@ class Init: self.ntfy = Ntfy(config["ntfy_server_url"], config["ntfy_logs_disabled"]) self.monitor_cpu_temp = cpu.Tempature(self.ntfy, config["cpu_warning_temp"]) - cpu.Tempature.cpu_temp_warning_message = config["cpu_temp_warning_message"] + cpu.Tempature.warning_message = config["cpu_temp_warning_message"] + cpu.Tempature.timeout_check = config["cpu_temp_warning_timeout"] def __listen(self): while True: @@ -44,8 +46,8 @@ class Init: 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(self.config["startup_ping_message"]) + if not self.config["startup_notify_disabled"]: + self.ntfy.send(self.config["startup_notify_message"]) self.__listen() @@ -53,12 +55,13 @@ if __name__ == "__main__": if package.installed("lm-sensors"): cli_args = cli.Interface() Init({ + "cpu_temp_warning_timeout": cli_args.cpu_temp_warning_timeout, "cpu_temp_warning_message": cli_args.cpu_temp_warning_message, "cpu_temp_check_disabled": cli_args.disable_cpu_temp, - "startup_ping_disabled": cli_args.disable_startup_ping, - "startup_ping_message": cli_args.startup_ping_message, + "startup_notify_disabled": cli_args.disable_startup_notify, + "startup_notify_message": cli_args.startup_notify_message, "ntfy_logs_disabled": cli_args.disable_ntfy_logs, "cpu_warning_temp": cli_args.cpu_temp_warning, "update_interval": cli_args.update_rate, - "ntfy_server_url": cli_args.server_address + "/" + cli_args.topic, #Heh. + "ntfy_server_url": cli_args.server_address_no_topic + "/" + cli_args.topic, #Heh. }).start() \ No newline at end of file diff --git a/src/ntfy.py b/src/ntfy.py index ee7b31f..e40e929 100644 --- a/src/ntfy.py +++ b/src/ntfy.py @@ -1,16 +1,35 @@ import requests +from typing import TypedDict, Optional from print_t import print_t +from enum import Enum + +class HeaderPriority(Enum): + Urgent = "5" + High = "4" + Medium = "3" + Low = "2" + Min = "1" + +class Headers(TypedDict): + Priority: HeaderPriority + Title: str + Tags: str class Ntfy: def __init__(self, server: str, logging_disabled: bool): self.server = server self.logging_disabled = logging_disabled - def send(self, message: str): + def send(self, message: str, title: Optional[str] = None, extra_headers: Optional[Headers] = None): + headers = {} + if title: + headers["Title"] = title.encode(encoding="utf-8") + if extra_headers: + headers.update(extra_headers) if not self.logging_disabled: print_t("Ntfy OUT: " + message) try: - requests.post(self.server, data=message.encode(encoding="utf-8")) + requests.post(self.server, data=message.encode(encoding="utf-8"), headers=headers) except Exception as err: print_t(f"\033[31m{err}\033[0m")