int instead of math.floor for the CPU tempature

README.md

@@ -2,11 +2,13 @@
 ### Monitor your proxmox datacenter with phone notifications
 <img src="docs/IMG_20250609_174259.jpg" width="500"/>
 
+Right now only CPU tempatures are monitored.
+
 # Prerequisites
 * [Ntfy](https://ntfy.sh/)
-* [Proxmoxer](https://pypi.org/project/proxmoxer/)
 * 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
 This example will utilize [tmux](https://github.com/tmux/tmux/wiki):
@@ -28,7 +30,7 @@ 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.
 
 ## Updating
-You can update the script without cloning the project:
+You can update the script without cloning the project using git:
 ```
-git pull --depth=1 --rebase --force
+git pull
 ```

src/address.py

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

src/cli.py

@@ -2,10 +2,31 @@ import argparse
 
 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(
 			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")
@@ -13,9 +34,13 @@ def Interface():
 
 		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-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-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")
@@ -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.")
 
-	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
+		}

src/command.py

@@ -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()

src/cpu.py

@@ -1,10 +1,8 @@
-import subprocess
+import psutil
 import time
-import math
-import re
 
+from typing  import Optional, Callable
 from print_t import print_t
-from typing  import Optional
 from ntfy    import Ntfy
 
 _init_run_critical: bool  = True
@@ -14,8 +12,7 @@ _time_now:          float = time.time()
 last_cpu_check_warning:  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:
 	timeout_check_critical: int = 600 # Seconds
@@ -25,16 +22,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 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
 
 	def ntfy_check(self):
@@ -45,7 +41,8 @@ class Tempature:
 			global last_cpu_check_warning
 			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)):
 				_init_run_warning = False
 				last_cpu_check_warning = time.time()
@@ -55,4 +52,4 @@ class Tempature:
 				last_cpu_check_crticial = time.time()
 				self.ntfy.send(message=f"{cpu_temp} C", title=Tempature.critical_message)
 		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")

src/main.py

@@ -6,7 +6,6 @@ import cpu
 
 from datetime import datetime
 from address  import Address
-from typing   import TypedDict
 from ntfy     import Ntfy
 
 class Prompt:
@@ -33,25 +32,11 @@ Address with a topic:
 \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"""
 
-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:
-	def __init__(self, config: Config):
+	def __init__(self, config: cli.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"]
@@ -60,6 +45,8 @@ class Init:
 		while True:
 			if not self.config["cpu_temp_check_disabled"]:
 				self.monitor_cpu_temp.ntfy_check()
+			if not self.config["daily_notifys_disabled"]:
+				...
 			time.sleep(self.config["update_interval"])
 
 	def __start_notify(self):
@@ -77,29 +64,17 @@ class Init:
 		self.__listen()
 
 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():
-		formatted_address = address.format(cli_args.topic)
+		formatted_address = address.format(interface_args.topic)
 		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:
-		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 command.package_installed("lm-sensors"):
 	main()

src/mem.py

@@ -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)