refactor(KIAUH): big refactor of instance handling

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

kiauh/core/instance_manager/base_instance.py

@@ -11,7 +11,7 @@
 
 from abc import abstractmethod, ABC
 from pathlib import Path
-from typing import List, Union, Optional, Type, TypeVar
+from typing import List, Type, TypeVar
 
 from kiauh.utils.constants import SYSTEMD, CURRENT_USER
 
@@ -25,7 +25,7 @@ class BaseInstance(ABC):
 
     def __init__(
         self,
-        suffix: Optional[str],
+        suffix: str,
         instance_type: B = B,
     ):
         self._instance_type = instance_type
@@ -52,7 +52,7 @@ class BaseInstance(ABC):
         return self._suffix
 
     @suffix.setter
-    def suffix(self, value: Union[str, None]) -> None:
+    def suffix(self, value: str) -> None:
         self._suffix = value
 
     @property
@@ -144,7 +144,7 @@ class BaseInstance(ABC):
 
     def get_service_file_name(self, extension: bool = False) -> str:
         name = f"{self.__class__.__name__.lower()}"
-        if self.suffix is not None:
+        if self.suffix != "":
             name += f"-{self.suffix}"
 
         return name if not extension else f"{name}.service"
@@ -153,7 +153,7 @@ class BaseInstance(ABC):
         return SYSTEMD.joinpath(self.get_service_file_name(extension=True))
 
     def get_data_dir_name_from_suffix(self) -> str:
-        if self._suffix is None:
+        if self._suffix == "":
             return "printer"
         elif self._suffix.isdigit():
             return f"printer_{self._suffix}"

kiauh/core/instance_manager/instance_manager.py

@@ -207,10 +207,8 @@ class InstanceManager:
 
         return instance_list
 
-    def _get_instance_suffix(self, file_path: Path) -> Union[str, None]:
-        full_name = file_path.name.split(".")[0]
-
-        return full_name.split("-")[-1] if "-" in full_name else None
+    def _get_instance_suffix(self, file_path: Path) -> str:
+        return file_path.stem.split("-")[-1] if "-" in file_path.stem else ""
 
     def _sort_instance_list(self, s: Union[int, str, None]):
         if s is None:

kiauh/core/instance_manager/name_scheme.py

@@ -0,0 +1,8 @@
+from enum import unique, Enum
+
+
+@unique
+class NameScheme(Enum):
+    SINGLE = "SINGLE"
+    INDEX = "INDEX"
+    CUSTOM = "CUSTOM"

kiauh/modules/klipper/klipper.py

@@ -9,10 +9,9 @@
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
 
-import shutil
 import subprocess
 from pathlib import Path
-from typing import List, Union
+from typing import List
 
 from kiauh.core.instance_manager.base_instance import BaseInstance
 from kiauh.modules.klipper import KLIPPER_DIR, KLIPPER_ENV_DIR, MODULE_PATH
@@ -26,7 +25,7 @@ class Klipper(BaseInstance):
     def blacklist(cls) -> List[str]:
         return ["None", "mcu"]
 
-    def __init__(self, suffix: str = None):
+    def __init__(self, suffix: str = ""):
         super().__init__(instance_type=self, suffix=suffix)
         self.klipper_dir: Path = KLIPPER_DIR
         self.env_dir: Path = KLIPPER_ENV_DIR

kiauh/modules/klipper/klipper_setup.py

@@ -10,12 +10,13 @@
 # ======================================================================= #
 
 from pathlib import Path
-from typing import List, Union
+from typing import List
 
 from kiauh import KIAUH_CFG
 from kiauh.core.backup_manager.backup_manager import BackupManager
 from kiauh.core.config_manager.config_manager import ConfigManager
 from kiauh.core.instance_manager.instance_manager import InstanceManager
+from kiauh.core.instance_manager.name_scheme import NameScheme
 from kiauh.modules.klipper import (
     EXIT_KLIPPER_SETUP,
     DEFAULT_KLIPPER_REPO_URL,
@@ -25,20 +26,21 @@ from kiauh.modules.klipper import (
 )
 from kiauh.modules.klipper.klipper import Klipper
 from kiauh.modules.klipper.klipper_dialogs import (
-    print_instance_overview,
-    print_select_instance_count_dialog,
     print_update_warn_dialog,
+    print_select_custom_name_dialog,
 )
 from kiauh.modules.klipper.klipper_utils import (
-    handle_convert_single_to_multi_instance_names,
-    handle_new_multi_instance_names,
-    handle_existing_multi_instance_names,
     handle_disruptive_system_packages,
     check_user_groups,
-    handle_single_to_multi_conversion,
+    handle_to_multi_instance_conversion,
     create_example_printer_cfg,
+    detect_name_scheme,
+    add_to_existing,
+    get_install_count,
+    assign_custom_name,
 )
 from kiauh.core.repo_manager.repo_manager import RepoManager
+from kiauh.modules.moonraker.moonraker import Moonraker
 from kiauh.utils.input_utils import get_confirm, get_number_input
 from kiauh.utils.logger import Logger
 from kiauh.utils.system_utils import (
@@ -50,51 +52,82 @@ from kiauh.utils.system_utils import (
 )
 
 
+# TODO: this method needs refactoring! (but it works for now)
 def install_klipper() -> None:
     im = InstanceManager(Klipper)
+    kl_instances: List[Klipper] = im.instances
 
-    add_additional = handle_existing_instances(im.instances)
-    if len(im.instances) > 0 and not add_additional:
+    # ask to add new instances, if there are existing ones
+    if kl_instances and not add_to_existing():
         Logger.print_status(EXIT_KLIPPER_SETUP)
         return
 
-    print_select_instance_count_dialog()
-    question = f"Number of{' additional' if len(im.instances) > 0 else ''} Klipper instances to set up"
-    install_count = get_number_input(question, 1, default=1, allow_go_back=True)
+    install_count = get_install_count()
+    # install_count = None -> user entered "b" to go back
     if install_count is None:
         Logger.print_status(EXIT_KLIPPER_SETUP)
         return
 
-    instance_names = set_instance_suffix(im.instances, install_count)
-    if instance_names is None:
-        Logger.print_status(EXIT_KLIPPER_SETUP)
-        return
+    # create a dict of the size of the existing instances + install count
+    name_scheme = NameScheme.SINGLE
+    single_to_multi = len(kl_instances) == 1 and kl_instances[0].suffix == ""
+    name_dict = {c: "" for c in range(len(kl_instances) + install_count)}
+
+    if (not kl_instances and install_count > 1) or single_to_multi:
+        print_select_custom_name_dialog()
+        if get_confirm("Assign custom names?", False, allow_go_back=True):
+            name_scheme = NameScheme.CUSTOM
+        else:
+            name_scheme = NameScheme.INDEX
+
+    # if there are more moonraker instances installed than klipper, we
+    # load their names into the name_dict, as we will detect and enforce that naming scheme
+    mr_instances: List[Moonraker] = InstanceManager(Moonraker).instances
+    if len(mr_instances) > len(kl_instances):
+        for k, v in enumerate(mr_instances):
+            name_dict[k] = v.suffix
+        name_scheme = detect_name_scheme(mr_instances)
+    elif len(kl_instances) > 1:
+        for k, v in enumerate(kl_instances):
+            name_dict[k] = v.suffix
+        name_scheme = detect_name_scheme(kl_instances)
+
+    # set instance names if multiple instances will be created
+    if name_scheme != NameScheme.SINGLE:
+        for k in name_dict:
+            if name_dict[k] == "" and name_scheme == NameScheme.INDEX:
+                name_dict[k] = str(k + 1)
+            elif name_dict[k] == "" and name_scheme == NameScheme.CUSTOM:
+                assign_custom_name(k, name_dict)
 
     create_example_cfg = get_confirm("Create example printer.cfg?")
 
-    if len(im.instances) < 1:
+    if not kl_instances:
         setup_klipper_prerequesites()
 
-    convert_single_to_multi = (
-        len(im.instances) == 1 and im.instances[0].suffix is None and install_count >= 1
-    )
-
-    for name in instance_names:
-        if convert_single_to_multi:
-            current_instance = handle_single_to_multi_conversion(im, name)
-            convert_single_to_multi = False
+    count = 0
+    for name in name_dict:
+        if name_dict[name] in [n.suffix for n in kl_instances]:
+            continue
         else:
-            current_instance = Klipper(suffix=name)
+            count += 1
 
-        im.current_instance = current_instance
+        if single_to_multi:
+            handle_to_multi_instance_conversion(name_dict[name])
+            single_to_multi = False
+            count -= 1
+        else:
+            new_instance = Klipper(suffix=name_dict[name])
+            im.current_instance = new_instance
             im.create_instance()
             im.enable_instance()
-
             if create_example_cfg:
-            create_example_printer_cfg(current_instance)
-
+                create_example_printer_cfg(new_instance)
             im.start_instance()
 
+        if count == install_count:
+            break
+
     im.reload_daemon()
 
     # step 4: check/handle conflicting packages/services
@@ -137,41 +170,6 @@ def install_klipper_packages(klipper_dir: Path) -> None:
     install_system_packages(packages)
 
 
-def handle_existing_instances(instance_list: List[Klipper]) -> bool:
-    instance_count = len(instance_list)
-
-    if instance_count > 0:
-        print_instance_overview(instance_list)
-        if not get_confirm("Add new instances?", allow_go_back=True):
-            return False
-
-    return True
-
-
-def set_instance_suffix(
-    instance_list: List[Klipper], install_count: int
-) -> List[Union[str, None]]:
-    instance_count = len(instance_list)
-
-    # new single instance install
-    if instance_count == 0 and install_count == 1:
-        return [None]
-
-    # convert single instance install to multi install
-    elif instance_count == 1 and install_count >= 1 and instance_list[0].suffix is None:
-        return handle_convert_single_to_multi_instance_names(install_count)
-
-    # new multi instance install
-    elif instance_count == 0 and install_count > 1:
-        return handle_new_multi_instance_names(instance_count, install_count)
-
-    # existing multi instance install
-    elif instance_count > 1:
-        return handle_existing_multi_instance_names(
-            instance_count, install_count, instance_list
-        )
-
-
 def update_klipper() -> None:
     print_update_warn_dialog()
     if not get_confirm("Update Klipper now?"):

kiauh/modules/klipper/klipper_utils.py

@@ -20,16 +20,20 @@ from pathlib import Path
 from typing import List, Union, Literal, Dict
 
 from kiauh.core.config_manager.config_manager import ConfigManager
+from kiauh.core.instance_manager.base_instance import BaseInstance
 from kiauh.core.instance_manager.instance_manager import InstanceManager
+from kiauh.core.instance_manager.name_scheme import NameScheme
 from kiauh.modules.klipper import MODULE_PATH, KLIPPER_DIR, KLIPPER_ENV_DIR
 from kiauh.modules.klipper.klipper import Klipper
 from kiauh.modules.klipper.klipper_dialogs import (
     print_missing_usergroup_dialog,
-    print_select_custom_name_dialog,
+    print_instance_overview,
+    print_select_instance_count_dialog,
 )
+from kiauh.modules.moonraker.moonraker_utils import moonraker_to_multi_conversion
 from kiauh.utils.common import get_install_status_common, get_repo_name
 from kiauh.utils.constants import CURRENT_USER
-from kiauh.utils.input_utils import get_confirm, get_string_input
+from kiauh.utils.input_utils import get_confirm, get_string_input, get_number_input
 from kiauh.utils.logger import Logger
 from kiauh.utils.system_utils import mask_system_service
 
@@ -41,84 +45,57 @@ def get_klipper_status() -> Dict[Literal["status", "repo"], str]:
     }
 
 
-def assign_custom_names(
-    instance_count: int, install_count: int, instance_list: List[Klipper] = None
-) -> List[str]:
-    instance_names = []
-    exclude = Klipper.blacklist()
-
-    # if an instance_list is provided, exclude all existing instance suffixes
-    if instance_list is not None:
-        for instance in instance_list:
-            exclude.append(instance.suffix)
-
-    for i in range(instance_count + install_count):
-        question = f"Enter name for instance {i + 1}"
-        name = get_string_input(question, exclude=exclude)
-        instance_names.append(name)
-        exclude.append(name)
-
-    return instance_names
+def add_to_existing() -> bool:
+    kl_instances = InstanceManager(Klipper).instances
+    print_instance_overview(kl_instances)
+    return get_confirm("Add new instances?", allow_go_back=True)
 
 
-def handle_convert_single_to_multi_instance_names(
-    install_count: int,
-) -> Union[List[str], None]:
-    print_select_custom_name_dialog()
-    choice = get_confirm("Assign custom names?", False, allow_go_back=True)
-    if choice is True:
-        # instance_count = 0 and install_count + 1 as we want to assign a new name to the existing single install
-        return assign_custom_names(0, install_count + 1)
-    elif choice is False:
-        # "install_count + 2" as we need to account for the existing single install
-        _range = range(1, install_count + 2)
-        return [str(i) for i in _range]
-
-    return None
+def get_install_count() -> Union[int, None]:
+    kl_instances = InstanceManager(Klipper).instances
+    print_select_instance_count_dialog()
+    question = f"Number of{' additional' if len(kl_instances) > 0 else ''} Klipper instances to set up"
+    return get_number_input(question, 1, default=1, allow_go_back=True)
 
 
-def handle_new_multi_instance_names(
-    instance_count: int, install_count: int
-) -> Union[List[str], None]:
-    print_select_custom_name_dialog()
-    choice = get_confirm("Assign custom names?", False, allow_go_back=True)
-    if choice is True:
-        return assign_custom_names(instance_count, install_count)
-    elif choice is False:
-        _range = range(1, install_count + 1)
-        return [str(i) for i in _range]
-
-    return None
+def assign_custom_name(key: int, name_dict: Dict[int, str]) -> None:
+    existing_names = []
+    existing_names.extend(Klipper.blacklist())
+    existing_names.extend(name_dict[n] for n in name_dict)
+    question = f"Enter name for instance {key + 1}"
+    name_dict[key] = get_string_input(question, exclude=existing_names)
 
 
-def handle_existing_multi_instance_names(
-    instance_count: int, install_count: int, instance_list: List[Klipper]
-) -> List[str]:
-    if has_custom_names(instance_list):
-        return assign_custom_names(instance_count, install_count, instance_list)
+def handle_to_multi_instance_conversion(new_name: str) -> None:
+    Logger.print_status("Converting single instance to multi instances ...")
+    klipper_to_multi_conversion(new_name)
+    moonraker_to_multi_conversion(new_name)
+
+
+def klipper_to_multi_conversion(new_name: str) -> None:
+    Logger.print_status("Convert Klipper single to multi instance ...")
+    im = InstanceManager(Klipper)
+    im.current_instance = im.instances[0]
+    # temporarily store the data dir path
+    old_data_dir = im.instances[0].data_dir
+    # remove the old single instance
+    im.stop_instance()
+    im.disable_instance()
+    im.delete_instance()
+    # create a new klipper instance with the new name
+    im.current_instance = Klipper(suffix=new_name)
+    new_data_dir: Path = im.current_instance.data_dir
+
+    # rename the old data dir and use it for the new instance
+    Logger.print_status(f"Rename '{old_data_dir}' to '{new_data_dir}' ...")
+    if not new_data_dir.is_dir():
+        old_data_dir.rename(new_data_dir)
     else:
-        start = get_highest_index(instance_list) + 1
-        _range = range(start, start + install_count)
-        return [str(i) for i in _range]
+        Logger.print_info(f"'{new_data_dir}' already exist. Skipped ...")
 
-
-def handle_single_to_multi_conversion(
-    instance_manager: InstanceManager, name: str
-) -> Klipper:
-    instance_list = instance_manager.instances
-    instance_manager.current_instance = instance_list[0]
-    old_data_dir_name = instance_manager.instances[0].data_dir
-    instance_manager.stop_instance()
-    instance_manager.disable_instance()
-    instance_manager.delete_instance()
-    instance_manager.current_instance = Klipper(suffix=name)
-    new_data_dir_name = instance_manager.current_instance.data_dir
-    try:
-        Path(old_data_dir_name).rename(new_data_dir_name)
-        return instance_manager.current_instance
-    except OSError as e:
-        log = f"Cannot rename {old_data_dir_name} to {new_data_dir_name}:\n{e}"
-        Logger.print_error(log)
+    im.create_instance()
+    im.enable_instance()
+    im.start_instance()
 
 
 def check_user_groups():
@@ -189,13 +166,13 @@ def handle_disruptive_system_packages() -> None:
             Logger.print_warn(warn_msg)
 
 
-def has_custom_names(instance_list: List[Klipper]) -> bool:
+def detect_name_scheme(instance_list: List[BaseInstance]) -> NameScheme:
     pattern = re.compile("^\d+$")
     for instance in instance_list:
         if not pattern.match(instance.suffix):
-            return True
+            return NameScheme.CUSTOM
 
-    return False
+    return NameScheme.INDEX
 
 
 def get_highest_index(instance_list: List[Klipper]) -> int:

kiauh/modules/moonraker/moonraker.py

@@ -9,7 +9,6 @@
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
 
-import shutil
 import subprocess
 from pathlib import Path
 from typing import List, Union
@@ -27,7 +26,7 @@ class Moonraker(BaseInstance):
     def blacklist(cls) -> List[str]:
         return ["None", "mcu"]
 
-    def __init__(self, suffix: str = None):
+    def __init__(self, suffix: str = ""):
         super().__init__(instance_type=self, suffix=suffix)
         self.moonraker_dir: Path = MOONRAKER_DIR
         self.env_dir: Path = MOONRAKER_ENV_DIR

kiauh/modules/moonraker/moonraker_setup.py

@@ -54,21 +54,23 @@ from kiauh.utils.system_utils import (
 
 
 def check_moonraker_install_requirements() -> bool:
-    kl_im = InstanceManager(Klipper)
-    kl_instance_list = kl_im.instances
-    kl_instance_count = len(kl_instance_list)
-
     if not (sys.version_info.major >= 3 and sys.version_info.minor >= 7):
         Logger.print_error("Versioncheck failed!")
         Logger.print_error("Python 3.7 or newer required to run Moonraker.")
         return False
 
-    is_klipper_installed = kl_instance_count > 0
-    if not is_klipper_installed:
+    kl_instance_count = len(InstanceManager(Klipper).instances)
+    if kl_instance_count < 1:
         Logger.print_warn("Klipper not installed!")
         Logger.print_warn("Moonraker cannot be installed! Install Klipper first.")
         return False
 
+    mr_instance_count = len(InstanceManager(Moonraker).instances)
+    if mr_instance_count >= kl_instance_count:
+        Logger.print_warn("Unable to install more Moonraker instances!")
+        Logger.print_warn("More Klipper instances required.")
+        return False
+
     return True
 
 

kiauh/modules/moonraker/moonraker_utils.py

@@ -10,9 +10,10 @@
 # ======================================================================= #
 
 import shutil
-from typing import Dict, Literal
+from typing import Dict, Literal, List
 
 from kiauh.core.config_manager.config_manager import ConfigManager
+from kiauh.core.instance_manager.instance_manager import InstanceManager
 from kiauh.modules.moonraker import (
     DEFAULT_MOONRAKER_PORT,
     MODULE_PATH,
@@ -83,3 +84,36 @@ def create_example_moonraker_conf(
 
     cm.write_config()
     Logger.print_ok(f"Example moonraker.conf created in '{instance.cfg_dir}'")
+
+
+def moonraker_to_multi_conversion(new_name: str) -> None:
+    """
+    Converts the first instance in the List of Moonraker instances to an instance
+    with a new name. This method will be called when converting from a single Klipper
+    instance install to a multi instance install when Moonraker is also already
+    installed with a single instance.
+    :param new_name: new name the previous single instance is renamed to
+    :return: None
+    """
+    im = InstanceManager(Moonraker)
+    instances: List[Moonraker] = im.instances
+    if not instances:
+        return
+
+    # in case there are multiple Moonraker instances, we don't want to do anything
+    if len(instances) > 1:
+        Logger.print_info("More than a single Moonraker instance found. Skipped ...")
+        return
+
+    Logger.print_status("Convert Moonraker single to multi instance ...")
+    # remove the old single instance
+    im.current_instance = im.instances[0]
+    im.stop_instance()
+    im.disable_instance()
+    im.delete_instance()
+    # create a new klipper instance with the new name
+    im.current_instance = Moonraker(suffix=new_name)
+    # create, enable and start the new moonraker instance
+    im.create_instance()
+    im.enable_instance()
+    im.start_instance()