Merge 2d4c952b4b into 234cf2c751

Signed-off-by: Aleksei Sviridkin <f@lex.la>

kiauh/components/klipper/klipper_setup.py

@@ -176,9 +176,6 @@ def install_klipper_packages() -> None:
     script = KLIPPER_INSTALL_SCRIPT
     packages = parse_packages_from_file(script)
 
-    # Add pkg-config for rp2040 build
-    packages.append("pkg-config")
-
     # Add dbus requirement for DietPi distro
     if Path("/boot/dietpi/.version").exists():
         packages.append("dbus")

kiauh/components/moonraker/moonraker_setup.py

@@ -27,9 +27,6 @@ from components.moonraker import (
 )
 from components.moonraker.moonraker import Moonraker
 from components.moonraker.moonraker_dialogs import print_moonraker_overview
-from components.moonraker.services.moonraker_instance_service import (
-    MoonrakerInstanceService,
-    )
 from components.moonraker.utils.sysdeps_parser import SysDepsParser
 from components.moonraker.utils.utils import (
     backup_moonraker_dir,
@@ -42,9 +39,8 @@ from components.webui_client.client_utils import (
 )
 from components.webui_client.mainsail_data import MainsailData
 from core.instance_manager.instance_manager import InstanceManager
-from core.logger import DialogType, Logger
+from core.logger import Logger
 from core.settings.kiauh_settings import KiauhSettings
-from core.types.color import Color
 from utils.common import check_install_dependencies
 from utils.fs_utils import check_file_exist
 from utils.git_utils import git_clone_wrapper, git_pull_wrapper
@@ -58,7 +54,6 @@ from utils.sys_utils import (
     cmd_sysctl_manage,
     cmd_sysctl_service,
     create_python_venv,
-    get_ipv4_addr,
     install_python_requirements,
     parse_packages_from_file,
 )
@@ -70,18 +65,12 @@ def install_moonraker() -> None:
     if not check_moonraker_install_requirements(klipper_list):
         return
 
-    instance_service = MoonrakerInstanceService()
+    moonraker_list: List[Moonraker] = get_instances(Moonraker)
-    instance_service.load_instances()
+    instances: List[Moonraker] = []
-
-    moonraker_list: List[Moonraker] = instance_service.get_all_instances()
-    new_instances: List[Moonraker] = []
     selected_option: str | Klipper
 
     if len(klipper_list) == 1:
-        suffix: str = klipper_list[0].suffix
+        instances.append(Moonraker(klipper_list[0].suffix))
-        new_inst = instance_service.create_new_instance(suffix)
-        new_instances.append(new_inst)
-
     else:
         print_moonraker_overview(
             klipper_list,
@@ -100,15 +89,12 @@ def install_moonraker() -> None:
             return
 
         if selected_option == "a":
-            new_inst_list: List[Moonraker] = (
+            instances.extend([Moonraker(k.suffix) for k in klipper_list])
-                [instance_service.create_new_instance(k.suffix) for k in klipper_list])
-            new_instances.extend(new_inst_list)
         else:
             klipper_instance: Klipper | None = options.get(selected_option)
             if klipper_instance is None:
                 raise Exception("Error selecting instance!")
-            new_inst = instance_service.create_new_instance(klipper_instance.suffix)
+            instances.append(Moonraker(klipper_instance.suffix))
-            new_instances.append(new_inst)
 
     create_example_cfg = get_confirm("Create example moonraker.conf?")
 
@@ -117,8 +103,8 @@ def install_moonraker() -> None:
         setup_moonraker_prerequesites()
         install_moonraker_polkit()
 
-        ports_map = instance_service.get_instance_port_map()
+        used_ports_map = {m.suffix: m.port for m in moonraker_list}
-        for instance in new_instances:
+        for instance in instances:
             instance.create()
             cmd_sysctl_service(instance.service_file_path.name, "enable")
 
@@ -126,7 +112,7 @@ def install_moonraker() -> None:
                 # if a webclient and/or it's config is installed, patch
                 # its update section to the config
                 clients = get_existing_clients()
-                create_example_moonraker_conf(instance, ports_map, clients)
+                create_example_moonraker_conf(instance, used_ports_map, clients)
 
             cmd_sysctl_service(instance.service_file_path.name, "start")
 
@@ -137,26 +123,6 @@ def install_moonraker() -> None:
         if MainsailData().client_dir.exists() and len(moonraker_list) > 1:
             enable_mainsail_remotemode()
 
-        instance_service.load_instances()
-        new_instances = [instance_service.get_instance_by_suffix(i.suffix) for i in
-                         new_instances]
-
-        ip: str = get_ipv4_addr()
-        # noinspection HttpUrlsUsage
-        url_list = [f"● {i.service_file_path.stem}: http://{ip}:{i.port}" for i in
-               new_instances if i.port]
-        dialog_content = []
-        if url_list:
-            dialog_content.append("You can access Moonraker via the following URL:")
-            dialog_content.extend(url_list)
-
-        Logger.print_dialog(
-            DialogType.CUSTOM,
-            custom_title="Moonraker successfully installed!",
-            custom_color=Color.GREEN,
-            content=dialog_content)
-
-
     except Exception as e:
         Logger.print_error(f"Error while installing Moonraker: {e}")
         return

kiauh/components/moonraker/services/moonraker_instance_service.py

@@ -1,41 +0,0 @@
-from __future__ import annotations
-
-from typing import Dict, List
-
-from components.moonraker.moonraker import Moonraker
-from utils.instance_utils import get_instances
-
-
-class MoonrakerInstanceService:
-    __cls_instance = None
-    __instances: List[Moonraker] = []
-
-    def __new__(cls) -> "MoonrakerInstanceService":
-        if cls.__cls_instance is None:
-            cls.__cls_instance = super(MoonrakerInstanceService, cls).__new__(cls)
-        return cls.__cls_instance
-
-    def __init__(self) -> None:
-        if not hasattr(self, "__initialized"):
-            self.__initialized = False
-        if self.__initialized:
-            return
-        self.__initialized = True
-
-    def load_instances(self) -> None:
-        self.__instances = get_instances(Moonraker)
-
-    def create_new_instance(self, suffix: str) -> Moonraker:
-        instance = Moonraker(suffix)
-        self.__instances.append(instance)
-        return instance
-
-    def get_all_instances(self) -> List[Moonraker]:
-        return self.__instances
-
-    def get_instance_by_suffix(self, suffix: str) -> Moonraker | None:
-        instances: List[Moonraker] = [i for i in self.__instances if i.suffix == suffix]
-        return instances[0] if instances else None
-
-    def get_instance_port_map(self) -> Dict[str, int]:
-        return {i.suffix: i.port for i in self.__instances}

kiauh/core/logger.py

@@ -27,12 +27,6 @@ class DialogType(Enum):
 LINE_WIDTH = 53
 
 
-BORDER_TOP: str = "┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓"
-BORDER_BOTTOM: str = "┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛"
-BORDER_TITLE: str = "┠───────────────────────────────────────────────────────┨"
-BORDER_LEFT: str = "┃"
-BORDER_RIGHT: str = "┃"
-
 class Logger:
     @staticmethod
     def print_info(msg, prefix=True, start="", end="\n") -> None:
@@ -87,27 +81,23 @@ class Logger:
         :param margin_top: The number of empty lines to print before the dialog.
         :param margin_bottom: The number of empty lines to print after the dialog.
         """
-        color = Logger._get_dialog_color(title, custom_color)
+        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_color)
-        print("\n" * margin_top)
+        dialog_content = Logger.format_content(
-
-        print(Color.apply(BORDER_TOP, color))
-
-        if dialog_title:
-            print(Color.apply(f"┃ {dialog_title:^{LINE_WIDTH}} ┃", color))
-            print(Color.apply(BORDER_TITLE, color))
-
-        if content:
-            print(Logger.format_content(
             content,
             LINE_WIDTH,
-                color,
+            dialog_color,
             center_content,
-                ))
+        )
-
+        top = Logger._format_top_border(dialog_color)
-        print(Color.apply(BORDER_BOTTOM, color))
+        bottom = Logger._format_bottom_border(dialog_color)
 
+        print("\n" * margin_top)
+        print(
+            f"{top}{dialog_title_formatted}{dialog_content}{bottom}",
+            end="",
+        )
         print("\n" * margin_bottom)
 
     @staticmethod
@@ -129,6 +119,31 @@ class Logger:
 
         return color
 
+    @staticmethod
+    def _format_top_border(color: Color) -> str:
+        _border = Color.apply(
+            "┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓\n", color
+        )
+        return _border
+
+    @staticmethod
+    def _format_bottom_border(color: Color) -> str:
+        _border = Color.apply(
+            "\n┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛", color
+        )
+        return _border
+
+    @staticmethod
+    def _format_dialog_title(title: str | None, color: Color) -> str:
+        if title is None:
+            return ""
+
+        _title = Color.apply(f"┃ {title:^{LINE_WIDTH}} ┃\n", color)
+        _title += Color.apply(
+            "┠───────────────────────────────────────────────────────┨\n", color
+        )
+        return _title
+
     @staticmethod
     def format_content(
         content: List[str],

kiauh/utils/git_utils.py

@@ -253,9 +253,29 @@ def get_remote_commit(repo: Path) -> str | None:
         return None
 
 
-def git_cmd_clone(repo: str, target_dir: Path) -> None:
+def git_cmd_clone(repo: str, target_dir: Path, depth: int = 0, single_branch: bool = False) -> None:
+    """
+    Clones a repository with optional depth and single-branch parameters.
+
+    :param repo: URL of the repository to clone.
+    :param target_dir: Path where the repository will be cloned.
+    :param depth: Clone depth. If 0, the depth option will be omitted.
+    :param single_branch: Clone only a single branch if True.
+    :return: None
+    """
     try:
-        command = ["git", "clone", repo, target_dir.as_posix()]
+        command = ["git", "clone"]
+
+        # Add --depth flag if depth > 0
+        if depth > 0:
+            command += ["--depth", str(depth)]
+
+        # Add --single-branch flag if single_branch is True
+        if single_branch:
+            command.append("--single-branch")
+
+        command += [repo, target_dir.as_posix()]
+
         run(command, check=True)
 
         Logger.print_ok("Clone successful!")

scripts/klipper.sh

@@ -304,8 +304,6 @@ function install_klipper_packages() {
   packages=$(grep "PKGLIST=" "${install_script}" | cut -d'"' -f2 | sed 's/\${PKGLIST}//g' | tr -d '\n')
   ### add dfu-util for octopi-images
   packages+=" dfu-util"
-  ### add pkg-config for rp2040 build
-  packages+=" pkg-config"
   ### add dbus requirement for DietPi distro
   [[ -e "/boot/dietpi/.version" ]] && packages+=" dbus"
 

scripts/moonraker.sh

@@ -147,177 +147,7 @@ function install_moonraker_dependencies() {
   ### read PKGLIST from official install-script
   status_msg "Reading dependencies..."
   # shellcheck disable=SC2016
-  packages=$(python3 - << EOF
+  packages=$(cat $package_json | tr -d ' \n{}' | cut -d "]" -f1 | cut -d":" -f2 | tr -d '"[' | sed 's/,/ /g')
-from __future__ import annotations
-import shlex
-import re
-import pathlib
-import logging
-import json
-
-from typing import Tuple, Dict, List, Any
-
-def _get_distro_info() -> Dict[str, Any]:
-    release_file = pathlib.Path("/etc/os-release")
-    release_info: Dict[str, str] = {}
-    with release_file.open("r") as f:
-        lexer = shlex.shlex(f, posix=True)
-        lexer.whitespace_split = True
-        for item in list(lexer):
-            if "=" in item:
-                key, val = item.split("=", maxsplit=1)
-                release_info[key] = val
-    return dict(
-        distro_id=release_info.get("ID", ""),
-        distro_version=release_info.get("VERSION_ID", ""),
-        aliases=release_info.get("ID_LIKE", "").split()
-    )
-
-def _convert_version(version: str) -> Tuple[str | int, ...]:
-    version = version.strip()
-    ver_match = re.match(r"\d+(\.\d+)*((?:-|\.).+)?", version)
-    if ver_match is not None:
-        return tuple([
-            int(part) if part.isdigit() else part
-            for part in re.split(r"\.|-", version)
-        ])
-    return (version,)
-
-class SysDepsParser:
-    def __init__(self, distro_info: Dict[str, Any] | None = None) -> None:
-        if distro_info is None:
-            distro_info = _get_distro_info()
-        self.distro_id: str = distro_info.get("distro_id", "")
-        self.aliases: List[str] = distro_info.get("aliases", [])
-        self.distro_version: Tuple[int | str, ...] = tuple()
-        version = distro_info.get("distro_version")
-        if version:
-            self.distro_version = _convert_version(version)
-
-    def _parse_spec(self, full_spec: str) -> str | None:
-        parts = full_spec.split(";", maxsplit=1)
-        if len(parts) == 1:
-            return full_spec
-        pkg_name = parts[0].strip()
-        expressions = re.split(r"( and | or )", parts[1].strip())
-        if not len(expressions) & 1:
-            logging.info(
-                f"Requirement specifier is missing an expression "
-                f"between logical operators : {full_spec}"
-            )
-            return None
-        last_result: bool = True
-        last_logical_op: str | None = "and"
-        for idx, exp in enumerate(expressions):
-            if idx & 1:
-                if last_logical_op is not None:
-                    logging.info(
-                        "Requirement specifier contains sequential logical "
-                        f"operators: {full_spec}"
-                    )
-                    return None
-                logical_op = exp.strip()
-                if logical_op not in ("and", "or"):
-                    logging.info(
-                        f"Invalid logical operator {logical_op} in requirement "
-                        f"specifier: {full_spec}")
-                    return None
-                last_logical_op = logical_op
-                continue
-            elif last_logical_op is None:
-                logging.info(
-                    f"Requirement specifier contains two seqential expressions "
-                    f"without a logical operator: {full_spec}")
-                return None
-            dep_parts = re.split(r"(==|!=|<=|>=|<|>)", exp.strip())
-            req_var = dep_parts[0].strip().lower()
-            if len(dep_parts) != 3:
-                logging.info(f"Invalid comparison, must be 3 parts: {full_spec}")
-                return None
-            elif req_var == "distro_id":
-                left_op: str | Tuple[int | str, ...] = self.distro_id
-                right_op = dep_parts[2].strip().strip("\"'")
-            elif req_var == "distro_version":
-                if not self.distro_version:
-                    logging.info(
-                        "Distro Version not detected, cannot satisfy requirement: "
-                        f"{full_spec}"
-                    )
-                    return None
-                left_op = self.distro_version
-                right_op = _convert_version(dep_parts[2].strip().strip("\"'"))
-            else:
-                logging.info(f"Invalid requirement specifier: {full_spec}")
-                return None
-            operator = dep_parts[1].strip()
-            try:
-                compfunc = {
-                    "<": lambda x, y: x < y,
-                    ">": lambda x, y: x > y,
-                    "==": lambda x, y: x == y,
-                    "!=": lambda x, y: x != y,
-                    ">=": lambda x, y: x >= y,
-                    "<=": lambda x, y: x <= y
-                }.get(operator, lambda x, y: False)
-                result = compfunc(left_op, right_op)
-                if last_logical_op == "and":
-                    last_result &= result
-                else:
-                    last_result |= result
-                last_logical_op = None
-            except Exception:
-                logging.exception(f"Error comparing requirements: {full_spec}")
-                return None
-        if last_result:
-            return pkg_name
-        return None
-
-    def parse_dependencies(self, sys_deps: Dict[str, List[str]]) -> List[str]:
-        if not self.distro_id:
-            logging.info(
-                "Failed to detect current distro ID, cannot parse dependencies"
-            )
-            return []
-        all_ids = [self.distro_id] + self.aliases
-        for distro_id in all_ids:
-            if distro_id in sys_deps:
-                if not sys_deps[distro_id]:
-                    logging.info(
-                        f"Dependency data contains an empty package definition "
-                        f"for linux distro '{distro_id}'"
-                    )
-                    continue
-                processed_deps: List[str] = []
-                for dep in sys_deps[distro_id]:
-                    parsed_dep = self._parse_spec(dep)
-                    if parsed_dep is not None:
-                        processed_deps.append(parsed_dep)
-                return processed_deps
-        else:
-            logging.info(
-                f"Dependency data has no package definition for linux "
-                f"distro '{self.distro_id}'"
-            )
-        return []
-# *** SYSTEM DEPENDENCIES START ***
-system_deps = {
-    "debian": [
-        "python3-virtualenv", "python3-dev", "libopenjp2-7", "libsodium-dev",
-        "zlib1g-dev", "libjpeg-dev", "packagekit",
-        "wireless-tools; distro_id != 'ubuntu' or distro_version <= '24.04'",
-        "iw; distro_id == 'ubuntu' and distro_version >= '24.10'", "curl",
-        "build-essential"
-    ],
-}
-system_deps_json = pathlib.Path("$package_json")
-system_deps = json.loads(system_deps_json.read_bytes())
-parser = SysDepsParser()
-pkgs = parser.parse_dependencies(system_deps)
-if pkgs:
-    print(' '.join(pkgs), end="")
-exit(0)
-EOF
-)
 
   echo "${cyan}${packages}${white}" | tr '[:space:]' '\n'
   read -r -a packages <<< "${packages}"