diff --git a/kiauh/components/klipper/klipper_utils.py b/kiauh/components/klipper/klipper_utils.py index e71cb3b..51dd8bd 100644 --- a/kiauh/components/klipper/klipper_utils.py +++ b/kiauh/components/klipper/klipper_utils.py @@ -48,10 +48,12 @@ from utils.logger import Logger from utils.system_utils import mask_system_service -def get_klipper_status() -> Dict[ - Literal["status", "status_code", "instances", "repo", "local", "remote"], - Union[str, int], -]: +def get_klipper_status() -> ( + Dict[ + Literal["status", "status_code", "instances", "repo", "local", "remote"], + Union[str, int], + ] +): status = get_install_status_common(Klipper, KLIPPER_DIR, KLIPPER_ENV_DIR) return { "status": status.get("status"), @@ -69,7 +71,9 @@ def check_is_multi_install( return not existing_instances and install_count > 1 -def check_is_single_to_multi_conversion(existing_instances: List[Klipper]) -> bool: +def check_is_single_to_multi_conversion( + existing_instances: List[Klipper], +) -> bool: return len(existing_instances) == 1 and existing_instances[0].suffix == "" diff --git a/kiauh/components/moonraker/moonraker_remove.py b/kiauh/components/moonraker/moonraker_remove.py index 7d469fb..dde6c0e 100644 --- a/kiauh/components/moonraker/moonraker_remove.py +++ b/kiauh/components/moonraker/moonraker_remove.py @@ -130,9 +130,15 @@ def remove_polkit_rules() -> None: return try: - command = [f"{MOONRAKER_DIR}/scripts/set-policykit-rules.sh", "--clear"] + command = [ + f"{MOONRAKER_DIR}/scripts/set-policykit-rules.sh", + "--clear", + ] subprocess.run( - command, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL, check=True + command, + stderr=subprocess.PIPE, + stdout=subprocess.DEVNULL, + check=True, ) except subprocess.CalledProcessError as e: Logger.print_error(f"Error while removing policykit rules: {e}") diff --git a/kiauh/components/moonraker/moonraker_setup.py b/kiauh/components/moonraker/moonraker_setup.py index d55fd63..ef8c5f6 100644 --- a/kiauh/components/moonraker/moonraker_setup.py +++ b/kiauh/components/moonraker/moonraker_setup.py @@ -12,7 +12,10 @@ import sys from pathlib import Path from components.webui_client import MAINSAIL_DIR -from components.webui_client.client_utils import enable_mainsail_remotemode, get_existing_clients +from components.webui_client.client_utils import ( + enable_mainsail_remotemode, + get_existing_clients, +) from kiauh import KIAUH_CFG from components.klipper.klipper import Klipper from components.moonraker import ( @@ -170,7 +173,10 @@ def install_moonraker_polkit() -> None: try: command = [POLKIT_SCRIPT, "--disable-systemctl"] result = subprocess.run( - command, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL, text=True + command, + stderr=subprocess.PIPE, + stdout=subprocess.DEVNULL, + text=True, ) if result.returncode != 0 or result.stderr: Logger.print_error(f"{result.stderr}", False) diff --git a/kiauh/components/moonraker/moonraker_utils.py b/kiauh/components/moonraker/moonraker_utils.py index ca1a15c..f8be7be 100644 --- a/kiauh/components/moonraker/moonraker_utils.py +++ b/kiauh/components/moonraker/moonraker_utils.py @@ -32,10 +32,12 @@ from utils.system_utils import ( ) -def get_moonraker_status() -> Dict[ - Literal["status", "status_code", "instances", "repo", "local", "remote"], - Union[str, int], -]: +def get_moonraker_status() -> ( + Dict[ + Literal["status", "status_code", "instances", "repo", "local", "remote"], + Union[str, int], + ] +): status = get_install_status_common(Moonraker, MOONRAKER_DIR, MOONRAKER_ENV_DIR) return { "status": status.get("status"), diff --git a/kiauh/components/webui_client/client_config/client_config_setup.py b/kiauh/components/webui_client/client_config/client_config_setup.py index 02f25a2..58d2f3b 100644 --- a/kiauh/components/webui_client/client_config/client_config_setup.py +++ b/kiauh/components/webui_client/client_config/client_config_setup.py @@ -16,7 +16,9 @@ from kiauh import KIAUH_CFG from components.klipper.klipper import Klipper from components.moonraker.moonraker import Moonraker from components.webui_client import ClientConfigData, ClientName, ClientData -from components.webui_client.client_dialogs import print_client_already_installed_dialog +from components.webui_client.client_dialogs import ( + print_client_already_installed_dialog, +) from components.webui_client.client_utils import ( load_client_data, backup_client_config_data, @@ -29,8 +31,9 @@ from core.repo_manager.repo_manager import RepoManager from utils.common import backup_printer_config_dir from utils.filesystem_utils import ( create_symlink, - add_config_section, add_config_section_at_top, - ) + add_config_section, + add_config_section_at_top, +) from utils.input_utils import get_confirm from utils.logger import Logger diff --git a/kiauh/components/webui_client/client_utils.py b/kiauh/components/webui_client/client_utils.py index d2c4e1f..23d34e3 100644 --- a/kiauh/components/webui_client/client_utils.py +++ b/kiauh/components/webui_client/client_utils.py @@ -112,7 +112,9 @@ def get_client_status(client: ClientData) -> str: ) -def get_client_config_status(client: ClientData) -> Dict[ +def get_client_config_status( + client: ClientData, +) -> Dict[ Literal["repo", "local", "remote"], Union[str, int], ]: diff --git a/kiauh/core/instance_manager/instance_manager.py b/kiauh/core/instance_manager/instance_manager.py index 07ecdd9..0879484 100644 --- a/kiauh/core/instance_manager/instance_manager.py +++ b/kiauh/core/instance_manager/instance_manager.py @@ -108,7 +108,12 @@ class InstanceManager: def enable_instance(self) -> None: Logger.print_status(f"Enabling {self.instance_service_full} ...") try: - command = ["sudo", "systemctl", "enable", self.instance_service_full] + command = [ + "sudo", + "systemctl", + "enable", + self.instance_service_full, + ] if subprocess.run(command, check=True): Logger.print_ok(f"{self.instance_service_full} enabled.") except subprocess.CalledProcessError as e: @@ -118,7 +123,12 @@ class InstanceManager: def disable_instance(self) -> None: Logger.print_status(f"Disabling {self.instance_service_full} ...") try: - command = ["sudo", "systemctl", "disable", self.instance_service_full] + command = [ + "sudo", + "systemctl", + "disable", + self.instance_service_full, + ] if subprocess.run(command, check=True): Logger.print_ok(f"{self.instance_service_full} disabled.") except subprocess.CalledProcessError as e: @@ -128,7 +138,12 @@ class InstanceManager: def start_instance(self) -> None: Logger.print_status(f"Starting {self.instance_service_full} ...") try: - command = ["sudo", "systemctl", "start", self.instance_service_full] + command = [ + "sudo", + "systemctl", + "start", + self.instance_service_full, + ] if subprocess.run(command, check=True): Logger.print_ok(f"{self.instance_service_full} started.") except subprocess.CalledProcessError as e: @@ -138,7 +153,12 @@ class InstanceManager: def restart_instance(self) -> None: Logger.print_status(f"Restarting {self.instance_service_full} ...") try: - command = ["sudo", "systemctl", "restart", self.instance_service_full] + command = [ + "sudo", + "systemctl", + "restart", + self.instance_service_full, + ] if subprocess.run(command, check=True): Logger.print_ok(f"{self.instance_service_full} restarted.") except subprocess.CalledProcessError as e: diff --git a/kiauh/core/menus/main_menu.py b/kiauh/core/menus/main_menu.py index 8d20036..5d017e3 100644 --- a/kiauh/core/menus/main_menu.py +++ b/kiauh/core/menus/main_menu.py @@ -68,7 +68,11 @@ class MainMenu(BaseMenu): def init_status(self) -> None: status_vars = ["kl", "mr", "ms", "fl", "ks", "mb", "cn"] for var in status_vars: - setattr(self, f"{var}_status", f"{COLOR_RED}Not installed!{RESET_FORMAT}") + setattr( + self, + f"{var}_status", + f"{COLOR_RED}Not installed!{RESET_FORMAT}", + ) def fetch_status(self) -> None: # klipper diff --git a/kiauh/core/menus/remove_menu.py b/kiauh/core/menus/remove_menu.py index 8c13efe..a7b61a4 100644 --- a/kiauh/core/menus/remove_menu.py +++ b/kiauh/core/menus/remove_menu.py @@ -10,7 +10,9 @@ import textwrap from components.klipper.menus.klipper_remove_menu import KlipperRemoveMenu -from components.moonraker.menus.moonraker_remove_menu import MoonrakerRemoveMenu +from components.moonraker.menus.moonraker_remove_menu import ( + MoonrakerRemoveMenu, +) from components.webui_client.client_utils import load_client_data from components.webui_client.menus.client_remove_menu import ClientRemoveMenu from core.menus import BACK_FOOTER diff --git a/kiauh/extensions/gcode_shell_cmd/gcode_shell_cmd_extension.py b/kiauh/extensions/gcode_shell_cmd/gcode_shell_cmd_extension.py index e726565..a18488f 100644 --- a/kiauh/extensions/gcode_shell_cmd/gcode_shell_cmd_extension.py +++ b/kiauh/extensions/gcode_shell_cmd/gcode_shell_cmd_extension.py @@ -44,7 +44,9 @@ class GcodeShellCmdExtension(BaseExtension): overwrite = True if extension_installed: overwrite = get_confirm( - "Extension seems to be installed already. Overwrite?", True, False + "Extension seems to be installed already. Overwrite?", + True, + False, ) if not overwrite: diff --git a/kiauh/extensions/mainsail_theme_installer/mainsail_theme_installer_extension.py b/kiauh/extensions/mainsail_theme_installer/mainsail_theme_installer_extension.py index 3074c81..546b6ac 100644 --- a/kiauh/extensions/mainsail_theme_installer/mainsail_theme_installer_extension.py +++ b/kiauh/extensions/mainsail_theme_installer/mainsail_theme_installer_extension.py @@ -15,7 +15,10 @@ from typing import List, Union from typing import TypedDict from components.klipper.klipper import Klipper -from components.klipper.klipper_dialogs import print_instance_overview, DisplayType +from components.klipper.klipper_dialogs import ( + print_instance_overview, + DisplayType, +) from core.base_extension import BaseExtension from core.instance_manager.base_instance import BaseInstance from core.instance_manager.instance_manager import InstanceManager @@ -50,7 +53,7 @@ class MainsailThemeInstallerExtension(BaseExtension): show_headline=True, show_index=True, show_select_all=True, - ) + ) printer_list = get_printer_selection(self.instances, True) if printer_list is None: return @@ -130,7 +133,7 @@ class MainsailThemeInstallMenu(BaseMenu): show_headline=True, show_index=True, show_select_all=True, - ) + ) printer_list = get_printer_selection(self.instances, True) if printer_list is None: @@ -146,7 +149,9 @@ class MainsailThemeInstallMenu(BaseMenu): Logger.print_info(theme_data.get("short_note"), prefix=False, end="\n\n") -def get_printer_selection(instances: List[BaseInstance], is_install: bool) -> Union[List[BaseInstance], None]: +def get_printer_selection( + instances: List[BaseInstance], is_install: bool +) -> Union[List[BaseInstance], None]: options = [str(i) for i in range(len(instances))] options.extend(["a", "A", "b", "B"]) diff --git a/kiauh/utils/input_utils.py b/kiauh/utils/input_utils.py index 25a1b71..88b9188 100644 --- a/kiauh/utils/input_utils.py +++ b/kiauh/utils/input_utils.py @@ -51,7 +51,11 @@ def get_confirm( def get_number_input( - question: str, min_count: int, max_count=None, default=None, allow_go_back=False + question: str, + min_count: int, + max_count=None, + default=None, + allow_go_back=False, ) -> Union[int, None]: """ Helper method to get a number input from the user diff --git a/kiauh/utils/system_utils.py b/kiauh/utils/system_utils.py index 3b46913..991e65e 100644 --- a/kiauh/utils/system_utils.py +++ b/kiauh/utils/system_utils.py @@ -130,7 +130,12 @@ def install_python_requirements(target: Path, requirements: Path) -> None: update_python_pip(target) Logger.print_status("Installing Python requirements ...") - command = [target.joinpath("bin/pip"), "install", "-r", f"{requirements}"] + command = [ + target.joinpath("bin/pip"), + "install", + "-r", + f"{requirements}", + ] result = subprocess.run(command, stderr=subprocess.PIPE, text=True) if result.returncode != 0 or result.stderr: @@ -196,7 +201,10 @@ def check_package_install(packages: List[str]) -> List[str]: for package in packages: command = ["dpkg-query", "-f'${Status}'", "--show", package] result = subprocess.run( - command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True + command, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, ) if "installed" not in result.stdout.strip("'").split(): not_installed.append(package) diff --git a/resources/gcode_shell_command.py b/resources/gcode_shell_command.py index bb38ae5..3f316e6 100755 --- a/resources/gcode_shell_command.py +++ b/resources/gcode_shell_command.py @@ -8,22 +8,26 @@ 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') + 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.timeout = config.getfloat("timeout", 2.0, above=0.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, + "RUN_SHELL_COMMAND", + "CMD", + self.name, self.cmd_RUN_SHELL_COMMAND, - desc=self.cmd_RUN_SHELL_COMMAND_help) + desc=self.cmd_RUN_SHELL_COMMAND_help, + ) def _process_output(self, eventime): if self.proc_fd is None: @@ -33,11 +37,11 @@ class ShellCommand: except Exception: pass data = self.partial_output + data.decode() - if '\n' not in data: + if "\n" not in data: self.partial_output = data return - elif data[-1] != '\n': - split = data.rfind('\n') + 1 + elif data[-1] != "\n": + split = data.rfind("\n") + 1 self.partial_output = data[split:] data = data[:split] else: @@ -45,16 +49,19 @@ class ShellCommand: 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 = 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) + self.command + gcode_params, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) except Exception: - logging.exception( - "shell_command: Command {%s} failed" % (self.name)) + 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() @@ -64,7 +71,7 @@ class ShellCommand: endtime = eventtime + self.timeout complete = False while eventtime < endtime: - eventtime = reactor.pause(eventtime + .05) + eventtime = reactor.pause(eventtime + 0.05) if proc.poll() is not None: complete = True break