Ntfy headers

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()

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")

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:
-<https://github.com/unixtensor/proxmox-ntfy>
+* <https://github.com/unixtensor/proxmox-ntfy>
 <https://git.rhpidfyre.io/rhpidfyre/proxmox-ntfy>
 ------"""
 
 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()

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")