Merge 6a766f7882 into a8a73249a5

kiauh/components/klipper_firmware/firmware_utils.py

@@ -6,16 +6,8 @@
 #                                                                         #
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
-import re
+
-from subprocess import (
+from subprocess import PIPE, STDOUT, CalledProcessError, Popen, check_output, run
-    DEVNULL,
-    PIPE,
-    STDOUT,
-    CalledProcessError,
-    Popen,
-    check_output,
-    run,
-)
 from typing import List
 
 from components.klipper import KLIPPER_DIR
@@ -40,18 +32,16 @@ def find_firmware_file() -> bool:
     f3 = "klipper.bin"
     f4 = "klipper.uf2"
     fw_file_exists: bool = (
-        (target.joinpath(f1).exists() and target.joinpath(f2).exists())
+        target.joinpath(f1).exists() and target.joinpath(f2).exists()
-        or target.joinpath(f3).exists()
+    ) or target.joinpath(f3).exists() or target.joinpath(f4).exists()
-        or target.joinpath(f4).exists()
-    )
 
     return target_exists and fw_file_exists
 
 
 def find_usb_device_by_id() -> List[str]:
     try:
-        command = "find /dev/serial/by-id/*"
+        command = "find /dev/serial/by-id/* 2>/dev/null"
-        output = check_output(command, shell=True, text=True, stderr=DEVNULL)
+        output = check_output(command, shell=True, text=True)
         return output.splitlines()
     except CalledProcessError as e:
         Logger.print_error("Unable to find a USB device!")
@@ -61,14 +51,9 @@ def find_usb_device_by_id() -> List[str]:
 
 def find_uart_device() -> List[str]:
     try:
-        cmd = "find /dev -maxdepth 1"
+        command = '"find /dev -maxdepth 1 -regextype posix-extended -regex "^\/dev\/tty(AMA0|S0)$" 2>/dev/null"'
-        output = check_output(cmd, shell=True, text=True, stderr=DEVNULL)
+        output = check_output(command, shell=True, text=True)
-        device_list = []
+        return output.splitlines()
-        if output:
-            pattern = r"^/dev/tty(AMA0|S0)$"
-            devices = output.splitlines()
-            device_list = [d for d in devices if re.search(pattern, d)]
-        return device_list
     except CalledProcessError as e:
         Logger.print_error("Unable to find a UART device!")
         Logger.print_error(e, prefix=False)
@@ -77,13 +62,9 @@ def find_uart_device() -> List[str]:
 
 def find_usb_dfu_device() -> List[str]:
     try:
-        output = check_output("lsusb", shell=True, text=True, stderr=DEVNULL)
+        command = '"lsusb | grep "DFU" | cut -d " " -f 6 2>/dev/null"'
-        device_list = []
+        output = check_output(command, shell=True, text=True)
-        if output:
+        return output.splitlines()
-            devices = output.splitlines()
-            device_list = [d.split(" ")[5] for d in devices if "DFU" in d]
-        return device_list
-
     except CalledProcessError as e:
         Logger.print_error("Unable to find a USB DFU device!")
         Logger.print_error(e, prefix=False)

kiauh/extensions/obico/moonraker_obico.py

@@ -132,7 +132,7 @@ class MoonrakerObico:
             raise
         env_file_content = env_template_file_content.replace(
             "%CFG%",
-            f"{self.cfg_file}",
+            f"{self.base.cfg_dir}/{self.cfg_file}",
         )
         return env_file_content
 

kiauh/extensions/obico/moonraker_obico_extension.py

@@ -12,7 +12,6 @@ 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.logger import DialogType, Logger
 from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
     SimpleConfigParser,
@@ -309,8 +308,7 @@ 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)
-        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/__init__.py

@@ -1,29 +0,0 @@
-# ======================================================================= #
-#  Copyright (C) 2020 - 2024 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  #
-# ======================================================================= #
-from pathlib import Path
-
-# repo
-OA_REPO = "https://github.com/crysxd/OctoApp-Plugin.git"
-
-# directories
-OA_DIR = Path.home().joinpath("octoapp")
-OA_ENV_DIR = Path.home().joinpath("octoapp-env")
-OA_STORE_DIR = OA_DIR.joinpath("octoapp-store")
-
-# files
-OA_REQ_FILE = OA_DIR.joinpath("requirements.txt")
-OA_DEPS_JSON_FILE = OA_DIR.joinpath("moonraker-system-dependencies.json")
-OA_INSTALL_SCRIPT = OA_DIR.joinpath("install.sh")
-OA_UPDATE_SCRIPT = OA_DIR.joinpath("update.sh")
-OA_INSTALLER_LOG_FILE = Path.home().joinpath("octoapp-installer.log")
-
-# filenames
-OA_CFG_NAME = "octoapp.conf"
-OA_LOG_NAME = "octoapp.log"
-OA_SYS_CFG_NAME = "octoapp-system.cfg"

