diff --git a/kiauh/core/base_extension.py b/kiauh/core/base_extension.py new file mode 100644 index 0000000..09245ee --- /dev/null +++ b/kiauh/core/base_extension.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python3 + +# ======================================================================= # +# Copyright (C) 2020 - 2024 Dominik Willner # +# # +# 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 abc import abstractmethod, ABC +from typing import Dict + + +# noinspection PyUnusedLocal +# noinspection PyMethodMayBeStatic +class BaseExtension(ABC): + def __init__(self, metadata: Dict[str, str]): + self.metadata = metadata + + @abstractmethod + def install_extension(self, **kwargs) -> None: + raise NotImplementedError( + "Subclasses must implement the install_extension method" + ) + + @abstractmethod + def remove_extension(self, **kwargs) -> None: + raise NotImplementedError( + "Subclasses must implement the remove_extension method" + ) diff --git a/kiauh/core/menus/extensions_menu.py b/kiauh/core/menus/extensions_menu.py new file mode 100644 index 0000000..ebbc4c5 --- /dev/null +++ b/kiauh/core/menus/extensions_menu.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 + +# ======================================================================= # +# Copyright (C) 2020 - 2024 Dominik Willner # +# # +# 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 +import textwrap +import importlib +import inspect +from pathlib import Path +from typing import List, Dict + +from kiauh.core.base_extension import BaseExtension +from kiauh.core.menus import BACK_FOOTER +from kiauh.core.menus.base_menu import BaseMenu +from kiauh.utils.constants import RESET_FORMAT, COLOR_CYAN, COLOR_YELLOW + + +# noinspection PyUnusedLocal +# noinspection PyMethodMayBeStatic +class ExtensionsMenu(BaseMenu): + def __init__(self): + self.extensions = self.discover_extensions() + super().__init__( + header=True, + options=self.get_options(), + footer_type=BACK_FOOTER, + ) + + def discover_extensions(self) -> List[BaseExtension]: + extensions = [] + extensions_dir = Path(__file__).resolve().parents[2].joinpath("extensions") + + for extension in extensions_dir.iterdir(): + metadata_json = Path(extension).joinpath("metadata.json") + if not metadata_json.exists(): + continue + + try: + with open(metadata_json, "r") as m: + metadata = json.load(m).get("metadata") + module_name = ( + f"kiauh.extensions.{extension.name}.{metadata.get('module')}" + ) + name, extension = inspect.getmembers( + importlib.import_module(module_name), + predicate=lambda o: inspect.isclass(o) + and issubclass(o, BaseExtension) + and o != BaseExtension, + )[0] + extensions.append(extension(metadata)) + except (IOError, json.JSONDecodeError, ImportError) as e: + print(f"Failed loading extension {extension}: {e}") + + return sorted(extensions, key=lambda ex: ex.metadata.get("index")) + + def get_options(self) -> Dict[str, BaseMenu]: + options = {} + for extension in self.extensions: + index = extension.metadata.get("index") + options[f"{index}"] = ExtensionSubmenu(extension) + + return options + + def print_menu(self): + header = " [ Extensions Menu ] " + color = COLOR_CYAN + line1 = f"{COLOR_YELLOW}Available Extensions:{RESET_FORMAT}" + count = 62 - len(color) - len(RESET_FORMAT) + menu = textwrap.dedent( + f""" + /=======================================================\\ + | {color}{header:~^{count}}{RESET_FORMAT} | + |-------------------------------------------------------| + | {line1:<62} | + | | + """ + )[1:] + print(menu, end="") + + for extension in self.extensions: + index = extension.metadata.get("index") + name = extension.metadata.get("display_name") + row = f"{index}) {name}" + print(f"| {row:<53} |") + + +# noinspection PyUnusedLocal +# noinspection PyMethodMayBeStatic +class ExtensionSubmenu(BaseMenu): + def __init__(self, extension: BaseExtension): + self.extension = extension + self.extension_name = extension.metadata.get("display_name") + self.extension_desc = extension.metadata.get("description") + super().__init__( + header=False, + options={ + "1": extension.install_extension, + "2": extension.remove_extension, + }, + footer_type=BACK_FOOTER, + ) + + def print_menu(self) -> None: + header = f" [ {self.extension_name} ] " + color = COLOR_YELLOW + count = 62 - len(color) - len(RESET_FORMAT) + + wrapper = textwrap.TextWrapper(55, initial_indent="| ", subsequent_indent="| ") + lines = wrapper.wrap(self.extension_desc) + formatted_lines = [f"{line:<55} |" for line in lines] + description_text = "\n".join(formatted_lines) + + menu = textwrap.dedent( + f""" + /=======================================================\\ + | {color}{header:~^{count}}{RESET_FORMAT} | + |-------------------------------------------------------| + """ + )[1:] + menu += f"{description_text}\n" + menu += textwrap.dedent( + """ + |-------------------------------------------------------| + | 1) Install | + | 2) Remove | + """ + )[1:] + print(menu, end="") diff --git a/kiauh/core/menus/main_menu.py b/kiauh/core/menus/main_menu.py index aa59d79..366b99a 100644 --- a/kiauh/core/menus/main_menu.py +++ b/kiauh/core/menus/main_menu.py @@ -14,6 +14,7 @@ import textwrap from kiauh.core.menus import QUIT_FOOTER from kiauh.core.menus.advanced_menu import AdvancedMenu from kiauh.core.menus.base_menu import BaseMenu +from kiauh.core.menus.extensions_menu import ExtensionsMenu from kiauh.core.menus.install_menu import InstallMenu from kiauh.core.menus.remove_menu import RemoveMenu from kiauh.core.menus.settings_menu import SettingsMenu @@ -43,7 +44,7 @@ class MainMenu(BaseMenu): "3": RemoveMenu, "4": AdvancedMenu, "5": None, - "e": None, + "e": ExtensionsMenu, "s": SettingsMenu, }, footer_type=QUIT_FOOTER, diff --git a/kiauh/extensions/__init__.py b/kiauh/extensions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/kiauh/extensions/gcode_shell_cmd/__init__.py b/kiauh/extensions/gcode_shell_cmd/__init__.py new file mode 100644 index 0000000..c0d921b --- /dev/null +++ b/kiauh/extensions/gcode_shell_cmd/__init__.py @@ -0,0 +1,22 @@ +#!/usr/bin/env python3 + +# ======================================================================= # +# Copyright (C) 2020 - 2024 Dominik Willner # +# # +# 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 + + +EXT_MODULE_NAME = "gcode_shell_command.py" +MODULE_PATH = Path(__file__).resolve().parent +MODULE_ASSETS = MODULE_PATH.joinpath("assets") +KLIPPER_DIR = Path.home().joinpath("klipper") +KLIPPER_EXTRAS = KLIPPER_DIR.joinpath("klippy/extras") +EXTENSION_SRC = MODULE_ASSETS.joinpath(EXT_MODULE_NAME) +EXTENSION_TARGET_PATH = KLIPPER_EXTRAS.joinpath(EXT_MODULE_NAME) +EXAMPLE_CFG_SRC = MODULE_ASSETS.joinpath("shell_command.cfg") diff --git a/kiauh/extensions/gcode_shell_cmd/assets/gcode_shell_command.py b/kiauh/extensions/gcode_shell_cmd/assets/gcode_shell_command.py new file mode 100644 index 0000000..bb38ae5 --- /dev/null +++ b/kiauh/extensions/gcode_shell_cmd/assets/gcode_shell_command.py @@ -0,0 +1,87 @@ +# Run a shell command via gcode +# +# Copyright (C) 2019 Eric Callahan +# +# This file may be distributed under the terms of the GNU GPLv3 license. +import os +import shlex +import subprocess +import logging + +class ShellCommand: + def __init__(self, config): + self.name = config.get_name().split()[-1] + self.printer = config.get_printer() + self.gcode = self.printer.lookup_object('gcode') + cmd = config.get('command') + cmd = os.path.expanduser(cmd) + self.command = shlex.split(cmd) + self.timeout = config.getfloat('timeout', 2., above=0.) + self.verbose = config.getboolean('verbose', True) + self.proc_fd = None + self.partial_output = "" + self.gcode.register_mux_command( + "RUN_SHELL_COMMAND", "CMD", self.name, + self.cmd_RUN_SHELL_COMMAND, + desc=self.cmd_RUN_SHELL_COMMAND_help) + + def _process_output(self, eventime): + if self.proc_fd is None: + return + try: + data = os.read(self.proc_fd, 4096) + except Exception: + pass + data = self.partial_output + data.decode() + if '\n' not in data: + self.partial_output = data + return + elif data[-1] != '\n': + split = data.rfind('\n') + 1 + self.partial_output = data[split:] + data = data[:split] + else: + self.partial_output = "" + self.gcode.respond_info(data) + + cmd_RUN_SHELL_COMMAND_help = "Run a linux shell command" + def cmd_RUN_SHELL_COMMAND(self, params): + gcode_params = params.get('PARAMS','') + gcode_params = shlex.split(gcode_params) + reactor = self.printer.get_reactor() + try: + proc = subprocess.Popen( + self.command + gcode_params, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + except Exception: + logging.exception( + "shell_command: Command {%s} failed" % (self.name)) + raise self.gcode.error("Error running command {%s}" % (self.name)) + if self.verbose: + self.proc_fd = proc.stdout.fileno() + self.gcode.respond_info("Running Command {%s}...:" % (self.name)) + hdl = reactor.register_fd(self.proc_fd, self._process_output) + eventtime = reactor.monotonic() + endtime = eventtime + self.timeout + complete = False + while eventtime < endtime: + eventtime = reactor.pause(eventtime + .05) + if proc.poll() is not None: + complete = True + break + if not complete: + proc.terminate() + if self.verbose: + if self.partial_output: + self.gcode.respond_info(self.partial_output) + self.partial_output = "" + if complete: + msg = "Command {%s} finished\n" % (self.name) + else: + msg = "Command {%s} timed out" % (self.name) + self.gcode.respond_info(msg) + reactor.unregister_fd(hdl) + self.proc_fd = None + + +def load_config_prefix(config): + return ShellCommand(config) diff --git a/kiauh/extensions/gcode_shell_cmd/assets/shell_command.cfg b/kiauh/extensions/gcode_shell_cmd/assets/shell_command.cfg new file mode 100644 index 0000000..34e7581 --- /dev/null +++ b/kiauh/extensions/gcode_shell_cmd/assets/shell_command.cfg @@ -0,0 +1,7 @@ +[gcode_shell_command hello_world] +command: echo hello world +timeout: 2. +verbose: True +[gcode_macro HELLO_WORLD] +gcode: + RUN_SHELL_COMMAND CMD=hello_world \ No newline at end of file diff --git a/kiauh/extensions/gcode_shell_cmd/gcode_shell_cmd_extension.py b/kiauh/extensions/gcode_shell_cmd/gcode_shell_cmd_extension.py new file mode 100644 index 0000000..d8ed49b --- /dev/null +++ b/kiauh/extensions/gcode_shell_cmd/gcode_shell_cmd_extension.py @@ -0,0 +1,127 @@ +#!/usr/bin/env python3 + +# ======================================================================= # +# Copyright (C) 2020 - 2024 Dominik Willner # +# # +# 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 os +import shutil +from typing import List + +from kiauh.components.klipper.klipper import Klipper +from kiauh.core.backup_manager.backup_manager import BackupManager +from kiauh.core.base_extension import BaseExtension +from kiauh.core.config_manager.config_manager import ConfigManager +from kiauh.core.instance_manager.instance_manager import InstanceManager +from kiauh.extensions.gcode_shell_cmd import ( + EXTENSION_TARGET_PATH, + EXTENSION_SRC, + KLIPPER_DIR, + EXAMPLE_CFG_SRC, + KLIPPER_EXTRAS, +) +from kiauh.utils.filesystem_utils import check_file_exist +from kiauh.utils.input_utils import get_confirm +from kiauh.utils.logger import Logger + + +# noinspection PyMethodMayBeStatic +class GcodeShellCmdExtension(BaseExtension): + def install_extension(self, **kwargs) -> None: + install_example = get_confirm("Create an example shell command?", False, False) + + klipper_dir_exists = check_file_exist(KLIPPER_DIR) + if not klipper_dir_exists: + Logger.print_warn( + "No Klipper directory found! Unable to install extension." + ) + return + + extension_installed = check_file_exist(EXTENSION_TARGET_PATH) + overwrite = True + if extension_installed: + overwrite = get_confirm( + "Extension seems to be installed already. Overwrite?", True, False + ) + + if not overwrite: + Logger.print_warn("Installation aborted due to user request.") + return + + im = InstanceManager(Klipper) + im.stop_all_instance() + + try: + Logger.print_status(f"Copy extension to '{KLIPPER_EXTRAS}' ...") + shutil.copy(EXTENSION_SRC, EXTENSION_TARGET_PATH) + except OSError as e: + Logger.print_error(f"Unable to install extension: {e}") + return + + if install_example: + self.install_example_cfg(im.instances) + + im.start_all_instance() + + Logger.print_ok("Installing G-Code Shell Command extension successfull!") + + def remove_extension(self, **kwargs) -> None: + extension_installed = check_file_exist(EXTENSION_TARGET_PATH) + if not extension_installed: + Logger.print_info("Extension does not seem to be installed! Skipping ...") + return + + question = "Do you really want to remove the extension?" + if get_confirm(question, True, False): + try: + Logger.print_status(f"Removing '{EXTENSION_TARGET_PATH}' ...") + os.remove(EXTENSION_TARGET_PATH) + Logger.print_ok("Extension successfully removed!") + except OSError as e: + Logger.print_error(f"Unable to remove extension: {e}") + + Logger.print_warn("PLEASE NOTE:") + Logger.print_warn( + "Remaining gcode shell command will cause Klipper to throw an error." + ) + Logger.print_warn("Make sure to remove them from the printer.cfg!") + + def install_example_cfg(self, instances: List[Klipper]): + cfg_dirs = [instance.cfg_dir for instance in instances] + # copy extension to klippy/extras + for cfg_dir in cfg_dirs: + Logger.print_status(f"Create shell_command.cfg in '{cfg_dir}' ...") + if check_file_exist(cfg_dir.joinpath("shell_command.cfg")): + Logger.print_info("File already exists! Skipping ...") + continue + try: + shutil.copy(EXAMPLE_CFG_SRC, cfg_dir) + Logger.print_ok("Done!") + except OSError as e: + Logger.warn(f"Unable to create example config: {e}") + + # backup each printer.cfg before modification + bm = BackupManager() + for instance in instances: + bm.backup_file( + [instance.cfg_file], + custom_filename=f"{instance.suffix}.printer.cfg", + ) + + # add section to printer.cfg if not already defined + section = "include shell_command.cfg" + cfg_files = [instance.cfg_file for instance in instances] + for cfg_file in cfg_files: + Logger.print_status(f"Include shell_command.cfg in '{cfg_file}' ...") + cm = ConfigManager(cfg_file) + if cm.config.has_section(section): + Logger.print_info("Section already defined! Skipping ...") + continue + cm.config.add_section(section) + cm.write_config() + Logger.print_ok("Done!") diff --git a/kiauh/extensions/gcode_shell_cmd/metadata.json b/kiauh/extensions/gcode_shell_cmd/metadata.json new file mode 100644 index 0000000..cfb38b4 --- /dev/null +++ b/kiauh/extensions/gcode_shell_cmd/metadata.json @@ -0,0 +1,9 @@ +{ + "metadata": { + "index": 1, + "module": "gcode_shell_cmd_extension", + "maintained_by": "dw-0", + "display_name": "G-Code Shell Command", + "description": "Allows to run a shell command from gcode." + } +}