mirror of
https://github.com/unixtensor/proxmox-ntfy.git
synced 2025-06-28 08:28:05 +00:00
Implament sys.argv
with argparse
This commit is contained in:
24
src/cli.py
24
src/cli.py
@ -10,20 +10,14 @@ python3 main.py ntfy.domain.com\033[0m
|
|||||||
|
|
||||||
Use \033[32m-h\033[0m or \033[32m--help\033[0m for a full list of options."""
|
Use \033[32m-h\033[0m or \033[32m--help\033[0m for a full list of options."""
|
||||||
|
|
||||||
class Interface:
|
def Interface():
|
||||||
def __init__(self):
|
parser = argparse.ArgumentParser()
|
||||||
self.parser = argparse.ArgumentParser()
|
parser.add_argument("server_address", help="The ntfy server address.")
|
||||||
self.parser.add_argument("--cpu-temp-disabled", action="store_true", help="Disable notifications for CPU tempature.")
|
|
||||||
self.parser.add_argument("--cpu-temp-critical", type=int, default=80, help="CPU tempature for the crtitical alert. default = 80")
|
|
||||||
self.parser.add_argument("--cpu-temp-warning", type=int, default=70, help="CPU tempature for the warning alert. default = 70")
|
|
||||||
self.parser.add_argument("--update-rate", type=int, default=1, help="How often updates happen in seconds. default = 1")
|
|
||||||
|
|
||||||
def parsed_args(self):
|
parser.add_argument("--disable-uptime-notifys", action="store_true", help="Disable uptime notifications.")
|
||||||
return self.parser.parse_args()
|
parser.add_argument("--disable-cpu-temp", action="store_true", help="Disable notifications for CPU tempature.")
|
||||||
|
parser.add_argument("--cpu-temp-critical", type=int, default=80, help="CPU tempature for the crtitical alert. default = 80")
|
||||||
|
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")
|
||||||
|
|
||||||
def argv_1(self) -> Optional[str]:
|
return parser.parse_args()
|
||||||
if len(sys.argv) > 1:
|
|
||||||
return sys.argv[1]
|
|
||||||
else:
|
|
||||||
print(_ntfy_configure_prompt)
|
|
||||||
return None
|
|
15
src/cpu.py
15
src/cpu.py
@ -5,7 +5,10 @@ import re
|
|||||||
from typing import Optional
|
from typing import Optional
|
||||||
from ntfy import Ntfy
|
from ntfy import Ntfy
|
||||||
|
|
||||||
last_cpu_check_unix: float = time.time()
|
_time_now = time.time()
|
||||||
|
last_cpu_check_critical: float = _time_now
|
||||||
|
last_cpu_check_warning: float = _time_now
|
||||||
|
|
||||||
last_cpu_check: int = 60 # Seconds
|
last_cpu_check: int = 60 # Seconds
|
||||||
|
|
||||||
class Tempature:
|
class Tempature:
|
||||||
@ -26,10 +29,14 @@ class Tempature:
|
|||||||
return float(match.group(1))
|
return float(match.group(1))
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
def __check_time(self, last_check: float) -> bool:
|
||||||
|
return (time.time() - last_check) > last_cpu_check * 1000
|
||||||
|
|
||||||
def ntfy_check(self):
|
def ntfy_check(self):
|
||||||
cpu_temp = self.get()
|
cpu_temp = self.get()
|
||||||
if cpu_temp and (time.time() - last_cpu_check_unix) > last_cpu_check * 1000:
|
if cpu_temp:
|
||||||
if cpu_temp >= self.cpu_critical_temp:
|
print(f"{cpu_temp}")
|
||||||
|
if cpu_temp >= self.cpu_critical_temp and self.__check_time(last_cpu_check_critical):
|
||||||
self.ntfy.send(f"{Tempature.cpu_temp_crtitical_message} {cpu_temp}")
|
self.ntfy.send(f"{Tempature.cpu_temp_crtitical_message} {cpu_temp}")
|
||||||
elif cpu_temp >= self.cpu_warning_temp:
|
if cpu_temp >= self.cpu_warning_temp and self.__check_time(last_cpu_check_warning):
|
||||||
self.ntfy.send(f"{Tempature.cpu_temp_warning_message} {cpu_temp}")
|
self.ntfy.send(f"{Tempature.cpu_temp_warning_message} {cpu_temp}")
|
||||||
|
52
src/main.py
52
src/main.py
@ -4,36 +4,42 @@ import package
|
|||||||
import cli
|
import cli
|
||||||
import cpu
|
import cpu
|
||||||
|
|
||||||
|
from datetime import datetime
|
||||||
|
from typing import TypedDict
|
||||||
from ntfy import Ntfy
|
from ntfy import Ntfy
|
||||||
|
|
||||||
def start(
|
_pretty_date_time = datetime.fromtimestamp(time.time())
|
||||||
cpu_temp_checking: bool,
|
_monitoring_prompt = f"""{_pretty_date_time}
|
||||||
cpu_critical_temp: int,
|
Ntfy monitoring software is now listening.
|
||||||
cpu_warning_temp: int,
|
|
||||||
ntfy_server: str,
|
|
||||||
interval: int
|
|
||||||
):
|
|
||||||
ntfy = Ntfy(ntfy_server)
|
|
||||||
ntfy_cpu_temp_monitor = cpu.Tempature(ntfy, cpu_critical_temp, cpu_warning_temp)
|
|
||||||
|
|
||||||
print(f"Started. {time.time()}")
|
Source code available at:
|
||||||
print("Ntfy monitoring software is now listening.")
|
<https://github.com/unixtensor/proxmox-ntfy>
|
||||||
|
<https://git.rhpidfyre.io/rhpidfyre/proxmox-ntfy>"""
|
||||||
|
|
||||||
|
class Config(TypedDict):
|
||||||
|
cpu_temp_check_disabled: bool
|
||||||
|
cpu_critical_temp: int
|
||||||
|
cpu_warning_temp: int
|
||||||
|
update_interval: int
|
||||||
|
ntfy_server_url: str
|
||||||
|
|
||||||
|
def start(config: Config):
|
||||||
|
ntfy = Ntfy(config["ntfy_server_url"])
|
||||||
|
ntfy_cpu_temp_monitor = cpu.Tempature(ntfy, config["cpu_critical_temp"], config["cpu_warning_temp"])
|
||||||
|
|
||||||
|
print(_monitoring_prompt)
|
||||||
while True:
|
while True:
|
||||||
if not cpu_temp_checking:
|
if not config["cpu_temp_check_disabled"]:
|
||||||
ntfy_cpu_temp_monitor.ntfy_check()
|
ntfy_cpu_temp_monitor.ntfy_check()
|
||||||
time.sleep(interval)
|
time.sleep(config["update_interval"])
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
if package.installed("lm-sensors"):
|
if package.installed("lm-sensors"):
|
||||||
cli_args = cli.Interface()
|
cli_args = cli.Interface()
|
||||||
ntfy_server = cli_args.argv_1()
|
start({
|
||||||
parsed = cli_args.parsed_args()
|
"cpu_temp_check_disabled": cli_args.disable_cpu_temp,
|
||||||
if ntfy_server:
|
"cpu_critical_temp": cli_args.cpu_temp_critical,
|
||||||
start(
|
"cpu_warning_temp": cli_args.cpu_temp_warning,
|
||||||
parsed.cpu_temp_checking,
|
"update_interval": cli_args.update_rate,
|
||||||
parsed.cpu_temp_critical,
|
"ntfy_server_url": cli_args.server_address,
|
||||||
parsed.cpu_temp_warning,
|
})
|
||||||
ntfy_server,
|
|
||||||
parsed.update_rate,
|
|
||||||
)
|
|
@ -1,4 +1,7 @@
|
|||||||
import requests
|
import requests
|
||||||
|
import time
|
||||||
|
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
class Ntfy:
|
class Ntfy:
|
||||||
def __init__(self, server: str):
|
def __init__(self, server: str):
|
||||||
@ -6,6 +9,8 @@ class Ntfy:
|
|||||||
|
|
||||||
def send(self, message: str):
|
def send(self, message: str):
|
||||||
try:
|
try:
|
||||||
|
pretty_date_time = datetime.fromtimestamp(time.time())
|
||||||
|
print(f"[{pretty_date_time}]: {message}")
|
||||||
requests.post(self.server, data=message)
|
requests.post(self.server, data=message)
|
||||||
except Exception as err:
|
except Exception as err:
|
||||||
print(f"Ntfy failed. {err}")
|
print(f"Ntfy failed. {err}")
|
||||||
|
Reference in New Issue
Block a user