kiauh/extensions/octoapp/metadata.json

@@ -1,17 +0,0 @@
-{
-    "metadata": {
-        "index": 9,
-        "module": "octoapp_extension",
-        "maintained_by": "crysxd",
-        "display_name": "OctoApp for Klipper",
-        "description": [
-            "Your favorite 3D printing app for iOS & Android",
-            "- Print notifications on your phone & watch",
-            "- Control and start prints from your phone",
-            "- Live webcam view",
-            "- Live Gcode preview",
-            "- And much much more!"
-        ],
-        "updates": true
-    }
-}

kiauh/extensions/octoapp/octoapp.py

@@ -1,75 +0,0 @@
-# ======================================================================= #
-#  Copyright (C) 2020 - 2024 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  #
-# ======================================================================= #
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from pathlib import Path
-from subprocess import CalledProcessError, run
-
-from components.moonraker import MOONRAKER_CFG_NAME
-from components.moonraker.moonraker import Moonraker
-from core.instance_manager.base_instance import BaseInstance
-from core.logger import Logger
-from extensions.octoapp import (
-    OA_CFG_NAME,
-    OA_DIR,
-    OA_ENV_DIR,
-    OA_INSTALL_SCRIPT,
-    OA_LOG_NAME,
-    OA_SYS_CFG_NAME,
-    OA_UPDATE_SCRIPT,
-)
-from utils.sys_utils import get_service_file_path
-
-
-@dataclass
-class Octoapp:
-    suffix: str
-    base: BaseInstance = field(init=False, repr=False)
-    service_file_path: Path = field(init=False)
-    log_file_name = OA_LOG_NAME
-    dir: Path = OA_DIR
-    env_dir: Path = OA_ENV_DIR
-    data_dir: Path = field(init=False)
-    store_dir: Path = field(init=False)
-    cfg_file: Path = field(init=False)
-    sys_cfg_file: Path = field(init=False)
-
-    def __post_init__(self):
-        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.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)
-        self.data_dir = self.base.data_dir
-        self.sys_cfg_file = self.base.cfg_dir.joinpath(OA_SYS_CFG_NAME)
-
-    def create(self) -> None:
-        Logger.print_status("Creating OctoApp for Klipper Instance ...")
-
-        try:
-            cmd = f"{OA_INSTALL_SCRIPT} {self.base.cfg_dir}/{MOONRAKER_CFG_NAME}"
-            run(cmd, check=True, shell=True)
-
-        except CalledProcessError as e:
-            Logger.print_error(f"Error creating instance: {e}")
-            raise
-
-    @staticmethod
-    def update() -> None:
-        try:
-            run(OA_UPDATE_SCRIPT.as_posix(), check=True, shell=True, cwd=OA_DIR)
-
-        except CalledProcessError as e:
-            Logger.print_error(f"Error updating OctoApp for Klipper: {e}")
-            raise

kiauh/extensions/octoapp/octoapp_extension.py

@@ -1,191 +0,0 @@
-# ======================================================================= #
-#  Copyright (C) 2020 - 2024 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 json
-from typing import List
-
-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
-from extensions.octoapp import (
-    OA_DEPS_JSON_FILE,
-    OA_DIR,
-    OA_ENV_DIR,
-    OA_INSTALL_SCRIPT,
-    OA_INSTALLER_LOG_FILE,
-    OA_REPO,
-    OA_REQ_FILE,
-    OA_SYS_CFG_NAME,
-)
-from extensions.octoapp.octoapp import Octoapp
-from utils.common import (
-    check_install_dependencies,
-    moonraker_exists,
-)
-from utils.config_utils import (
-    remove_config_section,
-)
-from utils.fs_utils import run_remove_routines
-from utils.git_utils import git_clone_wrapper
-from utils.input_utils import get_confirm
-from utils.instance_utils import get_instances
-from utils.sys_utils import (
-    install_python_requirements,
-    parse_packages_from_file,
-)
-
-
-# noinspection PyMethodMayBeStatic
-class OctoappExtension(BaseExtension):
-    def install_extension(self, **kwargs) -> None:
-        Logger.print_status("Installing OctoApp for Klipper ...")
-
-        # check if moonraker is installed. if not, notify the user and exit
-        if not moonraker_exists():
-            return
-
-        force_clone = False
-        OA_instances: List[Octoapp] = get_instances(Octoapp)
-        if OA_instances:
-            Logger.print_dialog(
-                DialogType.INFO,
-                [
-                    "OctoApp is already installed!",
-                    "It is safe to run the installer again to link your "
-                    "printer or repair any issues.",
-                ],
-            )
-            if not get_confirm("Re-run OctoApp installation?"):
-                Logger.print_info("Exiting OctoApp for Klipper installation ...")
-                return
-            else:
-                Logger.print_status("Re-Installing OctoApp for Klipper ...")
-                force_clone = True
-
-        mr_instances: List[Moonraker] = get_instances(Moonraker)
-
-        mr_names = [f"● {moonraker.data_dir.name}" for moonraker in mr_instances]
-        if len(mr_names) > 1:
-            Logger.print_dialog(
-                DialogType.INFO,
-                [
-                    "The following Moonraker instances were found:",
-                    *mr_names,
-                    "\n\n",
-                    "The setup will apply the same names to OctoApp!",
-                ],
-            )
-
-        if not get_confirm(
-            "Continue OctoApp for Klipper installation?",
-            default_choice=True,
-            allow_go_back=True,
-        ):
-            Logger.print_info("Exiting OctoApp for Klipper installation ...")
-            return
-
-        try:
-            git_clone_wrapper(OA_REPO, OA_DIR, force=force_clone)
-
-            for moonraker in mr_instances:
-                instance = Octoapp(suffix=moonraker.suffix)
-                instance.create()
-
-            InstanceManager.restart_all(mr_instances)
-
-            Logger.print_dialog(
-                DialogType.SUCCESS,
-                ["OctoApp for Klipper successfully installed!"],
-                center_content=True,
-            )
-
-        except Exception as 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 ...")
-        try:
-            Octoapp.update()
-            Logger.print_dialog(
-                DialogType.SUCCESS,
-                ["OctoApp for Klipper successfully updated!"],
-                center_content=True,
-            )
-
-        except Exception as e:
-            Logger.print_error(f"Error during OctoApp for Klipper update:\n{e}")
-
-    def remove_extension(self, **kwargs) -> None:
-        Logger.print_status("Removing OctoApp for Klipper ...")
-
-        mr_instances: List[Moonraker] = get_instances(Moonraker)
-        ob_instances: List[Octoapp] = get_instances(Octoapp)
-
-        try:
-            self._remove_OA_instances(ob_instances)
-            self._remove_OA_dir()
-            self._remove_OA_env()
-            remove_config_section(f"include {OA_SYS_CFG_NAME}", mr_instances)
-            run_remove_routines(OA_INSTALLER_LOG_FILE)
-            Logger.print_dialog(
-                DialogType.SUCCESS,
-                ["OctoApp for Klipper successfully removed!"],
-                center_content=True,
-            )
-
-        except Exception as e:
-            Logger.print_error(f"Error during OctoApp for Klipper removal:\n{e}")
-
-    def _install_OA_dependencies(self) -> None:
-        OA_deps = []
-        if OA_DEPS_JSON_FILE.exists():
-            with open(OA_DEPS_JSON_FILE, "r") as deps:
-                OA_deps = json.load(deps).get("debian", [])
-        elif OA_INSTALL_SCRIPT.exists():
-            OA_deps = parse_packages_from_file(OA_INSTALL_SCRIPT)
-
-        if not OA_deps:
-            raise ValueError("Error reading OctoApp dependencies!")
-
-        check_install_dependencies({*OA_deps})
-        install_python_requirements(OA_ENV_DIR, OA_REQ_FILE)
-
-    def _remove_OA_instances(
-        self,
-        instance_list: List[Octoapp],
-    ) -> None:
-        if not instance_list:
-            Logger.print_info("No OctoApp instances found. Skipped ...")
-            return
-
-        for instance in instance_list:
-            Logger.print_status(
-                f"Removing instance {instance.service_file_path.stem} ..."
-            )
-            InstanceManager.remove(instance)
-
-    def _remove_OA_dir(self) -> None:
-        Logger.print_status("Removing OctoApp for Klipper directory ...")
-
-        if not OA_DIR.exists():
-            Logger.print_info(f"'{OA_DIR}' does not exist. Skipped ...")
-            return
-
-        run_remove_routines(OA_DIR)
-
-    def _remove_OA_env(self) -> None:
-        Logger.print_status("Removing OctoApp for Klipper environment ...")
-
-        if not OA_ENV_DIR.exists():
-            Logger.print_info(f"'{OA_ENV_DIR}' does not exist. Skipped ...")
-            return
-
-        run_remove_routines(OA_ENV_DIR)

kiauh/extensions/telegram_bot/assets/moonraker-telegram-bot.service

@@ -1,5 +1,5 @@
 [Unit]
-Description=Moonraker Telegram Bot SV1
+Description=Moonraker Telegram Bot SV1 %INST%
 Documentation=https://github.com/nlef/moonraker-telegram-bot/wiki
 After=network-online.target
 

kiauh/extensions/telegram_bot/moonraker_telegram_bot_extension.py

@@ -161,11 +161,10 @@ class TelegramBotExtension(BaseExtension):
         # install dependencies
         script = TG_BOT_DIR.joinpath("scripts/install.sh")
         package_list = parse_packages_from_file(script)
-
         check_install_dependencies({*package_list})
 
         # create virtualenv
-        if create_python_venv(TG_BOT_ENV, allow_access_to_system_site_packages=True):
+        if create_python_venv(TG_BOT_ENV):
             install_python_requirements(TG_BOT_ENV, TG_BOT_REQ_FILE)
 
     def _patch_bot_update_manager(self, instances: List[Moonraker]) -> None:

kiauh/utils/instance_type.py

@@ -13,7 +13,6 @@ from components.klipper.klipper import Klipper
 from components.moonraker.moonraker import Moonraker
 from extensions.obico.moonraker_obico import MoonrakerObico
 from extensions.octoeverywhere.octoeverywhere import Octoeverywhere
-from extensions.octoapp.octoapp import Octoapp
 from extensions.telegram_bot.moonraker_telegram_bot import MoonrakerTelegramBot
 
 InstanceType = TypeVar(
@@ -23,5 +22,4 @@ InstanceType = TypeVar(
     MoonrakerTelegramBot,
     MoonrakerObico,
     Octoeverywhere,
-    Octoapp,
 )

kiauh/utils/instance_utils.py

@@ -17,7 +17,7 @@ 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) -> List[InstanceType]:
     from utils.common import convert_camelcase_to_kebabcase
 
     if not isinstance(instance_type, type):
@@ -30,7 +30,7 @@ def get_instances(instance_type: type, suffix_blacklist: List[str] = SUFFIX_BLAC
         Path(SYSTEMD, service)
         for service in SYSTEMD.iterdir()
         if pattern.search(service.name)
-        and not any(s in service.name for s in suffix_blacklist)
+        and not any(s in service.name for s in SUFFIX_BLACKLIST)
     ]
 
     instance_list = [

kiauh/utils/sys_utils.py

@@ -91,27 +91,19 @@ def parse_packages_from_file(source_file: Path) -> List[str]:
     return packages
 
 
-def create_python_venv(
+def create_python_venv(target: Path, force: bool = False) -> bool:
-    target: Path,
-    force: bool = False,
-    allow_access_to_system_site_packages: 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 target: Path where to create the virtualenv at
     :param force: Force recreation of the virtualenv
-    :param allow_access_to_system_site_packages: give the virtual environment access to the system site-packages dir
+    :param target: Path where to create the virtualenv at
     :return: bool
     """
     Logger.print_status("Set up Python virtual environment ...")
-    cmd = ["virtualenv", "-p", "/usr/bin/python3", target.as_posix()]
-    cmd.append(
-        "--system-site-packages"
-    ) if allow_access_to_system_site_packages else None
     if not target.exists():
         try:
+            cmd = ["virtualenv", "-p", "/usr/bin/python3", target.as_posix()]
             run(cmd, check=True)
             Logger.print_ok("Setup of virtualenv successful!")
             return True

scripts/flash_klipper.sh

@@ -358,6 +358,7 @@ function get_usb_id() {
   unset mcu_list
   sleep 1
   mcus=$(find /dev/serial/by-id/* 2>/dev/null)
+  mcus+=" $(find /dev/serial/by-path/* 2>/dev/null)"
 
   for mcu in ${mcus}; do
     mcu_list+=("${mcu}")