fix(moonraker): adapt to new moonraker system_dependency.json syntax (#644)

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

kiauh/components/moonraker/moonraker_setup.py

@@ -27,10 +27,11 @@ from components.moonraker import (
 )
 from components.moonraker.moonraker import Moonraker
 from components.moonraker.moonraker_dialogs import print_moonraker_overview
-from components.moonraker.moonraker_utils import (
+from components.moonraker.utils.sysdeps_parser import SysDepsParser
+from components.moonraker.utils.utils import (
     backup_moonraker_dir,
     create_example_moonraker_conf,
-    parse_sysdeps_file,
+    load_sysdeps_json,
 )
 from components.webui_client.client_utils import (
     enable_mainsail_remotemode,
@@ -53,7 +54,6 @@ from utils.sys_utils import (
     cmd_sysctl_manage,
     cmd_sysctl_service,
     create_python_venv,
-    get_distro_info,
     install_python_requirements,
     parse_packages_from_file,
 )
@@ -155,44 +155,24 @@ def setup_moonraker_prerequesites() -> None:
 
 
 def install_moonraker_packages() -> None:
+    Logger.print_status("Parsing Moonraker system dependencies  ...")
+
     moonraker_deps = []
-
     if MOONRAKER_DEPS_JSON_FILE.exists():
-        Logger.print_status(
-            f"Parsing system dependencies from {MOONRAKER_DEPS_JSON_FILE.name} ..."
-        )
-        parsed_sysdeps = parse_sysdeps_file(MOONRAKER_DEPS_JSON_FILE)
-        distro_name, distro_version = get_distro_info()
-
-        Logger.print_info(f"Distro name: {distro_name}")
-        Logger.print_info(f"Distro version: {distro_version}")
-
-        for dep in parsed_sysdeps.get(distro_name, []):
-            pkg = dep[0].strip()
-            comparator = dep[1].strip()
-            req_version = dep[2].strip()
-
-            comparisons = {
-                "": lambda x, y: True,
-                "<": 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,
-            }
-
-            if comparisons[comparator](float(distro_version), float(req_version or 0)):
-                moonraker_deps.append(pkg)
+        Logger.print_info(
+            f"Parsing system dependencies from {MOONRAKER_DEPS_JSON_FILE.name} ...")
+        parser = SysDepsParser()
+        sysdeps = load_sysdeps_json(MOONRAKER_DEPS_JSON_FILE)
+        moonraker_deps.extend(parser.parse_dependencies(sysdeps))
 
     elif MOONRAKER_INSTALL_SCRIPT.exists():
-        Logger.print_status(
-            f"Parsing system dependencies from {MOONRAKER_INSTALL_SCRIPT.name} ..."
-        )
+        Logger.print_warn(f"{MOONRAKER_DEPS_JSON_FILE.name} not found!")
+        Logger.print_info(
+            f"Parsing system dependencies from {MOONRAKER_INSTALL_SCRIPT.name} ...")
         moonraker_deps = parse_packages_from_file(MOONRAKER_INSTALL_SCRIPT)
 
     if not moonraker_deps:
-        raise ValueError("Error reading Moonraker dependencies!")
+        raise ValueError("Error parsing Moonraker dependencies!")
 
     check_install_dependencies({*moonraker_deps})
 

kiauh/components/moonraker/utils/sysdeps_parser.py

@@ -0,0 +1,167 @@
+# ======================================================================= #
+#  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                                          #
+#  It was modified by Dominik Willner <th33xitus@gmail.com>               #
+#                                                                         #
+#  The original file is part of Moonraker:                                #
+#  https://github.com/Arksine/moonraker                                   #
+#  Copyright (C) 2025 Eric Callahan <arksine.code@gmail.com>              #
+#                                                                         #
+#  This file may be distributed under the terms of the GNU GPLv3 license  #
+# ======================================================================= #
+
+from __future__ import annotations
+
+import logging
+import pathlib
+import re
+import shlex
+from typing import Any, Dict, List, Tuple
+
+
+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:
+            # There should always be an odd number of expressions.  Each
+            # expression is separated by an "and" or "or" operator
+            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 []

kiauh/components/moonraker/utils/utils.py

@@ -7,10 +7,9 @@
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
 import json
-import re
 import shutil
 from pathlib import Path
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, Optional
 
 from components.moonraker import (
     MODULE_PATH,
@@ -141,33 +140,11 @@ def backup_moonraker_db_dir() -> None:
             name, source=instance.db_dir, target=MOONRAKER_DB_BACKUP_DIR
         )
 
-
-# This function is from sync_dependencies.py script from the Moonraker project on GitHub:
-# https://github.com/Arksine/moonraker/blob/master/scripts/sync_dependencies.py
-# Thanks to Arksine for his work on this project!
-def parse_sysdeps_file(sysdeps_file: Path) -> Dict[str, List[Tuple[str, str, str]]]:
-    """
-    Parses the system dependencies file and returns a dictionary with the parsed dependencies.
-    :param sysdeps_file: The path to the system dependencies file.
-    :return: A dictionary with the parsed dependencies in the format {distro: [(package, comparator, version)]}.
-    """
-    base_deps: Dict[str, List[str]] = json.loads(sysdeps_file.read_bytes())
-    parsed_deps: Dict[str, List[Tuple[str, str, str]]] = {}
-
-    for distro, pkgs in base_deps.items():
-        parsed_deps[distro] = []
-        for dep in pkgs:
-            parts = dep.split(";", maxsplit=1)
-            if len(parts) == 1:
-                parsed_deps[distro].append((dep.strip(), "", ""))
-            else:
-                pkg_name = parts[0].strip()
-                dep_parts = re.split(r"(==|!=|<=|>=|<|>)", parts[1].strip())
-                comp_var = dep_parts[0].strip().lower()
-                if len(dep_parts) != 3 or comp_var != "distro_version":
-                    continue
-                operator = dep_parts[1].strip()
-                req_version = dep_parts[2].strip()
-                parsed_deps[distro].append((pkg_name, operator, req_version))
-
-    return parsed_deps
+def load_sysdeps_json(file: Path) -> Dict[str, List[str]]:
+    try:
+        sysdeps: Dict[str, List[str]] = json.loads(file.read_bytes())
+    except json.JSONDecodeError as e:
+        Logger.print_error(f"Unable to parse {file.name}:\n{e}")
+        return {}
+    else:
+        return sysdeps

kiauh/core/menus/backup_menu.py

@@ -13,7 +13,7 @@ from typing import Type
 
 from components.klipper.klipper_utils import backup_klipper_dir
 from components.klipperscreen.klipperscreen import backup_klipperscreen_dir
-from components.moonraker.moonraker_utils import (
+from components.moonraker.utils.utils import (
     backup_moonraker_db_dir,
     backup_moonraker_dir,
 )

kiauh/core/menus/main_menu.py

@@ -16,7 +16,7 @@ from components.crowsnest.crowsnest import get_crowsnest_status
 from components.klipper.klipper_utils import get_klipper_status
 from components.klipperscreen.klipperscreen import get_klipperscreen_status
 from components.log_uploads.menus.log_upload_menu import LogUploadMenu
-from components.moonraker.moonraker_utils import get_moonraker_status
+from components.moonraker.utils.utils import get_moonraker_status
 from components.webui_client.client_utils import (
     get_client_status,
     get_current_client_config,

kiauh/core/menus/settings_menu.py

@@ -15,7 +15,7 @@ from typing import Literal, Tuple, Type
 from components.klipper import KLIPPER_DIR, KLIPPER_REPO_URL
 from components.klipper.klipper_utils import get_klipper_status
 from components.moonraker import MOONRAKER_DIR, MOONRAKER_REPO_URL
-from components.moonraker.moonraker_utils import get_moonraker_status
+from components.moonraker.utils.utils import get_moonraker_status
 from core.logger import DialogType, Logger
 from core.menus import Option
 from core.menus.base_menu import BaseMenu

kiauh/core/menus/update_menu.py

@@ -21,7 +21,7 @@ from components.klipperscreen.klipperscreen import (
     update_klipperscreen,
 )
 from components.moonraker.moonraker_setup import update_moonraker
-from components.moonraker.moonraker_utils import get_moonraker_status
+from components.moonraker.utils.utils import get_moonraker_status
 from components.webui_client.client_config.client_config_setup import (
     update_client_config,
 )