diff --git a/src/cli.py b/src/cli.py index 51d4cb8..63d5525 100644 --- a/src/cli.py +++ b/src/cli.py @@ -1,12 +1,17 @@ import argparse +import cpu + def Interface(): parser = argparse.ArgumentParser() parser.add_argument("server_address", help="The ntfy server address.") 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-cpu-temp", action="store_true", help="Disable notifications for CPU tempature.") parser.add_argument("--cpu-temp-warning", type=int, default=70, help="CPU tempature for the warning alert. default = 70") parser.add_argument("--update-rate", type=int, default=1, help="How often updates happen in seconds. default = 1") + 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") + return parser.parse_args() \ No newline at end of file diff --git a/src/cpu.py b/src/cpu.py index 32a2a6c..ef7e98a 100644 --- a/src/cpu.py +++ b/src/cpu.py @@ -8,7 +8,7 @@ from ntfy import Ntfy _time_now = time.time() last_cpu_check_warning: float = _time_now -last_check_debounce: int = 120 # Seconds +cpu_temp_queue_check: int = 120 # Seconds class Tempature: cpu_temp_warning_message: str = "🌡️ CPU is at a high tempature." @@ -26,12 +26,12 @@ class Tempature: return float(match.group(1)) return None - def __check_time(self, last_check: float) -> bool: - return (time.time() - last_check) > last_check_debounce * 1000 + def __queue_time(self, last_check: float) -> bool: + return (time.time() - last_check) > cpu_temp_queue_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.__check_time(last_cpu_check_warning): + 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") diff --git a/src/main.py b/src/main.py index f408a57..8837ac1 100644 --- a/src/main.py +++ b/src/main.py @@ -8,12 +8,6 @@ from datetime import datetime from typing import TypedDict from ntfy import Ntfy -class Config(TypedDict): - cpu_temp_check_disabled: bool - cpu_warning_temp: int - update_interval: int - ntfy_server_url: str - def start_prompt(server_url: str) -> str: return f"""{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} Listening and sending notifications to: \033[32m{server_url}\033[0m. @@ -23,24 +17,43 @@ Source code available at: ------""" -def start(config: Config): - ntfy = Ntfy(config["ntfy_server_url"]) - ntfy_cpu_temp_monitor = cpu.Tempature(ntfy, config["cpu_warning_temp"]) +class Config(TypedDict): + cpu_temp_warning_message: str + cpu_temp_check_disabled: bool + startup_ping_disabled: bool + cpu_warning_temp: int + update_interval: int + ntfy_server_url: str - print(config) - print(start_prompt(config["ntfy_server_url"])) +class Init: + def __init__(self, config: Config): + self.config = config + self.ntfy = Ntfy(config["ntfy_server_url"]) + self.monitor_cpu_temp = cpu.Tempature(self.ntfy, config["cpu_warning_temp"]) - while True: - if not config["cpu_temp_check_disabled"]: - ntfy_cpu_temp_monitor.ntfy_check() - time.sleep(config["update_interval"]) + cpu.Tempature.cpu_temp_warning_message = config["cpu_temp_warning_message"] + + def __listen(self): + while True: + if not self.config["cpu_temp_check_disabled"]: + self.monitor_cpu_temp.ntfy_check() + time.sleep(self.config["update_interval"]) + + 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("started") + self.__listen() if __name__ == "__main__": if package.installed("lm-sensors"): cli_args = cli.Interface() - start({ - "cpu_temp_check_disabled": cli_args.disable_cpu_temp, - "cpu_warning_temp": cli_args.cpu_temp_warning, - "update_interval": cli_args.update_rate, - "ntfy_server_url": cli_args.server_address, - }) \ No newline at end of file + Init({ + "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, + "cpu_warning_temp": cli_args.cpu_temp_warning, + "update_interval": cli_args.update_rate, + "ntfy_server_url": cli_args.server_address, + }).start() \ No newline at end of file diff --git a/src/ntfy.py b/src/ntfy.py index 6b025dd..7125c1f 100644 --- a/src/ntfy.py +++ b/src/ntfy.py @@ -11,8 +11,8 @@ class Ntfy: self.server = server def send(self, message: str): - print_t(message) + print_t("Ntfy OUT: " + message) try: requests.post(self.server, data=message) except Exception as err: - print_t(f"Ntfy failed. \033[31m{err}\033[0m") + print_t(f"\033[31m{err}\033[0m")