diff --git a/kiauh/core/backup_manager/backup_manager.py b/kiauh/core/backup_manager/backup_manager.py index 824f58c..ad9ea5b 100644 --- a/kiauh/core/backup_manager/backup_manager.py +++ b/kiauh/core/backup_manager/backup_manager.py @@ -17,6 +17,10 @@ from core.logger import Logger from utils.common import get_current_date +class BackupManagerException(Exception): + pass + + # noinspection PyUnusedLocal # noinspection PyMethodMayBeStatic class BackupManager: @@ -65,7 +69,7 @@ class BackupManager: def backup_directory( self, name: str, source: Path, target: Path | None = None - ) -> None: + ) -> Path | None: Logger.print_status(f"Creating backup of {name} in {target} ...") if source is None or not Path(source).exists(): @@ -76,15 +80,15 @@ class BackupManager: try: date = get_current_date().get("date") time = get_current_date().get("time") - shutil.copytree( - source, - target.joinpath(f"{name.lower()}-{date}-{time}"), - ignore=self.ignore_folders_func, - ) + backup_target = target.joinpath(f"{name.lower()}-{date}-{time}") + shutil.copytree(source, backup_target, ignore=self.ignore_folders_func) Logger.print_ok("Backup successful!") + + return backup_target + except OSError as e: Logger.print_error(f"Unable to backup directory '{source}':\n{e}") - return + raise BackupManagerException(f"Unable to backup directory '{source}':\n{e}") def ignore_folders_func(self, dirpath, filenames) -> List[str]: return ( diff --git a/kiauh/core/menus/main_menu.py b/kiauh/core/menus/main_menu.py index b7dec43..7366053 100644 --- a/kiauh/core/menus/main_menu.py +++ b/kiauh/core/menus/main_menu.py @@ -57,7 +57,8 @@ class MainMenu(BaseMenu): self.footer_type: FooterType = FooterType.QUIT self.version = "" - self.kl_status = self.kl_repo = self.mr_status = self.mr_repo = "" + self.kl_status = self.kl_owner = self.kl_repo = "" + self.mr_status = self.mr_owner = self.mr_repo = "" self.ms_status = self.fl_status = self.ks_status = self.mb_status = "" self.cn_status = self.cc_status = self.oe_status = "" self._init_status() @@ -103,6 +104,7 @@ class MainMenu(BaseMenu): status_data: ComponentStatus = status_fn(*args) code: int = status_data.status status: StatusText = StatusMap[code] + owner: str = status_data.owner repo: str = status_data.repo instance_count: int = status_data.instances @@ -111,6 +113,7 @@ class MainMenu(BaseMenu): count_txt = f": {instance_count}" setattr(self, f"{name}_status", self._format_by_code(code, status, count_txt)) + setattr(self, f"{name}_owner", f"{COLOR_CYAN}{owner}{RESET_FORMAT}") setattr(self, f"{name}_repo", f"{COLOR_CYAN}{repo}{RESET_FORMAT}") def _format_by_code(self, code: int, status: str, count: str) -> str: @@ -140,17 +143,19 @@ class MainMenu(BaseMenu): ║ {color}{header:~^{count}}{RESET_FORMAT} ║ ╟──────────────────┬────────────────────────────────────╢ ║ 0) [Log-Upload] │ Klipper: {self.kl_status:<{pad1}} ║ - ║ │ Repo: {self.kl_repo:<{pad1}} ║ - ║ 1) [Install] ├────────────────────────────────────╢ - ║ 2) [Update] │ Moonraker: {self.mr_status:<{pad1}} ║ - ║ 3) [Remove] │ Repo: {self.mr_repo:<{pad1}} ║ - ║ 4) [Advanced] ├────────────────────────────────────╢ - ║ 5) [Backup] │ Mainsail: {self.ms_status:<{pad2}} ║ + ║ │ Owner: {self.kl_owner:<{pad1}} ║ + ║ 1) [Install] │ Repo: {self.kl_repo:<{pad1}} ║ + ║ 2) [Update] ├────────────────────────────────────╢ + ║ 3) [Remove] │ Moonraker: {self.mr_status:<{pad1}} ║ + ║ 4) [Advanced] │ Owner: {self.mr_owner:<{pad1}} ║ + ║ 5) [Backup] │ Repo: {self.mr_repo:<{pad1}} ║ + ║ ├────────────────────────────────────╢ + ║ S) [Settings] │ Mainsail: {self.ms_status:<{pad2}} ║ ║ │ Fluidd: {self.fl_status:<{pad2}} ║ - ║ S) [Settings] │ Client-Config: {self.cc_status:<{pad2}} ║ - ║ │ ║ - ║ Community: │ KlipperScreen: {self.ks_status:<{pad2}} ║ - ║ E) [Extensions] │ Mobileraker: {self.mb_status:<{pad2}} ║ + ║ Community: │ Client-Config: {self.cc_status:<{pad2}} ║ + ║ E) [Extensions] │ ║ + ║ │ KlipperScreen: {self.ks_status:<{pad2}} ║ + ║ │ Mobileraker: {self.mb_status:<{pad2}} ║ ║ │ OctoEverywhere: {self.oe_status:<{pad2}} ║ ║ │ Crowsnest: {self.cn_status:<{pad2}} ║ ╟──────────────────┼────────────────────────────────────╢ diff --git a/kiauh/core/menus/settings_menu.py b/kiauh/core/menus/settings_menu.py index 5989d96..86f8007 100644 --- a/kiauh/core/menus/settings_menu.py +++ b/kiauh/core/menus/settings_menu.py @@ -8,24 +8,16 @@ # ======================================================================= # from __future__ import annotations -import shutil import textwrap -from pathlib import Path -from typing import Tuple, Type +from typing import Literal, Tuple, Type -from components.klipper import KLIPPER_DIR -from components.klipper.klipper import Klipper -from components.moonraker import MOONRAKER_DIR -from components.moonraker.moonraker import Moonraker from core.constants import COLOR_CYAN, COLOR_GREEN, RESET_FORMAT -from core.instance_manager.instance_manager import InstanceManager from core.logger import DialogType, Logger from core.menus import Option from core.menus.base_menu import BaseMenu -from core.settings.kiauh_settings import KiauhSettings -from utils.git_utils import git_clone_wrapper +from core.settings.kiauh_settings import KiauhSettings, RepoSettings +from procedures.switch_repo import run_switch_repo_routine from utils.input_utils import get_confirm, get_string_input -from utils.instance_utils import get_instances # noinspection PyUnusedLocal @@ -105,22 +97,28 @@ class SettingsMenu(BaseMenu): self.mainsail_unstable = self.settings.mainsail.unstable_releases self.fluidd_unstable = self.settings.fluidd.unstable_releases - def _format_repo_str(self, repo_name: str) -> None: - repo = self.settings.get(repo_name, "repo_url") - repo = f"{'/'.join(repo.rsplit('/', 2)[-2:])}" - branch = self.settings.get(repo_name, "branch") - branch = f"({COLOR_CYAN}@ {branch}{RESET_FORMAT})" - setattr(self, f"{repo_name}_repo", f"{COLOR_CYAN}{repo}{RESET_FORMAT} {branch}") + def _format_repo_str(self, repo_name: Literal["klipper", "moonraker"]) -> None: + repo: RepoSettings = self.settings[repo_name] + repo_str = f"{'/'.join(repo.repo_url.rsplit('/', 2)[-2:])}" + branch_str = f"({COLOR_CYAN}@ {repo.branch}{RESET_FORMAT})" + + setattr( + self, + f"{repo_name}_repo", + f"{COLOR_CYAN}{repo_str}{RESET_FORMAT} {branch_str}", + ) def _gather_input(self) -> Tuple[str, str]: Logger.print_dialog( DialogType.ATTENTION, [ - "There is no input validation in place! Make sure your" - " input is valid and has no typos! For any change to" - " take effect, the repository must be cloned again. " - "Make sure you don't have any ongoing prints running, " - "as the services will be restarted!" + "There is no input validation in place! Make sure your the input is " + "valid and has no typos or invalid characters! For the change to take " + "effect, the new repository will be cloned. A backup of the old " + "repository will be created.", + "\n\n", + "Make sure you don't have any ongoing prints running, as the services " + "will be restarted during this process! You will loose any ongoing print!", ], ) repo = get_string_input( @@ -134,7 +132,7 @@ class SettingsMenu(BaseMenu): return repo, branch - def _set_repo(self, repo_name: str) -> None: + def _set_repo(self, repo_name: Literal["klipper", "moonraker"]) -> None: repo_url, branch = self._gather_input() display_name = repo_name.capitalize() Logger.print_dialog( @@ -148,10 +146,13 @@ class SettingsMenu(BaseMenu): ) if get_confirm("Apply changes?", allow_go_back=True): - self.settings.set(repo_name, "repo_url", repo_url) - self.settings.set(repo_name, "branch", branch) + repo: RepoSettings = self.settings[repo_name] + repo.repo_url = repo_url + repo.branch = branch + self.settings.save() self._load_settings() + Logger.print_ok("Changes saved!") else: Logger.print_info( @@ -161,31 +162,10 @@ class SettingsMenu(BaseMenu): Logger.print_status(f"Switching to {display_name}'s new source repository ...") self._switch_repo(repo_name) - Logger.print_ok(f"Switched to {repo_url} at branch {branch}!") - def _switch_repo(self, name: str) -> None: - target_dir: Path - if name == "klipper": - target_dir = KLIPPER_DIR - _type = Klipper - elif name == "moonraker": - target_dir = MOONRAKER_DIR - _type = Moonraker - else: - Logger.print_error("Invalid repository name!") - return - - if target_dir.exists(): - shutil.rmtree(target_dir) - - instances = get_instances(_type) - InstanceManager.stop_all(instances) - - repo = self.settings.get(name, "repo_url") - branch = self.settings.get(name, "branch") - git_clone_wrapper(repo, target_dir, branch) - - InstanceManager.start_all(instances) + def _switch_repo(self, name: Literal["klipper", "moonraker"]) -> None: + repo: RepoSettings = self.settings[name] + run_switch_repo_routine(name, repo) def set_klipper_repo(self, **kwargs) -> None: self._set_repo("klipper") diff --git a/kiauh/core/settings/kiauh_settings.py b/kiauh/core/settings/kiauh_settings.py index 2fad91d..ff2ac2b 100644 --- a/kiauh/core/settings/kiauh_settings.py +++ b/kiauh/core/settings/kiauh_settings.py @@ -8,6 +8,9 @@ # ======================================================================= # from __future__ import annotations +from dataclasses import dataclass, field +from typing import Any + from core.logger import DialogType, Logger from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import ( NoOptionError, @@ -22,33 +25,21 @@ DEFAULT_CFG = PROJECT_ROOT.joinpath("default.kiauh.cfg") CUSTOM_CFG = PROJECT_ROOT.joinpath("kiauh.cfg") +@dataclass class AppSettings: - def __init__(self) -> None: - self.backup_before_update = None + backup_before_update: bool | None = field(default=None) -class KlipperSettings: - def __init__(self) -> None: - self.repo_url = None - self.branch = None +@dataclass +class RepoSettings: + repo_url: str | None = field(default=None) + branch: str | None = field(default=None) -class MoonrakerSettings: - def __init__(self) -> None: - self.repo_url = None - self.branch = None - - -class MainsailSettings: - def __init__(self) -> None: - self.port = None - self.unstable_releases = None - - -class FluiddSettings: - def __init__(self) -> None: - self.port = None - self.unstable_releases = None +@dataclass +class WebUiSettings: + port: str | None = field(default=None) + unstable_releases: bool | None = field(default=None) # noinspection PyUnusedLocal @@ -61,6 +52,12 @@ class KiauhSettings: cls._instance = super(KiauhSettings, cls).__new__(cls, *args, **kwargs) return cls._instance + def __repr__(self) -> str: + return f"KiauhSettings(kiauh={self.kiauh}, klipper={self.klipper}, moonraker={self.moonraker}, mainsail={self.mainsail}, fluidd={self.fluidd})" + + def __getitem__(self, item: str) -> Any: + return getattr(self, item) + def __init__(self) -> None: if not hasattr(self, "__initialized"): self.__initialized = False @@ -69,20 +66,10 @@ class KiauhSettings: self.__initialized = True self.config = SimpleConfigParser() self.kiauh = AppSettings() - self.klipper = KlipperSettings() - self.moonraker = MoonrakerSettings() - self.mainsail = MainsailSettings() - self.fluidd = FluiddSettings() - - self.kiauh.backup_before_update = None - self.klipper.repo_url = None - self.klipper.branch = None - self.moonraker.repo_url = None - self.moonraker.branch = None - self.mainsail.port = None - self.mainsail.unstable_releases = None - self.fluidd.port = None - self.fluidd.unstable_releases = None + self.klipper = RepoSettings() + self.moonraker = RepoSettings() + self.mainsail = WebUiSettings() + self.fluidd = WebUiSettings() self._load_config() @@ -102,22 +89,8 @@ class KiauhSettings: except AttributeError: raise - def set(self, section: str, option: str, value: str | int | bool) -> None: - """ - Set a value in the settings state by providing the section and option name as strings. - Prefer direct access to the properties, as it is usually safer! - :param section: The section name as string. - :param option: The option name as string. - :param value: The value to set as string, int or bool. - """ - try: - section = getattr(self, section) - section.option = value # type: ignore - except AttributeError: - raise - def save(self) -> None: - self._set_config_options() + self._set_config_options_state() self.config.write(CUSTOM_CFG) self._load_config() @@ -129,7 +102,7 @@ class KiauhSettings: self.config.read(cfg) self._validate_cfg() - self._read_settings() + self._apply_settings_from_file() def _validate_cfg(self) -> None: try: @@ -171,7 +144,7 @@ class KiauhSettings: if v.isdigit() or v.lower() == "true" or v.lower() == "false": raise ValueError - def _read_settings(self) -> None: + def _apply_settings_from_file(self) -> None: self.kiauh.backup_before_update = self.config.getboolean( "kiauh", "backup_before_update" ) @@ -188,7 +161,7 @@ class KiauhSettings: "fluidd", "unstable_releases" ) - def _set_config_options(self) -> None: + def _set_config_options_state(self) -> None: self.config.set( "kiauh", "backup_before_update", diff --git a/kiauh/core/types.py b/kiauh/core/types.py index 274ef9d..6f5a0e6 100644 --- a/kiauh/core/types.py +++ b/kiauh/core/types.py @@ -23,6 +23,7 @@ StatusMap: Dict[StatusCode, StatusText] = { @dataclass class ComponentStatus: status: StatusCode + owner: str | None = None repo: str | None = None local: str | None = None remote: str | None = None diff --git a/kiauh/procedures/switch_repo.py b/kiauh/procedures/switch_repo.py new file mode 100644 index 0000000..4ddcab7 --- /dev/null +++ b/kiauh/procedures/switch_repo.py @@ -0,0 +1,154 @@ +# ======================================================================= # +# Copyright (C) 2020 - 2024 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 + +import shutil +from pathlib import Path +from typing import Literal + +from components.klipper import ( + KLIPPER_BACKUP_DIR, + KLIPPER_DIR, + KLIPPER_ENV_DIR, + KLIPPER_REQ_FILE, +) +from components.klipper.klipper import Klipper +from components.klipper.klipper_setup import install_klipper_packages +from components.moonraker import ( + MOONRAKER_BACKUP_DIR, + MOONRAKER_DIR, + MOONRAKER_ENV_DIR, + MOONRAKER_REQ_FILE, +) +from components.moonraker.moonraker import Moonraker +from components.moonraker.moonraker_setup import install_moonraker_packages +from core.backup_manager.backup_manager import BackupManager, BackupManagerException +from core.instance_manager.instance_manager import InstanceManager +from core.logger import Logger +from core.settings.kiauh_settings import RepoSettings +from utils.git_utils import GitException, get_repo_name, git_clone_wrapper +from utils.instance_utils import get_instances +from utils.sys_utils import ( + VenvCreationFailedException, + create_python_venv, + install_python_requirements, +) + + +class RepoSwitchFailedException(Exception): + pass + + +def run_switch_repo_routine( + name: Literal["klipper", "moonraker"], repo_settings: RepoSettings +) -> None: + repo_dir: Path = KLIPPER_DIR if name == "klipper" else MOONRAKER_DIR + env_dir: Path = KLIPPER_ENV_DIR if name == "klipper" else MOONRAKER_ENV_DIR + req_file = KLIPPER_REQ_FILE if name == "klipper" else MOONRAKER_REQ_FILE + backup_dir: Path = KLIPPER_BACKUP_DIR if name == "klipper" else MOONRAKER_BACKUP_DIR + _type = Klipper if name == "klipper" else Moonraker + + # step 1: stop all instances + Logger.print_status(f"Stopping all {_type.__name__} instances ...") + instances = get_instances(_type) + InstanceManager.stop_all(instances) + + repo_dir_backup_path: Path | None = None + env_dir_backup_path: Path | None = None + + try: + # step 2: backup old repo and env + org, repo = get_repo_name(repo_dir) + backup_dir = backup_dir.joinpath(org) + bm = BackupManager() + repo_dir_backup_path = bm.backup_directory( + repo_dir.name, + repo_dir, + backup_dir, + ) + env_dir_backup_path = bm.backup_directory( + env_dir.name, + env_dir, + backup_dir, + ) + + # step 3: read repo url and branch from settings + repo_url = repo_settings.repo_url + branch = repo_settings.branch + + if not (repo_url or branch): + error = f"Invalid repository URL ({repo_url}) or branch ({branch})!" + raise ValueError(error) + + # step 4: clone new repo + git_clone_wrapper(repo_url, repo_dir, branch, force=True) + + # step 5: install os dependencies + if name == "klipper": + install_klipper_packages() + elif name == "moonraker": + install_moonraker_packages() + + # step 6: recreate python virtualenv + Logger.print_status(f"Recreating {_type.__name__} virtualenv ...") + if not create_python_venv(env_dir, force=True): + raise GitException(f"Failed to recreate virtualenv for {_type.__name__}") + else: + install_python_requirements(env_dir, req_file) + + Logger.print_ok(f"Switched to {repo_url} at branch {branch}!") + + except BackupManagerException as e: + Logger.print_error(f"Error during backup of repository: {e}") + raise RepoSwitchFailedException(e) + + except (GitException, VenvCreationFailedException) as e: + # if something goes wrong during cloning or recreating the virtualenv, + # we restore the backup of the repo and env + Logger.print_error(f"Error during repository switch: {e}", start="\n") + Logger.print_status(f"Restoring last backup of {_type.__name__} ...") + _restore_repo_backup( + _type.__name__, + env_dir, + env_dir_backup_path, + repo_dir, + repo_dir_backup_path, + ) + + except RepoSwitchFailedException as e: + Logger.print_error(f"Something went wrong: {e}") + return + + Logger.print_status(f"Restarting all {_type.__name__} instances ...") + InstanceManager.start_all(instances) + + +def _restore_repo_backup( + name: str, + env_dir: Path, + env_dir_backup_path: Path | None, + repo_dir: Path, + repo_dir_backup_path: Path | None, +) -> None: + # if repo_dir_backup_path is not None and env_dir_backup_path is not None: + if not repo_dir_backup_path or not env_dir_backup_path: + raise RepoSwitchFailedException( + f"Unable to restore backup of {name}! Path of backups directory is None!" + ) + + try: + if repo_dir.exists(): + shutil.rmtree(repo_dir) + shutil.copytree(repo_dir_backup_path, repo_dir) + if env_dir.exists(): + shutil.rmtree(env_dir) + shutil.copytree(env_dir_backup_path, env_dir) + Logger.print_warn(f"Restored backup of {name} successfully!") + except Exception as e: + raise RepoSwitchFailedException(f"Error restoring backup: {e}") diff --git a/kiauh/utils/common.py b/kiauh/utils/common.py index 7c0ba22..6f5619a 100644 --- a/kiauh/utils/common.py +++ b/kiauh/utils/common.py @@ -124,10 +124,12 @@ def get_install_status( else: status = 1 # incomplete + org, repo = get_repo_name(repo_dir) return ComponentStatus( status=status, instances=instances, - repo=get_repo_name(repo_dir), + owner=org, + repo=repo, local=get_local_commit(repo_dir), remote=get_remote_commit(repo_dir), ) diff --git a/kiauh/utils/git_utils.py b/kiauh/utils/git_utils.py index 484f1a9..6e2977c 100644 --- a/kiauh/utils/git_utils.py +++ b/kiauh/utils/git_utils.py @@ -16,6 +16,10 @@ from utils.input_utils import get_confirm, get_number_input from utils.instance_utils import get_instances +class GitException(Exception): + pass + + def git_clone_wrapper( repo: str, target_dir: Path, branch: str | None = None, force: bool = False ) -> None: @@ -43,10 +47,10 @@ def git_clone_wrapper( except CalledProcessError: log = "An unexpected error occured during cloning of the repository." Logger.print_error(log) - return + raise GitException(log) except OSError as e: Logger.print_error(f"Error removing existing repository: {e.strerror}") - return + raise GitException(f"Error removing existing repository: {e.strerror}") def git_pull_wrapper(repo: str, target_dir: Path) -> None: @@ -66,20 +70,22 @@ def git_pull_wrapper(repo: str, target_dir: Path) -> None: return -def get_repo_name(repo: Path) -> str | None: +def get_repo_name(repo: Path) -> tuple[str, str] | None: """ Helper method to extract the organisation and name of a repository | :param repo: repository to extract the values from :return: String in form of "/" or None """ if not repo.exists() or not repo.joinpath(".git").exists(): - return "-" + return "-", "-" try: cmd = ["git", "-C", repo.as_posix(), "config", "--get", "remote.origin.url"] result: str = check_output(cmd, stderr=DEVNULL).decode(encoding="utf-8") substrings: List[str] = result.strip().split("/")[-2:] - return "/".join(substrings).replace(".git", "") + return substrings[0], substrings[1] + + # return "/".join(substrings).replace(".git", "") except CalledProcessError: return None diff --git a/kiauh/utils/sys_utils.py b/kiauh/utils/sys_utils.py index d2b7534..02b5096 100644 --- a/kiauh/utils/sys_utils.py +++ b/kiauh/utils/sys_utils.py @@ -39,6 +39,10 @@ SysCtlServiceAction = Literal[ SysCtlManageAction = Literal["daemon-reload", "reset-failed"] +class VenvCreationFailedException(Exception): + pass + + def kill(opt_err_msg: str = "") -> None: """ Kills the application | @@ -87,11 +91,12 @@ def parse_packages_from_file(source_file: Path) -> List[str]: return packages -def create_python_venv(target: Path) -> bool: +def create_python_venv(target: Path, force: bool = False) -> bool: """ Create a python 3 virtualenv at the provided target destination. Returns True if the virtualenv was created successfully. Returns False if the virtualenv already exists, recreation was declined or creation failed. + :param force: Force recreation of the virtualenv :param target: Path where to create the virtualenv at :return: bool """ @@ -106,7 +111,7 @@ def create_python_venv(target: Path) -> bool: Logger.print_error(f"Error setting up virtualenv:\n{e}") return False else: - if not get_confirm( + if not force and not get_confirm( "Virtualenv already exists. Re-create?", default_choice=False ): Logger.print_info("Skipping re-creation of virtualenv ...") @@ -174,14 +179,14 @@ def install_python_requirements(target: Path, requirements: Path) -> None: if result.returncode != 0 or result.stderr: Logger.print_error(f"{result.stderr}", False) - Logger.print_error("Installing Python requirements failed!") - return + raise VenvCreationFailedException("Installing Python requirements failed!") Logger.print_ok("Installing Python requirements successful!") - except CalledProcessError as e: - log = f"Error installing Python requirements:\n{e.output.decode()}" + + except Exception as e: + log = f"Error installing Python requirements: {e}" Logger.print_error(log) - raise + raise VenvCreationFailedException(log) def update_system_package_lists(silent: bool, rls_info_change=False) -> None: