feat: implement KiauhSettings and use it where appropriate

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

default_kiauh.cfg

@@ -0,0 +1,18 @@
+[kiauh]
+backup_before_update: False
+
+[klipper]
+repo_url: https://github.com/Klipper3d/klipper
+branch: master
+
+[moonraker]
+repo_url: https://github.com/Arksine/moonraker
+branch: master
+
+[mainsail]
+port: 80
+unstable_releases: False
+
+[fluidd]
+port: 81
+unstable_releases: False

kiauh/__init__.py

@@ -11,7 +11,5 @@ import sys
 from pathlib import Path
 
 PROJECT_ROOT = Path(__file__).resolve().parent.parent
-KIAUH_CFG = PROJECT_ROOT.joinpath("kiauh.cfg")
-
 APPLICATION_ROOT = Path(__file__).resolve().parent
 sys.path.append(str(APPLICATION_ROOT))

kiauh/components/klipper/__init__.py

@@ -17,6 +17,5 @@ KLIPPER_DIR = Path.home().joinpath("klipper")
 KLIPPER_ENV_DIR = Path.home().joinpath("klippy-env")
 KLIPPER_BACKUP_DIR = BACKUP_ROOT_DIR.joinpath("klipper-backups")
 KLIPPER_REQUIREMENTS_TXT = KLIPPER_DIR.joinpath("scripts/klippy-requirements.txt")
-DEFAULT_KLIPPER_REPO_URL = "https://github.com/Klipper3D/klipper"
 
 EXIT_KLIPPER_SETUP = "Exiting Klipper setup ..."

kiauh/components/klipper/klipper_setup.py

@@ -12,10 +12,9 @@ from pathlib import Path
 from components.webui_client.client_utils import (
     get_existing_clients,
 )
