Auto install packages

Moved subprocess method to the helper module

lnxlink/__main__.py

@@ -8,7 +8,6 @@
 import logging
 import argparse
 import platform
-import subprocess
 import importlib.metadata
 import ssl
 
@@ -59,20 +58,6 @@ def __init__(self, config_path):
         self.client = mqtt.Client()
         self.setup_mqtt()
 
-    def subprocess(self, command):
-        """Global subprocess command"""
-        result = subprocess.run(
-            command,
-            shell=True,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            check=False,
-            timeout=3,
-        )
-        stdout = result.stdout.decode("UTF-8").strip()
-        returncode = result.returncode
-        return stdout, returncode
-
     def publish_monitor_data(self, topic, pub_data):
         """Publish info data to mqtt in the correct format"""
         # logger.info(topic, pub_data, type(pub_data))

lnxlink/modules/bash.py

@@ -1,4 +1,5 @@
 """Run a terminal command"""
+from .scripts.helpers import syscommand
 
 
 class Addon:
@@ -20,5 +21,5 @@ def exposed_controls(self):
 
     def start_control(self, topic, data):
         """Control system"""
-        stdout, _ = self.lnxlink.subprocess(data)
+        stdout, _, _ = syscommand(data)
         return stdout

lnxlink/modules/battery.py

@@ -1,5 +1,6 @@
 """Gets the battery information of connected devices"""
 import jc
+from .scripts.helpers import syscommand
 
 
 class Addon:
@@ -25,7 +26,7 @@ def exposed_controls(self):
 
     def get_info(self):
         """Gather information from the system"""
-        stdout, _ = self.lnxlink.subprocess("upower --dump")
+        stdout, _, _ = syscommand("upower --dump")
         upower_json = jc.parse("upower", stdout)
 
         devices = {"status": None}

lnxlink/modules/boot_select.py

@@ -1,5 +1,6 @@
 """Selects the OS to boot from on GRUB at the next restart"""
 import re
+from .scripts.helpers import syscommand
 
 
 class Addon:
@@ -13,7 +14,7 @@ def __init__(self, lnxlink=None):
 
     def get_info(self):
         """Gather information from the system"""
-        stdout, _ = self.lnxlink.subprocess("grub-editenv list")
+        stdout, _, _ = syscommand("grub-editenv list")
         nextentry_pattern = re.compile(r"^next_entry=(\d+)")
         nextentry_match = re.match(nextentry_pattern, stdout)
         entry_ind = 0
@@ -35,7 +36,7 @@ def exposed_controls(self):
     def start_control(self, topic, data):
         """Control system"""
         ind = self.options.index(data)
-        self.lnxlink.subprocess(f"sudo grub-reboot {ind}")
+        syscommand(f"sudo grub-reboot {ind}")
 
     def _get_grub_entries(self):
         """Get the grub entities in the correct order"""

lnxlink/modules/brightness.py

@@ -2,6 +2,7 @@
 import subprocess
 import re
 import logging
+from .scripts.helpers import syscommand
 
 logger = logging.getLogger("lnxlink")
 
@@ -68,7 +69,7 @@ def start_control(self, topic, data):
 
     def _get_displays(self):
         """Get all the displays"""
-        stdout, _ = self.lnxlink.subprocess(
+        stdout, _, _ = syscommand(
             "xrandr --verbose --current",
         )
         pattern = re.compile(

lnxlink/modules/gpu.py

@@ -3,8 +3,7 @@
 import math
 import logging
 from shutil import which
-import pyamdgpuinfo
-import nvsmi
+from .scripts.helpers import import_install_package, syscommand
 
 logger = logging.getLogger("lnxlink")
 
@@ -16,21 +15,28 @@ def __init__(self, lnxlink):
         """Setup addon"""
         self.name = "GPU"
         self.lnxlink = lnxlink
-        self.gpu_ids = {"amd": pyamdgpuinfo.detect_gpus()}
+        self._requirements()
+        self.gpu_ids = {"amd": self.lib["amd"].detect_gpus()}
         if which("nvidia-smi") is not None:
             try:
-                self.gpu_ids["nvidia"] = len(list(nvsmi.get_gpus()))
+                self.gpu_ids["nvidia"] = len(list(self.lib["nvidia"].get_gpus()))
             except Exception as err:
                 logger.error("Found nvidia-smi, but there was an error: %s", err)
                 self.gpu_ids["nvidia"] = 0
         else:
             self.gpu_ids["nvidia"] = 0
 
+    def _requirements(self):
+        self.lib = {
+            "amd": import_install_package("pyamdgpuinfo", ">=2.1.4"),
+            "nvidia": import_install_package("nvsmi", ">=0.4.2"),
+        }
+
     def get_info(self):
         """Gather information from the system"""
         gpus = {}
         for gpu_id in range(self.gpu_ids["amd"]):
-            amd_gpu = pyamdgpuinfo.get_gpu(gpu_id)
+            amd_gpu = self.lib["amd"].get_gpu(gpu_id)
             gpus[f"amd_{gpu_id}"] = {
                 "Name": amd_gpu.name,
                 "VRAM usage": amd_gpu.query_vram_usage(),
@@ -41,7 +47,7 @@ def get_info(self):
                 "Voltage": amd_gpu.query_graphics_voltage(),
             }
         for gpu_id in range(self.gpu_ids["nvidia"]):
-            nvidia_gpu = list(nvsmi.get_gpus())[gpu_id]
+            nvidia_gpu = list(self.lib["nvidia"].get_gpus())[gpu_id]
             gpu_util = nvidia_gpu.gpu_util
             gpu_util = self._older_gpu_load(gpu_id, gpu_util)
             gpus[f"nvidia_{gpu_id}"] = {
@@ -57,7 +63,7 @@ def _older_gpu_load(self, gpu_id, gpu_util):
         if math.isnan(gpu_util):
             gpu_util = None
             if which("nvidia-settings") is not None:
-                settings_out, _ = self.lnxlink.subprocess(
+                settings_out, _, _ = syscommand(
                     f"nvidia-settings -q '[gpu:{gpu_id}]/GPUUtilization'"
                 )
                 match = re.findall(r"graphics=(\d+)", settings_out)

lnxlink/modules/keep_alive.py

@@ -1,5 +1,6 @@
 """Keeps display on"""
 import re
+from .scripts.helpers import syscommand
 
 
 class Addon:
@@ -25,10 +26,10 @@ def get_info(self):
         enabled_list = []
 
         # Check if Gnome Idle Time is active
-        stdout_dim, _ = self.lnxlink.subprocess(
+        stdout_dim, _, _ = syscommand(
             "gsettings get org.gnome.desktop.session idle-delay"
         )
-        stdout_suspend, _ = self.lnxlink.subprocess(
+        stdout_suspend, _, _ = syscommand(
             "gsettings get org.gnome.settings-daemon.plugins.power sleep-inactive-ac-type"
         )
         if stdout_dim != "":
@@ -37,7 +38,7 @@ def get_info(self):
                 enabled_list.append("nothing" in stdout_suspend)
 
         # Check if DPMS is active
-        stdout_xset, _ = self.lnxlink.subprocess("xset q")
+        stdout_xset, _, _ = syscommand("xset q")
         xset_pattern = re.compile(r"Standby: (\d+)\s+Suspend: (\d+)\s+Off: (\d+)")
         xset_match = re.findall(xset_pattern, stdout_xset)
         for nums in xset_match:
@@ -50,22 +51,22 @@ def get_info(self):
     def start_control(self, topic, data):
         """Control system"""
         if data.lower() == "off":
-            self.lnxlink.subprocess(
+            syscommand(
                 "gsettings set org.gnome.desktop.session idle-delay 600",
             )
-            self.lnxlink.subprocess(
+            syscommand(
                 'gsettings set org.gnome.settings-daemon.plugins.power sleep-inactive-ac-type "suspend"',
             )
-            self.lnxlink.subprocess(
+            syscommand(
                 "xset +dpms",
             )
         elif data.lower() == "on":
-            self.lnxlink.subprocess(
+            syscommand(
                 "gsettings set org.gnome.desktop.session idle-delay 0",
             )
-            self.lnxlink.subprocess(
+            syscommand(
                 'gsettings set org.gnome.settings-daemon.plugins.power sleep-inactive-ac-type "nothing"',
             )
-            self.lnxlink.subprocess(
+            syscommand(
                 "xset -dpms",
             )

lnxlink/modules/microphone_used.py

@@ -2,6 +2,7 @@
 import glob
 import json
 import re
+from .scripts.helpers import syscommand
 
 
 class Addon:
@@ -12,7 +13,7 @@ def __init__(self, lnxlink):
         self.name = "Microphone used"
         self.lnxlink = lnxlink
 
-        _, returncode = self.lnxlink.subprocess(
+        _, _, returncode = syscommand(
             "which pactl && pactl -f json list short source-outputs",
         )
         self.use_pactl = False
@@ -22,7 +23,7 @@ def __init__(self, lnxlink):
     def get_info(self):
         """Gather information from the system"""
         if self.use_pactl:
-            stdout, _ = self.lnxlink.subprocess(
+            stdout, _, _ = syscommand(
                 "pactl -f json list short source-outputs",
             )
             # Replace 0,00 values with 0.00

lnxlink/modules/power_profile.py

@@ -1,5 +1,6 @@
 """Selects the Power Profile"""
 import re
+from .scripts.helpers import syscommand
 
 
 class Addon:
@@ -13,7 +14,7 @@ def __init__(self, lnxlink=None):
 
     def get_info(self):
         """Gather information from the system"""
-        stdout, _ = self.lnxlink.subprocess("powerprofilesctl get")
+        stdout, _, _ = syscommand("powerprofilesctl get")
         return stdout
 
     def exposed_controls(self):
@@ -28,13 +29,13 @@ def exposed_controls(self):
 
     def start_control(self, topic, data):
         """Control system"""
-        self.lnxlink.subprocess(f"powerprofilesctl set {data}")
+        syscommand(f"powerprofilesctl set {data}")
 
     def _get_power_profiles(self):
         """Get the power profiles in the correct order"""
         profiles_pattern = re.compile(r"([\w-]+):\n")
 
-        stdout, _ = self.lnxlink.subprocess("powerprofilesctl list")
+        stdout, _, _ = syscommand("powerprofilesctl list")
         profiles = re.findall(profiles_pattern, stdout)
 
         return profiles

lnxlink/modules/screen_onoff.py

@@ -1,5 +1,6 @@
 """Turns on or off the screen"""
 import subprocess
+from .scripts.helpers import syscommand
 
 
 class Addon:
@@ -21,7 +22,7 @@ def exposed_controls(self):
 
     def get_info(self):
         """Gather information from the system"""
-        stdout, _ = self.lnxlink.subprocess(
+        stdout, _, _ = syscommand(
             "xset q | grep -i 'monitor is'",
         )
         results = stdout.lower()

lnxlink/modules/screenshot.py

@@ -1,8 +1,6 @@
 """Shows an image of the desktop as a camera entity"""
 import base64
-from mss import mss
-import numpy as np
-import cv2
+from .scripts.helpers import import_install_package
 
 
 class Addon:
@@ -12,14 +10,22 @@ def __init__(self, lnxlink):
         """Setup addon"""
         self.name = "Screenshot"
         self.run = False
+        self._requirements()
+
+    def _requirements(self):
+        self.lib = {
+            "cv2": import_install_package("opencv-python", ">=4.7.0.68", "cv2"),
+            "mss": import_install_package("mss", ">=7.0.1"),
+            "np": import_install_package("numpy", ">=1.24.0"),
+        }
 
     def get_camera_frame(self):
         """Convert screen image to Base64 text"""
         if self.run:
-            with mss() as sct:
+            with self.lib["mss"].mss() as sct:
                 sct_img = sct.grab(sct.monitors[1])
-                frame = np.array(sct_img)
-            _, buffer = cv2.imencode(".jpg", frame)
+                frame = self.lib["np"].array(sct_img)
+            _, buffer = self.lib["cv2"].imencode(".jpg", frame)
             frame = base64.b64encode(buffer)
             return frame
         return None

lnxlink/modules/scripts/helpers.py

@@ -0,0 +1,41 @@
+"""A collection of helper functions"""
+import sys
+import logging
+import subprocess
+
+logger = logging.getLogger("lnxlink")
+
+
+def syscommand(command):
+    """Global subprocess command"""
+    if isinstance(command, list):
+        command = " ".join(command)
+    result = subprocess.run(
+        command,
+        shell=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=False,
+        timeout=3,
+    )
+    stderr = result.stderr.decode("UTF-8").strip()
+    stdout = result.stdout.decode("UTF-8").strip()
+    returncode = result.returncode
+    return stdout, stderr, returncode
+
+
+def import_install_package(package, version="", syspackage=None):
+    """Imports a system package and if it doesn't exist, it gets installed"""
+    if syspackage is None:
+        syspackage = package
+    try:
+        return __import__(syspackage)
+    except ImportError:
+        package_version = package + version
+        logger.error("Package %s is not installed, installing now...", package_version)
+        args = [sys.executable, "-m", "pip", "install", "--quiet", package_version]
+        _, stderr, returncode = syscommand(args)
+        if returncode != 0:
+            logger.error("Can't install package %s: %s", package, stderr)
+            return None
+    return __import__(syspackage)