diff --git a/kiauh/components/moonraker/moonraker_dialogs.py b/kiauh/components/moonraker/moonraker_dialogs.py index e8e1666..aa7acf5 100644 --- a/kiauh/components/moonraker/moonraker_dialogs.py +++ b/kiauh/components/moonraker/moonraker_dialogs.py @@ -48,7 +48,7 @@ def print_moonraker_overview( for i, k in enumerate(instance_map): mr_name = instance_map.get(k) m = f"<-> {mr_name}" if mr_name != "" else "" - line = Color.apply(f"{f'{i+1})' if show_index else '●'} {k} {m}", Color.CYAN) + line = Color.apply(f"{f'{i + 1})' if show_index else '●'} {k} {m}", Color.CYAN) dialog += f"║ {line:<63}║\n" warn_l1 = Color.apply("PLEASE NOTE:", Color.YELLOW) diff --git a/kiauh/components/moonraker/moonraker_setup.py b/kiauh/components/moonraker/moonraker_setup.py index 23bbba8..6e69e68 100644 --- a/kiauh/components/moonraker/moonraker_setup.py +++ b/kiauh/components/moonraker/moonraker_setup.py @@ -29,7 +29,7 @@ from components.moonraker.moonraker import Moonraker from components.moonraker.moonraker_dialogs import print_moonraker_overview from components.moonraker.services.moonraker_instance_service import ( MoonrakerInstanceService, - ) +) from components.moonraker.utils.sysdeps_parser import SysDepsParser from components.moonraker.utils.utils import ( backup_moonraker_dir, @@ -100,8 +100,9 @@ def install_moonraker() -> None: return if selected_option == "a": - new_inst_list: List[Moonraker] = ( - [instance_service.create_new_instance(k.suffix) for k in klipper_list]) + new_inst_list: List[Moonraker] = [ + instance_service.create_new_instance(k.suffix) for k in klipper_list + ] new_instances.extend(new_inst_list) else: klipper_instance: Klipper | None = options.get(selected_option) @@ -138,13 +139,17 @@ def install_moonraker() -> None: enable_mainsail_remotemode() instance_service.load_instances() - new_instances = [instance_service.get_instance_by_suffix(i.suffix) for i in - new_instances] + new_instances = [ + instance_service.get_instance_by_suffix(i.suffix) for i in new_instances + ] ip: str = get_ipv4_addr() # noinspection HttpUrlsUsage - url_list = [f"● {i.service_file_path.stem}: http://{ip}:{i.port}" for i in - new_instances if i.port] + url_list = [ + f"● {i.service_file_path.stem}: http://{ip}:{i.port}" + for i in new_instances + if i.port + ] dialog_content = [] if url_list: dialog_content.append("You can access Moonraker via the following URL:") @@ -154,8 +159,8 @@ def install_moonraker() -> None: DialogType.CUSTOM, custom_title="Moonraker successfully installed!", custom_color=Color.GREEN, - content=dialog_content) - + content=dialog_content, + ) except Exception as e: Logger.print_error(f"Error while installing Moonraker: {e}") @@ -194,7 +199,8 @@ def install_moonraker_packages() -> None: moonraker_deps = [] if MOONRAKER_DEPS_JSON_FILE.exists(): Logger.print_info( - f"Parsing system dependencies from {MOONRAKER_DEPS_JSON_FILE.name} ...") + f"Parsing system dependencies from {MOONRAKER_DEPS_JSON_FILE.name} ..." + ) parser = SysDepsParser() sysdeps = load_sysdeps_json(MOONRAKER_DEPS_JSON_FILE) moonraker_deps.extend(parser.parse_dependencies(sysdeps)) @@ -202,7 +208,8 @@ def install_moonraker_packages() -> None: elif MOONRAKER_INSTALL_SCRIPT.exists(): Logger.print_warn(f"{MOONRAKER_DEPS_JSON_FILE.name} not found!") Logger.print_info( - f"Parsing system dependencies from {MOONRAKER_INSTALL_SCRIPT.name} ...") + f"Parsing system dependencies from {MOONRAKER_INSTALL_SCRIPT.name} ..." + ) moonraker_deps = parse_packages_from_file(MOONRAKER_INSTALL_SCRIPT) if not moonraker_deps: diff --git a/kiauh/components/moonraker/utils/sysdeps_parser.py b/kiauh/components/moonraker/utils/sysdeps_parser.py index bed902f..15dad17 100644 --- a/kiauh/components/moonraker/utils/sysdeps_parser.py +++ b/kiauh/components/moonraker/utils/sysdeps_parser.py @@ -34,19 +34,23 @@ def _get_distro_info() -> Dict[str, Any]: return dict( distro_id=release_info.get("ID", ""), distro_version=release_info.get("VERSION_ID", ""), - aliases=release_info.get("ID_LIKE", "").split() + aliases=release_info.get("ID_LIKE", "").split(), ) + def _convert_version(version: str) -> Tuple[str | int, ...]: version = version.strip() ver_match = re.match(r"\d+(\.\d+)*((?:-|\.).+)?", version) if ver_match is not None: - return tuple([ - int(part) if part.isdigit() else part - for part in re.split(r"\.|-", version) - ]) + return tuple( + [ + int(part) if part.isdigit() else part + for part in re.split(r"\.|-", version) + ] + ) return (version,) + class SysDepsParser: def __init__(self, distro_info: Dict[str, Any] | None = None) -> None: if distro_info is None: @@ -86,14 +90,16 @@ class SysDepsParser: if logical_op not in ("and", "or"): logging.info( f"Invalid logical operator {logical_op} in requirement " - f"specifier: {full_spec}") + f"specifier: {full_spec}" + ) return None last_logical_op = logical_op continue elif last_logical_op is None: logging.info( f"Requirement specifier contains two seqential expressions " - f"without a logical operator: {full_spec}") + f"without a logical operator: {full_spec}" + ) return None dep_parts = re.split(r"(==|!=|<=|>=|<|>)", exp.strip()) req_var = dep_parts[0].strip().lower() @@ -123,7 +129,7 @@ class SysDepsParser: "==": lambda x, y: x == y, "!=": lambda x, y: x != y, ">=": lambda x, y: x >= y, - "<=": lambda x, y: x <= y + "<=": lambda x, y: x <= y, }.get(operator, lambda x, y: False) result = compfunc(left_op, right_op) if last_logical_op == "and": diff --git a/kiauh/components/moonraker/utils/utils.py b/kiauh/components/moonraker/utils/utils.py index 7b09763..d5d9149 100644 --- a/kiauh/components/moonraker/utils/utils.py +++ b/kiauh/components/moonraker/utils/utils.py @@ -140,6 +140,7 @@ def backup_moonraker_db_dir() -> None: name, source=instance.db_dir, target=MOONRAKER_DB_BACKUP_DIR ) + def load_sysdeps_json(file: Path) -> Dict[str, List[str]]: try: sysdeps: Dict[str, List[str]] = json.loads(file.read_bytes()) diff --git a/kiauh/components/webui_client/menus/client_remove_menu.py b/kiauh/components/webui_client/menus/client_remove_menu.py index 57fa0a7..82ce69d 100644 --- a/kiauh/components/webui_client/menus/client_remove_menu.py +++ b/kiauh/components/webui_client/menus/client_remove_menu.py @@ -57,7 +57,7 @@ class ClientRemoveMenu(BaseMenu): o1 = checked if self.remove_client else unchecked o2 = checked if self.remove_client_cfg else unchecked o3 = checked if self.backup_config_json else unchecked - sel_state = f"{'Select'if not self.select_state else 'Deselect'} everything" + sel_state = f"{'Select' if not self.select_state else 'Deselect'} everything" menu = textwrap.dedent( f""" ╟───────────────────────────────────────────────────────╢ diff --git a/kiauh/core/backup_manager/backup_manager.py b/kiauh/core/backup_manager/backup_manager.py index 14d6c47..0497898 100644 --- a/kiauh/core/backup_manager/backup_manager.py +++ b/kiauh/core/backup_manager/backup_manager.py @@ -86,7 +86,12 @@ class BackupManager: date = get_current_date().get("date") time = get_current_date().get("time") backup_target = target.joinpath(f"{name.lower()}-{date}-{time}") - shutil.copytree(source, backup_target, ignore=self.ignore_folders_func, ignore_dangling_symlinks=True) + shutil.copytree( + source, + backup_target, + ignore=self.ignore_folders_func, + ignore_dangling_symlinks=True, + ) Logger.print_ok("Backup successful!") return backup_target diff --git a/kiauh/core/menus/settings_menu.py b/kiauh/core/menus/settings_menu.py index b836bc6..6fc8346 100644 --- a/kiauh/core/menus/settings_menu.py +++ b/kiauh/core/menus/settings_menu.py @@ -40,7 +40,6 @@ class SettingsMenu(BaseMenu): self._load_settings() print(self.klipper_status) - def set_previous_menu(self, previous_menu: Type[BaseMenu] | None) -> None: from core.menus.main_menu import MainMenu @@ -122,18 +121,24 @@ class SettingsMenu(BaseMenu): self.moonraker_status.owner = url_parts[-2] self.moonraker_status.branch = self.settings.moonraker.branch - def _gather_input(self, repo_name: Literal["klipper", "moonraker"], repo_dir: Path) -> Tuple[str, str]: + def _gather_input( + self, repo_name: Literal["klipper", "moonraker"], repo_dir: Path + ) -> Tuple[str, str]: warn_msg = [ "There is only basic input validation in place! " - "Make sure your the input is valid and has no typos or invalid characters!"] + "Make sure your the input is valid and has no typos or invalid characters!" + ] if repo_dir.exists(): - warn_msg.extend([ - "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!"]) + warn_msg.extend( + [ + "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!", + ] + ) Logger.print_dialog(DialogType.ATTENTION, warn_msg) @@ -143,14 +148,14 @@ class SettingsMenu(BaseMenu): default=KLIPPER_REPO_URL if repo_name == "klipper" else MOONRAKER_REPO_URL, ) branch = get_string_input( - "Enter new branch name", - regex=r"^.+$", - default="master" + "Enter new branch name", regex=r"^.+$", default="master" ) return repo, branch - def _set_repo(self, repo_name: Literal["klipper", "moonraker"], repo_dir: Path) -> None: + def _set_repo( + self, repo_name: Literal["klipper", "moonraker"], repo_dir: Path + ) -> None: repo_url, branch = self._gather_input(repo_name, repo_dir) display_name = repo_name.capitalize() Logger.print_dialog( @@ -180,11 +185,15 @@ class SettingsMenu(BaseMenu): self._switch_repo(repo_name, repo_dir) - def _switch_repo(self, name: Literal["klipper", "moonraker"], repo_dir: Path ) -> None: + def _switch_repo( + self, name: Literal["klipper", "moonraker"], repo_dir: Path + ) -> None: if not repo_dir.exists(): return - Logger.print_status(f"Switching to {name.capitalize()}'s new source repository ...") + Logger.print_status( + f"Switching to {name.capitalize()}'s new source repository ..." + ) repo: RepoSettings = self.settings[name] run_switch_repo_routine(name, repo) diff --git a/kiauh/extensions/klipper_backup/klipper_backup_extension.py b/kiauh/extensions/klipper_backup/klipper_backup_extension.py index 95b34ee..a3e1242 100644 --- a/kiauh/extensions/klipper_backup/klipper_backup_extension.py +++ b/kiauh/extensions/klipper_backup/klipper_backup_extension.py @@ -10,11 +10,11 @@ # ======================================================================= # import os -import shutil import subprocess +from pathlib import Path + from core.constants import SYSTEMD from core.logger import Logger -from pathlib import Path from extensions.base_extension import BaseExtension from extensions.klipper_backup import ( KLIPPERBACKUP_CONFIG_DIR, @@ -29,7 +29,6 @@ from utils.sys_utils import cmd_sysctl_manage, remove_system_service, unit_file_ class KlipperbackupExtension(BaseExtension): - def remove_extension(self, **kwargs) -> None: if not check_file_exist(KLIPPERBACKUP_DIR): Logger.print_info("Extension does not seem to be installed! Skipping ...") @@ -48,29 +47,44 @@ class KlipperbackupExtension(BaseExtension): cmd_sysctl_manage("daemon-reload") cmd_sysctl_manage("reset-failed") else: - Logger.print_error(f"Unknown unit type {unit_type} of {full_service_name}") + Logger.print_error( + f"Unknown unit type {unit_type} of {full_service_name}" + ) except: Logger.print_error(f"Failed to remove {full_service_name}: {str(e)}") def check_crontab_entry(entry) -> bool: try: - crontab_content = subprocess.check_output(["crontab", "-l"], stderr=subprocess.DEVNULL, text=True) + crontab_content = subprocess.check_output( + ["crontab", "-l"], stderr=subprocess.DEVNULL, text=True + ) except subprocess.CalledProcessError: return False return any(entry in line for line in crontab_content.splitlines()) def remove_moonraker_entry(): original_file_path = MOONRAKER_CONF - comparison_file_path = os.path.join(str(KLIPPERBACKUP_DIR), "install-files", "moonraker.conf") - if not (os.path.exists(original_file_path) and os.path.exists(comparison_file_path)): + comparison_file_path = os.path.join( + str(KLIPPERBACKUP_DIR), "install-files", "moonraker.conf" + ) + if not ( + os.path.exists(original_file_path) + and os.path.exists(comparison_file_path) + ): return False - with open(original_file_path, "r") as original_file, open(comparison_file_path, "r") as comparison_file: + with open(original_file_path, "r") as original_file, open( + comparison_file_path, "r" + ) as comparison_file: original_content = original_file.read() comparison_content = comparison_file.read() if comparison_content in original_content: Logger.print_status("Removing Klipper-Backup moonraker entry ...") - modified_content = original_content.replace(comparison_content, "").strip() - modified_content = "\n".join(line for line in modified_content.split("\n") if line.strip()) + modified_content = original_content.replace( + comparison_content, "" + ).strip() + modified_content = "\n".join( + line for line in modified_content.split("\n") if line.strip() + ) with open(original_file_path, "w") as original_file: original_file.write(modified_content) Logger.print_ok("Klipper-Backup moonraker entry successfully removed!") @@ -79,7 +93,11 @@ class KlipperbackupExtension(BaseExtension): if get_confirm("Do you really want to remove the extension?", True, False): # Remove systemd timer and services - service_names = ["klipper-backup-on-boot", "klipper-backup-filewatch", "klipper-backup"] + service_names = [ + "klipper-backup-on-boot", + "klipper-backup-filewatch", + "klipper-backup", + ] unit_types = ["timer", "service"] for service_name in service_names: @@ -91,10 +109,23 @@ class KlipperbackupExtension(BaseExtension): try: if check_crontab_entry("/klipper-backup/script.sh"): Logger.print_status("Removing Klipper-Backup crontab entry ...") - crontab_content = subprocess.check_output(["crontab", "-l"], text=True) - modified_content = "\n".join(line for line in crontab_content.splitlines() if "/klipper-backup/script.sh" not in line) - subprocess.run(["crontab", "-"], input=modified_content + "\n", text=True, check=True) - Logger.print_ok("Klipper-Backup crontab entry successfully removed!") + crontab_content = subprocess.check_output( + ["crontab", "-l"], text=True + ) + modified_content = "\n".join( + line + for line in crontab_content.splitlines() + if "/klipper-backup/script.sh" not in line + ) + subprocess.run( + ["crontab", "-"], + input=modified_content + "\n", + text=True, + check=True, + ) + Logger.print_ok( + "Klipper-Backup crontab entry successfully removed!" + ) except subprocess.CalledProcessError: Logger.print_error("Unable to remove the Klipper-Backup cron entry") @@ -102,7 +133,9 @@ class KlipperbackupExtension(BaseExtension): try: remove_moonraker_entry() except: - Logger.print_error("Unable to remove the Klipper-Backup moonraker entry") + Logger.print_error( + "Unable to remove the Klipper-Backup moonraker entry" + ) # Remove Klipper-backup extension Logger.print_status("Removing Klipper-Backup extension ...") @@ -112,7 +145,7 @@ class KlipperbackupExtension(BaseExtension): remove_with_sudo(KLIPPERBACKUP_CONFIG_DIR) Logger.print_ok("Extension Klipper-Backup successfully removed!") except: - Logger.print_error(f"Unable to remove Klipper-Backup extension") + Logger.print_error("Unable to remove Klipper-Backup extension") def install_extension(self, **kwargs) -> None: if not KLIPPERBACKUP_DIR.exists(): diff --git a/kiauh/extensions/obico/moonraker_obico_extension.py b/kiauh/extensions/obico/moonraker_obico_extension.py index c7fd130..e0bd009 100644 --- a/kiauh/extensions/obico/moonraker_obico_extension.py +++ b/kiauh/extensions/obico/moonraker_obico_extension.py @@ -11,8 +11,8 @@ from typing import List from components.klipper.klipper import Klipper from components.moonraker.moonraker import Moonraker -from core.instance_manager.instance_manager import InstanceManager from core.instance_manager.base_instance import SUFFIX_BLACKLIST +from core.instance_manager.instance_manager import InstanceManager from core.logger import DialogType, Logger from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import ( SimpleConfigParser, @@ -309,8 +309,12 @@ class ObicoExtension(BaseExtension): def _check_and_opt_link_instances(self) -> None: Logger.print_status("Checking link status of Obico instances ...") - suffix_blacklist: List[str] = [suffix for suffix in SUFFIX_BLACKLIST if suffix != 'obico'] - ob_instances: List[MoonrakerObico] = get_instances(MoonrakerObico, suffix_blacklist=suffix_blacklist) + suffix_blacklist: List[str] = [ + suffix for suffix in SUFFIX_BLACKLIST if suffix != "obico" + ] + ob_instances: List[MoonrakerObico] = get_instances( + MoonrakerObico, suffix_blacklist=suffix_blacklist + ) unlinked_instances: List[MoonrakerObico] = [ obico for obico in ob_instances if not obico.is_linked ] diff --git a/kiauh/extensions/octoapp/octoapp.py b/kiauh/extensions/octoapp/octoapp.py index b7a1da0..8480d70 100644 --- a/kiauh/extensions/octoapp/octoapp.py +++ b/kiauh/extensions/octoapp/octoapp.py @@ -45,9 +45,7 @@ class Octoapp: self.base: BaseInstance = BaseInstance(Moonraker, self.suffix) self.base.log_file_name = self.log_file_name - self.service_file_path: Path = get_service_file_path( - Octoapp, self.suffix - ) + self.service_file_path: Path = get_service_file_path(Octoapp, self.suffix) self.store_dir = self.base.data_dir.joinpath("store") self.cfg_file = self.base.cfg_dir.joinpath(OA_CFG_NAME) self.sys_cfg_file = self.base.cfg_dir.joinpath(OA_SYS_CFG_NAME) diff --git a/kiauh/extensions/octoapp/octoapp_extension.py b/kiauh/extensions/octoapp/octoapp_extension.py index 26bb748..ff20c52 100644 --- a/kiauh/extensions/octoapp/octoapp_extension.py +++ b/kiauh/extensions/octoapp/octoapp_extension.py @@ -9,8 +9,8 @@ import json from typing import List -from components.moonraker.moonraker import Moonraker from components.klipper.klipper import Klipper +from components.moonraker.moonraker import Moonraker from core.instance_manager.instance_manager import InstanceManager from core.logger import DialogType, Logger from extensions.base_extension import BaseExtension @@ -107,9 +107,7 @@ class OctoappExtension(BaseExtension): ) except Exception as e: - Logger.print_error( - f"Error during OctoApp for Klipper installation:\n{e}" - ) + Logger.print_error(f"Error during OctoApp for Klipper installation:\n{e}") def update_extension(self, **kwargs) -> None: Logger.print_status("Updating OctoApp for Klipper ...") @@ -183,7 +181,6 @@ class OctoappExtension(BaseExtension): run_remove_routines(OA_DIR) - def _remove_OA_store_dirs(self) -> None: Logger.print_status("Removing OctoApp for Klipper store directory ...") @@ -197,7 +194,6 @@ class OctoappExtension(BaseExtension): run_remove_routines(store_dir) - def _remove_OA_env(self) -> None: Logger.print_status("Removing OctoApp for Klipper environment ...") diff --git a/kiauh/extensions/telegram_bot/moonraker_telegram_bot.py b/kiauh/extensions/telegram_bot/moonraker_telegram_bot.py index a59ec61..e3d4191 100644 --- a/kiauh/extensions/telegram_bot/moonraker_telegram_bot.py +++ b/kiauh/extensions/telegram_bot/moonraker_telegram_bot.py @@ -116,10 +116,7 @@ class MoonrakerTelegramBot: "%TELEGRAM_BOT_DIR%", self.bot_dir.as_posix(), ) - env_file_content = env_file_content.replace( - "%CFG%", - self.cfg_file.as_posix() - ) + env_file_content = env_file_content.replace("%CFG%", self.cfg_file.as_posix()) env_file_content = env_file_content.replace( "%LOG%", self.base.log_dir.joinpath(self.log_file_name).as_posix(), diff --git a/kiauh/utils/common.py b/kiauh/utils/common.py index 2ace9fd..b23d258 100644 --- a/kiauh/utils/common.py +++ b/kiauh/utils/common.py @@ -192,5 +192,5 @@ def moonraker_exists(name: str = "") -> List[Moonraker]: def trunc_string(input_str: str, length: int) -> str: if len(input_str) > length: - return f"{input_str[:length - 3]}..." + return f"{input_str[: length - 3]}..." return input_str diff --git a/kiauh/utils/git_utils.py b/kiauh/utils/git_utils.py index a658607..6f509ac 100644 --- a/kiauh/utils/git_utils.py +++ b/kiauh/utils/git_utils.py @@ -132,8 +132,10 @@ def get_local_tags(repo_path: Path, _filter: str | None = None) -> List[str]: tags: List[str] = result.split("\n")[:-1] - return sorted(tags, key=lambda x: [int(i) if i.isdigit() else i for i in - re.split(r'(\d+)', x)]) + return sorted( + tags, + key=lambda x: [int(i) if i.isdigit() else i for i in re.split(r"(\d+)", x)], + ) except CalledProcessError: return [] diff --git a/kiauh/utils/instance_utils.py b/kiauh/utils/instance_utils.py index 92fd3a8..c73f690 100644 --- a/kiauh/utils/instance_utils.py +++ b/kiauh/utils/instance_utils.py @@ -17,7 +17,9 @@ from core.instance_manager.base_instance import SUFFIX_BLACKLIST from utils.instance_type import InstanceType -def get_instances(instance_type: type, suffix_blacklist: List[str] = SUFFIX_BLACKLIST) -> List[InstanceType]: +def get_instances( + instance_type: type, suffix_blacklist: List[str] = SUFFIX_BLACKLIST +) -> List[InstanceType]: from utils.common import convert_camelcase_to_kebabcase if not isinstance(instance_type, type): diff --git a/pyproject.toml b/pyproject.toml index 1881881..002b70d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,12 +5,13 @@ requires-python = ">=3.8" dev=["ruff", "mypy"] [tool.ruff] -required-version = ">=0.3.4" +required-version = ">=0.9.10" respect-gitignore = true -exclude = [".git",".github", "./docs"] +exclude = [".git",".github", "./docs", "kiauh/core/submodules"] line-length = 88 indent-width = 4 output-format = "full" +target-version = "py38" [tool.ruff.format] indent-style = "space"