-from kiauh import KIAUH_CFG
+from core.settings.kiauh_settings import KiauhSettings
 from components.klipper import (
     EXIT_KLIPPER_SETUP,
-    DEFAULT_KLIPPER_REPO_URL,
     KLIPPER_DIR,
     KLIPPER_ENV_DIR,
     KLIPPER_REQUIREMENTS_TXT,
@@ -36,7 +35,6 @@ from components.klipper.klipper_utils import (
     backup_klipper_dir,
 )
 from components.moonraker.moonraker import Moonraker
-from core.config_manager.config_manager import ConfigManager
 from core.instance_manager.instance_manager import InstanceManager
 from core.repo_manager.repo_manager import RepoManager
 from utils.input_utils import get_confirm
@@ -109,13 +107,10 @@ def install_klipper() -> None:
 
 
 def setup_klipper_prerequesites() -> None:
-    cm = ConfigManager(cfg_file=KIAUH_CFG)
+    settings = KiauhSettings()
-    repo = str(cm.get_value("klipper", "repository_url") or DEFAULT_KLIPPER_REPO_URL)
-    branch = str(cm.get_value("klipper", "branch") or "master")
-
     repo_manager = RepoManager(
-        repo=repo,
+        repo=settings.get("klipper", "repo_url"),
-        branch=branch,
+        branch=settings.get("klipper", "branch"),
         target_dir=KLIPPER_DIR,
     )
     repo_manager.clone_repo()
@@ -150,19 +145,16 @@ def update_klipper() -> None:
     if not get_confirm("Update Klipper now?"):
         return
 
-    cm = ConfigManager(cfg_file=KIAUH_CFG)
+    settings = KiauhSettings()
-    if cm.get_value("kiauh", "backup_before_update"):
+    if settings.get("kiauh", "backup_before_update"):
         backup_klipper_dir()
 
     instance_manager = InstanceManager(Klipper)
     instance_manager.stop_all_instance()
 
-    repo = str(cm.get_value("klipper", "repository_url") or DEFAULT_KLIPPER_REPO_URL)
-    branch = str(cm.get_value("klipper", "branch") or "master")
-
     repo_manager = RepoManager(
-        repo=repo,
+        repo=settings.get("klipper", "repo_url"),
-        branch=branch,
+        branch=settings.get("klipper", "branch"),
         target_dir=KLIPPER_DIR,
     )
     repo_manager.pull_repo()

kiauh/components/moonraker/__init__.py

@@ -20,7 +20,6 @@ MOONRAKER_DB_BACKUP_DIR = BACKUP_ROOT_DIR.joinpath("moonraker-db-backups")
 MOONRAKER_REQUIREMENTS_TXT = MOONRAKER_DIR.joinpath(
     "scripts/moonraker-requirements.txt"
 )
-DEFAULT_MOONRAKER_REPO_URL = "https://github.com/Arksine/moonraker"
 DEFAULT_MOONRAKER_PORT = 7125
 
 # introduced due to

kiauh/components/moonraker/moonraker_setup.py

@@ -16,11 +16,10 @@ from components.webui_client.client_utils import (
     get_existing_clients,
 )
 from components.webui_client.mainsail_data import MainsailData
-from kiauh import KIAUH_CFG
+from core.settings.kiauh_settings import KiauhSettings
 from components.klipper.klipper import Klipper
 from components.moonraker import (
     EXIT_MOONRAKER_SETUP,
-    DEFAULT_MOONRAKER_REPO_URL,
     MOONRAKER_DIR,
     MOONRAKER_ENV_DIR,
     MOONRAKER_REQUIREMENTS_TXT,
@@ -35,7 +34,6 @@ from components.moonraker.moonraker_utils import (
     create_example_moonraker_conf,
     backup_moonraker_dir,
 )
-from core.config_manager.config_manager import ConfigManager
 from core.instance_manager.instance_manager import InstanceManager
 from core.repo_manager.repo_manager import RepoManager
 from utils.filesystem_utils import check_file_exist
@@ -133,11 +131,9 @@ def check_moonraker_install_requirements() -> bool:
 
 
 def setup_moonraker_prerequesites() -> None:
-    cm = ConfigManager(cfg_file=KIAUH_CFG)
+    settings = KiauhSettings()
-    repo = str(
+    repo = settings.get("moonraker", "repo_url")
-        cm.get_value("moonraker", "repository_url") or DEFAULT_MOONRAKER_REPO_URL
+    branch = settings.get("moonraker", "branch")
-    )
-    branch = str(cm.get_value("moonraker", "branch") or "master")
 
     repo_manager = RepoManager(
         repo=repo,
@@ -193,21 +189,16 @@ def update_moonraker() -> None:
     if not get_confirm("Update Moonraker now?"):
         return
 
-    cm = ConfigManager(cfg_file=KIAUH_CFG)
+    settings = KiauhSettings()
-    if cm.get_value("kiauh", "backup_before_update"):
+    if settings.get("kiauh", "backup_before_update"):
         backup_moonraker_dir()
 
     instance_manager = InstanceManager(Moonraker)
     instance_manager.stop_all_instance()
 
-    repo = str(
-        cm.get_value("moonraker", "repository_url") or DEFAULT_MOONRAKER_REPO_URL
-    )
-    branch = str(cm.get_value("moonraker", "branch") or "master")
-
     repo_manager = RepoManager(
-        repo=repo,
+        repo=settings.get("moonraker", "repo_url"),
-        branch=branch,
+        branch=settings.get("moonraker", "branch"),
         target_dir=MOONRAKER_DIR,
     )
     repo_manager.pull_repo()

kiauh/components/webui_client/client_config/client_config_setup.py

@@ -13,7 +13,7 @@ from pathlib import Path
 from typing import List
 
 from components.webui_client.base_data import BaseWebClient, BaseWebClientConfig
-from kiauh import KIAUH_CFG
+from core.settings.kiauh_settings import KiauhSettings
 from components.klipper.klipper import Klipper
 from components.moonraker.moonraker import Moonraker
 from components.webui_client.client_dialogs import (
@@ -23,7 +23,6 @@ from components.webui_client.client_utils import (
     backup_client_config_data,
     config_for_other_client_exist,
 )
-from core.config_manager.config_manager import ConfigManager
 
 from core.instance_manager.instance_manager import InstanceManager
 from core.repo_manager.repo_manager import RepoManager
@@ -108,8 +107,8 @@ def update_client_config(client: BaseWebClient) -> None:
         )
         return
 
-    cm = ConfigManager(cfg_file=KIAUH_CFG)
+    settings = KiauhSettings()
-    if cm.get_value("kiauh", "backup_before_update"):
+    if settings.get("kiauh", "backup_before_update"):
         backup_client_config_data(client)
 
     repo_manager = RepoManager(

kiauh/components/webui_client/client_setup.py

@@ -33,9 +33,8 @@ from components.webui_client.client_utils import (
     symlink_webui_nginx_log,
     config_for_other_client_exist,
 )
-from core.config_manager.config_manager import ConfigManager
 from core.instance_manager.instance_manager import InstanceManager
-from kiauh import KIAUH_CFG
+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.filesystem_utils import (
@@ -104,24 +103,23 @@ def install_client(client: BaseWebClient) -> None:
         question = f"Download the recommended {client_config.display_name}?"
         install_client_cfg = get_confirm(question, allow_go_back=False)
 
-    cm = ConfigManager(cfg_file=KIAUH_CFG)
+    settings = KiauhSettings()
-    default_port = cm.get_value(client.name, "port")
+    port = settings.get(client.name, "port")
-    client_port = default_port if default_port and default_port.isdigit() else "80"
     ports_in_use = read_ports_from_nginx_configs()
 
     # check if configured port is a valid number and not in use already
-    valid_port = is_valid_port(client_port, ports_in_use)
+    valid_port = is_valid_port(port, ports_in_use)
     while not valid_port:
         next_port = get_next_free_port(ports_in_use)
         print_client_port_select_dialog(client.display_name, next_port, ports_in_use)
-        client_port = str(
+        port = str(
             get_number_input(
                 f"Configure {client.display_name} for port",
                 min_count=int(next_port),
                 default=next_port,
             )
         )
-        valid_port = is_valid_port(client_port, ports_in_use)
+        valid_port = is_valid_port(port, ports_in_use)
 
     check_install_dependencies(["nginx"])
 
@@ -146,7 +144,7 @@ def install_client(client: BaseWebClient) -> None:
 
         copy_upstream_nginx_cfg()
         copy_common_vars_nginx_cfg()
-        create_client_nginx_cfg(client, client_port)
+        create_client_nginx_cfg(client, port)
         if kl_instances:
             symlink_webui_nginx_log(kl_instances)
         control_systemd_service("nginx", "restart")
@@ -155,7 +153,7 @@ def install_client(client: BaseWebClient) -> None:
         Logger.print_error(f"{client.display_name} installation failed!\n{e}")
         return
 
-    log = f"Open {client.display_name} now on: http://{get_ipv4_addr()}:{client_port}"
+    log = f"Open {client.display_name} now on: http://{get_ipv4_addr()}:{port}"
     Logger.print_ok(f"{client.display_name} installation complete!", start="\n")
     Logger.print_ok(log, prefix=False, end="\n\n")
 

kiauh/core/settings/kiauh_settings.py

@@ -0,0 +1,117 @@
+# ======================================================================= #
+#  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 textwrap
+import configparser
+from configparser import ConfigParser
+from typing import Dict, Union
+from kiauh import PROJECT_ROOT
+from utils.constants import RESET_FORMAT, COLOR_RED
+from utils.logger import Logger
+from utils.system_utils import kill
+
+
+# noinspection PyUnusedLocal
+# noinspection PyMethodMayBeStatic
+class KiauhSettings:
+    _instance = None
+    _default_cfg = PROJECT_ROOT.joinpath("default_kiauh.cfg")
+    _custom_cfg = PROJECT_ROOT.joinpath("kiauh.cfg")
+
+    def __new__(cls, *args, **kwargs) -> "KiauhSettings":
+        if cls._instance is None:
+            cls._instance = super(KiauhSettings, cls).__new__(cls, *args, **kwargs)
+            cls._instance.__initialized = False
+        return cls._instance
+
+    def __init__(self) -> None:
+        if self.__initialized:
+            return
+        self.__initialized = True
+        self.settings: Dict[str, Dict[str, Union[str, int, bool]]] = {}
+
+        self.config = ConfigParser()
+        if self._custom_cfg.exists():
+            self.config.read(self._custom_cfg)
+        elif self._default_cfg.exists():
+            self.config.read(self._default_cfg)
+        else:
+            self._kill()
+
+        sections = self.config.sections()
+        for s in sections:
+            self.settings[s] = dict(self.config[s])
+
+        self._validate_cfg()
+
+    def get(self, key: str, value: Union[str, int, bool]) -> Union[str, int, bool]:
+        return self.settings[key][value]
+
+    def set(self, key: str, value: Union[str, int, bool]) -> None:
+        self.settings[key][value] = value
+
+    def save(self) -> None:
+        for section, option in self.settings.items():
+            self.config[section] = option
+        with open(self._custom_cfg, "w") as configfile:
+            self.config.write(configfile)
+
+    def _validate_cfg(self) -> None:
+        try:
+            self._validate_bool("kiauh", "backup_before_update")
+
+            self._validate_str("klipper", "repo_url")
+            self._validate_str("klipper", "branch")
+
+            self._validate_int("mainsail", "port")
+            self._validate_bool("mainsail", "unstable_releases")
+
+            self._validate_int("fluidd", "port")
+            self._validate_bool("fluidd", "unstable_releases")
+
+        except ValueError:
+            err = f"Invalid value for option '{self._v_option}' in section '{self._v_section}'"
+            Logger.print_error(err)
+            kill()
+        except configparser.NoSectionError:
+            err = f"Missing section '{self._v_section}' in config file"
+            Logger.print_error(err)
+            kill()
+        except configparser.NoOptionError:
+            err = f"Missing option '{self._v_option}' in section '{self._v_section}'"
+            Logger.print_error(err)
+            kill()
+
+    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.get(section, option)
+        if v.isdigit() or v.lower() == "true" or v.lower() == "false":
+            raise ValueError
+
+    def _kill(self) -> None:
+        l1 = "!!! ERROR !!!"
+        l2 = "No KIAUH configuration file found!"
+        error = textwrap.dedent(
+            f"""
+            {COLOR_RED}/=======================================================\\
+            | {l1:^53} |
+            | {l2:^53} |
+            \=======================================================/{RESET_FORMAT}
+            """
+        )[1:]
+        print(error, end="")
+        kill()

kiauh/main.py

@@ -8,11 +8,13 @@
 # ======================================================================= #
 
 from core.menus.main_menu import MainMenu
+from core.settings.kiauh_settings import KiauhSettings
 from utils.logger import Logger
 
 
 def main():
     try:
+        KiauhSettings()
         MainMenu().run()
     except KeyboardInterrupt:
         Logger.print_ok("\nHappy printing!\n", prefix=False)