diff --git a/kiauh/core/instance_manager/base_instance.py b/kiauh/core/instance_manager/base_instance.py index bccacee..903c76b 100644 --- a/kiauh/core/instance_manager/base_instance.py +++ b/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}" diff --git a/kiauh/core/instance_manager/instance_manager.py b/kiauh/core/instance_manager/instance_manager.py index 411a25b..4a45e39 100644 --- a/kiauh/core/instance_manager/instance_manager.py +++ b/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: diff --git a/kiauh/core/instance_manager/name_scheme.py b/kiauh/core/instance_manager/name_scheme.py new file mode 100644 index 0000000..bfd9e2c --- /dev/null +++ b/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" diff --git a/kiauh/modules/klipper/klipper.py b/kiauh/modules/klipper/klipper.py index 1f11fa2..6a43452 100644 --- a/kiauh/modules/klipper/klipper.py +++ b/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 diff --git a/kiauh/modules/klipper/klipper_setup.py b/kiauh/modules/klipper/klipper_setup.py index efa78e5..a30c27c 100644 --- a/kiauh/modules/klipper/klipper_setup.py +++ b/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,50 +52,81 @@ 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 - im.create_instance() - im.enable_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(new_instance) + im.start_instance() - if create_example_cfg: - create_example_printer_cfg(current_instance) - - im.start_instance() + if count == install_count: + break im.reload_daemon() @@ -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?"): diff --git a/kiauh/modules/klipper/klipper_utils.py b/kiauh/modules/klipper/klipper_utils.py index baf1b3a..b6980dc 100644 --- a/kiauh/modules/klipper/klipper_utils.py +++ b/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: diff --git a/kiauh/modules/moonraker/moonraker.py b/kiauh/modules/moonraker/moonraker.py index 78d60c7..904c813 100644 --- a/kiauh/modules/moonraker/moonraker.py +++ b/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 diff --git a/kiauh/modules/moonraker/moonraker_setup.py b/kiauh/modules/moonraker/moonraker_setup.py index 430e157..8208d16 100644 --- a/kiauh/modules/moonraker/moonraker_setup.py +++ b/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 diff --git a/kiauh/modules/moonraker/moonraker_utils.py b/kiauh/modules/moonraker/moonraker_utils.py index b9c8981..9045b42 100644 --- a/kiauh/modules/moonraker/moonraker_utils.py +++ b/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()