feat: add utils function to check for a specific service instance

Signed-off-by: Dominik Willner <th33xitus@gmail.com>

kiauh/components/klipper/klipper_remove.py

@@ -17,6 +17,7 @@ from core.instance_manager.instance_manager import InstanceManager
 from utils.fs_utils import remove_file
 from utils.input_utils import get_selection_input
 from utils.logger import Logger
+from utils.sys_utils import cmd_sysctl_manage
 
 
 def run_klipper_removal(
@@ -92,7 +93,7 @@ def remove_instances(
         instance_manager.disable_instance()
         instance_manager.delete_instance()
 
-    instance_manager.reload_daemon()
+    cmd_sysctl_manage("daemon-reload")
 
 
 def remove_klipper_dir() -> None:

kiauh/components/klipper/klipper_setup.py

@@ -41,6 +41,7 @@ from utils.git_utils import git_clone_wrapper, git_pull_wrapper
 from utils.input_utils import get_confirm
 from utils.logger import Logger
 from utils.sys_utils import (
+    cmd_sysctl_manage,
     create_python_venv,
     install_python_requirements,
     parse_packages_from_file,
@@ -92,7 +93,7 @@ def install_klipper() -> None:
             if count == install_count:
                 break
 
-        kl_im.reload_daemon()
+        cmd_sysctl_manage("daemon-reload")
 
     except Exception as e:
         Logger.print_error(e)

kiauh/components/moonraker/moonraker_remove.py

@@ -18,6 +18,7 @@ from core.instance_manager.instance_manager import InstanceManager
 from utils.fs_utils import remove_file
 from utils.input_utils import get_selection_input
 from utils.logger import Logger
+from utils.sys_utils import cmd_sysctl_manage
 
 
 def run_moonraker_removal(
@@ -98,7 +99,7 @@ def remove_instances(
         instance_manager.disable_instance()
         instance_manager.delete_instance()
 
-    instance_manager.reload_daemon()
+    cmd_sysctl_manage("daemon-reload")
 
 
 def remove_moonraker_dir() -> None:

kiauh/components/moonraker/moonraker_setup.py

@@ -44,6 +44,7 @@ from utils.input_utils import (
 from utils.logger import Logger
 from utils.sys_utils import (
     check_python_version,
+    cmd_sysctl_manage,
     create_python_venv,
     install_python_requirements,
     parse_packages_from_file,
@@ -110,7 +111,7 @@ def install_moonraker() -> None:
 
             mr_im.start_instance()
 
-        mr_im.reload_daemon()
+        cmd_sysctl_manage("daemon-reload")
 
         # if mainsail is installed, and we installed
         # multiple moonraker instances, we enable mainsails remote mode
@@ -153,7 +154,8 @@ def install_moonraker_packages(moonraker_dir: Path) -> None:
     moonraker_deps = []
 
     if deps_json.exists():
-        moonraker_deps = json.load(deps_json).get("debian", [])
+        with open(deps_json, "r") as deps:
+            moonraker_deps = json.load(deps).get("debian", [])
     elif install_script.exists():
         moonraker_deps = parse_packages_from_file(install_script)
 

kiauh/core/instance_manager/instance_manager.py

@@ -107,7 +107,6 @@ class InstanceManager:
             raise ValueError("current_instance cannot be None")
 
     def enable_instance(self) -> None:
-        Logger.print_status(f"Enabling {self.instance_service_full} ...")
         try:
             cmd_sysctl_service(self.instance_service_full, "enable")
         except subprocess.CalledProcessError as e:
@@ -115,7 +114,6 @@ class InstanceManager:
             Logger.print_error(f"{e}")
 
     def disable_instance(self) -> None:
-        Logger.print_status(f"Disabling {self.instance_service_full} ...")
         try:
             cmd_sysctl_service(self.instance_service_full, "disable")
         except subprocess.CalledProcessError as e:
@@ -123,7 +121,6 @@ class InstanceManager:
             Logger.print_error(f"{e}")
 
     def start_instance(self) -> None:
-        Logger.print_status(f"Starting {self.instance_service_full} ...")
         try:
             cmd_sysctl_service(self.instance_service_full, "start")
         except subprocess.CalledProcessError as e:
@@ -131,7 +128,6 @@ class InstanceManager:
             Logger.print_error(f"{e}")
 
     def restart_instance(self) -> None:
-        Logger.print_status(f"Restarting {self.instance_service_full} ...")
         try:
             cmd_sysctl_service(self.instance_service_full, "restart")
         except subprocess.CalledProcessError as e:
@@ -149,7 +145,6 @@ class InstanceManager:
             self.restart_instance()
 
     def stop_instance(self) -> None:
-        Logger.print_status(f"Stopping {self.instance_service_full} ...")
         try:
             cmd_sysctl_service(self.instance_service_full, "stop")
         except subprocess.CalledProcessError as e:
@@ -162,17 +157,6 @@ class InstanceManager:
             self.current_instance = instance
             self.stop_instance()
 
-    def reload_daemon(self) -> None:
-        Logger.print_status("Reloading systemd manager configuration ...")
-        try:
-            command = ["sudo", "systemctl", "daemon-reload"]
-            if subprocess.run(command, check=True):
-                Logger.print_ok("Systemd manager configuration reloaded")
-        except subprocess.CalledProcessError as e:
-            Logger.print_error("Error reloading systemd manager configuration:")
-            Logger.print_error(f"{e}")
-            raise
-
     def find_instances(self) -> List[T]:
         from utils.common import convert_camelcase_to_kebabcase
 

kiauh/extensions/extensions_menu.py

@@ -12,13 +12,14 @@ import inspect
 import json
 import textwrap
 from pathlib import Path
-from typing import Dict, Optional, Type
+from typing import Dict, List, Optional, Type
 
 from core.menus import Option
 from core.menus.base_menu import BaseMenu
 from extensions import EXTENSION_ROOT
 from extensions.base_extension import BaseExtension
 from utils.constants import COLOR_CYAN, COLOR_YELLOW, RESET_FORMAT
+from utils.logger import Logger
 
 
 # noinspection PyUnusedLocal
@@ -129,11 +130,14 @@ class ExtensionSubmenu(BaseMenu):
         header = f" [ {self.extension.metadata.get('display_name')} ] "
         color = COLOR_YELLOW
         count = 62 - len(color) - len(RESET_FORMAT)
-
+        line_width = 53
-        wrapper = textwrap.TextWrapper(55, initial_indent="| ", subsequent_indent="| ")
+        description: List[str] = self.extension.metadata.get("description", [])
-        lines = wrapper.wrap(self.extension.metadata.get("description"))
+        description_text = Logger.format_content(
-        formatted_lines = [f"{line:<55} |" for line in lines]
+            description,
-        description_text = "\n".join(formatted_lines)
+            line_width,
+            border_left="|",
+            border_right="|",
+        )
 
         menu = textwrap.dedent(
             f"""

kiauh/extensions/gcode_shell_cmd/metadata.json

@@ -4,6 +4,6 @@
     "module": "gcode_shell_cmd_extension",
     "maintained_by": "dw-0",
     "display_name": "G-Code Shell Command",
-    "description": "Allows to run a shell command from gcode."
+    "description": ["Run a shell commands from gcode."]
   }
 }

kiauh/extensions/klipper_backup/metadata.json

@@ -4,7 +4,7 @@
     "module": "klipper_backup_extension",
     "maintained_by": "Staubgeborener",
     "display_name": "Klipper-Backup",
-    "description": "Backup all your klipper files in GitHub",
+    "description": ["Backup all your Klipper files to GitHub"],
     "updates": true
   }
 }

kiauh/extensions/mainsail_theme_installer/metadata.json

@@ -4,6 +4,6 @@
     "module": "mainsail_theme_installer_extension",
     "maintained_by": "dw-0",
     "display_name": "Mainsail Theme Installer",
-    "description": "Install Mainsail Themes maintained by the community."
+    "description": ["Install Mainsail Themes maintained by the Mainsail community."]
   }
 }

kiauh/extensions/pretty_gcode/metadata.json

@@ -4,7 +4,7 @@
     "module": "pretty_gcode_extension",
     "maintained_by": "Kragrathea",
     "display_name": "PrettyGCode for Klipper",
-    "description": "3D G-Code viewer for Klipper",
+    "description": ["3D G-Code viewer for Klipper"],
     "updates": true
   }
 }

kiauh/extensions/telegram_bot/metadata.json

@@ -4,7 +4,7 @@
     "module": "moonraker_telegram_bot_extension",
     "maintained_by": "nlef",
     "display_name": "Moonraker Telegram Bot",
-    "description": "Allows to control your printer with the Telegram messenger app.",
+    "description": ["Control your printer with the Telegram messenger app."],
     "project_url": "https://github.com/nlef/moonraker-telegram-bot",
     "updates": true
   }

kiauh/extensions/telegram_bot/moonraker_telegram_bot_extension.py

@@ -196,7 +196,7 @@ class TelegramBotExtension(BaseExtension):
             instance_manager.disable_instance()
             instance_manager.delete_instance()
 
-        instance_manager.reload_daemon()
+        cmd_sysctl_manage("daemon-reload")
 
     def _remove_bot_dir(self) -> None:
         if not TELEGRAM_BOT_DIR.exists():

kiauh/utils/common.py

@@ -20,7 +20,7 @@ from utils.constants import (
     RESET_FORMAT,
 )
 from utils.git_utils import get_local_commit, get_remote_commit, get_repo_name
-from utils.logger import Logger
+from utils.logger import DialogType, Logger
 from utils.sys_utils import (
     check_package_install,
     install_system_packages,
@@ -124,3 +124,33 @@ def backup_printer_config_dir():
             source=instance.cfg_dir,
             target=PRINTER_CFG_BACKUP_DIR,
         )
+
+
+def moonraker_exists(name: str = "") -> bool:
+    """
+    Helper method to check if a Moonraker instance exists
+    :param name: Optional name of an installer where the check is performed
+    :return: True if at least one Moonraker instance exists, False otherwise
+    """
+    from components.moonraker.moonraker import Moonraker
+
+    mr_im = InstanceManager(Moonraker)
+    mr_instances: List[Moonraker] = mr_im.instances
+
+    info = (
+        f"{name} requires Moonraker to be installed"
+        if name
+        else "A Moonraker installation is required"
+    )
+
+    if not mr_instances:
+        Logger.print_dialog(
+            DialogType.WARNING,
+            [
+                "No Moonraker instances found!",
+                f"{info}. Please install Moonraker first!",
+            ],
+            end="",
+        )
+        return False
+    return True

kiauh/utils/input_utils.py

@@ -6,7 +6,7 @@
 #                                                                         #
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
-
+import re
 from typing import List, Union
 
 from utils import INVALID_CHOICE
@@ -84,6 +84,7 @@ def get_number_input(
 
 def get_string_input(
     question: str,
+    regex: Union[str, None] = None,
     exclude: Union[List, None] = None,
     allow_special_chars=False,
     default=None,
@@ -91,6 +92,7 @@ def get_string_input(
     """
     Helper method to get a string input from the user
     :param question: The question to display
+    :param regex: An optional regex pattern to validate the input against
     :param exclude: List of strings which are not allowed
     :param allow_special_chars: Wheter to allow special characters in the input
     :param default: Optional default value
@@ -98,11 +100,18 @@ def get_string_input(
     """
     _exclude = [] if exclude is None else exclude
     _question = format_question(question, default)
+    _pattern = re.compile(regex) if regex is not None else None
     while True:
         _input = input(_question)
 
+        print(_input)
+
         if _input.lower() in _exclude:
             Logger.print_error("This value is already in use/reserved.")
+        elif default is not None and _input == "":
+            return default
+        elif _pattern is not None and _pattern.match(_input):
+            return _input
         elif allow_special_chars:
             return _input
         elif not allow_special_chars and _input.isalnum():

kiauh/utils/logger.py

@@ -87,6 +87,7 @@ class Logger:
     def print_dialog(
         title: DialogType,
         content: List[str],
+        center_content: bool = False,
         custom_title: str = None,
         custom_color: DialogCustomColor = None,
         end: str = "\n",
@@ -94,7 +95,7 @@ class Logger:
         dialog_color = Logger._get_dialog_color(title, custom_color)
         dialog_title = Logger._get_dialog_title(title, custom_title)
         dialog_title_formatted = Logger._format_dialog_title(dialog_title)
-        dialog_content = Logger._format_dialog_content(content, LINE_WIDTH)
+        dialog_content = Logger.format_content(content, LINE_WIDTH, center_content)
         top = Logger._format_top_border(dialog_color)
         bottom = Logger._format_bottom_border()
 
@@ -140,9 +141,13 @@ class Logger:
             return "\n"
 
     @staticmethod
-    def _format_dialog_content(content: List[str], line_width: int) -> str:
+    def format_content(
-        border_left = "┃"
+        content: List[str],
-        border_right = "┃"
+        line_width: int,
+        center_content: bool = False,
+        border_left: str = "┃",
+        border_right: str = "┃",
+    ) -> str:
         wrapper = textwrap.TextWrapper(line_width)
 
         lines = []
@@ -155,7 +160,13 @@ class Logger:
             if c == "\n\n" and i < len(content) - 1:
                 lines.append(" " * line_width)
 
-        formatted_lines = [
+        if not center_content:
-            f"{border_left} {line:<{line_width}} {border_right}" for line in lines
+            formatted_lines = [
-        ]
+                f"{border_left} {line:<{line_width}} {border_right}" for line in lines
+            ]
+        else:
+            formatted_lines = [
+                f"{border_left} {line:^{line_width}} {border_right}" for line in lines
+            ]
+
         return "\n".join(formatted_lines)

kiauh/utils/sys_utils.py

@@ -8,6 +8,7 @@
 # ======================================================================= #
 
 import os
+import re
 import select
 import shutil
 import socket
@@ -20,6 +21,7 @@ from pathlib import Path
 from subprocess import DEVNULL, PIPE, CalledProcessError, Popen, run
 from typing import List, Literal
 
+from utils.constants import SYSTEMD
 from utils.fs_utils import check_file_exist
 from utils.input_utils import get_confirm
 from utils.logger import Logger
@@ -73,7 +75,6 @@ def parse_packages_from_file(source_file: Path) -> List[str]:
     """
 
     packages = []
-    print("Reading dependencies...")
     with open(source_file, "r") as file:
         for line in file:
             line = line.strip()
@@ -365,6 +366,23 @@ def cmd_sysctl_manage(action: SysCtlManageAction) -> None:
         raise
 
 
+def service_instance_exists(name: str, exclude: List[str] = None) -> bool:
+    """
+    Checks if a systemd service instance exists.
+    :param name: the service name
+    :param exclude: List of strings of service names to exclude
+    :return: True if the service exists, False otherwise
+    """
+    exclude = exclude or []
+    pattern = re.compile(f"^{name}(-[0-9a-zA-Z]+)?.service$")
+    service_list = [
+        Path(SYSTEMD, service)
+        for service in SYSTEMD.iterdir()
+        if pattern.search(service.name) and not any(s in service.name for s in exclude)
+    ]
+    return any(service_list)
+
+
 def log_process(process: Popen) -> None:
     """
     Helper method to print stdout of a process in near realtime to the console.