diff --git a/src/cli.py b/src/cli.py index 95c6c66..f535550 100644 --- a/src/cli.py +++ b/src/cli.py @@ -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() \ No newline at end of file diff --git a/src/command.py b/src/command.py index 3504cbf..bbc9b31 100644 --- a/src/command.py +++ b/src/command.py @@ -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() \ No newline at end of file diff --git a/src/cpu.py b/src/cpu.py index 0dd6796..9180e04 100644 --- a/src/cpu.py +++ b/src/cpu.py @@ -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") diff --git a/src/main.py b/src/main.py index 0a3cba6..7ed3daa 100644 --- a/src/main.py +++ b/src/main.py @@ -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()) diff --git a/src/ntfy.py b/src/ntfy.py index e40e929..41e8374 100644 --- a/src/ntfy.py +++ b/src/ntfy.py @@ -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: