Merge pull request #524 from dw-0/develop

Merge develop into master

kiauh/components/klipper_firmware/menus/klipper_build_menu.py

@@ -81,6 +81,7 @@ class KlipperBuildFirmwareMenu(BaseMenu):
             line = f"{COLOR_RED}Dependencies are missing!{RESET_FORMAT}"
 
         menu += f"║ {line:<62} ║\n"
+        menu += "╟───────────────────────────────────────────────────────╢\n"
 
         print(menu, end="")
 

kiauh/components/klipper_firmware/menus/klipper_flash_menu.py

@@ -249,7 +249,7 @@ class KlipperSelectMcuIdMenu(BaseMenu):
         self.flash_options = FlashOptions()
         self.mcu_list = self.flash_options.mcu_list
         self.input_label_txt = "Select MCU to flash"
-        self.footer_type = FooterType.BACK_HELP
+        self.footer_type = FooterType.BACK
 
     def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None:
         self.previous_menu = (
@@ -265,7 +265,7 @@ class KlipperSelectMcuIdMenu(BaseMenu):
 
     def print_menu(self) -> None:
         header = "!!! ATTENTION !!!"
-        header2 = f"[{COLOR_CYAN}List of available MCUs{RESET_FORMAT}]"
+        header2 = f"[{COLOR_CYAN}List of detected MCUs{RESET_FORMAT}]"
         color = COLOR_RED
         count = 62 - len(color) - len(RESET_FORMAT)
         menu = textwrap.dedent(
@@ -277,15 +277,21 @@ class KlipperSelectMcuIdMenu(BaseMenu):
             ║ ONLY flash a firmware created for the respective MCU! ║
             ║                                                       ║
             ╟{header2:─^64}╢
+            ║                                                       ║
             """
         )[1:]
 
         for i, mcu in enumerate(self.mcu_list):
             mcu = mcu.split("/")[-1]
-            menu += f"   ● MCU #{i}: {COLOR_CYAN}{mcu}{RESET_FORMAT}\n"
-        menu += "╟───────────────────────────┬───────────────────────────╢"
+            menu += f"║ {i}) {COLOR_CYAN}{mcu:<51}{RESET_FORMAT}║\n"
 
-        print(menu, end="\n")
+        menu += textwrap.dedent(
+            """
+            ║                                                       ║
+            ╟───────────────────────────────────────────────────────╢
+            """
+        )[1:]
+        print(menu, end="")
 
     def flash_mcu(self, **kwargs):
         try:
@@ -343,8 +349,8 @@ class KlipperSelectSDFlashBoardMenu(BaseMenu):
 
             for i, board in enumerate(self.available_boards):
                 line = f" {i}) {board}"
-                menu += f"|{line:<55}|\n"
-
+                menu += f"║{line:<55}║\n"
+            menu += "╟───────────────────────────────────────────────────────╢"
             print(menu, end="")
 
     def board_select(self, **kwargs):
@@ -392,8 +398,8 @@ class KlipperFlashOverviewMenu(BaseMenu):
 
     def set_options(self) -> None:
         self.options = {
-            "Y": Option(self.execute_flash),
-            "N": Option(self.abort_process),
+            "y": Option(self.execute_flash),
+            "n": Option(self.abort_process),
         }
 
         self.default_option = Option(self.execute_flash)
@@ -406,7 +412,7 @@ class KlipperFlashOverviewMenu(BaseMenu):
         method = self.flash_options.flash_method.value
         command = self.flash_options.flash_command.value
         conn_type = self.flash_options.connection_type.value
-        mcu = self.flash_options.selected_mcu
+        mcu = self.flash_options.selected_mcu.split("/")[-1]
         board = self.flash_options.selected_board
         baudrate = self.flash_options.selected_baudrate
         subheader = f"[{COLOR_CYAN}Overview{RESET_FORMAT}]"
@@ -420,26 +426,37 @@ class KlipperFlashOverviewMenu(BaseMenu):
             ║ sure everything is correct, start the process. If any ║
             ║ parameter needs to be changed, you can go back (B)    ║
             ║ step by step or abort and start from the beginning.   ║
-            ║{subheader:-^64}║
+            ║{subheader:─^64}║
+            ║                                                       ║
             """
         )[1:]
 
-        menu += f"   ● MCU: {COLOR_CYAN}{mcu}{RESET_FORMAT}\n"
-        menu += f"   ● Connection: {COLOR_CYAN}{conn_type}{RESET_FORMAT}\n"
-        menu += f"   ● Flash method: {COLOR_CYAN}{method}{RESET_FORMAT}\n"
-        menu += f"   ● Flash command: {COLOR_CYAN}{command}{RESET_FORMAT}\n"
+        menu += textwrap.dedent(
+            f"""
+            ║ MCU: {COLOR_CYAN}{mcu:<48}{RESET_FORMAT} ║
+            ║ Connection: {COLOR_CYAN}{conn_type:<41}{RESET_FORMAT} ║
+            ║ Flash method: {COLOR_CYAN}{method:<39}{RESET_FORMAT} ║
+            ║ Flash command: {COLOR_CYAN}{command:<38}{RESET_FORMAT} ║
+            """
+        )[1:]
 
         if self.flash_options.flash_method is FlashMethod.SD_CARD:
-            menu += f"   ● Board type: {COLOR_CYAN}{board}{RESET_FORMAT}\n"
-            menu += f"   ● Baudrate: {COLOR_CYAN}{baudrate}{RESET_FORMAT}\n"
+            menu += textwrap.dedent(
+                f"""
+                ║ Board type: {COLOR_CYAN}{board:<41}{RESET_FORMAT} ║
+                ║ Baudrate: {COLOR_CYAN}{baudrate:<43}{RESET_FORMAT} ║
+                """
+            )[1:]
 
         menu += textwrap.dedent(
             """
+            ║                                                       ║
             ╟───────────────────────────────────────────────────────╢
             ║  Y) Start flash process                               ║
             ║  N) Abort - Return to Advanced Menu                   ║
+            ╟───────────────────────────────────────────────────────╢
             """
-        )
+        )[1:]
         print(menu, end="")
 
     def execute_flash(self, **kwargs):

kiauh/components/klipperscreen/klipperscreen.py

@@ -103,8 +103,8 @@ def patch_klipperscreen_update_manager(instances: List[Moonraker]) -> None:
         options=[
             ("type", "git_repo"),
             ("path", KLIPPERSCREEN_DIR.as_posix()),
-            ("orgin", KLIPPERSCREEN_REPO),
-            ("manages_servcies", "KlipperScreen"),
+            ("origin", KLIPPERSCREEN_REPO),
+            ("managed_services", "KlipperScreen"),
             ("env", f"{KLIPPERSCREEN_ENV_DIR}/bin/python"),
             ("requirements", KLIPPERSCREEN_REQ_FILE.as_posix()),
             ("install_script", KLIPPERSCREEN_INSTALL_SCRIPT.as_posix()),

kiauh/core/backup_manager/backup_manager.py

@@ -17,6 +17,10 @@ from core.logger import Logger
 from utils.common import get_current_date
 
 
+class BackupManagerException(Exception):
+    pass
+
+
 # noinspection PyUnusedLocal
 # noinspection PyMethodMayBeStatic
 class BackupManager:
@@ -65,7 +69,7 @@ class BackupManager:
 
     def backup_directory(
         self, name: str, source: Path, target: Path | None = None
-    ) -> None:
+    ) -> Path | None:
         Logger.print_status(f"Creating backup of {name} in {target} ...")
 
         if source is None or not Path(source).exists():
@@ -76,15 +80,15 @@ class BackupManager:
         try:
             date = get_current_date().get("date")
             time = get_current_date().get("time")
-            shutil.copytree(
-                source,
-                target.joinpath(f"{name.lower()}-{date}-{time}"),
-                ignore=self.ignore_folders_func,
-            )
+            backup_target = target.joinpath(f"{name.lower()}-{date}-{time}")
+            shutil.copytree(source, backup_target, ignore=self.ignore_folders_func)
             Logger.print_ok("Backup successful!")
+
+            return backup_target
+
         except OSError as e:
             Logger.print_error(f"Unable to backup directory '{source}':\n{e}")
-            return
+            raise BackupManagerException(f"Unable to backup directory '{source}':\n{e}")
 
     def ignore_folders_func(self, dirpath, filenames) -> List[str]:
         return (

kiauh/core/menus/__init__.py

@@ -22,7 +22,10 @@ class Option:
     :param opt_data: Can be used to pass any additional data to the menu option
     """
 
-    method: Type[Callable] | None = None
+    def __repr__(self):
+        return f"Option(method={self.method.__name__}, opt_index={self.opt_index}, opt_data={self.opt_data})"
+
+    method: Type[Callable]
     opt_index: str = ""
     opt_data: Any = None
 

kiauh/core/menus/base_menu.py

@@ -25,6 +25,7 @@ from core.constants import (
 )
 from core.logger import Logger
 from core.menus import FooterType, Option
+from utils.input_utils import get_selection_input
 
 
 def clear() -> None:
@@ -141,7 +142,7 @@ class BaseMenu(metaclass=PostInitCaller):
     def __go_to_help(self, **kwargs) -> None:
         if self.help_menu is None:
             return
-        self.help_menu(previous_menu=self).run()
+        self.help_menu(previous_menu=self.__class__).run()
 
     def __exit(self, **kwargs) -> None:
         Logger.print_ok("###### Happy printing!", False)
@@ -177,46 +178,20 @@ class BaseMenu(metaclass=PostInitCaller):
         self.print_menu()
         self.print_footer()
 
-    def validate_user_input(self, usr_input: str) -> Option:
-        """
-        Validate the user input and either return an Option, a string or None
-        :param usr_input: The user input in form of a string
-        :return: Option, str or None
-        """
-        usr_input = usr_input.lower()
-        option = self.options.get(
-            usr_input,
-            Option(method=None, opt_index="", opt_data=None),
-        )
-
-        # if option/usr_input is None/empty string, we execute the menus default option if specified
-        if (option is None or usr_input == "") and self.default_option is not None:
-            self.default_option.opt_index = usr_input
-            return self.default_option
-
-        # user selected a regular option
-        option.opt_index = usr_input
-        return option
-
-    def handle_user_input(self) -> Option:
-        """Handle the user input, return the validated input or print an error."""
-        while True:
-            print(f"{COLOR_CYAN}###### {self.input_label_txt}: {RESET_FORMAT}", end="")
-            usr_input = input().lower()
-            validated_input = self.validate_user_input(usr_input)
-
-            if validated_input.method is not None:
-                return validated_input
-            else:
-                Logger.print_error("Invalid input!", False)
-
     def run(self) -> None:
         """Start the menu lifecycle. When this function returns, the lifecycle of the menu ends."""
         try:
             self.display_menu()
-            option = self.handle_user_input()
-            option.method(opt_index=option.opt_index, opt_data=option.opt_data)
+            option = get_selection_input(self.input_label_txt, self.options)
+            selected_option: Option = self.options.get(option)
+
+            selected_option.method(
+                opt_index=selected_option.opt_index,
+                opt_data=selected_option.opt_data,
+            )
+
             self.run()
+
         except Exception as e:
             Logger.print_error(
                 f"An unexpected error occured:\n{e}\n{traceback.format_exc()}"

kiauh/core/menus/main_menu.py

@@ -57,7 +57,8 @@ class MainMenu(BaseMenu):
         self.footer_type: FooterType = FooterType.QUIT
 
         self.version = ""
-        self.kl_status = self.kl_repo = self.mr_status = self.mr_repo = ""
+        self.kl_status = self.kl_owner = self.kl_repo = ""
+        self.mr_status = self.mr_owner = self.mr_repo = ""
         self.ms_status = self.fl_status = self.ks_status = self.mb_status = ""
         self.cn_status = self.cc_status = self.oe_status = ""
         self._init_status()
@@ -103,6 +104,7 @@ class MainMenu(BaseMenu):
         status_data: ComponentStatus = status_fn(*args)
         code: int = status_data.status
         status: StatusText = StatusMap[code]
+        owner: str = status_data.owner
         repo: str = status_data.repo
         instance_count: int = status_data.instances
 
@@ -111,6 +113,7 @@ class MainMenu(BaseMenu):
             count_txt = f": {instance_count}"
 
         setattr(self, f"{name}_status", self._format_by_code(code, status, count_txt))
+        setattr(self, f"{name}_owner", f"{COLOR_CYAN}{owner}{RESET_FORMAT}")
         setattr(self, f"{name}_repo", f"{COLOR_CYAN}{repo}{RESET_FORMAT}")
 
     def _format_by_code(self, code: int, status: str, count: str) -> str:
@@ -140,17 +143,19 @@ class MainMenu(BaseMenu):
             ║ {color}{header:~^{count}}{RESET_FORMAT} ║
             ╟──────────────────┬────────────────────────────────────╢
             ║  0) [Log-Upload] │   Klipper: {self.kl_status:<{pad1}} ║
-            ║                  │      Repo: {self.kl_repo:<{pad1}} ║
-            ║  1) [Install]    ├────────────────────────────────────╢
-            ║  2) [Update]     │ Moonraker: {self.mr_status:<{pad1}} ║
-            ║  3) [Remove]     │      Repo: {self.mr_repo:<{pad1}} ║
-            ║  4) [Advanced]   ├────────────────────────────────────╢
-            ║  5) [Backup]     │        Mainsail: {self.ms_status:<{pad2}} ║
+            ║                  │     Owner: {self.kl_owner:<{pad1}} ║
+            ║  1) [Install]    │      Repo: {self.kl_repo:<{pad1}} ║
+            ║  2) [Update]     ├────────────────────────────────────╢
+            ║  3) [Remove]     │ Moonraker: {self.mr_status:<{pad1}} ║
+            ║  4) [Advanced]   │     Owner: {self.mr_owner:<{pad1}} ║
+            ║  5) [Backup]     │      Repo: {self.mr_repo:<{pad1}} ║
+            ║                  ├────────────────────────────────────╢
+            ║  S) [Settings]   │        Mainsail: {self.ms_status:<{pad2}} ║
             ║                  │          Fluidd: {self.fl_status:<{pad2}} ║
-            ║  S) [Settings]   │   Client-Config: {self.cc_status:<{pad2}} ║
-            ║                  │                                    ║
-            ║ Community:       │   KlipperScreen: {self.ks_status:<{pad2}} ║
-            ║  E) [Extensions] │     Mobileraker: {self.mb_status:<{pad2}} ║
+            ║ Community:       │   Client-Config: {self.cc_status:<{pad2}} ║
+            ║  E) [Extensions] │                                    ║
+            ║                  │   KlipperScreen: {self.ks_status:<{pad2}} ║
+            ║                  │     Mobileraker: {self.mb_status:<{pad2}} ║
             ║                  │  OctoEverywhere: {self.oe_status:<{pad2}} ║
             ║                  │       Crowsnest: {self.cn_status:<{pad2}} ║
             ╟──────────────────┼────────────────────────────────────╢

kiauh/core/menus/settings_menu.py

@@ -8,24 +8,16 @@
 # ======================================================================= #
 from __future__ import annotations
 
-import shutil
 import textwrap
-from pathlib import Path
-from typing import Tuple, Type
+from typing import Literal, Tuple, Type
 
-from components.klipper import KLIPPER_DIR
-from components.klipper.klipper import Klipper
-from components.moonraker import MOONRAKER_DIR
-from components.moonraker.moonraker import Moonraker
 from core.constants import COLOR_CYAN, COLOR_GREEN, RESET_FORMAT
-from core.instance_manager.instance_manager import InstanceManager
 from core.logger import DialogType, Logger
 from core.menus import Option
 from core.menus.base_menu import BaseMenu
-from core.settings.kiauh_settings import KiauhSettings
-from utils.git_utils import git_clone_wrapper
+from core.settings.kiauh_settings import KiauhSettings, RepoSettings
+from procedures.switch_repo import run_switch_repo_routine
 from utils.input_utils import get_confirm, get_string_input
-from utils.instance_utils import get_instances
 
 
 # noinspection PyUnusedLocal
@@ -105,22 +97,28 @@ class SettingsMenu(BaseMenu):
         self.mainsail_unstable = self.settings.mainsail.unstable_releases
         self.fluidd_unstable = self.settings.fluidd.unstable_releases
 
-    def _format_repo_str(self, repo_name: str) -> None:
-        repo = self.settings.get(repo_name, "repo_url")
-        repo = f"{'/'.join(repo.rsplit('/', 2)[-2:])}"
-        branch = self.settings.get(repo_name, "branch")
-        branch = f"({COLOR_CYAN}@ {branch}{RESET_FORMAT})"
-        setattr(self, f"{repo_name}_repo", f"{COLOR_CYAN}{repo}{RESET_FORMAT} {branch}")
+    def _format_repo_str(self, repo_name: Literal["klipper", "moonraker"]) -> None:
+        repo: RepoSettings = self.settings[repo_name]
+        repo_str = f"{'/'.join(repo.repo_url.rsplit('/', 2)[-2:])}"
+        branch_str = f"({COLOR_CYAN}@ {repo.branch}{RESET_FORMAT})"
+
+        setattr(
+            self,
+            f"{repo_name}_repo",
+            f"{COLOR_CYAN}{repo_str}{RESET_FORMAT} {branch_str}",
+        )
 
     def _gather_input(self) -> Tuple[str, str]:
         Logger.print_dialog(
             DialogType.ATTENTION,
             [
-                "There is no input validation in place! Make sure your"
-                " input is valid and has no typos! For any change to"
-                " take effect, the repository must be cloned again. "
-                "Make sure you don't have any ongoing prints running, "
-                "as the services will be restarted!"
+                "There is no input validation in place! Make sure your the input is "
+                "valid and has no typos or invalid characters! For the change to take "
+                "effect, the new repository will be cloned. A backup of the old "
+                "repository will be created.",
+                "\n\n",
+                "Make sure you don't have any ongoing prints running, as the services "
+                "will be restarted during this process! You will loose any ongoing print!",
             ],
         )
         repo = get_string_input(
@@ -134,7 +132,7 @@ class SettingsMenu(BaseMenu):
 
         return repo, branch
 
-    def _set_repo(self, repo_name: str) -> None:
+    def _set_repo(self, repo_name: Literal["klipper", "moonraker"]) -> None:
         repo_url, branch = self._gather_input()
         display_name = repo_name.capitalize()
         Logger.print_dialog(
@@ -148,10 +146,13 @@ class SettingsMenu(BaseMenu):
         )
 
         if get_confirm("Apply changes?", allow_go_back=True):
-            self.settings.set(repo_name, "repo_url", repo_url)
-            self.settings.set(repo_name, "branch", branch)
+            repo: RepoSettings = self.settings[repo_name]
+            repo.repo_url = repo_url
+            repo.branch = branch
+
             self.settings.save()
             self._load_settings()
+
             Logger.print_ok("Changes saved!")
         else:
             Logger.print_info(
@@ -161,31 +162,10 @@ class SettingsMenu(BaseMenu):
 
         Logger.print_status(f"Switching to {display_name}'s new source repository ...")
         self._switch_repo(repo_name)
-        Logger.print_ok(f"Switched to {repo_url} at branch {branch}!")
 
-    def _switch_repo(self, name: str) -> None:
-        target_dir: Path
-        if name == "klipper":
-            target_dir = KLIPPER_DIR
-            _type = Klipper
-        elif name == "moonraker":
-            target_dir = MOONRAKER_DIR
-            _type = Moonraker
-        else:
-            Logger.print_error("Invalid repository name!")
-            return
-
-        if target_dir.exists():
-            shutil.rmtree(target_dir)
-
-        instances = get_instances(_type)
-        InstanceManager.stop_all(instances)
-
-        repo = self.settings.get(name, "repo_url")
-        branch = self.settings.get(name, "branch")
-        git_clone_wrapper(repo, target_dir, branch)
-
-        InstanceManager.start_all(instances)
+    def _switch_repo(self, name: Literal["klipper", "moonraker"]) -> None:
+        repo: RepoSettings = self.settings[name]
+        run_switch_repo_routine(name, repo)
 
     def set_klipper_repo(self, **kwargs) -> None:
         self._set_repo("klipper")

kiauh/core/settings/kiauh_settings.py

@@ -8,6 +8,9 @@
 # ======================================================================= #
 from __future__ import annotations
 
+from dataclasses import dataclass, field
+from typing import Any
+
 from core.logger import DialogType, Logger
 from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
     NoOptionError,
@@ -22,33 +25,21 @@ DEFAULT_CFG = PROJECT_ROOT.joinpath("default.kiauh.cfg")
 CUSTOM_CFG = PROJECT_ROOT.joinpath("kiauh.cfg")
 
 
+@dataclass
 class AppSettings:
-    def __init__(self) -> None:
-        self.backup_before_update = None
+    backup_before_update: bool | None = field(default=None)
 
 
-class KlipperSettings:
-    def __init__(self) -> None:
-        self.repo_url = None
-        self.branch = None
+@dataclass
+class RepoSettings:
+    repo_url: str | None = field(default=None)
+    branch: str | None = field(default=None)
 
 
-class MoonrakerSettings:
-    def __init__(self) -> None:
-        self.repo_url = None
-        self.branch = None
-
-
-class MainsailSettings:
-    def __init__(self) -> None:
-        self.port = None
-        self.unstable_releases = None
-
-
-class FluiddSettings:
-    def __init__(self) -> None:
-        self.port = None
-        self.unstable_releases = None
+@dataclass
+class WebUiSettings:
+    port: str | None = field(default=None)
+    unstable_releases: bool | None = field(default=None)
 
 
 # noinspection PyUnusedLocal
@@ -61,6 +52,12 @@ class KiauhSettings:
             cls._instance = super(KiauhSettings, cls).__new__(cls, *args, **kwargs)
         return cls._instance
 
+    def __repr__(self) -> str:
+        return f"KiauhSettings(kiauh={self.kiauh}, klipper={self.klipper}, moonraker={self.moonraker}, mainsail={self.mainsail}, fluidd={self.fluidd})"
+
+    def __getitem__(self, item: str) -> Any:
+        return getattr(self, item)
+
     def __init__(self) -> None:
         if not hasattr(self, "__initialized"):
             self.__initialized = False
@@ -69,20 +66,10 @@ class KiauhSettings:
         self.__initialized = True
         self.config = SimpleConfigParser()
         self.kiauh = AppSettings()
-        self.klipper = KlipperSettings()
-        self.moonraker = MoonrakerSettings()
-        self.mainsail = MainsailSettings()
-        self.fluidd = FluiddSettings()
-
-        self.kiauh.backup_before_update = None
-        self.klipper.repo_url = None
-        self.klipper.branch = None
-        self.moonraker.repo_url = None
-        self.moonraker.branch = None
-        self.mainsail.port = None
-        self.mainsail.unstable_releases = None
-        self.fluidd.port = None
-        self.fluidd.unstable_releases = None
+        self.klipper = RepoSettings()
+        self.moonraker = RepoSettings()
+        self.mainsail = WebUiSettings()
+        self.fluidd = WebUiSettings()
 
         self._load_config()
 
@@ -102,22 +89,8 @@ class KiauhSettings:
         except AttributeError:
             raise
 
-    def set(self, section: str, option: str, value: str | int | bool) -> None:
-        """
-        Set a value in the settings state by providing the section and option name as strings.
-        Prefer direct access to the properties, as it is usually safer!
-        :param section: The section name as string.
-        :param option: The option name as string.
-        :param value: The value to set as string, int or bool.
-        """
-        try:
-            section = getattr(self, section)
-            section.option = value  # type: ignore
-        except AttributeError:
-            raise
-
     def save(self) -> None:
-        self._set_config_options()
+        self._set_config_options_state()
         self.config.write(CUSTOM_CFG)
         self._load_config()
 
@@ -129,7 +102,7 @@ class KiauhSettings:
         self.config.read(cfg)
 
         self._validate_cfg()
-        self._read_settings()
+        self._apply_settings_from_file()
 
     def _validate_cfg(self) -> None:
         try:
@@ -171,7 +144,7 @@ class KiauhSettings:
         if v.isdigit() or v.lower() == "true" or v.lower() == "false":
             raise ValueError
 
-    def _read_settings(self) -> None:
+    def _apply_settings_from_file(self) -> None:
         self.kiauh.backup_before_update = self.config.getboolean(
             "kiauh", "backup_before_update"
         )
@@ -188,7 +161,7 @@ class KiauhSettings:
             "fluidd", "unstable_releases"
         )
 
-    def _set_config_options(self) -> None:
+    def _set_config_options_state(self) -> None:
         self.config.set(
             "kiauh",
             "backup_before_update",

kiauh/core/submodules/simple_config_parser/src/simple_config_parser/simple_config_parser.py

@@ -85,10 +85,46 @@ class DuplicateOptionError(Exception):
 class SimpleConfigParser:
     """A customized config parser targeted at handling Klipper style config files"""
 
-    _SECTION_RE = re.compile(r"\s*\[(\w+\s?.+)]\s*([#;].*)?$")
-    _OPTION_RE = re.compile(r"^\s*(\w+)\s*[:=]\s*([^=:].*)\s*([#;].*)?$")
-    _MLOPTION_RE = re.compile(r"^\s*(\w+)\s*[:=]\s*([#;].*)?$")
+    # definition of section line:
+    #  - then line MUST start with an opening square bracket - it is the first section marker
+    #  - the section marker MUST be followed by at least one character - it is the section name
+    #  - the section name MUST be followed by a closing square bracket - it is the second section marker
+    #  - the second section marker MAY be followed by any amount of whitespace characters
+    #  - the second section marker MAY be followed by a # or ; - it is the comment marker
+    #  - the inline comment MAY be of any length and character
+    _SECTION_RE = re.compile(r"\[(.+)]\s*([#;].*)?$")
+
+    # definition of option line:
+    #  - the line MUST start with a word - it is the option name
+    #  - the option name MUST be followed by a colon or an equal sign - it is the separator
+    #  - the separator MUST be followed by a value
+    #    - the separator MAY have any amount of leading or trailing whitespaces
+    #    - the separator MUST NOT be directly followed by a colon or equal sign
+    #  - the value MAY be of any length and character
+    #    - the value MAY contain any amount of trailing whitespaces
+    #    - the value MAY be followed by a # or ; - it is the comment marker
+    #  - the inline comment MAY be of any length and character
+    _OPTION_RE = re.compile(r"^([^:=\s]+)\s?[:=]\s*([^=:].*)\s*([#;].*)?$")
+
+    # definition of multiline option line:
+    #  - the line MUST start with a word - it is the option name
+    #  - the option name MUST be followed by a colon or an equal sign - it is the separator
+    #  - the separator MUST NOT be followed by a value
+    #    - the separator MAY have any amount of leading or trailing whitespaces
+    #    - the separator MUST NOT be directly followed by a colon or equal sign
+    #    - the separator MAY be followed by a # or ; - it is the comment marker
+    #  - the inline comment MAY be of any length and character
+    _MLOPTION_RE = re.compile(r"^([^:=\s]+)\s*[:=]\s*([#;].*)?$")
+
+    # definition of comment line:
+    #  - the line MAY start with any amount of whitespace characters
+    #  - the line MUST contain a # or ; - it is the comment marker
+    #  - the comment marker MAY be followed by any amount of whitespace characters
+    #  - the comment MAY be of any length and character
     _COMMENT_RE = re.compile(r"^\s*([#;].*)?$")
+
+    # definition of empty line:
+    #  - the line MUST contain only whitespace characters
     _EMPTY_LINE_RE = re.compile(r"^\s*$")
 
     BOOLEAN_STATES = {

kiauh/core/submodules/simple_config_parser/tests/features/line_parsing/data/case_parse_option.py

@@ -21,4 +21,8 @@ testcases = [
         "serial",
         "/dev/serial/by-id/<your-mcu-id>",
     ),
+    ("parameter_temperature_(°C): 155", "parameter_temperature_(°C)", "155"),
+    ("parameter_humidity_(%_RH): 45", "parameter_humidity_(%_RH)", "45"),
+    ("parameter_spool_weight_(%): 10", "parameter_spool_weight_(%)", "10"),
+    ("path: /dev/shm/drying_box.json", "path", "/dev/shm/drying_box.json"),
 ]

kiauh/core/submodules/simple_config_parser/tests/features/line_parsing/test_single_line_parsing.py

@@ -14,7 +14,7 @@ def parser():
     return SimpleConfigParser()
 
 
-class TestLineParsing:
+class TestSingleLineParsing:
     @pytest.mark.parametrize("given, expected", [*case_parse_section])
     def test_parse_section(self, parser, given, expected):
         parser._parse_section(given)

kiauh/core/submodules/simple_config_parser/tests/features/line_type_detection/data/case_line_is_multiline_option.py

@@ -14,4 +14,14 @@ testcases = [
     ("", False),
     ("# that's a comment", False),
     ("; that's a comment", False),
+    ("parameter_humidity_(%_RH):", True),
+    ("parameter_spool_weight_(%):", True),
+    ("parameter_temperature_(°C):", True),
+    ("parameter_humidity_(%_RH): 18.123", False),
+    ("parameter_spool_weight_(%): 150", False),
+    ("parameter_temperature_(°C): 30,5", False),
+    ("trusted_clients:", True),
+    ("trusted_clients: 192.168.1.0/24", False),
+    ("cors_domains:", True),
+    ("cors_domains: http://*.lan", False),
 ]

kiauh/core/submodules/simple_config_parser/tests/features/line_type_detection/data/case_line_is_option.py

@@ -26,5 +26,6 @@ testcases = [
     ("description: homing!", True),
     ("description: inline macro :-)", True),
     ("path: %GCODES_DIR%", True),
+    ("path: /dev/shm/drying_box.json", True),
     ("serial = /dev/serial/by-id/<your-mcu-id>", True),
 ]

kiauh/core/types.py

@@ -23,6 +23,7 @@ StatusMap: Dict[StatusCode, StatusText] = {
 @dataclass
 class ComponentStatus:
     status: StatusCode
+    owner: str | None = None
     repo: str | None = None
     local: str | None = None
     remote: str | None = None

kiauh/extensions/telegram_bot/moonraker_telegram_bot_extension.py

@@ -175,7 +175,7 @@ class TelegramBotExtension(BaseExtension):
             options=[
                 ("type", "git_repo"),
                 ("path", str(TG_BOT_DIR)),
-                ("orgin", TG_BOT_REPO),
+                ("origin", TG_BOT_REPO),
                 ("env", env_py),
                 ("requirements", "scripts/requirements.txt"),
                 ("install_script", "scripts/install.sh"),

kiauh/procedures/switch_repo.py

@@ -0,0 +1,154 @@
+# ======================================================================= #
+#  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  #
+# ======================================================================= #
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+from typing import Literal
+
+from components.klipper import (
+    KLIPPER_BACKUP_DIR,
+    KLIPPER_DIR,
+    KLIPPER_ENV_DIR,
+    KLIPPER_REQ_FILE,
+)
+from components.klipper.klipper import Klipper
+from components.klipper.klipper_setup import install_klipper_packages
+from components.moonraker import (
+    MOONRAKER_BACKUP_DIR,
+    MOONRAKER_DIR,
+    MOONRAKER_ENV_DIR,
+    MOONRAKER_REQ_FILE,
+)
+from components.moonraker.moonraker import Moonraker
+from components.moonraker.moonraker_setup import install_moonraker_packages
+from core.backup_manager.backup_manager import BackupManager, BackupManagerException
+from core.instance_manager.instance_manager import InstanceManager
+from core.logger import Logger
+from core.settings.kiauh_settings import RepoSettings
+from utils.git_utils import GitException, get_repo_name, git_clone_wrapper
+from utils.instance_utils import get_instances
+from utils.sys_utils import (
+    VenvCreationFailedException,
+    create_python_venv,
+    install_python_requirements,
+)
+
+
+class RepoSwitchFailedException(Exception):
+    pass
+
+
+def run_switch_repo_routine(
+    name: Literal["klipper", "moonraker"], repo_settings: RepoSettings
+) -> None:
+    repo_dir: Path = KLIPPER_DIR if name == "klipper" else MOONRAKER_DIR
+    env_dir: Path = KLIPPER_ENV_DIR if name == "klipper" else MOONRAKER_ENV_DIR
+    req_file = KLIPPER_REQ_FILE if name == "klipper" else MOONRAKER_REQ_FILE
+    backup_dir: Path = KLIPPER_BACKUP_DIR if name == "klipper" else MOONRAKER_BACKUP_DIR
+    _type = Klipper if name == "klipper" else Moonraker
+
+    # step 1: stop all instances
+    Logger.print_status(f"Stopping all {_type.__name__} instances ...")
+    instances = get_instances(_type)
+    InstanceManager.stop_all(instances)
+
+    repo_dir_backup_path: Path | None = None
+    env_dir_backup_path: Path | None = None
+
+    try:
+        # step 2: backup old repo and env
+        org, repo = get_repo_name(repo_dir)
+        backup_dir = backup_dir.joinpath(org)
+        bm = BackupManager()
+        repo_dir_backup_path = bm.backup_directory(
+            repo_dir.name,
+            repo_dir,
+            backup_dir,
+        )
+        env_dir_backup_path = bm.backup_directory(
+            env_dir.name,
+            env_dir,
+            backup_dir,
+        )
+
+        # step 3: read repo url and branch from settings
+        repo_url = repo_settings.repo_url
+        branch = repo_settings.branch
+
+        if not (repo_url or branch):
+            error = f"Invalid repository URL ({repo_url}) or branch ({branch})!"
+            raise ValueError(error)
+
+        # step 4: clone new repo
+        git_clone_wrapper(repo_url, repo_dir, branch, force=True)
+
+        # step 5: install os dependencies
+        if name == "klipper":
+            install_klipper_packages()
+        elif name == "moonraker":
+            install_moonraker_packages()
+
+        # step 6: recreate python virtualenv
+        Logger.print_status(f"Recreating {_type.__name__} virtualenv ...")
+        if not create_python_venv(env_dir, force=True):
+            raise GitException(f"Failed to recreate virtualenv for {_type.__name__}")
+        else:
+            install_python_requirements(env_dir, req_file)
+
+        Logger.print_ok(f"Switched to {repo_url} at branch {branch}!")
+
+    except BackupManagerException as e:
+        Logger.print_error(f"Error during backup of repository: {e}")
+        raise RepoSwitchFailedException(e)
+
+    except (GitException, VenvCreationFailedException) as e:
+        # if something goes wrong during cloning or recreating the virtualenv,
+        # we restore the backup of the repo and env
+        Logger.print_error(f"Error during repository switch: {e}", start="\n")
+        Logger.print_status(f"Restoring last backup of {_type.__name__} ...")
+        _restore_repo_backup(
+            _type.__name__,
+            env_dir,
+            env_dir_backup_path,
+            repo_dir,
+            repo_dir_backup_path,
+        )
+
+    except RepoSwitchFailedException as e:
+        Logger.print_error(f"Something went wrong: {e}")
+        return
+
+    Logger.print_status(f"Restarting all {_type.__name__} instances ...")
+    InstanceManager.start_all(instances)
+
+
+def _restore_repo_backup(
+    name: str,
+    env_dir: Path,
+    env_dir_backup_path: Path | None,
+    repo_dir: Path,
+    repo_dir_backup_path: Path | None,
+) -> None:
+    # if repo_dir_backup_path is not None and env_dir_backup_path is not None:
+    if not repo_dir_backup_path or not env_dir_backup_path:
+        raise RepoSwitchFailedException(
+            f"Unable to restore backup of {name}! Path of backups directory is None!"
+        )
+
+    try:
+        if repo_dir.exists():
+            shutil.rmtree(repo_dir)
+            shutil.copytree(repo_dir_backup_path, repo_dir)
+        if env_dir.exists():
+            shutil.rmtree(env_dir)
+            shutil.copytree(env_dir_backup_path, env_dir)
+        Logger.print_warn(f"Restored backup of {name} successfully!")
+    except Exception as e:
+        raise RepoSwitchFailedException(f"Error restoring backup: {e}")

kiauh/utils/common.py

@@ -124,10 +124,12 @@ def get_install_status(
     else:
         status = 1  # incomplete
 
+    org, repo = get_repo_name(repo_dir)
     return ComponentStatus(
         status=status,
         instances=instances,
-        repo=get_repo_name(repo_dir),
+        owner=org,
+        repo=repo,
         local=get_local_commit(repo_dir),
         remote=get_remote_commit(repo_dir),
     )

kiauh/utils/git_utils.py

@@ -16,6 +16,10 @@ from utils.input_utils import get_confirm, get_number_input
 from utils.instance_utils import get_instances
 
 
+class GitException(Exception):
+    pass
+
+
 def git_clone_wrapper(
     repo: str, target_dir: Path, branch: str | None = None, force: bool = False
 ) -> None:
@@ -43,10 +47,10 @@ def git_clone_wrapper(
     except CalledProcessError:
         log = "An unexpected error occured during cloning of the repository."
         Logger.print_error(log)
-        return
+        raise GitException(log)
     except OSError as e:
         Logger.print_error(f"Error removing existing repository: {e.strerror}")
-        return
+        raise GitException(f"Error removing existing repository: {e.strerror}")
 
 
 def git_pull_wrapper(repo: str, target_dir: Path) -> None:
@@ -66,20 +70,22 @@ def git_pull_wrapper(repo: str, target_dir: Path) -> None:
         return
 
 
-def get_repo_name(repo: Path) -> str | None:
+def get_repo_name(repo: Path) -> tuple[str, str] | None:
     """
     Helper method to extract the organisation and name of a repository |
     :param repo: repository to extract the values from
     :return: String in form of "<orga>/<name>" or None
     """
     if not repo.exists() or not repo.joinpath(".git").exists():
-        return "-"
+        return "-", "-"
 
     try:
         cmd = ["git", "-C", repo.as_posix(), "config", "--get", "remote.origin.url"]
         result: str = check_output(cmd, stderr=DEVNULL).decode(encoding="utf-8")
         substrings: List[str] = result.strip().split("/")[-2:]
-        return "/".join(substrings).replace(".git", "")
+        return substrings[0], substrings[1]
+
+        # return "/".join(substrings).replace(".git", "")
     except CalledProcessError:
         return None
 

kiauh/utils/input_utils.py

@@ -137,7 +137,7 @@ def get_selection_input(question: str, option_list: List | Dict, default=None) -
         else:
             raise ValueError("Invalid option_list type")
 
-        Logger.print_error(INVALID_CHOICE)
+        Logger.print_error("Invalid option! Please select a valid option.", False)
 
 
 def format_question(question: str, default=None) -> str:

kiauh/utils/sys_utils.py

@@ -39,6 +39,10 @@ SysCtlServiceAction = Literal[
 SysCtlManageAction = Literal["daemon-reload", "reset-failed"]
 
 
+class VenvCreationFailedException(Exception):
+    pass
+
+
 def kill(opt_err_msg: str = "") -> None:
     """
     Kills the application |
@@ -87,11 +91,12 @@ def parse_packages_from_file(source_file: Path) -> List[str]:
     return packages
 
 
-def create_python_venv(target: Path) -> bool:
+def create_python_venv(target: Path, force: bool = False) -> bool:
     """
     Create a python 3 virtualenv at the provided target destination.
     Returns True if the virtualenv was created successfully.
     Returns False if the virtualenv already exists, recreation was declined or creation failed.
+    :param force: Force recreation of the virtualenv
     :param target: Path where to create the virtualenv at
     :return: bool
     """
@@ -106,7 +111,7 @@ def create_python_venv(target: Path) -> bool:
             Logger.print_error(f"Error setting up virtualenv:\n{e}")
             return False
     else:
-        if not get_confirm(
+        if not force and not get_confirm(
             "Virtualenv already exists. Re-create?", default_choice=False
         ):
             Logger.print_info("Skipping re-creation of virtualenv ...")
@@ -174,14 +179,14 @@ def install_python_requirements(target: Path, requirements: Path) -> None:
 
         if result.returncode != 0 or result.stderr:
             Logger.print_error(f"{result.stderr}", False)
-            Logger.print_error("Installing Python requirements failed!")
-            return
+            raise VenvCreationFailedException("Installing Python requirements failed!")
 
         Logger.print_ok("Installing Python requirements successful!")
-    except CalledProcessError as e:
-        log = f"Error installing Python requirements:\n{e.output.decode()}"
+
+    except Exception as e:
+        log = f"Error installing Python requirements: {e}"
         Logger.print_error(log)
-        raise
+        raise VenvCreationFailedException(log)
 
 
 def update_system_package_lists(silent: bool, rls_info_change=False) -> None: