feat: implement moonraker telegram bot extension

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

kiauh/components/klipperscreen/klipperscreen.py

@@ -55,6 +55,7 @@ def install_klipperscreen() -> None:
         warn_msg = [
             "Moonraker not found! KlipperScreen will not properly work "
             "without a working Moonraker installation.",
+            "\n\n",
             "KlipperScreens update manager configuration for Moonraker "
             "will not be added to any moonraker.conf.",
         ]
@@ -180,7 +181,7 @@ def remove_klipperscreen() -> None:
             cmd_sysctl_service(service, "stop")
             cmd_sysctl_service(service, "disable")
             remove_with_sudo(service)
-            cmd_sysctl_manage("deamon-reload")
+            cmd_sysctl_manage("daemon-reload")
             cmd_sysctl_manage("reset-failed")
             Logger.print_ok("KlipperScreen service successfully removed!")
 

kiauh/components/mobileraker/mobileraker.py

@@ -180,7 +180,7 @@ def remove_mobileraker() -> None:
             cmd_sysctl_service(service, "stop")
             cmd_sysctl_service(service, "disable")
             remove_with_sudo(service)
-            cmd_sysctl_manage("deamon-reload")
+            cmd_sysctl_manage("daemon-reload")
             cmd_sysctl_manage("reset-failed")
             Logger.print_ok("Mobileraker's companion service successfully removed!")
 

kiauh/core/instance_manager/base_instance.py

@@ -140,7 +140,9 @@ class BaseInstance(ABC):
             _dir.mkdir(exist_ok=True)
 
     def get_service_file_name(self, extension: bool = False) -> str:
-        name = f"{self.__class__.__name__.lower()}"
+        from utils.common import convert_camelcase_to_kebabcase
+
+        name = convert_camelcase_to_kebabcase(self.__class__.__name__)
         if self.suffix != "":
             name += f"-{self.suffix}"
 

kiauh/core/instance_manager/instance_manager.py

@@ -174,7 +174,9 @@ class InstanceManager:
             raise
 
     def find_instances(self) -> List[T]:
-        name = self.instance_type.__name__.lower()
+        from utils.common import convert_camelcase_to_kebabcase
+
+        name = convert_camelcase_to_kebabcase(self.instance_type.__name__)
         pattern = re.compile(f"^{name}(-[0-9a-zA-Z]+)?.service$")
         excluded = self.instance_type.blacklist()
 

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

@@ -0,0 +1 @@
+TELEGRAM_BOT_ARGS="%TELEGRAM_BOT_DIR%/bot/main.py -c %CFG% -l %LOG%"

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

@@ -0,0 +1,16 @@
+[Unit]
+Description=Moonraker Telegram Bot SV1 %INST%
+Documentation=https://github.com/nlef/moonraker-telegram-bot/wiki
+After=network-online.target
+
+[Install]
+WantedBy=multi-user.target
+
+[Service]
+Type=simple
+User=%USER%
+WorkingDirectory=%TELEGRAM_BOT_DIR%
+EnvironmentFile=%ENV_FILE%
+ExecStart=%ENV%/bin/python $TELEGRAM_BOT_ARGS
+Restart=always
+RestartSec=10

kiauh/extensions/telegram_bot/metadata.json

@@ -0,0 +1,11 @@
+{
+  "metadata": {
+    "index": 4,
+    "module": "moonraker_telegram_bot_extension",
+    "maintained_by": "nlef",
+    "display_name": "Moonraker Telegram Bot",
+    "description": "Allows to control your printer with the Telegram messenger app.",
+    "project_url": "https://github.com/nlef/moonraker-telegram-bot",
+    "updates": true
+  }
+}

kiauh/extensions/telegram_bot/moonraker_telegram_bot.py

@@ -0,0 +1,154 @@
+# ======================================================================= #
+#  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 subprocess
+from pathlib import Path
+from typing import List
+
+from core.instance_manager.base_instance import BaseInstance
+from utils.constants import SYSTEMD
+from utils.logger import Logger
+
+
+MODULE_PATH = Path(__file__).resolve().parent
+
+TELEGRAM_BOT_DIR = Path.home().joinpath("moonraker-telegram-bot")
+TELEGRAM_BOT_ENV = Path.home().joinpath("moonraker-telegram-bot-env")
+TELEGRAM_BOT_REPO = "https://github.com/nlef/moonraker-telegram-bot.git"
+
+
+# noinspection PyMethodMayBeStatic
+class MoonrakerTelegramBot(BaseInstance):
+    @classmethod
+    def blacklist(cls) -> List[str]:
+        return ["None", "mcu"]
+
+    def __init__(self, suffix: str = ""):
+        super().__init__(instance_type=self, suffix=suffix)
+        self.bot_dir: Path = TELEGRAM_BOT_DIR
+        self.env_dir: Path = TELEGRAM_BOT_ENV
+        self._cfg_file = self.cfg_dir.joinpath("telegram.conf")
+        self._log = self.log_dir.joinpath("telegram.log")
+        self._assets_dir = MODULE_PATH.joinpath("assets")
+
+    @property
+    def cfg_file(self) -> Path:
+        return self._cfg_file
+
+    @property
+    def log(self) -> Path:
+        return self._log
+
+    def create(self) -> None:
+        Logger.print_status("Creating new Moonraker Telegram Bot Instance ...")
+        service_template_path = MODULE_PATH.joinpath(
+            "assets/moonraker-telegram-bot.service"
+        )
+        service_file_name = self.get_service_file_name(extension=True)
+        service_file_target = SYSTEMD.joinpath(service_file_name)
+        env_template_file_path = MODULE_PATH.joinpath(
+            "assets/moonraker-telegram-bot.env"
+        )
+        env_file_target = self.sysd_dir.joinpath("moonraker-telegram-bot.env")
+
+        try:
+            self.create_folders()
+            self.write_service_file(
+                service_template_path, service_file_target, env_file_target
+            )
+            self.write_env_file(env_template_file_path, env_file_target)
+
+        except subprocess.CalledProcessError as e:
+            Logger.print_error(
+                f"Error creating service file {service_file_target}: {e}"
+            )
+            raise
+        except OSError as e:
+            Logger.print_error(f"Error creating env file {env_file_target}: {e}")
+            raise
+
+    def delete(self) -> None:
+        service_file = self.get_service_file_name(extension=True)
+        service_file_path = self.get_service_file_path()
+
+        Logger.print_status(f"Deleting Moonraker Telegram Bot Instance: {service_file}")
+
+        try:
+            command = ["sudo", "rm", "-f", service_file_path]
+            subprocess.run(command, check=True)
+            Logger.print_ok(f"Service file deleted: {service_file_path}")
+        except subprocess.CalledProcessError as e:
+            Logger.print_error(f"Error deleting service file: {e}")
+            raise
+
+    def write_service_file(
+        self,
+        service_template_path: Path,
+        service_file_target: Path,
+        env_file_target: Path,
+    ) -> None:
+        service_content = self._prep_service_file(
+            service_template_path, env_file_target
+        )
+        command = ["sudo", "tee", service_file_target]
+        subprocess.run(
+            command,
+            input=service_content.encode(),
+            stdout=subprocess.DEVNULL,
+            check=True,
+        )
+        Logger.print_ok(f"Service file created: {service_file_target}")
+
+    def write_env_file(
+        self, env_template_file_path: Path, env_file_target: Path
+    ) -> None:
+        env_file_content = self._prep_env_file(env_template_file_path)
+        with open(env_file_target, "w") as env_file:
+            env_file.write(env_file_content)
+        Logger.print_ok(f"Env file created: {env_file_target}")
+
+    def _prep_service_file(
+        self, service_template_path: Path, env_file_path: Path
+    ) -> str:
+        try:
+            with open(service_template_path, "r") as template_file:
+                template_content = template_file.read()
+        except FileNotFoundError:
+            Logger.print_error(
+                f"Unable to open {service_template_path} - File not found"
+            )
+            raise
+        service_content = template_content.replace("%USER%", self.user)
+        service_content = service_content.replace(
+            "%TELEGRAM_BOT_DIR%",
+            str(self.bot_dir),
+        )
+        service_content = service_content.replace("%ENV%", str(self.env_dir))
+        service_content = service_content.replace("%ENV_FILE%", str(env_file_path))
+        return service_content
+
+    def _prep_env_file(self, env_template_file_path: Path) -> str:
+        try:
+            with open(env_template_file_path, "r") as env_file:
+                env_template_file_content = env_file.read()
+        except FileNotFoundError:
+            Logger.print_error(
+                f"Unable to open {env_template_file_path} - File not found"
+            )
+            raise
+        env_file_content = env_template_file_content.replace(
+            "%TELEGRAM_BOT_DIR%",
+            str(self.bot_dir),
+        )
+        env_file_content = env_file_content.replace(
+            "%CFG%",
+            f"{self.cfg_dir}/printer.cfg",
+        )
+        env_file_content = env_file_content.replace("%LOG%", str(self.log))
+        return env_file_content

kiauh/extensions/telegram_bot/moonraker_telegram_bot_extension.py

@@ -0,0 +1,230 @@
+# ======================================================================= #
+#  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 shutil
+from subprocess import run
+from typing import List
+
+from components.moonraker.moonraker import Moonraker
+from core.instance_manager.instance_manager import InstanceManager
+from extensions.base_extension import BaseExtension
+from extensions.telegram_bot.moonraker_telegram_bot import (
+    MoonrakerTelegramBot,
+    TELEGRAM_BOT_REPO,
+    TELEGRAM_BOT_DIR,
+    TELEGRAM_BOT_ENV,
+)
+from utils.common import check_install_dependencies
+from utils.config_utils import add_config_section, remove_config_section
+from utils.fs_utils import remove_file
+from utils.git_utils import git_clone_wrapper, git_pull_wrapper
+from utils.input_utils import get_confirm
+from utils.logger import Logger, DialogType
+from utils.sys_utils import (
+    parse_packages_from_file,
+    create_python_venv,
+    install_python_requirements,
+    cmd_sysctl_manage,
+)
+
+
+# noinspection PyMethodMayBeStatic
+class TelegramBotExtension(BaseExtension):
+    def install_extension(self, **kwargs) -> None:
+        Logger.print_status("Installing Moonraker Telegram Bot ...")
+        mr_im = InstanceManager(Moonraker)
+        mr_instances: List[Moonraker] = mr_im.instances
+        if not mr_instances:
+            Logger.print_dialog(
+                DialogType.WARNING,
+                [
+                    "No Moonraker instances found!",
+                    "Moonraker Telegram Bot requires Moonraker to be installed. Please install Moonraker first!",
+                ],
+            )
+            return
+
+        instance_names = [f"● {instance.data_dir_name}" for instance in mr_instances]
+        Logger.print_dialog(
+            DialogType.INFO,
+            [
+                "The following Moonraker instances were found:",
+                *instance_names,
+                "\n\n",
+                "The setup will apply the same names to Telegram Bot!",
+            ],
+        )
+        if not get_confirm(
+            "Continue Moonraker Telegram Bot installation?",
+            default_choice=True,
+            allow_go_back=True,
+        ):
+            return
+
+        create_example_cfg = get_confirm("Create example telegram.conf?")
+
+        try:
+            git_clone_wrapper(TELEGRAM_BOT_REPO, TELEGRAM_BOT_DIR)
+            self._install_dependencies()
+
+            # create and start services / create bot configs
+            show_config_dialog = False
+            tb_im = InstanceManager(MoonrakerTelegramBot)
+            tb_names = [mr_i.suffix for mr_i in mr_instances]
+            for name in tb_names:
+                current_instance = MoonrakerTelegramBot(suffix=name)
+
+                tb_im.current_instance = current_instance
+                tb_im.create_instance()
+                tb_im.enable_instance()
+
+                if create_example_cfg:
+                    Logger.print_status(
+                        f"Creating Telegram Bot config in {current_instance.cfg_dir} ..."
+                    )
+                    template = TELEGRAM_BOT_DIR.joinpath(
+                        "scripts/base_install_template"
+                    )
+                    target_file = current_instance.cfg_file
+                    if not target_file.exists():
+                        show_config_dialog = True
+                        run(["cp", template, target_file], check=True)
+                    else:
+                        Logger.print_info(
+                            f"Telegram Bot config in {current_instance.cfg_dir} already exists! Skipped ..."
+                        )
+
+                tb_im.start_instance()
+
+            cmd_sysctl_manage("daemon-reload")
+
+            # add to moonraker update manager
+            self._patch_bot_update_manager(mr_instances)
+
+            # restart moonraker
+            mr_im.restart_all_instance()
+
+            if show_config_dialog:
+                Logger.print_dialog(
+                    DialogType.ATTENTION,
+                    [
+                        "During the installation of the Moonraker Telegram Bot, "
+                        "a basic config was created per instance. You need to edit the "
+                        "config file to set up your Telegram Bot. Please refer to the "
+                        "following wiki page for further information:",
+                        "https://github.com/nlef/moonraker-telegram-bot/wiki",
+                    ],
+                )
+
+            Logger.print_ok("Telegram Bot installation complete!")
+        except Exception as e:
+            Logger.print_error(
+                f"Error during installation of Moonraker Telegram Bot:\n{e}"
+            )
+
+    def update_extension(self, **kwargs) -> None:
+        Logger.print_status("Updating Moonraker Telegram Bot ...")
+        tb_im = InstanceManager(MoonrakerTelegramBot)
+        tb_im.stop_all_instance()
+
+        git_pull_wrapper(TELEGRAM_BOT_REPO, TELEGRAM_BOT_DIR)
+        self._install_dependencies()
+
+        tb_im.start_all_instance()
+
+    def remove_extension(self, **kwargs) -> None:
+        Logger.print_status("Removing Moonraker Telegram Bot ...")
+        mr_im = InstanceManager(Moonraker)
+        mr_instances: List[Moonraker] = mr_im.instances
+        tb_im = InstanceManager(MoonrakerTelegramBot)
+        tb_instances: List[MoonrakerTelegramBot] = tb_im.instances
+
+        try:
+            self._remove_bot_instances(tb_im, tb_instances)
+            self._remove_bot_dir()
+            self._remove_bot_env()
+            remove_config_section("update_manager moonraker-telegram-bot", mr_instances)
+            self._delete_bot_logs(tb_instances)
+        except Exception as e:
+            Logger.print_error(f"Error during removal of Moonraker Telegram Bot:\n{e}")
+
+        Logger.print_ok("Moonraker Telegram Bot removed!")
+
+    def _install_dependencies(self) -> None:
+        # install dependencies
+        script = TELEGRAM_BOT_DIR.joinpath("scripts/install.sh")
+        package_list = parse_packages_from_file(script)
+        check_install_dependencies(package_list)
+
+        # create virtualenv
+        create_python_venv(TELEGRAM_BOT_ENV)
+        requirements = TELEGRAM_BOT_DIR.joinpath("scripts/requirements.txt")
+        install_python_requirements(TELEGRAM_BOT_ENV, requirements)
+
+    def _patch_bot_update_manager(self, instances: List[Moonraker]) -> None:
+        env_py = f"{TELEGRAM_BOT_ENV}/bin/python"
+        add_config_section(
+            section="update_manager moonraker-telegram-bot",
+            instances=instances,
+            options=[
+                ("type", "git_repo"),
+                ("path", str(TELEGRAM_BOT_DIR)),
+                ("orgin", TELEGRAM_BOT_REPO),
+                ("env", env_py),
+                ("requirements", "scripts/requirements.txt"),
+                ("install_script", "scripts/install.sh"),
+            ],
+        )
+
+    def _remove_bot_instances(
+        self,
+        instance_manager: InstanceManager,
+        instance_list: List[MoonrakerTelegramBot],
+    ) -> 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_bot_dir(self) -> None:
+        if not TELEGRAM_BOT_DIR.exists():
+            Logger.print_info(f"'{TELEGRAM_BOT_DIR}' does not exist. Skipped ...")
+            return
+
+        try:
+            shutil.rmtree(TELEGRAM_BOT_DIR)
+        except OSError as e:
+            Logger.print_error(f"Unable to delete '{TELEGRAM_BOT_DIR}':\n{e}")
+
+    def _remove_bot_env(self) -> None:
+        if not TELEGRAM_BOT_ENV.exists():
+            Logger.print_info(f"'{TELEGRAM_BOT_ENV}' does not exist. Skipped ...")
+            return
+
+        try:
+            shutil.rmtree(TELEGRAM_BOT_ENV)
+        except OSError as e:
+            Logger.print_error(f"Unable to delete '{TELEGRAM_BOT_ENV}':\n{e}")
+
+    def _delete_bot_logs(self, instances: List[MoonrakerTelegramBot]) -> None:
+        all_logfiles = []
+        for instance in instances:
+            all_logfiles = list(instance.log_dir.glob("telegram_bot.log*"))
+        if not all_logfiles:
+            Logger.print_info("No Moonraker Telegram Bot logs found. Skipped ...")
+            return
+
+        for log in all_logfiles:
+            Logger.print_status(f"Remove '{log}'")
+            remove_file(log)

