mirror of
https://github.com/unixtensor/proxmox-ntfy.git
synced 2025-06-28 08:48:04 +00:00
More configuration and packages_installed function
This commit is contained in:
35
src/main.py
35
src/main.py
@ -4,14 +4,16 @@ import re
|
|||||||
|
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
clock_interval_secs = 1
|
clock_interval_secs: int = 1
|
||||||
last_check = time.time()
|
last_check_debounce: int = 60 # Seconds
|
||||||
cpu_critical_temp = 80
|
cpu_critical_temp: int = 80
|
||||||
cpu_warning_temp = 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]:
|
def package_installed(package_name: str) -> Optional[bool]:
|
||||||
try:
|
try:
|
||||||
installed = subprocess.run(["pacman", "-Q", package_name], stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0
|
installed = subprocess.run(["dpkg", "-s", package_name], stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0
|
||||||
if not installed:
|
if not installed:
|
||||||
print(f"Package \"{package_name}\" not installed.")
|
print(f"Package \"{package_name}\" not installed.")
|
||||||
return installed
|
return installed
|
||||||
@ -19,6 +21,12 @@ def package_installed(package_name: str) -> Optional[bool]:
|
|||||||
print(err)
|
print(err)
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
def packages_installed(package_list: list[str]) -> bool:
|
||||||
|
for pkg_name in package_list:
|
||||||
|
if not package_installed(pkg_name):
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
def cpu_temperature() -> Optional[float]:
|
def cpu_temperature() -> Optional[float]:
|
||||||
sensors_out = subprocess.check_output(["sensors"]).decode()
|
sensors_out = subprocess.check_output(["sensors"]).decode()
|
||||||
for line in sensors_out.splitlines():
|
for line in sensors_out.splitlines():
|
||||||
@ -28,15 +36,24 @@ def cpu_temperature() -> Optional[float]:
|
|||||||
return float(match.group(1))
|
return float(match.group(1))
|
||||||
return None
|
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():
|
def clock_check_cpu():
|
||||||
cpu_temp = cpu_temperature()
|
cpu_temp = cpu_temperature()
|
||||||
if cpu_temp and (time.time() - last_check) > 1000:
|
if cpu_temp and (time.time() - last_check) > last_check_debounce * 1000:
|
||||||
if cpu_temp >= cpu_critical_temp:
|
if cpu_temp >= cpu_critical_temp:
|
||||||
print("")
|
ntfy_send(f"🔥 CPU tempature is at critical tempatures! {cpu_temp}")
|
||||||
elif cpu_temp >= cpu_warning_temp:
|
elif cpu_temp >= cpu_warning_temp:
|
||||||
print("")
|
ntfy_send(f"🌡️ CPU tempature is at a high tempature. {cpu_temp}")
|
||||||
|
|
||||||
if package_installed("lm_sensors") and package_installed("ntfy"):
|
if packages_installed(["lm-sensors", "ntfy", "curl"]):
|
||||||
|
print(f"Working! {time.time()}")
|
||||||
while True:
|
while True:
|
||||||
clock_check_cpu()
|
clock_check_cpu()
|
||||||
time.sleep(clock_interval_secs)
|
time.sleep(clock_interval_secs)
|
Reference in New Issue
Block a user