rhpidfyre/proxmox-ntfy
1
0
Fork 0
mirror of https://github.com/unixtensor/proxmox-ntfy.git synced 2025-06-27 23:58:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Redisgned the way CPU tempature is detected and the start up notification

Browse Source
This commit is contained in:
rhpidfyre
2025-06-12 01:44:21 -04:00
parent 97e3348bd4
commit 3dbc105ae3
5 changed files with 75 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/cli.py
View File

@ -16,10 +16,14 @@ def Interface():
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-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", 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.")
return parser.parse_args()

4
src/command.py
View File

@ -9,6 +9,8 @@ def package_installed(package_name: str) -> Optional[bool]:
print(f"Package \"{package_name}\" not installed.")
return installed
except Exception as err:
print(err)
print(f"\033[31m{err}\033[0m")
return None
def uname() -> str:
return subprocess.run(["uname", "-a"], capture_output=True, text=True).stdout.strip()

53
src/cpu.py
View File

@ -7,21 +7,25 @@ from print_t import print_t
from typing import Optional
from ntfy import Ntfy
_time_now = time.time()
_init_run_critical: bool = True
_init_run_warning: bool = True
_time_now: float = time.time()
last_cpu_check_warning: float = _time_now
last_cpu_warning_set: float = _time_now
last_warning_temp: int = 0
last_cpu_check_warning: float = _time_now
last_cpu_check_crticial: float = _time_now
def timeout_expired(check: float, timeout: int) -> bool:
return (time.time() - check) > timeout * 1000
return (time.time() - check) > timeout
class Tempature:
warning_message: str = "🌡️ CPU is at a high tempature."
timeout_check: int = 60 # Seconds
timeout_check_critical: int = 120 # Seconds
timeout_check_warn: int = 300 # Seconds
critical_message: str = "🔥 CPU is at a very high tempature."
warning_message: str = "🌡️ CPU is at a high tempature."
thermal_critical_c: int = 80
thermal_warn_c: int = 70
def __init__(self, ntfy_instance: Ntfy, warning_temp: int):
self.warning_temp = warning_temp
def __init__(self, ntfy_instance: Ntfy):
self.ntfy = ntfy_instance
def get(self) -> Optional[float]:
@ -33,27 +37,22 @@ class Tempature:
return float(match.group(1))
return None
def __warn(self, cpu_temp: int):
pad = 0
if timeout_expired(last_cpu_warning_set, Tempature.timeout_check * 5):
#If the timeout expired, send a notify immediately
global last_warning_temp
last_warning_temp = 0
else:
#If the timeout is still active, apply padding to the current CPU tempature so its less spammy
pad = 5
if cpu_temp + pad > last_warning_temp:
message = f"{cpu_temp} C."
if last_warning_temp != 0:
message += f" The tempature has risen by: {cpu_temp - last_warning_temp} C. 📈"
last_warning_temp = cpu_temp
self.ntfy.send(message=message, title=Tempature.warning_message)
def ntfy_check(self):
cpu_temp = self.get()
if cpu_temp:
global _init_run_warning
global _init_run_critical
global last_cpu_check_warning
global last_cpu_check_crticial
cpu_temp = math.floor(cpu_temp)
if cpu_temp >= self.warning_temp and timeout_expired(last_cpu_check_warning, Tempature.timeout_check):
self.__warn(cpu_temp)
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
last_cpu_check_warning = time.time()
self.ntfy.send(message=f"{cpu_temp} C", title=Tempature.warning_message)
if cpu_temp >= Tempature.thermal_critical_c and (_init_run_critical or timeout_expired(last_cpu_check_crticial, Tempature.timeout_check_critical)):
_init_run_critical = False
last_cpu_check_crticial = time.time()
self.ntfy.send(message=f"{cpu_temp} C", title=Tempature.critical_message)
else:
print_t("\033[31mCannot get a feasible tempature value for the CPU. (lm-sensors)\033[0m")

61
src/main.py
View File

@ -44,23 +44,27 @@ Address with a topic:
\033[32mhttps://domain.com\033[0m -t|--topic \033[32mexample_topic\033[0m"""
class Config(TypedDict):
cpu_temp_warning_timeout: int
cpu_temp_warning_message: str
cpu_temp_check_disabled: bool
startup_notify_disabled: bool
startup_notify_message: str
ntfy_logs_disabled: bool
cpu_warning_temp: int
update_interval: int
ntfy_server_url: str
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
startup_notify_disabled: bool
startup_notify_message: str
ntfy_logs_disabled: bool
cpu_temp_critical: int
cpu_warning_temp: int
update_interval: int
ntfy_server_url: str
class Init:
def __init__(self, config: Config):
self.config = config
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.warning_message = config["cpu_temp_warning_message"]
cpu.Tempature.timeout_check = config["cpu_temp_warning_timeout"]
self.monitor_cpu_temp = cpu.Tempature(self.ntfy)
cpu.Tempature.warning_message = config["cpu_temp_warning_message"]
cpu.Tempature.timeout_check_warn = config["cpu_temp_warning_timeout"]
cpu.Tempature.thermal_warn_c = config["cpu_warning_temp"]
def __listen(self):
while True:
@ -68,11 +72,21 @@ class Init:
self.monitor_cpu_temp.ntfy_check()
time.sleep(self.config["update_interval"])
def __start_notify(self):
t = datetime.now()
self.ntfy.send(
message=command.uname(),
title=self.config["startup_notify_message"] + f" {t.strftime('%Y-%m-%d')} - {t.strftime('%H:%M:%S')}")
def start(self):
if self.config["cpu_warning_temp"] >= self.config["cpu_temp_critical"]:
print("CPU warning tempature cannot be greater than or equal to the crtitical tempature.")
return
print(f"{self.config}\n" + start_prompt(self.config["ntfy_server_url"]))
if not self.config["startup_notify_disabled"]:
self.ntfy.send(self.config["startup_notify_message"])
self.__start_notify()
self.__listen()
def main():
@ -80,15 +94,18 @@ def main():
address = Address(cli_args.server_address_no_topic)
if address.is_valid():
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_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": address.format(cli_args.topic)
"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,
"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_temp_critical": cli_args.cpu_temp_critical,
"cpu_warning_temp": cli_args.cpu_temp_warning,
"update_interval": cli_args.update_rate,
"ntfy_server_url": address.format(cli_args.topic)
}).start()
else:
print(address.not_valid_prompt())

2
src/ntfy.py
View File

@ -28,7 +28,7 @@ class Ntfy:
if extra_headers:
headers.update(extra_headers)
if not self.logging_disabled:
print_t("Ntfy OUT: " + message)
print_t(f"Ntfy OUT: \033[0;35mtitle\033[0m={title} \033[0;35mmessage\033[0m={message} \033[0;35mheaders\033[0m={headers}")
try:
requests.post(self.server, data=message.encode(encoding="utf-8"), headers=headers)
except Exception as err: