Clean up the main file a little

This commit is contained in:
2025-06-18 16:06:56 -04:00
parent a0091fd1e2
commit 6a50df8f89
2 changed files with 72 additions and 59 deletions

View File

@ -2,31 +2,72 @@ import argparse
import cpu import cpu
def Interface(): from typing import TypedDict
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.") class Config(TypedDict):
cpu_temp_critical_timeout: int
cpu_temp_critical_message: str
cpu_temp_warning_timeout: int
cpu_temp_warning_message: str
cpu_temp_check_disabled: bool
cpu_temp_critical: int
cpu_warning_temp: int
cpu_temp_zone_label: str
cpu_temp_zone: str
startup_notify_disabled: bool
startup_notify_message: str
daily_notifys_disabled: bool
ntfy_logs_disabled: bool
update_interval: int
ntfy_server_url: str
parser.add_argument("-t", "--topic", default="proxmox", help="The ntfy topic name that notifications will be sent to. Default = proxmox") class Interface:
parser.add_argument("--update-rate", type=int, default=1, help="How often updates happen in seconds. default = 1") def __init__(self):
parser = argparse.ArgumentParser(
prog="Proxmox monitoring",
description="Proxmox monitoring tool for phone notifications using ntfy.sh")
parser.add_argument("--disable-uptime-notifys", action="store_true", help="Disable uptime notifications.") parser.add_argument("server_address_no_topic", help="The ntfy server address.")
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-zone", default="k10temp", help="The tempature zone for getting CPU info. default = k10temp") parser.add_argument("-t", "--topic", default="proxmox", help="The ntfy topic name that notifications will be sent to. Default = proxmox")
parser.add_argument("--cpu-temp-zone-label", default="Tctl", help="The label for getting the current CPU tempature. default = Tctl") parser.add_argument("--update-rate", type=int, default=1, help="How often updates happen in seconds. default = 1")
parser.add_argument("--cpu-temp-warning", type=int, default=cpu.Tempature.thermal_warn_c, help=f"CPU tempature for the warning alert. default = {cpu.Tempature.thermal_warn_c}")
parser.add_argument("--cpu-temp-warning-timeout", type=int, default=cpu.Tempature.timeout_check_warn, help=f"Timeout in seconds until another CPU tempature related notification can be pushed. default = {cpu.Tempature.timeout_check_warn}")
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("--cpu-temp-critical", type=int, default=cpu.Tempature.thermal_critical_c, help=f"CPU tempature for the critical alert. default = {cpu.Tempature.thermal_critical_c}") parser.add_argument("--disable-uptime-notifys", action="store_true", help="Disable uptime notifications.")
parser.add_argument("--cpu-temp-critical-timeout", type=int, default=cpu.Tempature.timeout_check_critical, help=f"Timeout in seconds until another CPU tempature related notification can be pushed. default = {cpu.Tempature.timeout_check_critical}") parser.add_argument("--disable-startup-notify", action="store_true", help="Disable the start up notify.")
parser.add_argument("--cpu-temp-critical-message", default=cpu.Tempature.critical_message, help="The notification message if the CPU is at a high tempature. (message) [TEMP] C") parser.add_argument("--disable-daily-notifys", action="store_true", help="Disable the daily notifys on the system's highest stats.")
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("--startup-notify-message", default="🖥️ Ntfy proxmox monitoring started.", help="The notification message when the program is started.") parser.add_argument("--cpu-temp-zone", default="k10temp", help="The tempature zone for getting CPU info. default = k10temp")
parser.add_argument("--cpu-temp-zone-label", default="Tctl", help="The label for getting the current CPU tempature. default = Tctl")
return parser.parse_args() parser.add_argument("--cpu-temp-warning", type=int, default=cpu.Tempature.thermal_warn_c, help=f"CPU tempature for the warning alert. default = {cpu.Tempature.thermal_warn_c}")
parser.add_argument("--cpu-temp-warning-timeout", type=int, default=cpu.Tempature.timeout_check_warn, help=f"Timeout in seconds until another CPU tempature related notification can be pushed. default = {cpu.Tempature.timeout_check_warn}")
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("--cpu-temp-critical", type=int, default=cpu.Tempature.thermal_critical_c, help=f"CPU tempature for the critical alert. default = {cpu.Tempature.thermal_critical_c}")
parser.add_argument("--cpu-temp-critical-timeout", type=int, default=cpu.Tempature.timeout_check_critical, help=f"Timeout in seconds until another CPU tempature related notification can be pushed. default = {cpu.Tempature.timeout_check_critical}")
parser.add_argument("--cpu-temp-critical-message", default=cpu.Tempature.critical_message, help="The notification message if the CPU is at a high tempature. (message) [TEMP] C")
parser.add_argument("--startup-notify-message", default="🖥️ Ntfy proxmox monitoring started.", help="The notification message when the program is started.")
self.cli_args = parser.parse_args()
def to_config(self, formatted_address: str) -> Config:
return {
"cpu_temp_critical_timeout": self.cli_args.cpu_temp_critical_timeout,
"cpu_temp_critical_message": self.cli_args.cpu_temp_critical_message,
"cpu_temp_warning_timeout": self.cli_args.cpu_temp_warning_timeout,
"cpu_temp_warning_message": self.cli_args.cpu_temp_warning_message,
"cpu_temp_check_disabled": self.cli_args.disable_cpu_temp,
"cpu_temp_critical": self.cli_args.cpu_temp_critical,
"cpu_warning_temp": self.cli_args.cpu_temp_warning,
"cpu_temp_zone_label": self.cli_args.cpu_temp_zone_label,
"cpu_temp_zone": self.cli_args.cpu_temp_zone,
"startup_notify_disabled": self.cli_args.disable_startup_notify,
"daily_notifys_disabled": self.cli_args.disable_daily_notifys,
"startup_notify_message": self.cli_args.startup_notify_message,
"ntfy_logs_disabled": self.cli_args.disable_ntfy_logs,
"update_interval": self.cli_args.update_rate,
"ntfy_server_url": formatted_address
}

