Compare commits

...

12 Commits

7 changed files with 114 additions and 87 deletions

View File

@ -2,11 +2,13 @@
### Monitor your proxmox datacenter with phone notifications ### Monitor your proxmox datacenter with phone notifications
<img src="docs/IMG_20250609_174259.jpg" width="500"/> <img src="docs/IMG_20250609_174259.jpg" width="500"/>
Right now only CPU tempatures are monitored.
# Prerequisites # Prerequisites
* [Ntfy](https://ntfy.sh/) * [Ntfy](https://ntfy.sh/)
* [Proxmoxer](https://pypi.org/project/proxmoxer/)
* python3 * python3
* python3-requests (It's possible to use `curl` instead with Python [subprocess](https://docs.python.org/3/library/subprocess.html)) [*Optional] * python3-requests
* python3-psutil
# Deploying # Deploying
This example will utilize [tmux](https://github.com/tmux/tmux/wiki): This example will utilize [tmux](https://github.com/tmux/tmux/wiki):
@ -26,3 +28,9 @@ python3 ./proxmox-ntfy/src/main.py http://your.domain.com/
``` ```
Detatch from the current tmux session with `<Ctrl b><d>`, the script should now be deployed and running in the background. Detatch from the current tmux session with `<Ctrl b><d>`, the script should now be deployed and running in the background.
## Updating
You can update the script without cloning the project using git:
```
git pull
```

View File

@ -11,4 +11,4 @@ class Address:
addr = self.address addr = self.address
if self.address[len(self.address)-1] != "/": if self.address[len(self.address)-1] != "/":
addr += "/" addr += "/"
return addr + topic return addr + topic.replace(" ", "_")

View File

@ -2,10 +2,31 @@ import argparse
import cpu import cpu
def Interface(): from typing import TypedDict
class Config(TypedDict):
cpu_temp_critical_timeout: int
cpu_temp_critical_message: str
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
daily_notifys_disabled: bool
ntfy_logs_disabled: bool
update_interval: int
ntfy_server_url: str
class Interface:
def __init__(self):
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="Proxmox monitoring", prog="Proxmox monitoring",
description="Proxmox monitoring tool for phone notifications using ntfy.sh") 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("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") parser.add_argument("-t", "--topic", default="proxmox", help="The ntfy topic name that notifications will be sent to. Default = proxmox")
@ -13,9 +34,13 @@ 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-startup-notify", action="store_true", help="Disable the start up notify.") parser.add_argument("--disable-startup-notify", action="store_true", help="Disable the start up notify.")
parser.add_argument("--disable-daily-notifys", action="store_true", help="Disable the daily notifys on the system's highest stats.")
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("--disable-ntfy-logs", action="store_true", help="Disable logging ntfy activity to the output.") 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", 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-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") 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")
@ -26,4 +51,23 @@ def Interface():
parser.add_argument("--startup-notify-message", default="🖥️ Ntfy proxmox monitoring started.", help="The notification message when the program is started.") parser.add_argument("--startup-notify-message", default="🖥️ Ntfy proxmox monitoring started.", help="The notification message when the program is started.")
return parser.parse_args() self.cli_args = parser.parse_args()
def to_config(self, formatted_address: str) -> Config:
return {
"cpu_temp_critical_timeout": self.cli_args.cpu_temp_critical_timeout,
"cpu_temp_critical_message": self.cli_args.cpu_temp_critical_message,
"cpu_temp_warning_timeout": self.cli_args.cpu_temp_warning_timeout,
"cpu_temp_warning_message": self.cli_args.cpu_temp_warning_message,
"cpu_temp_check_disabled": self.cli_args.disable_cpu_temp,
"cpu_temp_critical": self.cli_args.cpu_temp_critical,
"cpu_warning_temp": self.cli_args.cpu_temp_warning,
"cpu_temp_zone_label": self.cli_args.cpu_temp_zone_label,
"cpu_temp_zone": self.cli_args.cpu_temp_zone,
"startup_notify_disabled": self.cli_args.disable_startup_notify,
"daily_notifys_disabled": self.cli_args.disable_daily_notifys,
"startup_notify_message": self.cli_args.startup_notify_message,
"ntfy_logs_disabled": self.cli_args.disable_ntfy_logs,
"update_interval": self.cli_args.update_rate,
"ntfy_server_url": formatted_address
}

View File

@ -1,16 +1,4 @@
import subprocess 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: def uname() -> str:
return subprocess.run(["uname", "-a"], capture_output=True, text=True).stdout.strip() return subprocess.run(["uname", "-a"], capture_output=True, text=True).stdout.strip()

View File

@ -1,10 +1,8 @@
import subprocess import psutil
import time import time
import math
import re
from typing import Optional, Callable
from print_t import print_t from print_t import print_t
from typing import Optional
from ntfy import Ntfy from ntfy import Ntfy
_init_run_critical: bool = True _init_run_critical: bool = True
@ -14,8 +12,7 @@ _time_now: float = time.time()
last_cpu_check_warning: float = _time_now last_cpu_check_warning: float = _time_now
last_cpu_check_crticial: float = _time_now last_cpu_check_crticial: float = _time_now
def timeout_expired(check: float, timeout: int) -> bool: timeout_expired: Callable[[float, int], bool] = lambda c, t: (time.time() - c) > t
return (time.time() - check) > timeout
class Tempature: class Tempature:
timeout_check_critical: int = 600 # Seconds timeout_check_critical: int = 600 # Seconds
@ -25,16 +22,15 @@ class Tempature:
thermal_critical_c: int = 80 thermal_critical_c: int = 80
thermal_warn_c: int = 70 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.ntfy = ntfy_instance
self.cpu_temp_zone = cpu_temp_zone
self.cpu_temp_zone_label = cpu_temp_zone_label
def get(self) -> Optional[float]: def get(self) -> Optional[float]:
sensors_out = subprocess.check_output(["sensors"]).decode() for entry in psutil.sensors_temperatures().get(self.cpu_temp_zone, []):
for line in sensors_out.splitlines(): if entry.label == self.cpu_temp_zone_label:
if "Tctl" in line: return entry.current
match = re.search(r"(\d+\.\d+)°C", line)
if match:
return float(match.group(1))
return None return None
def ntfy_check(self): def ntfy_check(self):
@ -45,7 +41,8 @@ class Tempature:
global last_cpu_check_warning global last_cpu_check_warning
global last_cpu_check_crticial global last_cpu_check_crticial
cpu_temp = math.floor(cpu_temp) cpu_temp = int(cpu_temp)
if cpu_temp >= Tempature.thermal_warn_c and (_init_run_warning or timeout_expired(last_cpu_check_warning, Tempature.timeout_check_warn)): if cpu_temp >= Tempature.thermal_warn_c and (_init_run_warning or timeout_expired(last_cpu_check_warning, Tempature.timeout_check_warn)):
_init_run_warning = False _init_run_warning = False
last_cpu_check_warning = time.time() last_cpu_check_warning = time.time()
@ -55,4 +52,4 @@ class Tempature:
last_cpu_check_crticial = time.time() last_cpu_check_crticial = time.time()
self.ntfy.send(message=f"{cpu_temp} C", title=Tempature.critical_message) self.ntfy.send(message=f"{cpu_temp} C", title=Tempature.critical_message)
else: else:
print_t("\033[31mCannot get a feasible tempature value for the CPU. (lm-sensors)\033[0m") print_t(f"\033[31mCannot get a feasible tempature value for the CPU. cpu_temp_zone={self.cpu_temp_zone} cpu_temp_zone_label={self.cpu_temp_zone_label}\033[0m")

View File

@ -6,7 +6,6 @@ import cpu
from datetime import datetime from datetime import datetime
from address import Address from address import Address
from typing import TypedDict
from ntfy import Ntfy from ntfy import Ntfy
class Prompt: class Prompt:
@ -33,25 +32,11 @@ Address with a topic:
\033[32mhttp://domain.com\033[0m -t|--topic \033[32mexample_topic\033[0m \033[32mhttp://domain.com\033[0m -t|--topic \033[32mexample_topic\033[0m
\033[32mhttps://domain.com\033[0m -t|--topic \033[32mexample_topic\033[0m""" \033[32mhttps://domain.com\033[0m -t|--topic \033[32mexample_topic\033[0m"""
class Config(TypedDict):
cpu_temp_critical_timeout: int
cpu_temp_critical_message: str
cpu_temp_warning_timeout: int
cpu_temp_warning_message: str
cpu_temp_check_disabled: bool
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
class Init: class Init:
def __init__(self, config: Config): def __init__(self, config: cli.Config):
self.config = config self.config = config
self.ntfy = Ntfy(config["ntfy_server_url"], config["ntfy_logs_disabled"]) 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.warning_message = config["cpu_temp_warning_message"]
cpu.Tempature.timeout_check_warn = config["cpu_temp_warning_timeout"] cpu.Tempature.timeout_check_warn = config["cpu_temp_warning_timeout"]
cpu.Tempature.thermal_warn_c = config["cpu_warning_temp"] cpu.Tempature.thermal_warn_c = config["cpu_warning_temp"]
@ -60,6 +45,8 @@ class Init:
while True: while True:
if not self.config["cpu_temp_check_disabled"]: if not self.config["cpu_temp_check_disabled"]:
self.monitor_cpu_temp.ntfy_check() self.monitor_cpu_temp.ntfy_check()
if not self.config["daily_notifys_disabled"]:
...
time.sleep(self.config["update_interval"]) time.sleep(self.config["update_interval"])
def __start_notify(self): def __start_notify(self):
@ -77,29 +64,17 @@ class Init:
self.__listen() self.__listen()
def main(): def main():
cli_args = cli.Interface() interface = cli.Interface()
address = Address(cli_args.server_address_no_topic) interface_args = interface.cli_args
address = Address(interface_args.server_address_no_topic)
if address.is_valid(): if address.is_valid():
formatted_address = address.format(cli_args.topic) formatted_address = address.format(interface_args.topic)
print(Prompt.start(formatted_address)) print(Prompt.start(formatted_address))
Init({ Init(interface.to_config(formatted_address)).start()
"cpu_temp_critical_timeout": cli_args.cpu_temp_critical_timeout,
"cpu_temp_critical_message": cli_args.cpu_temp_critical_message,
"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,
"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()
else: else:
print(Prompt.address_not_valid(cli_args.server_address_no_topic)) print(Prompt.address_not_valid(interface_args.server_address_no_topic))
if __name__ == "__main__": if __name__ == "__main__":
if command.package_installed("lm-sensors"):
main() main()

15
src/mem.py Normal file
View File

@ -0,0 +1,15 @@
import psutil
import math
peak: float = 0
check: int = 3600 # Seconds
def usage() -> float:
global peak
mem = round(psutil.virtual_memory().used / 1048576000, 1)
if mem > peak:
peak = mem
return mem
def total() -> int:
return math.floor(psutil.virtual_memory().total / 1048576000)