refactor: use sane fallbacks on missing kiauh config options

for some options a warning is print if the fallback is used

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

default.kiauh.cfg

@@ -13,6 +13,10 @@ repositories:
     https://github.com/Klipper3d/klipper
 
 [moonraker]
+# Moonraker supports two optional Python packages that can be used to reduce its CPU load
+# If set to true, those packages will be installed during the Moonraker installation
+optional_speedups: True
+
 # add custom repositories here, if at least one is given, the first in the list will be used by default
 # otherwise the official repository is used
 #

kiauh/components/klipper/services/klipper_setup_service.py

@@ -277,7 +277,7 @@ class KlipperSetupService:
 
         try:
             install_klipper_packages()
-            if create_python_venv(KLIPPER_ENV_DIR):
+            if create_python_venv(KLIPPER_ENV_DIR, False, False, self.settings.klipper.use_python_binary):
                 install_python_requirements(KLIPPER_ENV_DIR, KLIPPER_REQ_FILE)
         except Exception:
             Logger.print_error("Error during installation of Klipper requirements!")

kiauh/components/moonraker/services/moonraker_setup_service.py

@@ -315,11 +315,12 @@ class MoonrakerSetupService:
 
         try:
             install_moonraker_packages()
-            if create_python_venv(MOONRAKER_ENV_DIR):
+            if create_python_venv(MOONRAKER_ENV_DIR, False, False, self.settings.moonraker.use_python_binary):
                 install_python_requirements(MOONRAKER_ENV_DIR, MOONRAKER_REQ_FILE)
-                install_python_requirements(
+                if self.settings.moonraker.optional_speedups:
-                    MOONRAKER_ENV_DIR, MOONRAKER_SPEEDUPS_REQ_FILE
+                    install_python_requirements(
-                )
+                        MOONRAKER_ENV_DIR, MOONRAKER_SPEEDUPS_REQ_FILE
+                    )
             self.__install_polkit()
         except Exception:
             Logger.print_error("Error during installation of Moonraker requirements!")

kiauh/core/settings/kiauh_settings.py

@@ -8,14 +8,15 @@
 # ======================================================================= #
 from __future__ import annotations
 
+import shutil
 from dataclasses import dataclass, field
-from typing import Any, List
+from typing import Any, List, Literal
 
+from components.klipper import KLIPPER_REPO_URL
+from components.moonraker import MOONRAKER_REPO_URL
 from core.backup_manager.backup_manager import BackupManager
 from core.logger import DialogType, Logger
 from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
-    NoOptionError,
-    NoSectionError,
     SimpleConfigParser,
 )
 from utils.input_utils import get_confirm
@@ -27,14 +28,6 @@ DEFAULT_CFG = PROJECT_ROOT.joinpath("default.kiauh.cfg")
 CUSTOM_CFG = PROJECT_ROOT.joinpath("kiauh.cfg")
 
 
-class NoValueError(Exception):
-    """Raised when a required value is not defined for an option"""
-
-    def __init__(self, section: str, option: str):
-        msg = f"Missing value for option '{option}' in section '{section}'"
-        super().__init__(msg)
-
-
 class InvalidValueError(Exception):
     """Raised when a value is invalid for an option"""
 
@@ -55,8 +48,16 @@ class Repository:
 
 
 @dataclass
-class RepoSettings:
+class KlipperSettings:
     repositories: List[Repository] | None = field(default=None)
+    use_python_binary: str | None = field(default=None)
+
+
+@dataclass
+class MoonrakerSettings:
+    optional_speedups: bool | None = field(default=None)
+    repositories: List[Repository] | None = field(default=None)
+    use_python_binary: str | None = field(default=None)
 
 
 @dataclass
@@ -69,6 +70,7 @@ class WebUiSettings:
 # noinspection PyMethodMayBeStatic
 class KiauhSettings:
     __instance = None
+    __initialized = False
 
     def __new__(cls, *args, **kwargs) -> "KiauhSettings":
         if cls.__instance is None:
