style: ruff format

Signed-off-by: Dominik Willner <th33xitus@gmail.com>

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)

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] = (
+            new_inst_list: List[Moonraker] = [
-                [instance_service.create_new_instance(k.suffix) for k in klipper_list])
+                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
+        url_list = [
-               new_instances if i.port]
+            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:

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([
+        return tuple(
-            int(part) if part.isdigit() else part
+            [
-            for part in re.split(r"\.|-", version)
+                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":

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())

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"""
             ╟───────────────────────────────────────────────────────╢

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

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([
+            warn_msg.extend(
-                "For the change to take effect, the new repository will be cloned. "
+                [
-                "A backup of the old repository will be created.",
+                    "For the change to take effect, the new repository will be cloned. "
-                "\n\n",
+                    "A backup of the old repository will be created.",
-                "Make sure you don't have any ongoing prints running, as the services "
+                    "\n\n",
-                "will be restarted during this process! You will loose any ongoing print!"])
+                    "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",
+            "Enter new branch name", regex=r"^.+$", default="master"
-            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)

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")
+            comparison_file_path = os.path.join(
-            if not (os.path.exists(original_file_path) and os.path.exists(comparison_file_path)):
+                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 = original_content.replace(
-                modified_content = "\n".join(line for line in modified_content.split("\n") if line.strip())
+                    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)
+                    crontab_content = subprocess.check_output(
-                    modified_content = "\n".join(line for line in crontab_content.splitlines() if "/klipper-backup/script.sh" not in line)
+                        ["crontab", "-l"], text=True
-                    subprocess.run(["crontab", "-"], input=modified_content + "\n", text=True, check=True)
+                    )
-                    Logger.print_ok("Klipper-Backup crontab entry successfully removed!")
+                    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():

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']
+        suffix_blacklist: List[str] = [
-        ob_instances: List[MoonrakerObico] = get_instances(MoonrakerObico, suffix_blacklist=suffix_blacklist)
+            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
         ]

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(
+        self.service_file_path: Path = get_service_file_path(Octoapp, self.suffix)
-            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)

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(
+            Logger.print_error(f"Error during OctoApp for Klipper installation:\n{e}")
-                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 ...")
 

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(
+        env_file_content = env_file_content.replace("%CFG%", self.cfg_file.as_posix())
-            "%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(),

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

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
+        return sorted(
-                                                re.split(r'(\d+)', x)])
+            tags,
+            key=lambda x: [int(i) if i.isdigit() else i for i in re.split(r"(\d+)", x)],
+        )
 
     except CalledProcessError:
         return []

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):

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"