Merge 8c3397ea78 into f2691f33d3

kiauh/components/crowsnest/__init__.py

@@ -1,13 +0,0 @@
-# ======================================================================= #
-#  Copyright (C) 2020 - 2024 Dominik Willner <th33xitus@gmail.com>        #
-#                                                                         #
-#  This file is part of KIAUH - Klipper Installation And Update Helper    #
-#  https://github.com/dw-0/kiauh                                          #
-#                                                                         #
-#  This file may be distributed under the terms of the GNU GPLv3 license  #
-# ======================================================================= #
-
-from pathlib import Path
-
-CROWSNEST_DIR = Path.home().joinpath("crowsnest")
-CROWSNEST_REPO = "https://github.com/mainsail-crew/crowsnest.git"

kiauh/components/crowsnest/crowsnest.py

@@ -1,168 +0,0 @@
-# ======================================================================= #
-#  Copyright (C) 2020 - 2024 Dominik Willner <th33xitus@gmail.com>        #
-#                                                                         #
-#  This file is part of KIAUH - Klipper Installation And Update Helper    #
-#  https://github.com/dw-0/kiauh                                          #
-#                                                                         #
-#  This file may be distributed under the terms of the GNU GPLv3 license  #
-# ======================================================================= #
-from __future__ import annotations
-
-import shutil
-import textwrap
-from pathlib import Path
-from subprocess import run, CalledProcessError
-from typing import List, Dict, Literal, Union
-
-from components.crowsnest import CROWSNEST_REPO, CROWSNEST_DIR
-from components.klipper.klipper import Klipper
-from core.instance_manager.instance_manager import InstanceManager
-from utils.common import get_install_status
-from utils.constants import COLOR_CYAN, RESET_FORMAT, CURRENT_USER
-from utils.git_utils import (
-    git_clone_wrapper,
-    get_repo_name,
-    get_local_commit,
-    get_remote_commit,
-    git_pull_wrapper,
-)
-from utils.input_utils import get_confirm
-from utils.logger import Logger
-from utils.system_utils import (
-    check_package_install,
-    install_system_packages,
-    parse_packages_from_file,
-    update_system_package_lists,
-    control_systemd_service,
-)
-
-
-def install_crowsnest() -> None:
-    # Step 1: Clone crowsnest repo
-    git_clone_wrapper(CROWSNEST_REPO, "master", CROWSNEST_DIR)
-
-    # Step 2: Install dependencies
-    requirements: List[str] = check_package_install(["make"])
-    if requirements:
-        install_system_packages(requirements)
-
-    # Step 3: Check for Multi Instance
-    im = InstanceManager(Klipper)
-    instances: List[Klipper] = im.find_instances()
-
-    if len(instances) > 1:
-        Logger.print_status("Multi instance install detected ...")
-        info = textwrap.dedent("""
-            Crowsnest is NOT designed to support multi instances.
-            A workaround for this is to choose the most used instance as a 'master'
-            Use this instance to set up your 'crowsnest.conf' and steering it's service.
-            Found the following instances:
-            """)[:-1]
-        print(info, end="")
-        for instance in instances:
-            print(f"● {instance.data_dir_name}")
-
-        Logger.print_status("\nLaunching crowsnest's configuration tool ...")
-
-    if not get_confirm("Continue with configuration?", False, allow_go_back=True):
-        Logger.print_info("Installation aborted by user ... Exiting!")
-        return
-
-    config = Path(CROWSNEST_DIR).joinpath("tools/.config")
-    try:
-        run(
-            "make config",
-            cwd=CROWSNEST_DIR,
-            shell=True,
-            check=True,
-        )
-    except CalledProcessError as e:
-        Logger.print_error(f"Something went wrong! Please try again...\n{e}")
-        if config.exists():
-            Path.unlink(config)
-        return
-
-    if not config.exists():
-        Logger.print_error("Generating .config failed, installation aborted")
-        return
-
-    # Step 4: Launch crowsnest installer
-    print(f"{COLOR_CYAN}Installer will prompt you for sudo password!{RESET_FORMAT}")
-    Logger.print_status("Launching crowsnest installer ...")
-    try:
-        run(
-            f"sudo make install BASE_USER={CURRENT_USER}",
-            cwd=CROWSNEST_DIR,
-            shell=True,
-            check=True,
-        )
-    except CalledProcessError as e:
-        Logger.print_error(f"Something went wrong! Please try again...\n{e}")
-        return
-
-
-def update_crowsnest() -> None:
-    try:
-        control_systemd_service("crowsnest", "stop")
-
-        if not CROWSNEST_DIR.exists():
-            git_clone_wrapper(CROWSNEST_REPO, "master", CROWSNEST_DIR)
-        else:
-            Logger.print_status("Updating Crowsnest ...")
-
-            git_pull_wrapper(CROWSNEST_REPO, CROWSNEST_DIR)
-
-            script = CROWSNEST_DIR.joinpath("tools/install.sh")
-            deps = parse_packages_from_file(script)
-            packages = check_package_install(deps)
-            update_system_package_lists(silent=False)
-            install_system_packages(packages)
-
-        control_systemd_service("crowsnest", "restart")
-
-        Logger.print_ok("Crowsnest updated successfully.", end="\n\n")
-    except CalledProcessError as e:
-        Logger.print_error(f"Something went wrong! Please try again...\n{e}")
-        return
-
-
-def get_crowsnest_status() -> (
-    Dict[
-        Literal["status", "status_code", "repo", "local", "remote"],
-        Union[str, int],
-    ]
-):
-    files = [
-        Path("/usr/local/bin/crowsnest"),
-        Path("/etc/logrotate.d/crowsnest"),
-        Path("/etc/systemd/system/crowsnest.service"),
-    ]
-    status = get_install_status(CROWSNEST_DIR, files)
-    return {
-        "status": status.get("status"),
-        "status_code": status.get("status_code"),
-        "repo": get_repo_name(CROWSNEST_DIR),
-        "local": get_local_commit(CROWSNEST_DIR),
-        "remote": get_remote_commit(CROWSNEST_DIR),
-    }
-
-
-def remove_crowsnest() -> None:
-    if not CROWSNEST_DIR.exists():
-        Logger.print_info("Crowsnest does not seem to be installed! Skipping ...")
-        return
-
-    try:
-        run(
-            "make uninstall",
-            cwd=CROWSNEST_DIR,
-            shell=True,
-            check=True,
-        )
-    except CalledProcessError as e:
-        Logger.print_error(f"Something went wrong! Please try again...\n{e}")
-        return
-
-    Logger.print_status("Removing crowsnest directory ...")
-    shutil.rmtree(CROWSNEST_DIR)
-    Logger.print_ok("Directory removed!")

kiauh/components/webui_client/client_config/client_config_remove.py

@@ -16,8 +16,7 @@ from components.klipper.klipper import Klipper
 from components.moonraker.moonraker import Moonraker
 from components.webui_client.base_data import BaseWebClientConfig
 from core.instance_manager.instance_manager import InstanceManager
-from utils.config_utils import remove_config_section
-from utils.filesystem_utils import remove_file
+from utils.filesystem_utils import remove_file, remove_config_section
 from utils.logger import Logger
 
 

kiauh/components/webui_client/client_config/client_config_setup.py

@@ -26,8 +26,11 @@ from components.webui_client.client_utils import (
 
 from core.instance_manager.instance_manager import InstanceManager
 from utils.common import backup_printer_config_dir
-from utils.config_utils import add_config_section, add_config_section_at_top
-from utils.filesystem_utils import create_symlink
+from utils.filesystem_utils import (
+    create_symlink,
+    add_config_section,
+    add_config_section_at_top,
+)
 from utils.git_utils import git_clone_wrapper, git_pull_wrapper
 from utils.input_utils import get_confirm
 from utils.logger import Logger

kiauh/components/webui_client/client_remove.py

@@ -23,10 +23,10 @@ from components.webui_client.client_config.client_config_remove import (
 from components.webui_client.client_utils import backup_mainsail_config_json
 
 from core.instance_manager.instance_manager import InstanceManager
-from utils.config_utils import remove_config_section
 from utils.filesystem_utils import (
     remove_nginx_config,
     remove_nginx_logs,
+    remove_config_section,
 )
 from utils.logger import Logger
 
@@ -49,7 +49,7 @@ def run_client_removal(
         client_name = client.name
         remove_client_dir(client)
         remove_nginx_config(client_name)
-        remove_nginx_logs(client_name, kl_instances)
+        remove_nginx_logs(client_name)
 
         section = f"update_manager {client_name}"
         remove_config_section(section, mr_instances)

kiauh/components/webui_client/client_setup.py

@@ -37,7 +37,6 @@ from core.instance_manager.instance_manager import InstanceManager
 from core.settings.kiauh_settings import KiauhSettings
 from utils import NGINX_SITES_AVAILABLE, NGINX_SITES_ENABLED
 from utils.common import check_install_dependencies
-from utils.config_utils import add_config_section
 from utils.filesystem_utils import (
     unzip,
     copy_upstream_nginx_cfg,
@@ -45,6 +44,7 @@ from utils.filesystem_utils import (
     create_nginx_cfg,
     create_symlink,
     remove_file,
+    add_config_section,
     read_ports_from_nginx_configs,
     is_valid_port,
     get_next_free_port,

kiauh/core/instance_manager/instance_manager.py

@@ -15,7 +15,6 @@ from typing import List, Optional, Union, TypeVar
 from core.instance_manager.base_instance import BaseInstance
 from utils.constants import SYSTEMD
 from utils.logger import Logger
-from utils.system_utils import control_systemd_service
 
 T = TypeVar(name="T", bound=BaseInstance, covariant=True)
 
@@ -109,7 +108,14 @@ class InstanceManager:
     def enable_instance(self) -> None:
         Logger.print_status(f"Enabling {self.instance_service_full} ...")
         try:
-            control_systemd_service(self.instance_service_full, "enable")
+            command = [
+                "sudo",
+                "systemctl",
+                "enable",
+                self.instance_service_full,
+            ]
+            if subprocess.run(command, check=True):
+                Logger.print_ok(f"{self.instance_service_full} enabled.")
         except subprocess.CalledProcessError as e:
             Logger.print_error(f"Error enabling service {self.instance_service_full}:")
             Logger.print_error(f"{e}")
@@ -117,7 +123,14 @@ class InstanceManager:
     def disable_instance(self) -> None:
         Logger.print_status(f"Disabling {self.instance_service_full} ...")
         try:
-            control_systemd_service(self.instance_service_full, "disable")
+            command = [
+                "sudo",
+                "systemctl",
+                "disable",
+                self.instance_service_full,
+            ]
+            if subprocess.run(command, check=True):
+                Logger.print_ok(f"{self.instance_service_full} disabled.")
         except subprocess.CalledProcessError as e:
             Logger.print_error(f"Error disabling {self.instance_service_full}:")
             Logger.print_error(f"{e}")
@@ -125,7 +138,14 @@ class InstanceManager:
     def start_instance(self) -> None:
         Logger.print_status(f"Starting {self.instance_service_full} ...")
         try:
-            control_systemd_service(self.instance_service_full, "start")
+            command = [
+                "sudo",
+                "systemctl",
+                "start",
+                self.instance_service_full,
+            ]
+            if subprocess.run(command, check=True):
+                Logger.print_ok(f"{self.instance_service_full} started.")
         except subprocess.CalledProcessError as e:
             Logger.print_error(f"Error starting {self.instance_service_full}:")
             Logger.print_error(f"{e}")
@@ -133,7 +153,14 @@ class InstanceManager:
     def restart_instance(self) -> None:
         Logger.print_status(f"Restarting {self.instance_service_full} ...")
         try:
-            control_systemd_service(self.instance_service_full, "restart")
+            command = [
+                "sudo",
+                "systemctl",
+                "restart",
+                self.instance_service_full,
+            ]
+            if subprocess.run(command, check=True):
+                Logger.print_ok(f"{self.instance_service_full} restarted.")
         except subprocess.CalledProcessError as e:
             Logger.print_error(f"Error restarting {self.instance_service_full}:")
             Logger.print_error(f"{e}")
@@ -151,7 +178,9 @@ class InstanceManager:
     def stop_instance(self) -> None:
         Logger.print_status(f"Stopping {self.instance_service_full} ...")
         try:
-            control_systemd_service(self.instance_service_full, "stop")
+            command = ["sudo", "systemctl", "stop", self.instance_service_full]
+            if subprocess.run(command, check=True):
+                Logger.print_ok(f"{self.instance_service_full} stopped.")
         except subprocess.CalledProcessError as e:
             Logger.print_error(f"Error stopping {self.instance_service_full}:")
             Logger.print_error(f"{e}")

kiauh/core/menus/install_menu.py

@@ -10,7 +10,6 @@
 import textwrap
 from typing import Type, Optional
 
-from components.crowsnest.crowsnest import install_crowsnest
 from components.klipper import klipper_setup
 from components.moonraker import moonraker_setup
 from components.webui_client import client_setup
@@ -45,7 +44,6 @@ class InstallMenu(BaseMenu):
             "4": Option(method=self.install_fluidd, menu=False),
             "5": Option(method=self.install_mainsail_config, menu=False),
             "6": Option(method=self.install_fluidd_config, menu=False),
-            "9": Option(method=self.install_crowsnest, menu=False),
         }
 
     def print_menu(self):
@@ -90,6 +88,3 @@ class InstallMenu(BaseMenu):
 
     def install_fluidd_config(self, **kwargs):
         client_config_setup.install_client_config(FluiddData())
-
-    def install_crowsnest(self, **kwargs):
-        install_crowsnest()

kiauh/core/menus/main_menu.py

@@ -10,7 +10,6 @@
 import textwrap
 from typing import Type, Optional
 
-from components.crowsnest.crowsnest import get_crowsnest_status
 from components.klipper.klipper_utils import get_klipper_status
 from components.log_uploads.menus.log_upload_menu import LogUploadMenu
 from components.moonraker.moonraker_utils import get_moonraker_status
@@ -91,23 +90,16 @@ class MainMenu(BaseMenu):
         self.ms_status = get_client_status(MainsailData())
         self.fl_status = get_client_status(FluiddData())
         self.cc_status = get_current_client_config([MainsailData(), FluiddData()])
-        self._update_status("cn", get_crowsnest_status)
 
     def _update_status(self, status_name: str, status_fn: callable) -> None:
         status_data = status_fn()
         status = status_data.get("status")
         code = status_data.get("status_code")
-
-        instance_count = status_data.get("instances")
-
-        count: str = ""
-        if instance_count and code == 1:
-            count = f" {instance_count}"
-
+        instances = f" {status_data.get('instances')}" if code == 1 else ""
         setattr(
             self,
             f"{status_name}_status",
-            self._format_status_by_code(code, status, count),
+            self._format_status_by_code(code, status, instances),
         )
         setattr(
             self,

kiauh/core/menus/remove_menu.py

@@ -10,7 +10,6 @@
 import textwrap
 from typing import Type, Optional
 
-from components.crowsnest.crowsnest import remove_crowsnest
 from components.klipper.menus.klipper_remove_menu import KlipperRemoveMenu
 from components.moonraker.menus.moonraker_remove_menu import (
     MoonrakerRemoveMenu,
@@ -43,7 +42,6 @@ class RemoveMenu(BaseMenu):
             "2": Option(method=self.remove_moonraker, menu=True),
             "3": Option(method=self.remove_mainsail, menu=True),
             "4": Option(method=self.remove_fluidd, menu=True),
-            "6": Option(method=self.remove_crowsnest, menu=True),
         }
 
     def print_menu(self):
@@ -80,6 +78,3 @@ class RemoveMenu(BaseMenu):
 
     def remove_fluidd(self, **kwargs):
         ClientRemoveMenu(previous_menu=self.__class__, client=FluiddData()).run()
-
-    def remove_crowsnest(self, **kwargs):
-        remove_crowsnest()

kiauh/core/menus/update_menu.py

@@ -10,7 +10,6 @@
 import textwrap
 from typing import Type, Optional
 
-from components.crowsnest.crowsnest import get_crowsnest_status, update_crowsnest
 from components.klipper.klipper_setup import update_klipper
 from components.klipper.klipper_utils import (
     get_klipper_status,
@@ -58,8 +57,6 @@ class UpdateMenu(BaseMenu):
         self.mc_remote = f"{COLOR_WHITE}{RESET_FORMAT}"
         self.fc_local = f"{COLOR_WHITE}{RESET_FORMAT}"
         self.fc_remote = f"{COLOR_WHITE}{RESET_FORMAT}"
-        self.cn_local = f"{COLOR_WHITE}{RESET_FORMAT}"
-        self.cn_remote = f"{COLOR_WHITE}{RESET_FORMAT}"
 
         self.mainsail_client = MainsailData()
         self.fluidd_client = FluiddData()
@@ -114,7 +111,7 @@ class UpdateMenu(BaseMenu):
             | Other:                |---------------|---------------|
             |  7) KlipperScreen     |               |               |
             |  8) Mobileraker       |               |               |
-            |  9) Crowsnest         | {self.cn_local:<22} | {self.cn_remote:<22} |
+            |  9) Crowsnest         |               |               |
             |                       |-------------------------------|
             | 10) System            |                               |
             """
@@ -146,8 +143,7 @@ class UpdateMenu(BaseMenu):
 
     def update_mobileraker(self, **kwargs): ...
 
-    def update_crowsnest(self, **kwargs):
-        update_crowsnest()
+    def update_crowsnest(self, **kwargs): ...
 
     def upgrade_system_packages(self, **kwargs): ...
 
@@ -193,13 +189,6 @@ class UpdateMenu(BaseMenu):
         )
         self.fc_remote = f"{COLOR_GREEN}{fc_status.get('remote')}{RESET_FORMAT}"
 
-        # crowsnest
-        cn_status = get_crowsnest_status()
-        self.cn_local = self.format_local_status(
-            cn_status.get("local"), cn_status.get("remote")
-        )
-        self.cn_remote = f"{COLOR_GREEN}{cn_status.get('remote')}{RESET_FORMAT}"
-
     def format_local_status(self, local_version, remote_version) -> str:
         if local_version == remote_version:
             return f"{COLOR_GREEN}{local_version}{RESET_FORMAT}"

kiauh/utils/common.py

@@ -55,22 +55,6 @@ def check_install_dependencies(deps: List[str]) -> None:
         install_system_packages(requirements)
 
 
-def get_install_status(
-    repo_dir: Path, opt_files: List[Path]
-) -> Dict[Literal["status", "status_code", "instances"], Union[str, int]]:
-    status = [repo_dir.exists()]
-
-    for f in opt_files:
-        status.append(f.exists())
-
-    if all(status):
-        return {"status": "Installed!", "status_code": 1}
-    elif not any(status):
-        return {"status": "Not installed!", "status_code": 2}
-    else:
-        return {"status": "Incomplete!", "status_code": 3}
-
-
 def get_install_status_common(
     instance_type: Type[BaseInstance], repo_dir: Path, env_dir: Path
 ) -> Dict[Literal["status", "status_code", "instances"], Union[str, int]]:

kiauh/utils/config_utils.py

@@ -1,83 +0,0 @@
-# ======================================================================= #
-#  Copyright (C) 2020 - 2024 Dominik Willner <th33xitus@gmail.com>        #
-#                                                                         #
-#  This file is part of KIAUH - Klipper Installation And Update Helper    #
-#  https://github.com/dw-0/kiauh                                          #
-#                                                                         #
-#  This file may be distributed under the terms of the GNU GPLv3 license  #
-# ======================================================================= #
-import tempfile
-from pathlib import Path
-from typing import List, TypeVar, Tuple, Optional
-
-from components.klipper.klipper import Klipper
-from components.moonraker.moonraker import Moonraker
-from core.config_manager.config_manager import ConfigManager
-from utils.logger import Logger
-
-B = TypeVar("B", Klipper, Moonraker)
-ConfigOption = Tuple[str, str]
-
-
-def add_config_section(
-    section: str,
-    instances: List[B],
-    options: Optional[List[ConfigOption]] = None,
-) -> None:
-    for instance in instances:
-        cfg_file = instance.cfg_file
-        Logger.print_status(f"Add section '[{section}]' to '{cfg_file}' ...")
-
-        if not Path(cfg_file).exists():
-            Logger.print_warn(f"'{cfg_file}' not found!")
-            continue
-
-        cm = ConfigManager(cfg_file)
-        if cm.config.has_section(section):
-            Logger.print_info("Section already exist. Skipped ...")
-            continue
-
-        cm.config.add_section(section)
-
-        if options is not None:
-            for option in options:
-                cm.config.set(section, option[0], option[1])
-
-        cm.write_config()
-
-
-def add_config_section_at_top(section: str, instances: List[B]):
-    for instance in instances:
-        tmp_cfg = tempfile.NamedTemporaryFile(mode="w", delete=False)
-        tmp_cfg_path = Path(tmp_cfg.name)
-        cmt = ConfigManager(tmp_cfg_path)
-        cmt.config.add_section(section)
-        cmt.write_config()
-        tmp_cfg.close()
-
-        cfg_file = instance.cfg_file
-        with open(cfg_file, "r") as org:
-            org_content = org.readlines()
-        with open(tmp_cfg_path, "a") as tmp:
-            tmp.writelines(org_content)
-
-        cfg_file.unlink()
-        tmp_cfg_path.rename(cfg_file)
-
-
-def remove_config_section(section: str, instances: List[B]) -> None:
-    for instance in instances:
-        cfg_file = instance.cfg_file
-        Logger.print_status(f"Remove section '[{section}]' from '{cfg_file}' ...")
-
-        if not Path(cfg_file).exists():
-            Logger.print_warn(f"'{cfg_file}' not found!")
-            continue
-
-        cm = ConfigManager(cfg_file)
-        if not cm.config.has_section(section):
-            Logger.print_info("Section does not exist. Skipped ...")
-            continue
-
-        cm.config.remove_section(section)
-        cm.write_config()

kiauh/utils/filesystem_utils.py

@@ -11,13 +11,17 @@
 
 import re
 import shutil
+import subprocess
+import tempfile
 from pathlib import Path
 from zipfile import ZipFile
-from subprocess import run, check_output, CalledProcessError, PIPE, DEVNULL
 
-from typing import List
+from typing import List, TypeVar, Tuple, Optional
 
 from components.klipper.klipper import Klipper
+from components.moonraker.moonraker import Moonraker
+from core.config_manager.config_manager import ConfigManager
+from core.instance_manager.instance_manager import InstanceManager
 from utils import (
     NGINX_SITES_AVAILABLE,
     MODULE_PATH,
@@ -27,6 +31,10 @@ from utils import (
 from utils.logger import Logger
 
 
+B = TypeVar("B", Klipper, Moonraker)
+ConfigOption = Tuple[str, str]
+
+
 def check_file_exist(file_path: Path, sudo=False) -> bool:
     """
     Helper function for checking the existence of a file |
@@ -37,9 +45,9 @@ def check_file_exist(file_path: Path, sudo=False) -> bool:
     if sudo:
         try:
             command = ["sudo", "find", file_path]
-            check_output(command, stderr=DEVNULL)
+            subprocess.check_output(command, stderr=subprocess.DEVNULL)
             return True
-        except CalledProcessError:
+        except subprocess.CalledProcessError:
             return False
     else:
         if file_path.exists():
@@ -53,8 +61,8 @@ def create_symlink(source: Path, target: Path, sudo=False) -> None:
         cmd = ["ln", "-sf", source, target]
         if sudo:
             cmd.insert(0, "sudo")
-        run(cmd, stderr=PIPE, check=True)
-    except CalledProcessError as e:
+        subprocess.run(cmd, stderr=subprocess.PIPE, check=True)
+    except subprocess.CalledProcessError as e:
         Logger.print_error(f"Failed to create symlink: {e}")
         raise
 
@@ -62,8 +70,8 @@ def create_symlink(source: Path, target: Path, sudo=False) -> None:
 def remove_file(file_path: Path, sudo=False) -> None:
     try:
         cmd = f"{'sudo ' if sudo else ''}rm -f {file_path}"
-        run(cmd, stderr=PIPE, check=True, shell=True)
-    except CalledProcessError as e:
+        subprocess.run(cmd, stderr=subprocess.PIPE, check=True, shell=True)
+    except subprocess.CalledProcessError as e:
         log = f"Cannot remove file {file_path}: {e.stderr.decode()}"
         Logger.print_error(log)
         raise
@@ -89,8 +97,8 @@ def copy_upstream_nginx_cfg() -> None:
     target = NGINX_CONFD.joinpath("upstreams.conf")
     try:
         command = ["sudo", "cp", source, target]
-        run(command, stderr=PIPE, check=True)
-    except CalledProcessError as e:
+        subprocess.run(command, stderr=subprocess.PIPE, check=True)
+    except subprocess.CalledProcessError as e:
         log = f"Unable to create upstreams.conf: {e.stderr.decode()}"
         Logger.print_error(log)
         raise
@@ -105,8 +113,8 @@ def copy_common_vars_nginx_cfg() -> None:
     target = NGINX_CONFD.joinpath("common_vars.conf")
     try:
         command = ["sudo", "cp", source, target]
-        run(command, stderr=PIPE, check=True)
-    except CalledProcessError as e:
+        subprocess.run(command, stderr=subprocess.PIPE, check=True)
+    except subprocess.CalledProcessError as e:
         log = f"Unable to create upstreams.conf: {e.stderr.decode()}"
         Logger.print_error(log)
         raise
@@ -134,8 +142,8 @@ def create_nginx_cfg(name: str, port: int, root_dir: Path) -> None:
     target = NGINX_SITES_AVAILABLE.joinpath(name)
     try:
         command = ["sudo", "mv", tmp, target]
-        run(command, stderr=PIPE, check=True)
-    except CalledProcessError as e:
+        subprocess.run(command, stderr=subprocess.PIPE, check=True)
+    except subprocess.CalledProcessError as e:
         log = f"Unable to create '{target}': {e.stderr.decode()}"
         Logger.print_error(log)
         raise
@@ -175,23 +183,117 @@ def get_next_free_port(ports_in_use: List[int]) -> int:
     return min(valid_ports - used_ports)
 
 
+def add_config_section(
+    section: str,
+    instances: List[B],
+    options: Optional[List[ConfigOption]] = None,
+) -> None:
+    for instance in instances:
+        cfg_file = instance.cfg_file
+        Logger.print_status(f"Add section '[{section}]' to '{cfg_file}' ...")
+
+        if not Path(cfg_file).exists():
+            Logger.print_warn(f"'{cfg_file}' not found!")
+            continue
+
+        cm = ConfigManager(cfg_file)
+        if cm.config.has_section(section):
+            Logger.print_info("Section already exist. Skipped ...")
+            continue
+
+        cm.config.add_section(section)
+
+        if options is not None:
+            for option in options:
+                cm.config.set(section, option[0], option[1])
+
+        cm.write_config()
+
+
+def add_config_section_at_top(section: str, instances: List[B]):
+    for instance in instances:
+        tmp_cfg = tempfile.NamedTemporaryFile(mode="w", delete=False)
+        tmp_cfg_path = Path(tmp_cfg.name)
+        cmt = ConfigManager(tmp_cfg_path)
+        cmt.config.add_section(section)
+        cmt.write_config()
+        tmp_cfg.close()
+
+        cfg_file = instance.cfg_file
+        with open(cfg_file, "r") as org:
+            org_content = org.readlines()
+        with open(tmp_cfg_path, "a") as tmp:
+            tmp.writelines(org_content)
+
+        cfg_file.unlink()
+        tmp_cfg_path.rename(cfg_file)
+
+
+def remove_config_section(section: str, instances: List[B]) -> None:
+    for instance in instances:
+        cfg_file = instance.cfg_file
+        Logger.print_status(f"Remove section '[{section}]' from '{cfg_file}' ...")
+
+        if not Path(cfg_file).exists():
+            Logger.print_warn(f"'{cfg_file}' not found!")
+            continue
+
+        cm = ConfigManager(cfg_file)
+        if not cm.config.has_section(section):
+            Logger.print_info("Section does not exist. Skipped ...")
+            continue
+
+        cm.config.remove_section(section)
+        cm.write_config()
+
+
+def patch_moonraker_conf(
+    moonraker_instances: List[Moonraker],
+    name: str,
+    section_name: str,
+    template_file: str,
+) -> None:
+    for instance in moonraker_instances:
+        cfg_file = instance.cfg_file
+        Logger.print_status(f"Add {name} update section to '{cfg_file}' ...")
+
+        if not Path(cfg_file).exists():
+            Logger.print_warn(f"'{cfg_file}' not found!")
+            return
+
+        cm = ConfigManager(cfg_file)
+        if cm.config.has_section(section_name):
+            Logger.print_info("Section already exist. Skipped ...")
+            return
+
+        template = MODULE_PATH.joinpath("assets", template_file)
+        with open(template, "r") as t:
+            template_content = "\n"
+            template_content += t.read()
+
+        with open(cfg_file, "a") as f:
+            f.write(template_content)
+
+
 def remove_nginx_config(name: str) -> None:
     Logger.print_status(f"Removing NGINX config for {name.capitalize()} ...")
     try:
         remove_file(NGINX_SITES_AVAILABLE.joinpath(name), True)
         remove_file(NGINX_SITES_ENABLED.joinpath(name), True)
 
-    except CalledProcessError as e:
+    except subprocess.CalledProcessError as e:
         log = f"Unable to remove NGINX config '{name}':\n{e.stderr.decode()}"
         Logger.print_error(log)
 
 
-def remove_nginx_logs(name: str, instances: List[Klipper]) -> None:
+def remove_nginx_logs(name: str) -> None:
     Logger.print_status(f"Removing NGINX logs for {name.capitalize()} ...")
     try:
         remove_file(Path(f"/var/log/nginx/{name}-access.log"), True)
         remove_file(Path(f"/var/log/nginx/{name}-error.log"), True)
 
+        im = InstanceManager(Klipper)
+        instances: List[Klipper] = im.instances
         if not instances:
             return
 
@@ -199,5 +301,5 @@ def remove_nginx_logs(name: str, instances: List[Klipper]) -> None:
             remove_file(instance.log_dir.joinpath(f"{name}-access.log"))
             remove_file(instance.log_dir.joinpath(f"{name}-error.log"))
 
-    except (OSError, CalledProcessError) as e:
+    except (OSError, subprocess.CalledProcessError) as e:
         Logger.print_error(f"Unable to remove NGINX logs:\n{e}")

kiauh/utils/system_utils.py

@@ -21,9 +21,9 @@ from typing import List, Literal
 
 import select
 
-from utils.filesystem_utils import check_file_exist
 from utils.input_utils import get_confirm
 from utils.logger import Logger
+from utils.filesystem_utils import check_file_exist
 
 
 def kill(opt_err_msg: str = "") -> None:
@@ -240,7 +240,8 @@ def mask_system_service(service_name: str) -> None:
     :return: None
     """
     try:
-        control_systemd_service(service_name, "mask")
+        command = ["sudo", "systemctl", "mask", service_name]
+        run(command, stderr=PIPE, check=True)
     except CalledProcessError as e:
         log = f"Unable to mask system service {service_name}: {e.stderr.decode()}"
         Logger.print_error(log)
@@ -329,7 +330,7 @@ def set_nginx_permissions() -> None:
 
 
 def control_systemd_service(
-    name: str, action: Literal["start", "stop", "restart", "enable", "disable", "mask"]
+    name: str, action: Literal["start", "stop", "restart", "disable"]
 ) -> None:
     """
     Helper method to execute several actions for a specific systemd service. |
@@ -338,12 +339,12 @@ def control_systemd_service(
     :return: None
     """
     try:
-        Logger.print_status(f"{action.capitalize()} {name} ...")
-        command = ["sudo", "systemctl", action, name]
+        Logger.print_status(f"{action.capitalize()} {name}.service ...")
+        command = ["sudo", "systemctl", action, f"{name}.service"]
         run(command, stderr=PIPE, check=True)
         Logger.print_ok("OK!")
     except CalledProcessError as e:
-        log = f"Failed to {action} {name}: {e.stderr.decode()}"
+        log = f"Failed to {action} {name}.service: {e.stderr.decode()}"
         Logger.print_error(log)
         raise