Compare commits

..

2 Commits

Author SHA1 Message Date
f735fe50ba int instead of math.floor for the CPU tempature 2025-06-19 15:25:15 -04:00
6a50df8f89 Clean up the main file a little 2025-06-18 16:06:56 -04:00
3 changed files with 77 additions and 65 deletions

View File

@ -2,7 +2,27 @@ import argparse
import cpu import cpu
def Interface(): from typing import TypedDict
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
class Interface:
def __init__(self):
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="Proxmox monitoring", prog="Proxmox monitoring",
description="Proxmox monitoring tool for phone notifications using ntfy.sh") description="Proxmox monitoring tool for phone notifications using ntfy.sh")
@ -14,11 +34,13 @@ def Interface():
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-notify", action="store_true", help="Disable the start up notify.") parser.add_argument("--disable-startup-notify", action="store_true", help="Disable the start up notify.")
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-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("--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("--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") parser.add_argument("--cpu-temp-zone-label", default="Tctl", help="The label for getting the current CPU tempature. default = Tctl")
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", 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-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-warning-message", default=cpu.Tempature.warning_message, help="The notification message if the CPU is at a high tempature. (message) [TEMP] C")
@ -29,4 +51,23 @@ def Interface():
parser.add_argument("--startup-notify-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() 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

@ -1,9 +1,8 @@
import psutil import psutil
import time import time
import math
from typing import Optional, Callable
from print_t import print_t from print_t import print_t
from typing import Optional
from ntfy import Ntfy from ntfy import Ntfy
_init_run_critical: bool = True _init_run_critical: bool = True
@ -13,8 +12,7 @@ _time_now: float = time.time()
last_cpu_check_warning: float = _time_now last_cpu_check_warning: float = _time_now
last_cpu_check_crticial: float = _time_now last_cpu_check_crticial: float = _time_now
def timeout_expired(check: float, timeout: int) -> bool: timeout_expired: Callable[[float, int], bool] = lambda c, t: (time.time() - c) > t
return (time.time() - check) > timeout
class Tempature: class Tempature:
timeout_check_critical: int = 600 # Seconds timeout_check_critical: int = 600 # Seconds
@ -43,7 +41,8 @@ class Tempature:
global last_cpu_check_warning global last_cpu_check_warning
global last_cpu_check_crticial global last_cpu_check_crticial
cpu_temp = math.floor(cpu_temp) cpu_temp = int(cpu_temp)
if cpu_temp >= Tempature.thermal_warn_c and (_init_run_warning or timeout_expired(last_cpu_check_warning, Tempature.timeout_check_warn)): if cpu_temp >= Tempature.thermal_warn_c and (_init_run_warning or timeout_expired(last_cpu_check_warning, Tempature.timeout_check_warn)):
_init_run_warning = False _init_run_warning = False
last_cpu_check_warning = time.time() last_cpu_check_warning = time.time()
@ -53,4 +52,4 @@ class Tempature:
last_cpu_check_crticial = time.time() last_cpu_check_crticial = time.time()
self.ntfy.send(message=f"{cpu_temp} C", title=Tempature.critical_message) self.ntfy.send(message=f"{cpu_temp} C", title=Tempature.critical_message)
else: else:
print_t("\033[31mCannot get a feasible tempature value for the CPU. (lm-sensors)\033[0m") print_t(f"\033[31mCannot get a feasible tempature value for the CPU. cpu_temp_zone={self.cpu_temp_zone} cpu_temp_zone_label={self.cpu_temp_zone_label}\033[0m")

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