From 68369753fdb51f71e0cadea988ba5f10f33b811f Mon Sep 17 00:00:00 2001 From: dw-0 Date: Sat, 11 Nov 2023 23:56:27 +0100 Subject: [PATCH] refactor(InstanceManager): rework Signed-off-by: Dominik Willner --- kiauh/instance_manager/base_instance.py | 116 +++++++++++----- kiauh/instance_manager/instance_manager.py | 154 +++++++++++++-------- kiauh/modules/klipper/klipper.py | 32 +---- kiauh/modules/klipper/klipper_setup.py | 70 ++++------ kiauh/modules/klipper/klipper_utils.py | 18 +-- 5 files changed, 227 insertions(+), 163 deletions(-) diff --git a/kiauh/instance_manager/base_instance.py b/kiauh/instance_manager/base_instance.py index 3dcda97..df7e92f 100644 --- a/kiauh/instance_manager/base_instance.py +++ b/kiauh/instance_manager/base_instance.py @@ -11,7 +11,11 @@ from abc import abstractmethod, ABC from pathlib import Path -from typing import List, Optional +from typing import List, Union, Optional, Type, TypeVar + +from kiauh.utils.constants import SYSTEMD, CURRENT_USER + +B = TypeVar(name="B", bound="BaseInstance", covariant=True) class BaseInstance(ABC): @@ -21,43 +25,41 @@ class BaseInstance(ABC): def __init__( self, - prefix: Optional[str], - name: Optional[str], - user: Optional[str], - data_dir_name: Optional[str], + suffix: Optional[str], + instance_type: B = B, ): - self._prefix = prefix - self._name = name - self._user = user - self._data_dir_name = data_dir_name - self.data_dir = f"{Path.home()}/{self._data_dir_name}_data" - self.cfg_dir = f"{self.data_dir}/config" - self.log_dir = f"{self.data_dir}/logs" - self.comms_dir = f"{self.data_dir}/comms" - self.sysd_dir = f"{self.data_dir}/systemd" + self._instance_type = instance_type + self._suffix = suffix + self._user = CURRENT_USER + self._data_dir_name = self.get_data_dir_from_suffix() + self._data_dir = f"{Path.home()}/{self._data_dir_name}_data" + self._cfg_dir = f"{self.data_dir}/config" + self._log_dir = f"{self.data_dir}/logs" + self._comms_dir = f"{self.data_dir}/comms" + self._sysd_dir = f"{self.data_dir}/systemd" @property - def prefix(self) -> str: - return self._prefix + def instance_type(self) -> Type["BaseInstance"]: + return self._instance_type - @prefix.setter - def prefix(self, value) -> None: - self._prefix = value + @instance_type.setter + def instance_type(self, value: Type["BaseInstance"]) -> None: + self._instance_type = value @property - def name(self) -> str: - return self._name + def suffix(self) -> str: + return self._suffix - @name.setter - def name(self, value) -> None: - self._name = value + @suffix.setter + def suffix(self, value: Union[str, None]) -> None: + self._suffix = value @property def user(self) -> str: return self._user @user.setter - def user(self, value) -> None: + def user(self, value: str) -> None: self._user = value @property @@ -65,9 +67,49 @@ class BaseInstance(ABC): return self._data_dir_name @data_dir_name.setter - def data_dir_name(self, value) -> None: + def data_dir_name(self, value: str) -> None: self._data_dir_name = value + @property + def data_dir(self): + return self._data_dir + + @data_dir.setter + def data_dir(self, value: str): + self._data_dir = value + + @property + def cfg_dir(self): + return self._cfg_dir + + @cfg_dir.setter + def cfg_dir(self, value: str): + self._cfg_dir = value + + @property + def log_dir(self): + return self._log_dir + + @log_dir.setter + def log_dir(self, value: str): + self._log_dir = value + + @property + def comms_dir(self): + return self._comms_dir + + @comms_dir.setter + def comms_dir(self, value: str): + self._comms_dir = value + + @property + def sysd_dir(self): + return self._sysd_dir + + @sysd_dir.setter + def sysd_dir(self, value: str): + self._sysd_dir = value + @abstractmethod def create(self) -> None: raise NotImplementedError("Subclasses must implement the create method") @@ -76,8 +118,20 @@ class BaseInstance(ABC): def delete(self, del_remnants: bool) -> None: raise NotImplementedError("Subclasses must implement the delete method") - @abstractmethod - def get_service_file_name(self) -> str: - raise NotImplementedError( - "Subclasses must implement the get_service_file_name method" - ) + def get_service_file_name(self, extension: bool = False) -> str: + name = f"{self.__class__.__name__.lower()}" + if self.suffix is not None: + name += f"-{self.suffix}" + + return name if not extension else f"{name}.service" + + def get_service_file_path(self) -> str: + return f"{SYSTEMD}/{self.get_service_file_name(extension=True)}" + + def get_data_dir_from_suffix(self) -> str: + if self._suffix is None: + return "printer" + elif self._suffix.isdigit(): + return f"printer_{self._suffix}" + else: + return self._suffix diff --git a/kiauh/instance_manager/instance_manager.py b/kiauh/instance_manager/instance_manager.py index 5a775c1..56b886b 100644 --- a/kiauh/instance_manager/instance_manager.py +++ b/kiauh/instance_manager/instance_manager.py @@ -12,34 +12,85 @@ import os import re import subprocess -from pathlib import Path -from typing import Optional, List, Type, Union +from typing import List, Optional, Union, TypeVar from kiauh.instance_manager.base_instance import BaseInstance from kiauh.utils.constants import SYSTEMD from kiauh.utils.logger import Logger +I = TypeVar(name="I", bound=BaseInstance, covariant=True) + + # noinspection PyMethodMayBeStatic class InstanceManager: - def __init__( - self, - instance_type: Type[BaseInstance], - current_instance: Optional[BaseInstance] = None, - ) -> None: - self.instance_type = instance_type - self.current_instance = current_instance - self.instance_name = ( - current_instance.name if current_instance is not None else None - ) - self.instances = [] + def __init__(self, instance_type: I) -> None: + self._instance_type = instance_type + self._current_instance: Optional[I] = None + self._instance_suffix: Optional[str] = None + self._instance_service: Optional[str] = None + self._instance_service_path: Optional[str] = None + self._instances: List[I] = [] - def get_current_instance(self) -> BaseInstance: - return self.current_instance + @property + def instance_type(self) -> I: + return self._instance_type - def set_current_instance(self, instance: BaseInstance) -> None: - self.current_instance = instance - self.instance_name = instance.name + @instance_type.setter + def instance_type(self, value: I): + self._instance_type = value + + @property + def current_instance(self) -> I: + return self._current_instance + + @current_instance.setter + def current_instance(self, value: I) -> None: + self._current_instance = value + self.instance_suffix = value.suffix + self.instance_service = value.get_service_file_path() + self.instance_service_path = value.get_service_file_path() + + @property + def instance_suffix(self) -> str: + return self._instance_suffix + + @instance_suffix.setter + def instance_suffix(self, value: str): + self._instance_suffix = value + + @property + def instance_service(self) -> str: + return self._instance_service + + @instance_service.setter + def instance_service(self, value: str): + self._instance_service = value + + @property + def instance_service_path(self) -> str: + return self._instance_service_path + + @instance_service_path.setter + def instance_service_path(self, value: str): + self._instance_service_path = value + + @property + def instances(self) -> List[I]: + print("instances getter called") + if not self._instances: + print("instances none") + self._instances = self._find_instances() + + print("return instances") + print(self._instances) + for instance in self._instances: + print(type(instance), instance.suffix) + return sorted(self._instances, key=lambda x: self._sort_instance_list(x.suffix)) + + @instances.setter + def instances(self, value: List[I]): + self._instances = value def create_instance(self) -> None: if self.current_instance is not None: @@ -62,54 +113,56 @@ class InstanceManager: raise ValueError("current_instance cannot be None") def enable_instance(self) -> None: - Logger.print_info(f"Enabling {self.instance_name}.service ...") + Logger.print_info(f"Enabling {self.instance_service} ...") try: - command = ["sudo", "systemctl", "enable", f"{self.instance_name}.service"] + command = ["sudo", "systemctl", "enable", self.instance_service] if subprocess.run(command, check=True): - Logger.print_ok(f"{self.instance_name}.service enabled.") + Logger.print_ok(f"{self.instance_suffix}.service enabled.") except subprocess.CalledProcessError as e: - Logger.print_error(f"Error enabling service {self.instance_name}.service:") + Logger.print_error( + f"Error enabling service {self.instance_suffix}.service:" + ) Logger.print_error(f"{e}") def disable_instance(self) -> None: - Logger.print_info(f"Disabling {self.instance_name}.service ...") + Logger.print_info(f"Disabling {self.instance_service} ...") try: - command = ["sudo", "systemctl", "disable", f"{self.instance_name}.service"] + command = ["sudo", "systemctl", "disable", self.instance_service] if subprocess.run(command, check=True): - Logger.print_ok(f"{self.instance_name}.service disabled.") + Logger.print_ok(f"{self.instance_service} disabled.") except subprocess.CalledProcessError as e: - Logger.print_error(f"Error disabling service {self.instance_name}.service:") + Logger.print_error(f"Error disabling service {self.instance_service}:") Logger.print_error(f"{e}") def start_instance(self) -> None: - Logger.print_info(f"Starting {self.instance_name}.service ...") + Logger.print_info(f"Starting {self.instance_service} ...") try: - command = ["sudo", "systemctl", "start", f"{self.instance_name}.service"] + command = ["sudo", "systemctl", "start", self.instance_service] if subprocess.run(command, check=True): - Logger.print_ok(f"{self.instance_name}.service started.") + Logger.print_ok(f"{self.instance_service} started.") except subprocess.CalledProcessError as e: - Logger.print_error(f"Error starting service {self.instance_name}.service:") + Logger.print_error(f"Error starting service {self.instance_service}:") Logger.print_error(f"{e}") def start_all_instance(self) -> None: for instance in self.instances: - self.set_current_instance(instance) + self.current_instance = instance self.start_instance() def stop_instance(self) -> None: - Logger.print_info(f"Stopping {self.instance_name}.service ...") + Logger.print_info(f"Stopping {self.instance_service} ...") try: - command = ["sudo", "systemctl", "stop", f"{self.instance_name}.service"] + command = ["sudo", "systemctl", "stop", self.instance_service] if subprocess.run(command, check=True): - Logger.print_ok(f"{self.instance_name}.service stopped.") + Logger.print_ok(f"{self.instance_service} stopped.") except subprocess.CalledProcessError as e: - Logger.print_error(f"Error stopping service {self.instance_name}.service:") + Logger.print_error(f"Error stopping service {self.instance_service}:") Logger.print_error(f"{e}") raise def stop_all_instance(self) -> None: for instance in self.instances: - self.set_current_instance(instance) + self.current_instance = instance self.stop_instance() def reload_daemon(self) -> None: @@ -123,39 +176,30 @@ class InstanceManager: Logger.print_error(f"{e}") raise - def get_instances(self) -> List[BaseInstance]: - if not self.instances: - self._find_instances() - - return sorted(self.instances, key=lambda x: self._sort_instance_list(x.name)) - - def _find_instances(self) -> None: - prefix = self.instance_type.__name__.lower() - single_pattern = re.compile(f"^{prefix}.service$") - multi_pattern = re.compile(f"^{prefix}(-[0-9a-zA-Z]+)?.service$") - + def _find_instances(self) -> List[I]: + name = self.instance_type.__name__.lower() + pattern = re.compile(f"^{name}(-[0-9a-zA-Z]+)?.service$") excluded = self.instance_type.blacklist() + service_list = [ os.path.join(SYSTEMD, service) for service in os.listdir(SYSTEMD) - if multi_pattern.search(service) - and not any(s in service for s in excluded) - or single_pattern.search(service) + if pattern.search(service) and not any(s in service for s in excluded) ] instance_list = [ - self.instance_type(name=self._get_instance_name(Path(service))) + self.instance_type(suffix=self._get_instance_suffix(service)) for service in service_list ] - self.instances = instance_list + return instance_list - def _get_instance_name(self, file_path: Path) -> Union[str, None]: - full_name = str(file_path).split("/")[-1].split(".")[0] + def _get_instance_suffix(self, file_path: str) -> Union[str, None]: + full_name = file_path.split("/")[-1].split(".")[0] return full_name.split("-")[-1] if "-" in full_name else full_name - def _sort_instance_list(self, s): + def _sort_instance_list(self, s: Union[int, str, None]): if s is None: return diff --git a/kiauh/modules/klipper/klipper.py b/kiauh/modules/klipper/klipper.py index 326269f..33c4ec3 100644 --- a/kiauh/modules/klipper/klipper.py +++ b/kiauh/modules/klipper/klipper.py @@ -17,7 +17,7 @@ from typing import List from kiauh.instance_manager.base_instance import BaseInstance from kiauh.modules.klipper import KLIPPER_DIR, KLIPPER_ENV_DIR -from kiauh.utils.constants import CURRENT_USER, SYSTEMD +from kiauh.utils.constants import SYSTEMD from kiauh.utils.logger import Logger from kiauh.utils.system_utils import create_directory @@ -26,15 +26,10 @@ from kiauh.utils.system_utils import create_directory class Klipper(BaseInstance): @classmethod def blacklist(cls) -> List[str]: - return ["None", "klipper", "mcu"] + return ["None", "mcu"] - def __init__(self, name: str): - super().__init__( - name=name, - prefix="klipper", - user=CURRENT_USER, - data_dir_name=self._get_data_dir_from_name(name), - ) + def __init__(self, suffix: str = None): + super().__init__(instance_type=self, suffix=suffix) self.klipper_dir = KLIPPER_DIR self.env_dir = KLIPPER_ENV_DIR self.cfg_file = f"{self.cfg_dir}/printer.cfg" @@ -43,7 +38,7 @@ class Klipper(BaseInstance): self.uds = f"{self.comms_dir}/klippy.sock" def create(self) -> None: - Logger.print_info("Creating Klipper Instance") + Logger.print_info("Creating new Klipper Instance ...") module_path = os.path.dirname(os.path.abspath(__file__)) service_template_path = os.path.join(module_path, "res", "klipper.service") env_template_file_path = os.path.join(module_path, "res", "klipper.env") @@ -69,7 +64,7 @@ class Klipper(BaseInstance): def delete(self, del_remnants: bool) -> None: service_file = self.get_service_file_name(extension=True) - service_file_path = self._get_service_file_path() + service_file_path = self.get_service_file_path() Logger.print_info(f"Deleting Klipper Instance: {service_file}") @@ -116,13 +111,6 @@ class Klipper(BaseInstance): env_file.write(env_file_content) Logger.print_ok(f"Env file created: {env_file_target}") - def get_service_file_name(self, extension=False) -> str: - name = "klipper" if self.name == self.prefix else self.prefix + "-" + self.name - return name if not extension else f"{name}.service" - - def _get_service_file_path(self): - return f"{SYSTEMD}/{self.get_service_file_name(extension=True)}" - def _delete_klipper_remnants(self) -> None: try: Logger.print_info(f"Delete {self.klipper_dir} ...") @@ -137,14 +125,6 @@ class Klipper(BaseInstance): Logger.print_ok("Directories successfully deleted.") - def _get_data_dir_from_name(self, name: str) -> str: - if name == "klipper": - return "printer" - elif int(name.isdigit()): - return f"printer_{name}" - else: - return name - def _prep_service_file(self, service_template_path, env_file_path): try: with open(service_template_path, "r") as template_file: diff --git a/kiauh/modules/klipper/klipper_setup.py b/kiauh/modules/klipper/klipper_setup.py index d38b89f..e19fb99 100644 --- a/kiauh/modules/klipper/klipper_setup.py +++ b/kiauh/modules/klipper/klipper_setup.py @@ -55,41 +55,30 @@ from kiauh.utils.system_utils import ( def run_klipper_setup(install: bool) -> None: instance_manager = InstanceManager(Klipper) - instance_list = instance_manager.get_instances() + instance_list = instance_manager.instances instances_installed = len(instance_list) - is_klipper_installed = check_klipper_installation(instance_manager) + is_klipper_installed = instances_installed > 0 if not install and not is_klipper_installed: Logger.print_warn("Klipper not installed!") return if install: - add_additional = handle_existing_instances(instance_manager) + add_additional = handle_existing_instances(instance_list) if is_klipper_installed and not add_additional: Logger.print_info(EXIT_KLIPPER_SETUP) return - install_klipper(instance_manager) + install_klipper(instance_manager, instance_list) if not install: if instances_installed == 1: - remove_single_instance(instance_manager) + remove_single_instance(instance_manager, instance_list) else: - remove_multi_instance(instance_manager) + remove_multi_instance(instance_manager, instance_list) -def check_klipper_installation(instance_manager: InstanceManager) -> bool: - instance_list = instance_manager.get_instances() - instances_installed = len(instance_list) - - if instances_installed < 1: - return False - - return True - - -def handle_existing_instances(instance_manager: InstanceManager) -> bool: - instance_list = instance_manager.get_instances() +def handle_existing_instances(instance_list: List[Klipper]) -> bool: instance_count = len(instance_list) if instance_count > 0: @@ -100,9 +89,9 @@ def handle_existing_instances(instance_manager: InstanceManager) -> bool: return True -def install_klipper(instance_manager: InstanceManager) -> None: - instance_list = instance_manager.get_instances() - +def install_klipper( + instance_manager: InstanceManager, instance_list: List[Klipper] +) -> None: print_select_instance_count_dialog() question = f"Number of{' additional' if len(instance_list) > 0 else ''} Klipper instances to set up" install_count = get_number_input(question, 1, default=1, allow_go_back=True) @@ -110,7 +99,7 @@ def install_klipper(instance_manager: InstanceManager) -> None: Logger.print_info(EXIT_KLIPPER_SETUP) return - instance_names = set_instance_names(instance_list, install_count) + instance_names = set_instance_suffix(instance_list, install_count) if instance_names is None: Logger.print_info(EXIT_KLIPPER_SETUP) return @@ -119,11 +108,9 @@ def install_klipper(instance_manager: InstanceManager) -> None: setup_klipper_prerequesites() convert_single_to_multi = ( - True - if len(instance_list) == 1 - and instance_list[0].name is None + len(instance_list) == 1 + and instance_list[0].suffix is None and install_count >= 1 - else False ) for name in instance_names: @@ -131,7 +118,7 @@ def install_klipper(instance_manager: InstanceManager) -> None: handle_single_to_multi_conversion(instance_manager, name) convert_single_to_multi = False else: - instance_manager.set_current_instance(Klipper(name=name)) + instance_manager.current_instance = Klipper(suffix=name) instance_manager.create_instance() instance_manager.enable_instance() @@ -181,19 +168,17 @@ def install_klipper_packages(klipper_dir: Path) -> None: install_system_packages(packages) -def set_instance_names(instance_list, install_count: int) -> List[Union[str, None]]: +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 ["klipper"] + return [None] # convert single instance install to multi install - elif ( - instance_count == 1 - and instance_list[0].name == "klipper" - and install_count >= 1 - ): + 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 @@ -207,10 +192,11 @@ def set_instance_names(instance_list, install_count: int) -> List[Union[str, Non ) -def remove_single_instance(instance_manager: InstanceManager) -> None: - instance_list = instance_manager.get_instances() +def remove_single_instance( + instance_manager: InstanceManager, instance_list: List[Klipper] +) -> None: try: - instance_manager.set_current_instance(instance_list[0]) + instance_manager.current_instance = instance_list[0] instance_manager.stop_instance() instance_manager.disable_instance() instance_manager.delete_instance(del_remnants=True) @@ -220,8 +206,9 @@ def remove_single_instance(instance_manager: InstanceManager) -> None: return -def remove_multi_instance(instance_manager: InstanceManager) -> None: - instance_list = instance_manager.get_instances() +def remove_multi_instance( + instance_manager: InstanceManager, instance_list: List[Klipper] +) -> None: print_instance_overview(instance_list, show_index=True, show_select_all=True) options = [str(i) for i in range(len(instance_list))] @@ -235,7 +222,7 @@ def remove_multi_instance(instance_manager: InstanceManager) -> None: elif selection == "a".lower(): Logger.print_info("Removing all Klipper instances ...") for instance in instance_list: - instance_manager.set_current_instance(instance) + instance_manager.current_instance = instance instance_manager.stop_instance() instance_manager.disable_instance() instance_manager.delete_instance(del_remnants=True) @@ -244,7 +231,7 @@ def remove_multi_instance(instance_manager: InstanceManager) -> None: Logger.print_info( f"Removing Klipper instance: {instance.get_service_file_name()}" ) - instance_manager.set_current_instance(instance) + instance_manager.current_instance = instance instance_manager.stop_instance() instance_manager.disable_instance() instance_manager.delete_instance(del_remnants=False) @@ -259,7 +246,6 @@ def update_klipper() -> None: return instance_manager = InstanceManager(Klipper) - instance_manager.get_instances() instance_manager.stop_all_instance() cm = ConfigManager() diff --git a/kiauh/modules/klipper/klipper_utils.py b/kiauh/modules/klipper/klipper_utils.py index e018250..3bb8b04 100644 --- a/kiauh/modules/klipper/klipper_utils.py +++ b/kiauh/modules/klipper/klipper_utils.py @@ -35,10 +35,10 @@ def assign_custom_names( instance_names = [] exclude = Klipper.blacklist() - # if an instance_list is provided, exclude all existing instance names + # 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.name) + exclude.append(instance.suffix) for i in range(instance_count + install_count): question = f"Enter name for instance {i + 1}" @@ -93,14 +93,14 @@ def handle_existing_multi_instance_names( def handle_single_to_multi_conversion( instance_manager: InstanceManager, name: str ) -> None: - instance_list = instance_manager.get_instances() - instance_manager.set_current_instance(instance_list[0]) - old_data_dir_name = instance_manager.get_instances()[0].data_dir + 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(del_remnants=False) - instance_manager.set_current_instance(Klipper(name=name)) - new_data_dir_name = instance_manager.get_current_instance().data_dir + instance_manager.current_instance = Klipper(suffix=name) + new_data_dir_name = instance_manager.current_instance.data_dir try: os.rename(old_data_dir_name, new_data_dir_name) except OSError as e: @@ -175,12 +175,12 @@ def handle_disruptive_system_packages() -> None: def has_custom_names(instance_list: List[Klipper]) -> bool: pattern = re.compile("^\d+$") for instance in instance_list: - if not pattern.match(instance.name): + if not pattern.match(instance.suffix): return True return False def get_highest_index(instance_list: List[Klipper]) -> int: - indices = [int(instance.name.split("-")[-1]) for instance in instance_list] + indices = [int(instance.suffix.split("-")[-1]) for instance in instance_list] return max(indices)