@@ -86,20 +88,20 @@ class KiauhSettings:
         return getattr(self, item)
 
     def __init__(self) -> None:
-        if not hasattr(self, "__initialized"):
-            self.__initialized = False
         if self.__initialized:
             return
         self.__initialized = True
+
         self.config = SimpleConfigParser()
         self.kiauh = AppSettings()
-        self.klipper = RepoSettings()
+        self.klipper = KlipperSettings()
-        self.moonraker = RepoSettings()
+        self.moonraker = MoonrakerSettings()
         self.mainsail = WebUiSettings()
         self.fluidd = WebUiSettings()
 
-        self._load_config()
+        self.__read_config_set_internal_state()
 
+    # todo: refactor this, at least rename to something else!
     def get(self, section: str, option: str) -> str | int | bool:
         """
         Get a value from the settings state by providing the section and option name as
@@ -117,128 +119,178 @@ class KiauhSettings:
             raise
 
     def save(self) -> None:
-        self._set_config_options_state()
+        self.__write_internal_state_to_cfg()
-        self.config.write_file(CUSTOM_CFG)
+        self.__read_config_set_internal_state()
-        self._load_config()
 
-    def _load_config(self) -> None:
+    def __read_config_set_internal_state(self) -> None:
         if not CUSTOM_CFG.exists() and not DEFAULT_CFG.exists():
-            self.__kill()
+            Logger.print_dialog(
-
+                DialogType.ERROR,
-        cfg = CUSTOM_CFG if CUSTOM_CFG.exists() else DEFAULT_CFG
+                [
-        self.config.read_file(cfg)
+                    "No KIAUH configuration file found! Please make sure you have at least "
-
+                    "one of the following configuration files in KIAUH's root directory:",
-        needs_migration = self._check_deprecated_repo_config()
+                    "● default.kiauh.cfg",
-        if needs_migration:
+                    "● kiauh.cfg",
-            self._prompt_migration_dialog()
+                ],
-            return
+            )
-        else:
-            # Only validate if no migration was needed
-            self._validate_cfg()
-            self.__set_internal_state()
-
-    def _validate_cfg(self) -> None:
-        def __err_and_kill(error: str) -> None:
-            Logger.print_error(f"Error validating kiauh.cfg: {error}")
             kill()
 
-        try:
+        # copy default config to custom config if it does not exist
-            self._validate_bool("kiauh", "backup_before_update")
+        if not CUSTOM_CFG.exists():
+            shutil.copyfile(DEFAULT_CFG, CUSTOM_CFG)
 
-            self._validate_repositories("klipper", "repositories")
+        self.config.read_file(CUSTOM_CFG)
-            self._validate_repositories("moonraker", "repositories")
 
-            self._validate_int("mainsail", "port")
+        # check if there are deprecated repo_url and branch options in the kiauh.cfg
-            self._validate_bool("mainsail", "unstable_releases")
+        if self._check_deprecated_repo_config():
+            self._prompt_migration_dialog()
 
-            self._validate_int("fluidd", "port")
+        self.__set_internal_state()
-            self._validate_bool("fluidd", "unstable_releases")
-
-        except ValueError:
-            err = f"Invalid value for option '{self._v_option}' in section '{self._v_section}'"
-            __err_and_kill(err)
-        except NoSectionError:
-            err = f"Missing section '{self._v_section}' in config file"
-            __err_and_kill(err)
-        except NoOptionError:
-            err = f"Missing option '{self._v_option}' in section '{self._v_section}'"
-            __err_and_kill(err)
-        except NoValueError:
-            err = f"Missing value for option '{self._v_option}' in section '{self._v_section}'"
-            __err_and_kill(err)
-
-    def _validate_bool(self, section: str, option: str) -> None:
-        self._v_section, self._v_option = (section, option)
-        (bool(self.config.getboolean(section, option)))
-
-    def _validate_int(self, section: str, option: str) -> None:
-        self._v_section, self._v_option = (section, option)
-        int(self.config.getint(section, option))
-
-    def _validate_str(self, section: str, option: str) -> None:
-        self._v_section, self._v_option = (section, option)
-        v = self.config.getval(section, option)
-
-        if not v:
-            raise ValueError
-
-    def _validate_repositories(self, section: str, option: str) -> None:
-        self._v_section, self._v_option = (section, option)
-        repos = self.config.getval(section, option)
-        if not repos:
-            raise NoValueError(section, option)
-
-        for repo in repos:
-            if repo.strip().startswith("#") or repo.strip().startswith(";"):
-                continue
-            try:
-                if "," in repo:
-                    url, branch = repo.strip().split(",")
-                    if not url:
-                        raise InvalidValueError(section, option, repo)
-                else:
-                    url = repo.strip()
-                    if not url:
-                        raise InvalidValueError(section, option, repo)
-            except ValueError:
-                raise InvalidValueError(section, option, repo)
 
     def __set_internal_state(self) -> None:
-        self.kiauh.backup_before_update = self.config.getboolean(
+        # parse Kiauh options
-            "kiauh", "backup_before_update"
+        self.kiauh.backup_before_update = self.__read_from_cfg(
+            "kiauh",
+            "backup_before_update",
+            "bool",
+            False,
+            False,
         )
 
-        kl_repos = self.config.getval("klipper", "repositories")
+        # parse Klipper options
-        self.klipper.repositories = self.__set_repo_state(kl_repos)
+        self.klipper.use_python_binary = self.__read_from_cfg(
-
+            "klipper",
-        mr_repos = self.config.getval("moonraker", "repositories")
+            "use_python_binary",
-        self.moonraker.repositories = self.__set_repo_state(mr_repos)
+            "str",
-
+            None,
-        self.mainsail.port = self.config.getint("mainsail", "port")
+            True,
-        self.mainsail.unstable_releases = self.config.getboolean(
-            "mainsail", "unstable_releases"
         )
-        self.fluidd.port = self.config.getint("fluidd", "port")
+        kl_repos: List[str] = self.__read_from_cfg(
-        self.fluidd.unstable_releases = self.config.getboolean(
+            "klipper",
-            "fluidd", "unstable_releases"
+            "repositories",
+            "list",
+            [KLIPPER_REPO_URL],
+            False,
+        )
+        self.klipper.repositories = self.__set_repo_state("klipper", kl_repos)
+
+        # parse Moonraker options
+        self.moonraker.use_python_binary = self.__read_from_cfg(
+            "moonraker",
+            "use_python_binary",
+            "str",
+            None,
+            True,
+        )
+        self.moonraker.optional_speedups = self.__read_from_cfg(
+            "moonraker",
+            "optional_speedups",
+            "bool",
+            True,
+            False,
+        )
+        mr_repos: List[str] = self.__read_from_cfg(
+            "moonraker",
+            "repositories",
+            "list",
+            [MOONRAKER_REPO_URL],
+            False,
+        )
+        self.moonraker.repositories = self.__set_repo_state("moonraker", mr_repos)
+
+        # parse Mainsail options
+        self.mainsail.port = self.__read_from_cfg(
+            "mainsail",
+            "port",
+            "int",
+            80,
+            False,
+        )
+        self.mainsail.unstable_releases = self.__read_from_cfg(
+            "mainsail",
+            "unstable_releases",
+            "bool",
+            False,
+            False,
         )
 
-    def __set_repo_state(self, repos: List[str]) -> List[Repository]:
+        # parse Fluidd options
+        self.fluidd.port = self.__read_from_cfg(
+            "fluidd",
+            "port",
+            "int",
+            80,
+            False,
+        )
+        self.fluidd.unstable_releases = self.__read_from_cfg(
+            "fluidd",
+            "unstable_releases",
+            "bool",
+            False,
+            False,
+        )
+
+    def __read_from_cfg(
+        self,
+        section: str,
+        option: str,
+        value_type: Literal["str", "int", "bool", "list"] = "str",
+        fallback: str | int | bool | List[str] | None = None,
+        silent_fallback: bool = False,
+    ) -> str | int | bool | List[str] | None:
+        has_section = self.config.has_section(section)
+        has_option = self.config.has_option(section, option)
+
+        if not (has_section and has_option):
+            if not silent_fallback:
+                Logger.print_warn(
+                    f"Option '{option}' in section '{section}' not defined. Falling back to '{fallback}'."
+                )
+            return fallback
+
+        try:
+            return {
+                "int": self.config.getint,
+                "bool": self.config.getboolean,
+                "list": self.config.getval,
+                "str": self.config.getval,
+            }[value_type](section, option)
+        except (ValueError, TypeError) as e:
+            if not silent_fallback:
+                Logger.print_warn(
+                    f"Failed to parse value of option '{option}' in section '{section}' as {value_type}. Falling back to '{fallback}'. Error: {e}"
+                )
+            return fallback
+
+    def __set_repo_state(self, section: str, repos: List[str]) -> List[Repository]:
         _repos: List[Repository] = []
         for repo in repos:
-            if repo.strip().startswith("#") or repo.strip().startswith(";"):
+            try:
-                continue
+                if repo.strip().startswith("#") or repo.strip().startswith(";"):
-            if "," in repo:
+                    continue
-                url, branch = repo.strip().split(",")
+                if "," in repo:
-                if not branch:
+                    url, branch = repo.strip().split(",")
+
+                    if not branch:
+                        branch = "master"
+                else:
+                    url = repo.strip()
                     branch = "master"
-            else:
+
-                url = repo.strip()
+                # url must not be empty otherwise it's considered
-                branch = "master"
+                # as an unrecoverable, invalid configuration
-            _repos.append(Repository(url.strip(), branch.strip()))
+                if not url:
+                    raise InvalidValueError(section, "repositories", repo)
+
+                _repos.append(Repository(url.strip(), branch.strip()))
+
+            except InvalidValueError as e:
+                Logger.print_error(f"Error parsing kiauh.cfg: {e}")
+                kill()
+
         return _repos
 
-    def _set_config_options_state(self) -> None:
+    def __write_internal_state_to_cfg(self) -> None:
         """Updates the config with current settings, preserving values that haven't been modified"""
         if self.kiauh.backup_before_update is not None:
             self.config.set_option(
@@ -276,6 +328,8 @@ class KiauhSettings:
                 "fluidd", "unstable_releases", str(self.fluidd.unstable_releases)
             )
 
+        self.config.write_file(CUSTOM_CFG)
+
     def _check_deprecated_repo_config(self) -> bool:
         # repo_url and branch are deprecated - 2025.03.23
         for section in ["klipper", "moonraker"]:
@@ -287,22 +341,23 @@ class KiauhSettings:
 
     def _prompt_migration_dialog(self) -> None:
         migration_1: List[str] = [
-            "The old 'repo_url' and 'branch' options are now combined under 'repositories'.",
+            "Options 'repo_url' and 'branch' are now combined into a 'repositories' option.",
             "\n\n",
-            "Example format:",
+            "● Old format:",
-            "[klipper]",
+            "  [klipper]",
-            "repositories:",
+            "  repo_url: https://github.com/Klipper3d/klipper",
-            "    https://github.com/Klipper3d/klipper, master",
+            "  branch: master",
             "\n\n",
-            "[moonraker]",
+            "● New format:",
-            "repositories:",
+            "  [klipper]",
-            "    https://github.com/Arksine/moonraker, master",
+            "  repositories:",
+            "      https://github.com/Klipper3d/klipper, master",
         ]
         Logger.print_dialog(
             DialogType.ATTENTION,
             [
-                "Deprecated repository configuration found!",
+                "Deprecated kiauh.cfg configuration found!",
-                "KAIUH can now attempt to automatically migrate your configuration.",
+                "KAIUH can now attempt to automatically migrate the configuration.",
                 "\n\n",
                 *migration_1,
             ],
@@ -313,7 +368,7 @@ class KiauhSettings:
             Logger.print_dialog(
                 DialogType.ERROR,
                 [
-                    "Please update your configuration file manually.",
+                    "Please update the configuration file manually.",
                 ],
                 center_content=True,
             )
@@ -354,23 +409,7 @@ class KiauhSettings:
             self.config.write_file(CUSTOM_CFG)
             self.config.read_file(CUSTOM_CFG)  # reload config
 
-            # Validate the migrated config
-            self._validate_cfg()
-            self.__set_internal_state()
-
         except Exception as e:
             Logger.print_error(f"Error migrating configuration: {e}")
             Logger.print_error("Please migrate manually.")
             kill()
-
-    def __kill(self) -> None:
-        Logger.print_dialog(
-            DialogType.ERROR,
-            [
-                "No KIAUH configuration file found! Please make sure you have at least "
-                "one of the following configuration files in KIAUH's root directory:",
-                "● default.kiauh.cfg",
-                "● kiauh.cfg",
-            ],
-        )
-        kill()

kiauh/utils/sys_utils.py

@@ -95,6 +95,7 @@ def create_python_venv(
     target: Path,
     force: bool = False,
     allow_access_to_system_site_packages: bool = False,
+    use_python_binary: str | None = None
 ) -> bool:
     """
     Create a python 3 virtualenv at the provided target destination.
@@ -103,36 +104,46 @@ def create_python_venv(
     :param target: Path where to create the virtualenv at
     :param force: Force recreation of the virtualenv
     :param allow_access_to_system_site_packages: give the virtual environment access to the system site-packages dir
+    :param use_python_binary: allows to override default python binary
     :return: bool
     """
     Logger.print_status("Set up Python virtual environment ...")
-    cmd = ["virtualenv", "-p", "/usr/bin/python3", target.as_posix()]
+    # If binarry override is not set, we use default defined here
+    python_binary = use_python_binary if use_python_binary else "/usr/bin/python3"
+    cmd = ["virtualenv", "-p", python_binary, target.as_posix()]
     cmd.append(
         "--system-site-packages"
     ) if allow_access_to_system_site_packages else None
-    if not target.exists():
-        try:
-            run(cmd, check=True)
-            Logger.print_ok("Setup of virtualenv successful!")
-            return True
-        except CalledProcessError as e:
-            Logger.print_error(f"Error setting up virtualenv:\n{e}")
-            return False
-    else:
-        if not force and not get_confirm(
-            "Virtualenv already exists. Re-create?", default_choice=False
-        ):
-            Logger.print_info("Skipping re-creation of virtualenv ...")
-            return False
 
-        try:
+    n = 2
-            shutil.rmtree(target)
+    while(n > 0):
-            create_python_venv(target)
+        if not target.exists():
-            return True
+            try:
-        except OSError as e:
+                run(cmd, check=True)
-            log = f"Error removing existing virtualenv: {e.strerror}"
+                Logger.print_ok("Setup of virtualenv successful!")
-            Logger.print_error(log, False)
+                return True
-            return False
+            except CalledProcessError as e:
+                Logger.print_error(f"Error setting up virtualenv:\n{e}")
+                return False
+        else:
+            if n == 1:
+                # This case should never happen, 
+                # but the function should still behave correctly
+                Logger.print_error("Virtualenv still exists after deletion.")
+                return False                            
+            if not force and not get_confirm(
+                "Virtualenv already exists. Re-create?", default_choice=False
+            ):
+                Logger.print_info("Skipping re-creation of virtualenv ...")
+                return False
+
+            try:
+                shutil.rmtree(target)
+                n -= 1
+            except OSError as e:
+                log = f"Error removing existing virtualenv: {e.strerror}"
+                Logger.print_error(log, False)
+                return False
 
 
 def update_python_pip(target: Path) -> None: