diff --git a/kiauh/components/klipper/klipper_remove.py b/kiauh/components/klipper/klipper_remove.py deleted file mode 100644 index 3f9cc41..0000000 --- a/kiauh/components/klipper/klipper_remove.py +++ /dev/null @@ -1,117 +0,0 @@ -# ======================================================================= # -# Copyright (C) 2020 - 2025 Dominik Willner # -# # -# 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 - -from typing import List - -from components.klipper import KLIPPER_DIR, KLIPPER_ENV_DIR -from components.klipper.klipper import Klipper -from components.klipper.klipper_dialogs import print_instance_overview -from core.instance_manager.instance_manager import InstanceManager -from core.logger import Logger -from core.services.message_service import Message -from core.types.color import Color -from utils.fs_utils import run_remove_routines -from utils.input_utils import get_selection_input -from utils.instance_utils import get_instances -from utils.sys_utils import unit_file_exists - - -def run_klipper_removal( - remove_service: bool, - remove_dir: bool, - remove_env: bool, -) -> Message: - completion_msg = Message( - title="Klipper Removal Process completed", - color=Color.GREEN, - ) - klipper_instances: List[Klipper] = get_instances(Klipper) - - if remove_service: - Logger.print_status("Removing Klipper instances ...") - if klipper_instances: - instances_to_remove = select_instances_to_remove(klipper_instances) - remove_instances(instances_to_remove) - instance_names = [i.service_file_path.stem for i in instances_to_remove] - txt = f"● Klipper instances removed: {', '.join(instance_names)}" - completion_msg.text.append(txt) - else: - Logger.print_info("No Klipper Services installed! Skipped ...") - - if (remove_dir or remove_env) and unit_file_exists("klipper", suffix="service"): - completion_msg.text = [ - "Some Klipper services are still installed:", - f"● '{KLIPPER_DIR}' was not removed, even though selected for removal.", - f"● '{KLIPPER_ENV_DIR}' was not removed, even though selected for removal.", - ] - else: - if remove_dir: - Logger.print_status("Removing Klipper local repository ...") - if run_remove_routines(KLIPPER_DIR): - completion_msg.text.append("● Klipper local repository removed") - if remove_env: - Logger.print_status("Removing Klipper Python environment ...") - if run_remove_routines(KLIPPER_ENV_DIR): - completion_msg.text.append("● Klipper Python environment removed") - - if completion_msg.text: - completion_msg.text.insert(0, "The following actions were performed:") - else: - completion_msg.color = Color.YELLOW - completion_msg.centered = True - completion_msg.text = ["Nothing to remove."] - - return completion_msg - - -def select_instances_to_remove(instances: List[Klipper]) -> List[Klipper] | None: - start_index = 1 - options = [str(i + start_index) for i in range(len(instances))] - options.extend(["a", "b"]) - instance_map = {options[i]: instances[i] for i in range(len(instances))} - - print_instance_overview( - instances, - start_index=start_index, - show_index=True, - show_select_all=True, - ) - selection = get_selection_input("Select Klipper instance to remove", options) - - instances_to_remove = [] - if selection == "b": - return None - elif selection == "a": - instances_to_remove.extend(instances) - else: - instances_to_remove.append(instance_map[selection]) - - return instances_to_remove - - -def remove_instances( - instance_list: List[Klipper] | None, -) -> None: - if not instance_list: - return - - for instance in instance_list: - Logger.print_status(f"Removing instance {instance.service_file_path.stem} ...") - InstanceManager.remove(instance) - delete_klipper_env_file(instance) - - -def delete_klipper_env_file(instance: Klipper): - Logger.print_status(f"Remove '{instance.env_file}'") - if not instance.env_file.exists(): - msg = f"Env file in {instance.base.sysd_dir} not found. Skipped ..." - Logger.print_info(msg) - return - run_remove_routines(instance.env_file) diff --git a/kiauh/components/klipper/klipper_setup.py b/kiauh/components/klipper/klipper_setup.py deleted file mode 100644 index 3dcee38..0000000 --- a/kiauh/components/klipper/klipper_setup.py +++ /dev/null @@ -1,242 +0,0 @@ -# ======================================================================= # -# Copyright (C) 2020 - 2025 Dominik Willner # -# # -# 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 - -from pathlib import Path -from typing import Dict, List, Tuple - -from components.klipper import ( - EXIT_KLIPPER_SETUP, - KLIPPER_DIR, - KLIPPER_ENV_DIR, - KLIPPER_INSTALL_SCRIPT, - KLIPPER_REQ_FILE, -) -from components.klipper.klipper import Klipper -from components.klipper.klipper_dialogs import ( - print_select_custom_name_dialog, -) -from components.klipper.klipper_utils import ( - assign_custom_name, - backup_klipper_dir, - check_user_groups, - create_example_printer_cfg, - get_install_count, - handle_disruptive_system_packages, -) -from components.moonraker.moonraker import Moonraker -from components.webui_client.client_utils import ( - get_existing_clients, -) -from core.instance_manager.instance_manager import InstanceManager -from core.logger import DialogType, Logger -from core.settings.kiauh_settings import KiauhSettings -from utils.common import check_install_dependencies -from utils.git_utils import git_clone_wrapper, git_pull_wrapper -from utils.input_utils import get_confirm -from utils.instance_utils import get_instances -from utils.sys_utils import ( - cmd_sysctl_manage, - cmd_sysctl_service, - create_python_venv, - install_python_requirements, - parse_packages_from_file, -) - - -def install_klipper() -> None: - Logger.print_status("Installing Klipper ...") - - klipper_list: List[Klipper] = get_instances(Klipper) - moonraker_list: List[Moonraker] = get_instances(Moonraker) - match_moonraker: bool = False - - # if there are more moonraker instances than klipper instances, ask the user to - # match the klipper instance count to the count of moonraker instances with the same suffix - if len(moonraker_list) > len(klipper_list): - is_confirmed = display_moonraker_info(moonraker_list) - if not is_confirmed: - Logger.print_status(EXIT_KLIPPER_SETUP) - return - match_moonraker = True - - install_count, name_dict = get_install_count_and_name_dict( - klipper_list, moonraker_list - ) - - if install_count == 0: - Logger.print_status(EXIT_KLIPPER_SETUP) - return - - is_multi_install = install_count > 1 or (len(name_dict) >= 1 and install_count >= 1) - if not name_dict and install_count == 1: - name_dict = {0: ""} - elif is_multi_install and not match_moonraker: - custom_names = use_custom_names_or_go_back() - if custom_names is None: - Logger.print_status(EXIT_KLIPPER_SETUP) - return - - handle_instance_names(install_count, name_dict, custom_names) - - create_example_cfg = get_confirm("Create example printer.cfg?") - # run the actual installation - try: - run_klipper_setup(klipper_list, name_dict, create_example_cfg) - except Exception as e: - Logger.print_error(e) - Logger.print_error("Klipper installation failed!") - return - - -def run_klipper_setup( - klipper_list: List[Klipper], name_dict: Dict[int, str], create_example_cfg: bool -) -> None: - if not klipper_list: - setup_klipper_prerequesites() - - for i in name_dict: - # skip this iteration if there is already an instance with the name - if name_dict[i] in [n.suffix for n in klipper_list]: - continue - - instance = Klipper(suffix=name_dict[i]) - instance.create() - cmd_sysctl_service(instance.service_file_path.name, "enable") - - if create_example_cfg: - # if a client-config is installed, include it in the new example cfg - clients = get_existing_clients() - create_example_printer_cfg(instance, clients) - - cmd_sysctl_service(instance.service_file_path.name, "start") - - cmd_sysctl_manage("daemon-reload") - - # step 4: check/handle conflicting packages/services - handle_disruptive_system_packages() - - # step 5: check for required group membership - check_user_groups() - - -def handle_instance_names( - install_count: int, name_dict: Dict[int, str], custom_names: bool -) -> None: - for i in range(install_count): # 3 - key: int = len(name_dict.keys()) + 1 - if custom_names: - assign_custom_name(key, name_dict) - else: - name_dict[key] = str(len(name_dict) + 1) - - -def get_install_count_and_name_dict( - klipper_list: List[Klipper], moonraker_list: List[Moonraker] -) -> Tuple[int, Dict[int, str]]: - install_count: int | None - if len(moonraker_list) > len(klipper_list): - install_count = len(moonraker_list) - name_dict = {i: moonraker.suffix for i, moonraker in enumerate(moonraker_list)} - else: - install_count = get_install_count() - name_dict = {i: klipper.suffix for i, klipper in enumerate(klipper_list)} - - if install_count is None: - Logger.print_status(EXIT_KLIPPER_SETUP) - return 0, {} - - return install_count, name_dict - - -def setup_klipper_prerequesites() -> None: - settings = KiauhSettings() - repo = settings.klipper.repo_url - branch = settings.klipper.branch - - git_clone_wrapper(repo, KLIPPER_DIR, branch) - - # install klipper dependencies and create python virtualenv - try: - install_klipper_packages() - if create_python_venv(KLIPPER_ENV_DIR): - install_python_requirements(KLIPPER_ENV_DIR, KLIPPER_REQ_FILE) - except Exception: - Logger.print_error("Error during installation of Klipper requirements!") - raise - - -def install_klipper_packages() -> None: - script = KLIPPER_INSTALL_SCRIPT - packages = parse_packages_from_file(script) - - # Add pkg-config for rp2040 build - packages.append("pkg-config") - - # Add dbus requirement for DietPi distro - if Path("/boot/dietpi/.version").exists(): - packages.append("dbus") - - check_install_dependencies({*packages}) - - -def update_klipper() -> None: - Logger.print_dialog( - DialogType.WARNING, - [ - "Do NOT continue if there are ongoing prints running!", - "All Klipper instances will be restarted during the update process and " - "ongoing prints WILL FAIL.", - ], - ) - - if not get_confirm("Update Klipper now?"): - return - - settings = KiauhSettings() - if settings.kiauh.backup_before_update: - backup_klipper_dir() - - instances = get_instances(Klipper) - InstanceManager.stop_all(instances) - - git_pull_wrapper(repo=settings.klipper.repo_url, target_dir=KLIPPER_DIR) - - # install possible new system packages - install_klipper_packages() - # install possible new python dependencies - install_python_requirements(KLIPPER_ENV_DIR, KLIPPER_REQ_FILE) - - InstanceManager.start_all(instances) - - -def use_custom_names_or_go_back() -> bool | None: - print_select_custom_name_dialog() - _input: bool | None = get_confirm( - "Assign custom names?", - False, - allow_go_back=True, - ) - return _input - - -def display_moonraker_info(moonraker_list: List[Moonraker]) -> bool: - # todo: only show the klipper instances that are not already installed - Logger.print_dialog( - DialogType.INFO, - [ - "Existing Moonraker instances detected:", - *[f"● {m.service_file_path.stem}" for m in moonraker_list], - "\n\n", - "The following Klipper instances will be installed:", - *[f"● klipper-{m.suffix}" for m in moonraker_list], - ], - ) - _input: bool = get_confirm("Proceed with installation?") - return _input diff --git a/kiauh/components/klipper/klipper_utils.py b/kiauh/components/klipper/klipper_utils.py index 32a3941..f4026e0 100644 --- a/kiauh/components/klipper/klipper_utils.py +++ b/kiauh/components/klipper/klipper_utils.py @@ -11,6 +11,7 @@ from __future__ import annotations import grp import os import shutil +from pathlib import Path from subprocess import CalledProcessError, run from typing import Dict, List @@ -18,6 +19,7 @@ from components.klipper import ( KLIPPER_BACKUP_DIR, KLIPPER_DIR, KLIPPER_ENV_DIR, + KLIPPER_INSTALL_SCRIPT, MODULE_PATH, ) from components.klipper.klipper import Klipper @@ -37,10 +39,10 @@ from core.submodules.simple_config_parser.src.simple_config_parser.simple_config SimpleConfigParser, ) from core.types.component_status import ComponentStatus -from utils.common import get_install_status +from utils.common import check_install_dependencies, get_install_status from utils.input_utils import get_confirm, get_number_input, get_string_input from utils.instance_utils import get_instances -from utils.sys_utils import cmd_sysctl_service +from utils.sys_utils import cmd_sysctl_service, parse_packages_from_file def get_klipper_status() -> ComponentStatus: @@ -194,3 +196,17 @@ def backup_klipper_dir() -> None: bm = BackupManager() bm.backup_directory("klipper", source=KLIPPER_DIR, target=KLIPPER_BACKUP_DIR) bm.backup_directory("klippy-env", source=KLIPPER_ENV_DIR, target=KLIPPER_BACKUP_DIR) + + +def install_klipper_packages() -> None: + script = KLIPPER_INSTALL_SCRIPT + packages = parse_packages_from_file(script) + + # Add pkg-config for rp2040 build + packages.append("pkg-config") + + # Add dbus requirement for DietPi distro + if Path("/boot/dietpi/.version").exists(): + packages.append("dbus") + + check_install_dependencies({*packages}) diff --git a/kiauh/components/klipper/menus/klipper_remove_menu.py b/kiauh/components/klipper/menus/klipper_remove_menu.py index 329ce62..8ec3bcd 100644 --- a/kiauh/components/klipper/menus/klipper_remove_menu.py +++ b/kiauh/components/klipper/menus/klipper_remove_menu.py @@ -11,7 +11,7 @@ from __future__ import annotations import textwrap from typing import Type -from components.klipper import klipper_remove +from components.klipper.services.klipper_setup_service import KlipperSetupService from core.menus import FooterType, Option from core.menus.base_menu import BaseMenu from core.types.color import Color @@ -27,11 +27,13 @@ class KlipperRemoveMenu(BaseMenu): self.previous_menu: Type[BaseMenu] | None = previous_menu self.footer_type = FooterType.BACK - self.remove_klipper_service = False - self.remove_klipper_dir = False - self.remove_klipper_env = False + self.rm_svc = False + self.rm_dir = False + self.rm_env = False self.select_state = False + self.klsvc = KlipperSetupService() + def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None: from core.menus.remove_menu import RemoveMenu @@ -49,10 +51,10 @@ class KlipperRemoveMenu(BaseMenu): def print_menu(self) -> None: checked = f"[{Color.apply('x', Color.CYAN)}]" unchecked = "[ ]" - o1 = checked if self.remove_klipper_service else unchecked - o2 = checked if self.remove_klipper_dir else unchecked - o3 = checked if self.remove_klipper_env else unchecked - sel_state = f"{'Select'if not self.select_state else 'Deselect'} everything" + o1 = checked if self.rm_svc else unchecked + o2 = checked if self.rm_dir else unchecked + o3 = checked if self.rm_env else unchecked + sel_state = f"{'Select' if not self.select_state else 'Deselect'} everything" menu = textwrap.dedent( f""" ╟───────────────────────────────────────────────────────╢ @@ -73,37 +75,28 @@ class KlipperRemoveMenu(BaseMenu): def toggle_all(self, **kwargs) -> None: self.select_state = not self.select_state - self.remove_klipper_service = self.select_state - self.remove_klipper_dir = self.select_state - self.remove_klipper_env = self.select_state + self.rm_svc = self.select_state + self.rm_dir = self.select_state + self.rm_env = self.select_state def toggle_remove_klipper_service(self, **kwargs) -> None: - self.remove_klipper_service = not self.remove_klipper_service + self.rm_svc = not self.rm_svc def toggle_remove_klipper_dir(self, **kwargs) -> None: - self.remove_klipper_dir = not self.remove_klipper_dir + self.rm_dir = not self.rm_dir def toggle_remove_klipper_env(self, **kwargs) -> None: - self.remove_klipper_env = not self.remove_klipper_env + self.rm_env = not self.rm_env def run_removal_process(self, **kwargs) -> None: - if ( - not self.remove_klipper_service - and not self.remove_klipper_dir - and not self.remove_klipper_env - ): + if not self.rm_svc and not self.rm_dir and not self.rm_env: msg = "Nothing selected! Select options to remove first." print(Color.apply(msg, Color.RED)) return - completion_msg = klipper_remove.run_klipper_removal( - self.remove_klipper_service, - self.remove_klipper_dir, - self.remove_klipper_env, - ) - self.message_service.set_message(completion_msg) + self.klsvc.remove(self.rm_svc, self.rm_dir, self.rm_env) - self.remove_klipper_service = False - self.remove_klipper_dir = False - self.remove_klipper_env = False + self.rm_svc = False + self.rm_dir = False + self.rm_env = False self.select_state = False diff --git a/kiauh/components/klipper/services/__init__.py b/kiauh/components/klipper/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kiauh/components/klipper/services/klipper_instance_service.py b/kiauh/components/klipper/services/klipper_instance_service.py new file mode 100644 index 0000000..593e173 --- /dev/null +++ b/kiauh/components/klipper/services/klipper_instance_service.py @@ -0,0 +1,46 @@ +# ======================================================================= # +# Copyright (C) 2020 - 2025 Dominik Willner # +# # +# 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 + +from typing import List + +from components.klipper.klipper import Klipper +from utils.instance_utils import get_instances + + +class KlipperInstanceService: + __cls_instance = None + __instances: List[Klipper] = [] + + def __new__(cls) -> "KlipperInstanceService": + if cls.__cls_instance is None: + cls.__cls_instance = super(KlipperInstanceService, cls).__new__(cls) + return cls.__cls_instance + + def __init__(self) -> None: + if not hasattr(self, "__initialized"): + self.__initialized = False + if self.__initialized: + return + self.__initialized = True + + def load_instances(self) -> None: + self.__instances = get_instances(Klipper) + + def create_new_instance(self, suffix: str) -> Klipper: + instance = Klipper(suffix) + self.__instances.append(instance) + return instance + + def get_all_instances(self) -> List[Klipper]: + return self.__instances + + def get_instance_by_suffix(self, suffix: str) -> Klipper | None: + instances: List[Klipper] = [i for i in self.__instances if i.suffix == suffix] + return instances[0] if instances else None diff --git a/kiauh/components/klipper/services/klipper_setup_service.py b/kiauh/components/klipper/services/klipper_setup_service.py new file mode 100644 index 0000000..bcf5f9d --- /dev/null +++ b/kiauh/components/klipper/services/klipper_setup_service.py @@ -0,0 +1,362 @@ +# ======================================================================= # +# Copyright (C) 2020 - 2025 Dominik Willner # +# # +# 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 + +from copy import copy +from typing import Dict, List, Tuple + +from components.klipper import ( + EXIT_KLIPPER_SETUP, + KLIPPER_DIR, + KLIPPER_ENV_DIR, + KLIPPER_REQ_FILE, +) +from components.klipper.klipper import Klipper +from components.klipper.klipper_dialogs import ( + print_instance_overview, + print_select_custom_name_dialog, +) +from components.klipper.klipper_utils import ( + assign_custom_name, + backup_klipper_dir, + check_user_groups, + create_example_printer_cfg, + get_install_count, + handle_disruptive_system_packages, + install_klipper_packages, +) +from components.klipper.services.klipper_instance_service import KlipperInstanceService +from components.moonraker.moonraker import Moonraker +from components.moonraker.services.moonraker_instance_service import ( + MoonrakerInstanceService, +) +from components.webui_client.client_utils import ( + get_existing_clients, +) +from core.instance_manager.instance_manager import InstanceManager +from core.logger import DialogType, Logger +from core.services.message_service import Message, MessageService +from core.settings.kiauh_settings import KiauhSettings +from core.types.color import Color +from utils.fs_utils import run_remove_routines +from utils.git_utils import git_clone_wrapper, git_pull_wrapper +from utils.input_utils import get_confirm, get_selection_input +from utils.sys_utils import ( + cmd_sysctl_manage, + create_python_venv, + install_python_requirements, + unit_file_exists, +) + + +# noinspection PyMethodMayBeStatic +class KlipperSetupService: + __cls_instance = None + + kisvc: KlipperInstanceService + misvc: MoonrakerInstanceService + msgsvc = MessageService + + settings: KiauhSettings + klipper_list: List[Klipper] + moonraker_list: List[Moonraker] + + def __new__(cls) -> "KlipperSetupService": + if cls.__cls_instance is None: + cls.__cls_instance = super(KlipperSetupService, cls).__new__(cls) + return cls.__cls_instance + + def __init__(self) -> None: + if not hasattr(self, "__initialized"): + self.__initialized = False + if self.__initialized: + return + self.__initialized = True + self.__init_state() + + def __init_state(self) -> None: + self.settings = KiauhSettings() + + self.kisvc = KlipperInstanceService() + self.kisvc.load_instances() + self.klipper_list = self.kisvc.get_all_instances() + + self.misvc = MoonrakerInstanceService() + self.misvc.load_instances() + self.moonraker_list = self.misvc.get_all_instances() + + self.msgsvc = MessageService() + + def __refresh_state(self) -> None: + self.kisvc.load_instances() + self.klipper_list = self.kisvc.get_all_instances() + + self.misvc.load_instances() + self.moonraker_list = self.misvc.get_all_instances() + + def install(self) -> None: + Logger.print_status("Installing Klipper ...") + + match_moonraker: bool = False + + # if there are more moonraker instances than klipper instances, ask the user to + # match the klipper instance count to the count of moonraker instances with the same suffix + if len(self.moonraker_list) > len(self.klipper_list): + is_confirmed = self.__display_moonraker_info() + if not is_confirmed: + Logger.print_status(EXIT_KLIPPER_SETUP) + return + match_moonraker = True + + install_count, name_dict = self.__get_install_count_and_name_dict() + + if install_count == 0: + Logger.print_status(EXIT_KLIPPER_SETUP) + return + + is_multi_install = install_count > 1 or ( + len(name_dict) >= 1 and install_count >= 1 + ) + if not name_dict and install_count == 1: + name_dict = {0: ""} + elif is_multi_install and not match_moonraker: + custom_names = self.__use_custom_names_or_go_back() + if custom_names is None: + Logger.print_status(EXIT_KLIPPER_SETUP) + return + + self.__handle_instance_names(install_count, name_dict, custom_names) + + create_example_cfg = get_confirm("Create example printer.cfg?") + # run the actual installation + try: + self.__run_setup(name_dict, create_example_cfg) + except Exception as e: + Logger.print_error(e) + Logger.print_error("Klipper installation failed!") + return + + def update(self) -> None: + Logger.print_dialog( + DialogType.WARNING, + [ + "Do NOT continue if there are ongoing prints running!", + "All Klipper instances will be restarted during the update process and " + "ongoing prints WILL FAIL.", + ], + ) + + if not get_confirm("Update Klipper now?"): + return + + self.__refresh_state() + + if self.settings.kiauh.backup_before_update: + backup_klipper_dir() + + InstanceManager.stop_all(self.klipper_list) + git_pull_wrapper(self.settings.klipper.repo_url, KLIPPER_DIR) + install_klipper_packages() + install_python_requirements(KLIPPER_ENV_DIR, KLIPPER_REQ_FILE) + InstanceManager.start_all(self.klipper_list) + + def remove( + self, + remove_service: bool, + remove_dir: bool, + remove_env: bool, + ) -> None: + self.__refresh_state() + + completion_msg = Message( + title="Klipper Removal Process completed", + color=Color.GREEN, + ) + + if remove_service: + Logger.print_status("Removing Klipper instances ...") + if self.klipper_list: + instances_to_remove = self.__get_instances_to_remove() + self.__remove_instances(instances_to_remove) + if instances_to_remove: + instance_names = [ + i.service_file_path.stem for i in instances_to_remove + ] + txt = f"● Klipper instances removed: {', '.join(instance_names)}" + completion_msg.text.append(txt) + else: + Logger.print_info("No Klipper Services installed! Skipped ...") + + if (remove_dir or remove_env) and unit_file_exists("klipper", suffix="service"): + completion_msg.text = [ + "Some Klipper services are still installed:", + f"● '{KLIPPER_DIR}' was not removed, even though selected for removal.", + f"● '{KLIPPER_ENV_DIR}' was not removed, even though selected for removal.", + ] + else: + if remove_dir: + Logger.print_status("Removing Klipper local repository ...") + if run_remove_routines(KLIPPER_DIR): + completion_msg.text.append("● Klipper local repository removed") + if remove_env: + Logger.print_status("Removing Klipper Python environment ...") + if run_remove_routines(KLIPPER_ENV_DIR): + completion_msg.text.append("● Klipper Python environment removed") + + if completion_msg.text: + completion_msg.text.insert(0, "The following actions were performed:") + else: + completion_msg.color = Color.YELLOW + completion_msg.centered = True + completion_msg.text = ["Nothing to remove."] + + self.msgsvc.set_message(completion_msg) + + def __get_install_count_and_name_dict(self) -> Tuple[int, Dict[int, str]]: + install_count: int | None + if len(self.moonraker_list) > len(self.klipper_list): + install_count = len(self.moonraker_list) + name_dict = { + i: moonraker.suffix for i, moonraker in enumerate(self.moonraker_list) + } + else: + install_count = get_install_count() + name_dict = { + i: klipper.suffix for i, klipper in enumerate(self.klipper_list) + } + + if install_count is None: + Logger.print_status(EXIT_KLIPPER_SETUP) + return 0, {} + + return install_count, name_dict + + def __run_setup(self, name_dict: Dict[int, str], create_example_cfg: bool) -> None: + if not self.klipper_list: + self.__install_deps() + + for i in name_dict: + # skip this iteration if there is already an instance with the name + if name_dict[i] in [n.suffix for n in self.klipper_list]: + continue + + instance = Klipper(suffix=name_dict[i]) + instance.create() + InstanceManager.enable(instance) + + if create_example_cfg: + # if a client-config is installed, include it in the new example cfg + clients = get_existing_clients() + create_example_printer_cfg(instance, clients) + + InstanceManager.start(instance) + + cmd_sysctl_manage("daemon-reload") + + # step 4: check/handle conflicting packages/services + handle_disruptive_system_packages() + + # step 5: check for required group membership + check_user_groups() + + def __install_deps(self) -> None: + repo = self.settings.klipper.repo_url + branch = self.settings.klipper.branch + + git_clone_wrapper(repo, KLIPPER_DIR, branch) + + try: + install_klipper_packages() + if create_python_venv(KLIPPER_ENV_DIR): + install_python_requirements(KLIPPER_ENV_DIR, KLIPPER_REQ_FILE) + except Exception: + Logger.print_error("Error during installation of Klipper requirements!") + raise + + def __display_moonraker_info(self) -> bool: + # todo: only show the klipper instances that are not already installed + Logger.print_dialog( + DialogType.INFO, + [ + "Existing Moonraker instances detected:", + *[f"● {m.service_file_path.stem}" for m in self.moonraker_list], + "\n\n", + "The following Klipper instances will be installed:", + *[f"● klipper-{m.suffix}" for m in self.moonraker_list], + ], + ) + _input: bool = get_confirm("Proceed with installation?") + return _input + + def __handle_instance_names( + self, install_count: int, name_dict: Dict[int, str], custom_names: bool + ) -> None: + for i in range(install_count): # 3 + key: int = len(name_dict.keys()) + 1 + if custom_names: + assign_custom_name(key, name_dict) + else: + name_dict[key] = str(len(name_dict) + 1) + + def __use_custom_names_or_go_back(self) -> bool | None: + print_select_custom_name_dialog() + _input: bool | None = get_confirm( + "Assign custom names?", + False, + allow_go_back=True, + ) + return _input + + def __get_instances_to_remove(self) -> List[Klipper] | None: + start_index = 1 + curr_instances: List[Klipper] = self.klipper_list + instance_count = len(curr_instances) + + options = [str(i + start_index) for i in range(instance_count)] + options.extend(["a", "b"]) + instance_map = {options[i]: self.klipper_list[i] for i in range(instance_count)} + + print_instance_overview( + self.klipper_list, + start_index=start_index, + show_index=True, + show_select_all=True, + ) + selection = get_selection_input("Select Klipper instance to remove", options) + + if selection == "b": + return None + elif selection == "a": + return copy(self.klipper_list) + + return [instance_map[selection]] + + def __remove_instances( + self, + instance_list: List[Klipper] | None, + ) -> None: + if not instance_list: + return + + for instance in instance_list: + Logger.print_status( + f"Removing instance {instance.service_file_path.stem} ..." + ) + InstanceManager.remove(instance) + self.__delete_klipper_env_file(instance) + + self.__refresh_state() + + def __delete_klipper_env_file(self, instance: Klipper): + Logger.print_status(f"Remove '{instance.env_file}'") + if not instance.env_file.exists(): + msg = f"Env file in {instance.base.sysd_dir} not found. Skipped ..." + Logger.print_info(msg) + return + run_remove_routines(instance.env_file) diff --git a/kiauh/components/moonraker/services/moonraker_instance_service.py b/kiauh/components/moonraker/services/moonraker_instance_service.py index a08457a..86b8b19 100644 --- a/kiauh/components/moonraker/services/moonraker_instance_service.py +++ b/kiauh/components/moonraker/services/moonraker_instance_service.py @@ -1,3 +1,11 @@ +# ======================================================================= # +# Copyright (C) 2020 - 2025 Dominik Willner # +# # +# 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 from typing import Dict, List diff --git a/kiauh/core/logger.py b/kiauh/core/logger.py index 67689c9..559b1af 100644 --- a/kiauh/core/logger.py +++ b/kiauh/core/logger.py @@ -33,6 +33,7 @@ BORDER_TITLE: str = "┠────────────────── BORDER_LEFT: str = "┃" BORDER_RIGHT: str = "┃" + class Logger: @staticmethod def print_info(msg, prefix=True, start="", end="\n") -> None: @@ -99,12 +100,14 @@ class Logger: print(Color.apply(BORDER_TITLE, color)) if content: - print(Logger.format_content( - content, - LINE_WIDTH, - color, - center_content, - )) + print( + Logger.format_content( + content, + LINE_WIDTH, + color, + center_content, + ) + ) print(Color.apply(BORDER_BOTTOM, color)) diff --git a/kiauh/core/menus/install_menu.py b/kiauh/core/menus/install_menu.py index 7ae1afe..831a355 100644 --- a/kiauh/core/menus/install_menu.py +++ b/kiauh/core/menus/install_menu.py @@ -12,7 +12,7 @@ import textwrap from typing import Type from components.crowsnest.crowsnest import install_crowsnest -from components.klipper import klipper_setup +from components.klipper.services.klipper_setup_service import KlipperSetupService from components.klipperscreen.klipperscreen import install_klipperscreen from components.moonraker import moonraker_setup from components.webui_client.client_config.client_config_setup import ( @@ -36,6 +36,7 @@ class InstallMenu(BaseMenu): self.title = "Installation Menu" self.title_color = Color.GREEN self.previous_menu: Type[BaseMenu] | None = previous_menu + self.klsvc = KlipperSetupService() def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None: from core.menus.main_menu import MainMenu @@ -75,7 +76,7 @@ class InstallMenu(BaseMenu): print(menu, end="") def install_klipper(self, **kwargs) -> None: - klipper_setup.install_klipper() + self.klsvc.install() def install_moonraker(self, **kwargs) -> None: moonraker_setup.install_moonraker() diff --git a/kiauh/core/menus/update_menu.py b/kiauh/core/menus/update_menu.py index f46144a..5b6c77c 100644 --- a/kiauh/core/menus/update_menu.py +++ b/kiauh/core/menus/update_menu.py @@ -12,10 +12,10 @@ import textwrap from typing import Callable, List, Type from components.crowsnest.crowsnest import get_crowsnest_status, update_crowsnest -from components.klipper.klipper_setup import update_klipper from components.klipper.klipper_utils import ( get_klipper_status, ) +from components.klipper.services.klipper_setup_service import KlipperSetupService from components.klipperscreen.klipperscreen import ( get_klipperscreen_status, update_klipperscreen, @@ -193,7 +193,8 @@ class UpdateMenu(BaseMenu): self.upgrade_system_packages() def update_klipper(self, **kwargs) -> None: - self._run_update_routine("klipper", update_klipper) + klsvc = KlipperSetupService() + self._run_update_routine("klipper", klsvc.update) def update_moonraker(self, **kwargs) -> None: self._run_update_routine("moonraker", update_moonraker) diff --git a/kiauh/core/services/message_service.py b/kiauh/core/services/message_service.py index f95c521..bf7edda 100644 --- a/kiauh/core/services/message_service.py +++ b/kiauh/core/services/message_service.py @@ -6,6 +6,7 @@ # # # This file may be distributed under the terms of the GNU GPLv3 license # # ======================================================================= # +from __future__ import annotations from dataclasses import dataclass, field from typing import List @@ -23,12 +24,13 @@ class Message: class MessageService: - _instance = None + __cls_instance = None + __message: Message | None def __new__(cls) -> "MessageService": - if cls._instance is None: - cls._instance = super(MessageService, cls).__new__(cls) - return cls._instance + if cls.__cls_instance is None: + cls.__cls_instance = super(MessageService, cls).__new__(cls) + return cls.__cls_instance def __init__(self) -> None: if not hasattr(self, "__initialized"): @@ -36,24 +38,24 @@ class MessageService: if self.__initialized: return self.__initialized = True - self.message = None + self.__message = None def set_message(self, message: Message) -> None: - self.message = message + self.__message = message def display_message(self) -> None: - if self.message is None: + if self.__message is None: return Logger.print_dialog( title=DialogType.CUSTOM, - content=self.message.text, - custom_title=self.message.title, - custom_color=self.message.color, - center_content=self.message.centered, + content=self.__message.text, + custom_title=self.__message.title, + custom_color=self.__message.color, + center_content=self.__message.centered, ) self.__clear_message() def __clear_message(self) -> None: - self.message = None + self.__message = None diff --git a/kiauh/procedures/switch_repo.py b/kiauh/procedures/switch_repo.py index d6a8240..d490f31 100644 --- a/kiauh/procedures/switch_repo.py +++ b/kiauh/procedures/switch_repo.py @@ -19,7 +19,7 @@ from components.klipper import ( KLIPPER_REQ_FILE, ) from components.klipper.klipper import Klipper -from components.klipper.klipper_setup import install_klipper_packages +from components.klipper.klipper_utils import install_klipper_packages from components.moonraker import ( MOONRAKER_BACKUP_DIR, MOONRAKER_DIR,