refactor(Klipper): rework remove process

Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2023-12-25 15:35:13 +01:00
parent 012b6c4bb7
commit 142b4498a3
7 changed files with 273 additions and 135 deletions

View File

@@ -64,7 +64,7 @@ class InstallMenu(BaseMenu):
print(menu, end="")
def install_klipper(self):
klipper_setup.run_klipper_setup(install=True)
klipper_setup.install_klipper()
def install_moonraker(self):
moonraker_setup.install_moonraker()

View File

@@ -13,7 +13,7 @@ import textwrap
from kiauh.core.menus import BACK_FOOTER
from kiauh.core.menus.base_menu import BaseMenu
from kiauh.modules.klipper import klipper_setup
from kiauh.modules.klipper.menus.klipper_remove_menu import KlipperRemoveMenu
from kiauh.modules.mainsail.menus.mainsail_remove_menu import MainsailRemoveMenu
from kiauh.modules.moonraker.menus.moonraker_remove_menu import MoonrakerRemoveMenu
from kiauh.utils.constants import COLOR_RED, RESET_FORMAT
@@ -25,7 +25,7 @@ class RemoveMenu(BaseMenu):
super().__init__(
header=True,
options={
1: self.remove_klipper,
1: KlipperRemoveMenu,
2: MoonrakerRemoveMenu,
3: MainsailRemoveMenu,
5: self.remove_fluidd,
@@ -69,9 +69,6 @@ class RemoveMenu(BaseMenu):
)[1:]
print(menu, end="")
def remove_klipper(self):
klipper_setup.run_klipper_setup(install=False)
def remove_fluidd(self):
print("remove_fluidd")

View File

@@ -75,7 +75,7 @@ class Klipper(BaseInstance):
Logger.print_error(f"Error creating env file {env_file_target}: {e}")
raise
def delete(self, del_remnants: bool) -> None:
def delete(self) -> None:
service_file = self.get_service_file_name(extension=True)
service_file_path = self.get_service_file_path()
@@ -89,9 +89,6 @@ class Klipper(BaseInstance):
Logger.print_error(f"Error deleting service file: {e}")
raise
if del_remnants:
self._delete_klipper_remnants()
def write_service_file(
self,
service_template_path: Path,
@@ -118,20 +115,6 @@ class Klipper(BaseInstance):
env_file.write(env_file_content)
Logger.print_ok(f"Env file created: {env_file_target}")
def _delete_klipper_remnants(self) -> None:
try:
Logger.print_status(f"Delete {self.klipper_dir} ...")
shutil.rmtree(Path(self.klipper_dir))
Logger.print_status(f"Delete {self.env_dir} ...")
shutil.rmtree(Path(self.env_dir))
except FileNotFoundError:
Logger.print_status("Cannot delete Klipper directories. Not found.")
except PermissionError as e:
Logger.print_error(f"Error deleting Klipper directories: {e}")
raise
Logger.print_ok("Directories successfully deleted.")
def _prep_service_file(
self, service_template_path: Path, env_file_path: Path
) -> str:

View File

@@ -0,0 +1,133 @@
#!/usr/bin/env python3
# ======================================================================= #
# Copyright (C) 2020 - 2023 Dominik Willner <th33xitus@gmail.com> #
# #
# 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 #
# ======================================================================= #
import shutil
from typing import List, Union
from kiauh.core.instance_manager.instance_manager import InstanceManager
from kiauh.modules.klipper import KLIPPER_DIR, KLIPPER_ENV_DIR
from kiauh.modules.klipper.klipper import Klipper
from kiauh.modules.klipper.klipper_dialogs import print_instance_overview
from kiauh.utils.filesystem_utils import remove_file
from kiauh.utils.input_utils import get_selection_input
from kiauh.utils.logger import Logger
def run_klipper_removal(
remove_service: bool,
remove_dir: bool,
remove_env: bool,
delete_logs: bool,
) -> None:
im = InstanceManager(Klipper)
if remove_service:
Logger.print_status("Removing Klipper instances ...")
if im.instances:
instances_to_remove = select_instances_to_remove(im.instances)
remove_instances(im, instances_to_remove)
else:
Logger.print_info("No Klipper Services installed! Skipped ...")
im.find_instances()
if (remove_dir or remove_env) and im.instances:
Logger.print_warn("There are still other Klipper services installed!")
Logger.print_warn("Therefor the following parts cannot be removed:")
Logger.print_warn(
"""
● Klipper local repository
● Klipper Python environment
""",
False,
)
else:
if remove_dir:
Logger.print_status("Removing Klipper local repository ...")
remove_klipper_dir()
if remove_env:
Logger.print_status("Removing Klipper Python environment ...")
remove_klipper_env()
# delete klipper logs of all instances
if delete_logs:
Logger.print_status("Removing all Klipper logs ...")
delete_klipper_logs(im.instances)
def select_instances_to_remove(
instances: List[Klipper],
) -> Union[List[Klipper], None]:
print_instance_overview(instances, True, True)
options = [str(i) for i in range(len(instances))]
options.extend(["a", "A", "b", "B"])
selection = get_selection_input("Select Klipper instance to remove", options)
instances_to_remove = []
if selection == "b".lower():
return None
elif selection == "a".lower():
instances_to_remove.extend(instances)
else:
instance = instances[int(selection)]
instances_to_remove.append(instance)
return instances_to_remove
def remove_instances(
instance_manager: InstanceManager,
instance_list: List[Klipper],
) -> None:
for instance in instance_list:
Logger.print_status(f"Removing instance {instance.get_service_file_name()} ...")
instance_manager.current_instance = instance
instance_manager.stop_instance()
instance_manager.disable_instance()
instance_manager.delete_instance()
instance_manager.reload_daemon()
def remove_klipper_dir() -> None:
if not KLIPPER_DIR.exists():
Logger.print_info(f"'{KLIPPER_DIR}' does not exist. Skipped ...")
return
try:
shutil.rmtree(KLIPPER_DIR)
except OSError as e:
Logger.print_error(f"Unable to delete '{KLIPPER_DIR}':\n{e}")
def remove_klipper_env() -> None:
if not KLIPPER_ENV_DIR.exists():
Logger.print_info(f"'{KLIPPER_ENV_DIR}' does not exist. Skipped ...")
return
try:
shutil.rmtree(KLIPPER_ENV_DIR)
except OSError as e:
Logger.print_error(f"Unable to delete '{KLIPPER_ENV_DIR}':\n{e}")
def delete_klipper_logs(instances: List[Klipper]) -> None:
all_logfiles = []
for instance in instances:
all_logfiles = list(instance.log_dir.glob("klippy.log*"))
if not all_logfiles:
Logger.print_info("No Klipper logs found. Skipped ...")
return
for log in all_logfiles:
Logger.print_status(f"Remove '{log}'")
remove_file(log)

View File

@@ -9,7 +9,6 @@
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
import subprocess
from pathlib import Path
from typing import List, Union
@@ -40,11 +39,7 @@ from kiauh.modules.klipper.klipper_utils import (
create_example_printer_cfg,
)
from kiauh.core.repo_manager.repo_manager import RepoManager
from kiauh.utils.input_utils import (
get_confirm,
get_number_input,
get_selection_input,
)
from kiauh.utils.input_utils import get_confirm, get_number_input
from kiauh.utils.logger import Logger
from kiauh.utils.system_utils import (
parse_packages_from_file,
@@ -55,85 +50,52 @@ from kiauh.utils.system_utils import (
)
def run_klipper_setup(install: bool) -> None:
instance_manager = InstanceManager(Klipper)
instance_list = instance_manager.instances
instances_installed = len(instance_list)
def install_klipper() -> None:
im = InstanceManager(Klipper)
is_klipper_installed = instances_installed > 0
if not install and not is_klipper_installed:
Logger.print_warn("Klipper not installed!")
add_additional = handle_existing_instances(im.instances)
if len(im.instances) > 0 and not add_additional:
Logger.print_status(EXIT_KLIPPER_SETUP)
return
if install:
add_additional = handle_existing_instances(instance_list)
if is_klipper_installed and not add_additional:
Logger.print_status(EXIT_KLIPPER_SETUP)
return
install_klipper(instance_manager, instance_list)
if not install:
if instances_installed == 1:
remove_single_instance(instance_manager, instance_list)
else:
remove_multi_instance(instance_manager, instance_list)
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 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"
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)
if install_count is None:
Logger.print_status(EXIT_KLIPPER_SETUP)
return
instance_names = set_instance_suffix(instance_list, install_count)
instance_names = set_instance_suffix(im.instances, install_count)
if instance_names is None:
Logger.print_status(EXIT_KLIPPER_SETUP)
return
create_example_cfg = get_confirm("Create example printer.cfg?")
if len(instance_list) < 1:
if len(im.instances) < 1:
setup_klipper_prerequesites()
convert_single_to_multi = (
len(instance_list) == 1
and instance_list[0].suffix is None
and install_count >= 1
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(instance_manager, name)
current_instance = handle_single_to_multi_conversion(im, name)
convert_single_to_multi = False
else:
current_instance = Klipper(suffix=name)
instance_manager.current_instance = current_instance
instance_manager.create_instance()
instance_manager.enable_instance()
im.current_instance = current_instance
im.create_instance()
im.enable_instance()
if create_example_cfg:
create_example_printer_cfg(current_instance)
instance_manager.start_instance()
im.start_instance()
instance_manager.reload_daemon()
im.reload_daemon()
# step 4: check/handle conflicting packages/services
handle_disruptive_system_packages()
@@ -175,6 +137,17 @@ 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]]:
@@ -199,63 +172,6 @@ def set_instance_suffix(
)
def remove_single_instance(
instance_manager: InstanceManager, instance_list: List[Klipper]
) -> None:
question = f"Delete {KLIPPER_DIR} and {KLIPPER_ENV_DIR}?"
del_remnants = get_confirm(question, allow_go_back=True)
if del_remnants is None:
Logger.print_status("Exiting Klipper Uninstaller ...")
return
try:
instance_manager.current_instance = instance_list[0]
instance_manager.stop_instance()
instance_manager.disable_instance()
instance_manager.delete_instance(del_remnants=del_remnants)
instance_manager.reload_daemon()
except (OSError, subprocess.CalledProcessError):
Logger.print_error("Removing instance failed!")
return
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))]
options.extend(["a", "A", "b", "B"])
selection = get_selection_input("Select Klipper instance to remove", options)
if selection == "b".lower():
return
elif selection == "a".lower():
question = f"Delete {KLIPPER_DIR} and {KLIPPER_ENV_DIR}?"
del_remnants = get_confirm(question, allow_go_back=True)
if del_remnants is None:
Logger.print_status("Exiting Klipper Uninstaller ...")
return
Logger.print_status("Removing all Klipper instances ...")
for instance in instance_list:
instance_manager.current_instance = instance
instance_manager.stop_instance()
instance_manager.disable_instance()
instance_manager.delete_instance(del_remnants=del_remnants)
else:
instance = instance_list[int(selection)]
log = f"Removing Klipper instance: {instance.get_service_file_name()}"
Logger.print_status(log)
instance_manager.current_instance = instance
instance_manager.stop_instance()
instance_manager.disable_instance()
instance_manager.delete_instance(del_remnants=False)
instance_manager.reload_daemon()
def update_klipper() -> None:
print_update_warn_dialog()
if not get_confirm("Update Klipper now?"):

View File

View File

@@ -0,0 +1,109 @@
#!/usr/bin/env python3
# ======================================================================= #
# Copyright (C) 2020 - 2023 Dominik Willner <th33xitus@gmail.com> #
# #
# 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 #
# ======================================================================= #
import textwrap
from kiauh.core.menus import BACK_HELP_FOOTER
from kiauh.core.menus.base_menu import BaseMenu
from kiauh.modules.klipper import klipper_remove
from kiauh.modules.moonraker import moonraker_remove
from kiauh.utils.constants import RESET_FORMAT, COLOR_RED, COLOR_CYAN
class KlipperRemoveMenu(BaseMenu):
def __init__(self):
super().__init__(
header=False,
options={
0: self.toggle_all,
1: self.toggle_remove_klipper_service,
2: self.toggle_remove_klipper_dir,
3: self.toggle_remove_klipper_env,
4: self.toggle_delete_klipper_logs,
5: self.run_removal_process,
},
footer_type=BACK_HELP_FOOTER,
)
self.remove_klipper_service = False
self.remove_klipper_dir = False
self.remove_klipper_env = False
self.delete_klipper_logs = False
def print_menu(self) -> None:
header = " [ Remove Klipper ] "
color = COLOR_RED
count = 62 - len(color) - len(RESET_FORMAT)
checked = f"[{COLOR_CYAN}x{RESET_FORMAT}]"
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
o4 = checked if self.delete_klipper_logs else unchecked
menu = textwrap.dedent(
f"""
/=======================================================\\
| {color}{header:~^{count}}{RESET_FORMAT} |
|-------------------------------------------------------|
| Enter a number and hit enter to select / deselect |
| the specific option for removal. |
|-------------------------------------------------------|
| 0) Select everything |
|-------------------------------------------------------|
| 1) {o1} Remove Service |
| 2) {o2} Remove Local Repository |
| 3) {o3} Remove Python Environment |
| 4) {o4} Delete all Log-Files |
|-------------------------------------------------------|
| 5) Continue |
"""
)[1:]
print(menu, end="")
def toggle_all(self) -> None:
self.remove_klipper_service = True
self.remove_klipper_dir = True
self.remove_klipper_env = True
self.delete_klipper_logs = True
def toggle_remove_klipper_service(self) -> None:
self.remove_klipper_service = not self.remove_klipper_service
def toggle_remove_klipper_dir(self) -> None:
self.remove_klipper_dir = not self.remove_klipper_dir
def toggle_remove_klipper_env(self) -> None:
self.remove_klipper_env = not self.remove_klipper_env
def toggle_delete_klipper_logs(self) -> None:
self.delete_klipper_logs = not self.delete_klipper_logs
def run_removal_process(self) -> None:
if (
not self.remove_klipper_service
and not self.remove_klipper_dir
and not self.remove_klipper_env
and not self.delete_klipper_logs
):
error = f"{COLOR_RED}Nothing selected! Select options to remove first.{RESET_FORMAT}"
print(error)
return
klipper_remove.run_klipper_removal(
self.remove_klipper_service,
self.remove_klipper_dir,
self.remove_klipper_env,
self.delete_klipper_logs,
)
self.remove_klipper_service = False
self.remove_klipper_dir = False
self.remove_klipper_env = False
self.delete_klipper_logs = False