mirror of
https://github.com/unixtensor/proxmox-ntfy.git
synced 2025-06-27 23:28:05 +00:00
Replace the sensors
subprocess with psutil
for the CPU tempature
This commit is contained in:
@ -6,6 +6,7 @@ def Interface():
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="Proxmox monitoring",
|
||||
description="Proxmox monitoring tool for phone notifications using ntfy.sh")
|
||||
|
||||
parser.add_argument("server_address_no_topic", help="The ntfy server address.")
|
||||
|
||||
parser.add_argument("-t", "--topic", default="proxmox", help="The ntfy topic name that notifications will be sent to. Default = proxmox")
|
||||
@ -16,6 +17,8 @@ def Interface():
|
||||
parser.add_argument("--disable-cpu-temp", action="store_true", help="Disable notifications for CPU tempature.")
|
||||
parser.add_argument("--disable-ntfy-logs", action="store_true", help="Disable logging ntfy activity to the output.")
|
||||
|
||||
parser.add_argument("--cpu-temp-zone", default="k10temp", help="The tempature zone for getting CPU info. default = k10temp")
|
||||
parser.add_argument("--cpu-temp-zone-label", default="Tctl", help="The label for getting the current CPU tempature. default = Tctl")
|
||||
parser.add_argument("--cpu-temp-warning", type=int, default=cpu.Tempature.thermal_warn_c, help=f"CPU tempature for the warning alert. default = {cpu.Tempature.thermal_warn_c}")
|
||||
parser.add_argument("--cpu-temp-warning-timeout", type=int, default=cpu.Tempature.timeout_check_warn, help=f"Timeout in seconds until another CPU tempature related notification can be pushed. default = {cpu.Tempature.timeout_check_warn}")
|
||||
parser.add_argument("--cpu-temp-warning-message", default=cpu.Tempature.warning_message, help="The notification message if the CPU is at a high tempature. (message) [TEMP] C")
|
||||
|
@ -1,16 +1,4 @@
|
||||
import subprocess
|
||||
|
||||
from typing import Optional
|
||||
|
||||
def package_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(f"\033[31m{err}\033[0m")
|
||||
return None
|
||||
|
||||
def uname() -> str:
|
||||
return subprocess.run(["uname", "-a"], capture_output=True, text=True).stdout.strip()
|
16
src/cpu.py
16
src/cpu.py
@ -1,7 +1,6 @@
|
||||
import subprocess
|
||||
import psutil
|
||||
import time
|
||||
import math
|
||||
import re
|
||||
|
||||
from print_t import print_t
|
||||
from typing import Optional
|
||||
@ -25,16 +24,15 @@ class Tempature:
|
||||
thermal_critical_c: int = 80
|
||||
thermal_warn_c: int = 70
|
||||
|
||||
def __init__(self, ntfy_instance: Ntfy):
|
||||
def __init__(self, ntfy_instance: Ntfy, cpu_temp_zone: str, cpu_temp_zone_label: str):
|
||||
self.ntfy = ntfy_instance
|
||||
self.cpu_temp_zone = cpu_temp_zone
|
||||
self.cpu_temp_zone_label = cpu_temp_zone_label
|
||||
|
||||
def get(self) -> 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))
|
||||
for entry in psutil.sensors_temperatures().get(self.cpu_temp_zone, []):
|
||||
if entry.label == self.cpu_temp_zone_label:
|
||||
return entry.current
|
||||
return None
|
||||
|
||||
def ntfy_check(self):
|
||||
|
17
src/main.py
17
src/main.py
@ -39,11 +39,13 @@ class Config(TypedDict):
|
||||
cpu_temp_warning_timeout: int
|
||||
cpu_temp_warning_message: str
|
||||
cpu_temp_check_disabled: bool
|
||||
cpu_temp_critical: int
|
||||
cpu_warning_temp: int
|
||||
cpu_temp_zone_label: str
|
||||
cpu_temp_zone: str
|
||||
startup_notify_disabled: bool
|
||||
startup_notify_message: str
|
||||
ntfy_logs_disabled: bool
|
||||
cpu_temp_critical: int
|
||||
cpu_warning_temp: int
|
||||
update_interval: int
|
||||
ntfy_server_url: str
|
||||
|
||||
@ -51,7 +53,7 @@ class Init:
|
||||
def __init__(self, config: Config):
|
||||
self.config = config
|
||||
self.ntfy = Ntfy(config["ntfy_server_url"], config["ntfy_logs_disabled"])
|
||||
self.monitor_cpu_temp = cpu.Tempature(self.ntfy)
|
||||
self.monitor_cpu_temp = cpu.Tempature(self.ntfy, config["cpu_temp_zone"], config["cpu_temp_zone_label"])
|
||||
cpu.Tempature.warning_message = config["cpu_temp_warning_message"]
|
||||
cpu.Tempature.timeout_check_warn = config["cpu_temp_warning_timeout"]
|
||||
cpu.Tempature.thermal_warn_c = config["cpu_warning_temp"]
|
||||
@ -89,11 +91,13 @@ def main():
|
||||
"cpu_temp_warning_timeout": cli_args.cpu_temp_warning_timeout,
|
||||
"cpu_temp_warning_message": cli_args.cpu_temp_warning_message,
|
||||
"cpu_temp_check_disabled": cli_args.disable_cpu_temp,
|
||||
"cpu_temp_critical": cli_args.cpu_temp_critical,
|
||||
"cpu_warning_temp": cli_args.cpu_temp_warning,
|
||||
"cpu_temp_zone_label": cli_args.cpu_temp_zone_label,
|
||||
"cpu_temp_zone": cli_args.cpu_temp_zone,
|
||||
"startup_notify_disabled": cli_args.disable_startup_notify,
|
||||
"startup_notify_message": cli_args.startup_notify_message,
|
||||
"ntfy_logs_disabled": cli_args.disable_ntfy_logs,
|
||||
"cpu_temp_critical": cli_args.cpu_temp_critical,
|
||||
"cpu_warning_temp": cli_args.cpu_temp_warning,
|
||||
"update_interval": cli_args.update_rate,
|
||||
"ntfy_server_url": formatted_address
|
||||
}).start()
|
||||
@ -101,5 +105,4 @@ def main():
|
||||
print(Prompt.address_not_valid(cli_args.server_address_no_topic))
|
||||
|
||||
if __name__ == "__main__":
|
||||
if command.package_installed("lm-sensors"):
|
||||
main()
|
||||
main()
|
Reference in New Issue
Block a user