View File

@ -6,7 +6,6 @@ import cpu
from datetime import datetime from datetime import datetime
from address import Address from address import Address
from typing import TypedDict
from ntfy import Ntfy from ntfy import Ntfy
class Prompt: class Prompt:
@ -33,24 +32,8 @@ Address with a topic:
\033[32mhttp://domain.com\033[0m -t|--topic \033[32mexample_topic\033[0m \033[32mhttp://domain.com\033[0m -t|--topic \033[32mexample_topic\033[0m
\033[32mhttps://domain.com\033[0m -t|--topic \033[32mexample_topic\033[0m""" \033[32mhttps://domain.com\033[0m -t|--topic \033[32mexample_topic\033[0m"""
class Config(TypedDict):
cpu_temp_critical_timeout: int
cpu_temp_critical_message: str
cpu_temp_warning_timeout: int
cpu_temp_warning_message: str
cpu_temp_check_disabled: bool
cpu_temp_critical: int
cpu_warning_temp: int
cpu_temp_zone_label: str
cpu_temp_zone: str
startup_notify_disabled: bool
startup_notify_message: str
ntfy_logs_disabled: bool
update_interval: int
ntfy_server_url: str
class Init: class Init:
def __init__(self, config: Config): def __init__(self, config: cli.Config):
self.config = config self.config = config
self.ntfy = Ntfy(config["ntfy_server_url"], config["ntfy_logs_disabled"]) self.ntfy = Ntfy(config["ntfy_server_url"], config["ntfy_logs_disabled"])
self.monitor_cpu_temp = cpu.Tempature(self.ntfy, config["cpu_temp_zone"], config["cpu_temp_zone_label"]) self.monitor_cpu_temp = cpu.Tempature(self.ntfy, config["cpu_temp_zone"], config["cpu_temp_zone_label"])
@ -62,6 +45,8 @@ class Init:
while True: while True:
if not self.config["cpu_temp_check_disabled"]: if not self.config["cpu_temp_check_disabled"]:
self.monitor_cpu_temp.ntfy_check() self.monitor_cpu_temp.ntfy_check()
if not self.config["daily_notifys_disabled"]:
...
time.sleep(self.config["update_interval"]) time.sleep(self.config["update_interval"])
def __start_notify(self): def __start_notify(self):
@ -79,30 +64,17 @@ class Init:
self.__listen() self.__listen()
def main(): def main():
cli_args = cli.Interface() interface = cli.Interface()
address = Address(cli_args.server_address_no_topic) interface_args = interface.cli_args
address = Address(interface_args.server_address_no_topic)
if address.is_valid(): if address.is_valid():
formatted_address = address.format(cli_args.topic) formatted_address = address.format(interface_args.topic)
print(Prompt.start(formatted_address)) print(Prompt.start(formatted_address))
Init({ Init(interface.to_config(formatted_address)).start()
"cpu_temp_critical_timeout": cli_args.cpu_temp_critical_timeout,
"cpu_temp_critical_message": cli_args.cpu_temp_critical_message,
"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,
"cpu_temp_critical": cli_args.cpu_temp_critical,
"cpu_warning_temp": cli_args.cpu_temp_warning,
"cpu_temp_zone_label": cli_args.cpu_temp_zone_label,
"cpu_temp_zone": cli_args.cpu_temp_zone,
"startup_notify_disabled": cli_args.disable_startup_notify,
"startup_notify_message": cli_args.startup_notify_message,
"ntfy_logs_disabled": cli_args.disable_ntfy_logs,
"update_interval": cli_args.update_rate,
"ntfy_server_url": formatted_address
}).start()
else: else:
print(Prompt.address_not_valid(cli_args.server_address_no_topic)) print(Prompt.address_not_valid(interface_args.server_address_no_topic))
if __name__ == "__main__": if __name__ == "__main__":
main() main()