mirror of
https://github.com/unixtensor/proxmox-ntfy.git
synced 2025-06-28 08:28:05 +00:00
Split everything into modules
This commit is contained in:
26
src/cpu.py
Normal file
26
src/cpu.py
Normal file
@ -0,0 +1,26 @@
|
|||||||
|
import subprocess
|
||||||
|
import main
|
||||||
|
import ntfy
|
||||||
|
import time
|
||||||
|
import re
|
||||||
|
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
last_cpu_check: float = time.time()
|
||||||
|
|
||||||
|
def temperature() -> Optional[float]:
|
||||||
|
sensors_out = subprocess.check_output(["sensors"]).decode()
|
||||||
|
for line in sensors_out.splitlines():
|
||||||
|
if "Tctl" in line:
|
||||||
|
match = re.search(r"(\d+\.\d+)°C", line)
|
||||||
|
if match:
|
||||||
|
return float(match.group(1))
|
||||||
|
return None
|
||||||
|
|
||||||
|
def temperature_check():
|
||||||
|
cpu_temp = temperature()
|
||||||
|
if cpu_temp and (time.time() - last_cpu_check) > main.last_check_debounce * 1000:
|
||||||
|
if cpu_temp >= main.cpu_critical_temp:
|
||||||
|
ntfy.send(f"🔥 CPU tempature is at critical tempatures! {cpu_temp}")
|
||||||
|
elif cpu_temp >= main.cpu_warning_temp:
|
||||||
|
ntfy.send(f"🌡️ CPU tempature is at a high tempature. {cpu_temp}")
|
76
src/main.py
76
src/main.py
@ -1,60 +1,34 @@
|
|||||||
import subprocess
|
import argparse
|
||||||
import time
|
import time
|
||||||
import re
|
|
||||||
|
import package
|
||||||
|
import cpu
|
||||||
|
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
clock_interval_secs: int = 1
|
clock_interval_secs: int = 1
|
||||||
last_check_debounce: int = 60 # Seconds
|
last_check_debounce: int = 60 # Seconds
|
||||||
cpu_critical_temp: int = 80
|
cpu_critical_temp: int = 80
|
||||||
cpu_warning_temp: int = 70
|
cpu_warning_temp: int = 70
|
||||||
last_check: float = time.time()
|
|
||||||
ntfy_url: str = "10.0.0.69"
|
|
||||||
|
|
||||||
def package_installed(package_name: str) -> Optional[bool]:
|
ntfy_url: Optional[str] = None
|
||||||
try:
|
_ntfy_configure_prompt = """Please configure an ntfy url before starting.
|
||||||
installed = subprocess.run(["dpkg", "-s", package_name], stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0
|
\033[4mExamples:\033[0m
|
||||||
if not installed:
|
\033[32mpython3 main.py --url=10.0.13.37:42069
|
||||||
print(f"Package \"{package_name}\" not installed.")
|
python3 main.py --url=ntfy.domain.com\033[0m"""
|
||||||
return installed
|
|
||||||
except Exception as err:
|
|
||||||
print(err)
|
|
||||||
return None
|
|
||||||
|
|
||||||
def packages_installed(package_list: list[str]) -> bool:
|
def cli_interface() -> bool:
|
||||||
for pkg_name in package_list:
|
...
|
||||||
if not package_installed(pkg_name):
|
|
||||||
return False
|
|
||||||
return True
|
|
||||||
|
|
||||||
def cpu_temperature() -> Optional[float]:
|
def start():
|
||||||
sensors_out = subprocess.check_output(["sensors"]).decode()
|
print(f"Working! {time.time()}")
|
||||||
for line in sensors_out.splitlines():
|
while True:
|
||||||
if "Tctl" in line:
|
cpu.temperature_check()
|
||||||
match = re.search(r"(\d+\.\d+)°C", line)
|
time.sleep(clock_interval_secs)
|
||||||
if match:
|
|
||||||
return float(match.group(1))
|
|
||||||
return None
|
|
||||||
|
|
||||||
def ntfy_send(message: str):
|
|
||||||
try:
|
|
||||||
global last_check
|
|
||||||
last_check = time.time()
|
|
||||||
subprocess.run(["curl", "-d", f"\"{message}\"", ntfy_url], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
||||||
except Exception as err:
|
|
||||||
print(f"Ntfy failed. {err}")
|
|
||||||
|
|
||||||
def clock_check_cpu():
|
|
||||||
cpu_temp = cpu_temperature()
|
|
||||||
if cpu_temp and (time.time() - last_check) > last_check_debounce * 1000:
|
|
||||||
if cpu_temp >= cpu_critical_temp:
|
|
||||||
ntfy_send(f"🔥 CPU tempature is at critical tempatures! {cpu_temp}")
|
|
||||||
elif cpu_temp >= cpu_warning_temp:
|
|
||||||
ntfy_send(f"🌡️ CPU tempature is at a high tempature. {cpu_temp}")
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
if packages_installed(["lm-sensors", "ntfy", "curl"]):
|
if package.installed_list(["lm-sensors", "ntfy", "curl"]):
|
||||||
print(f"Working! {time.time()}")
|
if cli_interface():
|
||||||
while True:
|
start()
|
||||||
clock_check_cpu()
|
else:
|
||||||
time.sleep(clock_interval_secs)
|
print(_ntfy_configure_prompt)
|
16
src/ntfy.py
Normal file
16
src/ntfy.py
Normal file
@ -0,0 +1,16 @@
|
|||||||
|
import subprocess
|
||||||
|
import main
|
||||||
|
import time
|
||||||
|
|
||||||
|
last_ntfy_send: float = time.time()
|
||||||
|
|
||||||
|
def send(message: str):
|
||||||
|
try:
|
||||||
|
if main.ntfy_url:
|
||||||
|
global last_ntfy_send
|
||||||
|
last_ntfy_send = time.time()
|
||||||
|
subprocess.run(["curl", "-d", f"\"{message}\"", main.ntfy_url], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||||
|
else:
|
||||||
|
print("Ntfy send: url is not configured.")
|
||||||
|
except Exception as err:
|
||||||
|
print(f"Ntfy failed. {err}")
|
19
src/package.py
Normal file
19
src/package.py
Normal file
@ -0,0 +1,19 @@
|
|||||||
|
import subprocess
|
||||||
|
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
def installed(package_name: str) -> Optional[bool]:
|
||||||
|
try:
|
||||||
|
installed = subprocess.run(["dpkg", "-s", package_name], stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0
|
||||||
|
if not installed:
|
||||||
|
print(f"Package \"{package_name}\" not installed.")
|
||||||
|
return installed
|
||||||
|
except Exception as err:
|
||||||
|
print(err)
|
||||||
|
return None
|
||||||
|
|
||||||
|
def installed_list(package_name_list: list[str]) -> bool:
|
||||||
|
for pkg_name in package_name_list:
|
||||||
|
if not installed(pkg_name):
|
||||||
|
return False
|
||||||
|
return True
|
Reference in New Issue
Block a user