kiauh/utils/common.py

@@ -6,7 +6,7 @@
 #                                                                         #
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
-
+import re
 from datetime import datetime
 from pathlib import Path
 from typing import Dict, Literal, List, Type, Union
@@ -30,6 +30,10 @@ from utils.sys_utils import (
 )
 
 
+def convert_camelcase_to_kebabcase(name: str) -> str:
+    return re.sub(r"(?<!^)(?=[A-Z])", "-", name).lower()
+
+
 def get_current_date() -> Dict[Literal["date", "time"], str]:
     """
     Get the current date |

kiauh/utils/logger.py

@@ -89,6 +89,7 @@ class Logger:
         content: List[str],
         custom_title: str = None,
         custom_color: DialogCustomColor = None,
+        end: str = "\n",
     ) -> None:
         dialog_color = Logger._get_dialog_color(title, custom_color)
         dialog_title = Logger._get_dialog_title(title, custom_title)
@@ -99,7 +100,7 @@ class Logger:
 
         print(
             f"{top}{dialog_title_formatted}{dialog_content}{bottom}",
-            end="",
+            end=end,
         )
 
     @staticmethod
@@ -148,7 +149,10 @@ class Logger:
         for i, c in enumerate(content):
             paragraph = wrapper.wrap(c)
             lines.extend(paragraph)
-            if i < len(content) - 1:
+
+            # add a full blank line if we have a double newline
+            # character unless we are at the end of the list
+            if c == "\n\n" and i < len(content) - 1:
                 lines.append(" " * line_width)
 
         formatted_lines = [

kiauh/utils/sys_utils.py

@@ -36,7 +36,7 @@ SysCtlServiceAction = Literal[
     "mask",
     "unmask",
 ]
-SysCtlManageAction = Literal["deamon-reload", "reset-failed"]
+SysCtlManageAction = Literal["daemon-reload", "reset-failed"]
 
 
 def kill(opt_err_msg: str = "") -> None: