feat(moonraker): display moonraker ip address after install

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

kiauh/components/klipper/klipper_setup.py

@@ -176,6 +176,9 @@ 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,6 +27,9 @@ 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,
@@ -39,8 +42,9 @@ 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 Logger
+from core.logger import DialogType, 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
@@ -54,6 +58,7 @@ from utils.sys_utils import (
     cmd_sysctl_manage,
     cmd_sysctl_service,
     create_python_venv,
+    get_ipv4_addr,
     install_python_requirements,
     parse_packages_from_file,
 )
@@ -65,12 +70,18 @@ def install_moonraker() -> None:
     if not check_moonraker_install_requirements(klipper_list):
         return
 
-    moonraker_list: List[Moonraker] = get_instances(Moonraker)
-    instances: List[Moonraker] = []
+    instance_service = MoonrakerInstanceService()
+    instance_service.load_instances()
+
+    moonraker_list: List[Moonraker] = instance_service.get_all_instances()
+    new_instances: List[Moonraker] = []
     selected_option: str | Klipper
 
     if len(klipper_list) == 1:
-        instances.append(Moonraker(klipper_list[0].suffix))
+        suffix: str = klipper_list[0].suffix
+        new_inst = instance_service.create_new_instance(suffix)
+        new_instances.append(new_inst)
+
     else:
         print_moonraker_overview(
             klipper_list,
@@ -89,12 +100,15 @@ def install_moonraker() -> None:
             return
 
         if selected_option == "a":
-            instances.extend([Moonraker(k.suffix) for k in klipper_list])
+            new_inst_list: List[Moonraker] = (
+                [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!")
-            instances.append(Moonraker(klipper_instance.suffix))
+            new_inst = instance_service.create_new_instance(klipper_instance.suffix)
+            new_instances.append(new_inst)
 
     create_example_cfg = get_confirm("Create example moonraker.conf?")
 
@@ -103,8 +117,8 @@ def install_moonraker() -> None:
         setup_moonraker_prerequesites()
         install_moonraker_polkit()
 
-        used_ports_map = {m.suffix: m.port for m in moonraker_list}
-        for instance in instances:
+        ports_map = instance_service.get_instance_port_map()
+        for instance in new_instances:
             instance.create()
             cmd_sysctl_service(instance.service_file_path.name, "enable")
 
@@ -112,7 +126,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, used_ports_map, clients)
+                create_example_moonraker_conf(instance, ports_map, clients)
 
             cmd_sysctl_service(instance.service_file_path.name, "start")
 
@@ -123,6 +137,26 @@ 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

@@ -0,0 +1,41 @@
+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,6 +27,12 @@ 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:
@@ -81,23 +87,27 @@ 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.
         """
-        dialog_color = Logger._get_dialog_color(title, custom_color)
+        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)
-        dialog_content = Logger.format_content(
-            content,
-            LINE_WIDTH,
-            dialog_color,
-            center_content,
-        )
-        top = Logger._format_top_border(dialog_color)
-        bottom = Logger._format_bottom_border(dialog_color)
 
         print("\n" * margin_top)
-        print(
-            f"{top}{dialog_title_formatted}{dialog_content}{bottom}",
-            end="",
-        )
+
+        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,
+                center_content,
+                ))
+
+        print(Color.apply(BORDER_BOTTOM, color))
+
         print("\n" * margin_bottom)
 
     @staticmethod
@@ -119,31 +129,6 @@ 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],

scripts/klipper.sh

@@ -304,6 +304,8 @@ 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,7 +147,177 @@ function install_moonraker_dependencies() {
   ### read PKGLIST from official install-script
   status_msg "Reading dependencies..."
   # shellcheck disable=SC2016
-  packages=$(cat $package_json | tr -d ' \n{}' | cut -d "]" -f1 | cut -d":" -f2 | tr -d '"[' | sed 's/,/ /g')
+  packages=$(python3 - << EOF
+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}"