Replace the sensors subprocess with psutil for the CPU tempature

README.md

@@ -6,9 +6,9 @@ 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):

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

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

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,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 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):

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

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)