mirror of
https://github.com/unixtensor/proxmox-ntfy.git
synced 2025-06-28 03:28:05 +00:00
Remove CPU critical tempature warning
This commit is contained in:
@ -1,7 +1,4 @@
|
|||||||
import argparse
|
import argparse
|
||||||
import sys
|
|
||||||
|
|
||||||
from typing import Optional
|
|
||||||
|
|
||||||
_ntfy_configure_prompt = """\033[4mPlease configure an ntfy url before starting.\033[0m
|
_ntfy_configure_prompt = """\033[4mPlease configure an ntfy url before starting.\033[0m
|
||||||
Examples:
|
Examples:
|
||||||
@ -16,8 +13,7 @@ def Interface():
|
|||||||
|
|
||||||
parser.add_argument("--disable-uptime-notifys", action="store_true", help="Disable uptime notifications.")
|
parser.add_argument("--disable-uptime-notifys", action="store_true", help="Disable uptime notifications.")
|
||||||
parser.add_argument("--disable-cpu-temp", action="store_true", help="Disable notifications for CPU tempature.")
|
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("--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("--update-rate", type=int, default=1, help="How often updates happen in seconds. default = 1")
|
|
||||||
|
|
||||||
return parser.parse_args()
|
return parser.parse_args()
|
18
src/cpu.py
18
src/cpu.py
@ -1,22 +1,20 @@
|
|||||||
import subprocess
|
import subprocess
|
||||||
import time
|
import time
|
||||||
|
import math
|
||||||
import re
|
import re
|
||||||
|
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
from ntfy import Ntfy
|
from ntfy import Ntfy
|
||||||
|
|
||||||
_time_now = time.time()
|
_time_now = time.time()
|
||||||
last_cpu_check_critical: float = _time_now
|
|
||||||
last_cpu_check_warning: float = _time_now
|
last_cpu_check_warning: float = _time_now
|
||||||
|
|
||||||
last_cpu_check: int = 60 # Seconds
|
last_check_debounce: int = 120 # Seconds
|
||||||
|
|
||||||
class Tempature:
|
class Tempature:
|
||||||
cpu_temp_crtitical_message: str = "🔥 CPU tempature is at critical tempatures!"
|
cpu_temp_warning_message: str = "🌡️ CPU is at a high tempature."
|
||||||
cpu_temp_warning_message: str = "🌡️ CPU tempature is at a high tempature."
|
|
||||||
|
|
||||||
def __init__(self, ntfy_instance: Ntfy, cpu_critical_temp: int, cpu_warning_temp: int):
|
def __init__(self, ntfy_instance: Ntfy, cpu_warning_temp: int):
|
||||||
self.cpu_critical_temp = cpu_critical_temp
|
|
||||||
self.cpu_warning_temp = cpu_warning_temp
|
self.cpu_warning_temp = cpu_warning_temp
|
||||||
self.ntfy = ntfy_instance
|
self.ntfy = ntfy_instance
|
||||||
|
|
||||||
@ -30,13 +28,11 @@ class Tempature:
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
def __check_time(self, last_check: float) -> bool:
|
def __check_time(self, last_check: float) -> bool:
|
||||||
return (time.time() - last_check) > last_cpu_check * 1000
|
return (time.time() - last_check) > last_check_debounce * 1000
|
||||||
|
|
||||||
def ntfy_check(self):
|
def ntfy_check(self):
|
||||||
cpu_temp = self.get()
|
cpu_temp = self.get()
|
||||||
if cpu_temp:
|
if cpu_temp:
|
||||||
print(f"{cpu_temp}")
|
cpu_temp = math.floor(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}")
|
|
||||||
if cpu_temp >= self.cpu_warning_temp and self.__check_time(last_cpu_check_warning):
|
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} C")
|
||||||
|
@ -8,8 +8,7 @@ from datetime import datetime
|
|||||||
from typing import TypedDict
|
from typing import TypedDict
|
||||||
from ntfy import Ntfy
|
from ntfy import Ntfy
|
||||||
|
|
||||||
_pretty_date_time = datetime.fromtimestamp(time.time())
|
_monitoring_prompt = f"""{datetime.fromtimestamp(time.time())}
|
||||||
_monitoring_prompt = f"""{_pretty_date_time}
|
|
||||||
Ntfy monitoring software is now listening.
|
Ntfy monitoring software is now listening.
|
||||||
|
|
||||||
Source code available at:
|
Source code available at:
|
||||||
@ -18,14 +17,13 @@ Source code available at:
|
|||||||
|
|
||||||
class Config(TypedDict):
|
class Config(TypedDict):
|
||||||
cpu_temp_check_disabled: bool
|
cpu_temp_check_disabled: bool
|
||||||
cpu_critical_temp: int
|
|
||||||
cpu_warning_temp: int
|
cpu_warning_temp: int
|
||||||
update_interval: int
|
update_interval: int
|
||||||
ntfy_server_url: str
|
ntfy_server_url: str
|
||||||
|
|
||||||
def start(config: Config):
|
def start(config: Config):
|
||||||
ntfy = Ntfy(config["ntfy_server_url"])
|
ntfy = Ntfy(config["ntfy_server_url"])
|
||||||
ntfy_cpu_temp_monitor = cpu.Tempature(ntfy, config["cpu_critical_temp"], config["cpu_warning_temp"])
|
ntfy_cpu_temp_monitor = cpu.Tempature(ntfy, config["cpu_warning_temp"])
|
||||||
|
|
||||||
print(_monitoring_prompt)
|
print(_monitoring_prompt)
|
||||||
while True:
|
while True:
|
||||||
@ -38,7 +36,6 @@ if __name__ == "__main__":
|
|||||||
cli_args = cli.Interface()
|
cli_args = cli.Interface()
|
||||||
start({
|
start({
|
||||||
"cpu_temp_check_disabled": cli_args.disable_cpu_temp,
|
"cpu_temp_check_disabled": cli_args.disable_cpu_temp,
|
||||||
"cpu_critical_temp": cli_args.cpu_temp_critical,
|
|
||||||
"cpu_warning_temp": cli_args.cpu_temp_warning,
|
"cpu_warning_temp": cli_args.cpu_temp_warning,
|
||||||
"update_interval": cli_args.update_rate,
|
"update_interval": cli_args.update_rate,
|
||||||
"ntfy_server_url": cli_args.server_address,
|
"ntfy_server_url": cli_args.server_address,
|
||||||
|
@ -8,9 +8,8 @@ class Ntfy:
|
|||||||
self.server = server
|
self.server = server
|
||||||
|
|
||||||
def send(self, message: str):
|
def send(self, message: str):
|
||||||
|
print(f"[{datetime.fromtimestamp(time.time())}]: {message}")
|
||||||
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