From 1cd9414caea2c25faac963f13d05a2836be13e60 Mon Sep 17 00:00:00 2001 From: dw-0 Date: Mon, 1 Jul 2024 21:04:15 +0200 Subject: [PATCH] refactor: extract redundant code into shared methods Signed-off-by: Dominik Willner --- kiauh/components/klipper/klipper.py | 54 ++++++-------------- kiauh/components/moonraker/moonraker.py | 51 +++++------------- kiauh/core/instance_manager/base_instance.py | 10 ++++ kiauh/utils/sys_utils.py | 36 +++++++++++++ 4 files changed, 76 insertions(+), 75 deletions(-) diff --git a/kiauh/components/klipper/klipper.py b/kiauh/components/klipper/klipper.py index 90d2506..72e65cb 100644 --- a/kiauh/components/klipper/klipper.py +++ b/kiauh/components/klipper/klipper.py @@ -8,7 +8,7 @@ # ======================================================================= # from pathlib import Path -from subprocess import DEVNULL, CalledProcessError, run +from subprocess import CalledProcessError, run from typing import List from components.klipper import ( @@ -23,7 +23,6 @@ from components.klipper import ( KLIPPER_UDS_NAME, ) from core.instance_manager.base_instance import BaseInstance -from utils.constants import SYSTEMD from utils.logger import Logger @@ -59,12 +58,22 @@ class Klipper(BaseInstance): return self._uds def create(self) -> None: + from utils.sys_utils import create_env_file, create_service_file + Logger.print_status("Creating new Klipper Instance ...") try: self.create_folders() - self._write_service_file() - self._write_env_file() + + create_service_file( + name=self.get_service_file_name(extension=True), + content=self._prep_service_file_content(), + ) + + create_env_file( + path=self.sysd_dir.joinpath(KLIPPER_ENV_FILE_NAME), + content=self._prep_env_file_content(), + ) except CalledProcessError as e: Logger.print_error(f"Error creating instance: {e}") @@ -82,35 +91,13 @@ class Klipper(BaseInstance): try: command = ["sudo", "rm", "-f", service_file_path] run(command, check=True) - self._delete_logfiles() + self.delete_logfiles(KLIPPER_LOG_NAME) Logger.print_ok("Instance successfully removed!") except CalledProcessError as e: Logger.print_error(f"Error removing instance: {e}") raise - def _write_service_file(self) -> None: - service_file_name = self.get_service_file_name(extension=True) - service_file_target = SYSTEMD.joinpath(service_file_name) - service_content = self._prep_service_file() - command = ["sudo", "tee", service_file_target] - run( - command, - input=service_content.encode(), - stdout=DEVNULL, - check=True, - ) - Logger.print_ok(f"Service file created: {service_file_target}") - - def _write_env_file(self) -> None: - env_file_content = self._prep_env_file() - env_file_target = self.sysd_dir.joinpath(KLIPPER_ENV_FILE_NAME) - - 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) -> str: + def _prep_service_file_content(self) -> str: template = KLIPPER_SERVICE_TEMPLATE try: @@ -138,7 +125,7 @@ class Klipper(BaseInstance): ) return service_content - def _prep_env_file(self) -> str: + def _prep_env_file_content(self) -> str: template = KLIPPER_ENV_FILE_TEMPLATE try: @@ -169,12 +156,3 @@ class Klipper(BaseInstance): ) return env_file_content - - def _delete_logfiles(self) -> None: - from utils.fs_utils import run_remove_routines - - files = self.log_dir.iterdir() - logs = [f for f in files if f.name.startswith(KLIPPER_LOG_NAME)] - for log in logs: - Logger.print_status(f"Remove '{log}'") - run_remove_routines(log) diff --git a/kiauh/components/moonraker/moonraker.py b/kiauh/components/moonraker/moonraker.py index 68f8596..3dd5131 100644 --- a/kiauh/components/moonraker/moonraker.py +++ b/kiauh/components/moonraker/moonraker.py @@ -9,7 +9,7 @@ from __future__ import annotations from pathlib import Path -from subprocess import DEVNULL, CalledProcessError, run +from subprocess import CalledProcessError, run from typing import List from components.moonraker import ( @@ -25,7 +25,6 @@ from core.instance_manager.base_instance import BaseInstance from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import ( SimpleConfigParser, ) -from utils.constants import SYSTEMD from utils.logger import Logger @@ -56,12 +55,20 @@ class Moonraker(BaseInstance): return self._comms_dir def create(self, create_example_cfg: bool = False) -> None: + from utils.sys_utils import create_env_file, create_service_file + Logger.print_status("Creating new Moonraker Instance ...") try: self.create_folders([self.backup_dir, self.certs_dir, self._db_dir]) - self._write_service_file() - self._write_env_file() + create_service_file( + name=self.get_service_file_name(extension=True), + content=self._prep_service_file_content(), + ) + create_env_file( + path=self.sysd_dir.joinpath(MOONRAKER_ENV_FILE_NAME), + content=self._prep_env_file_content(), + ) except CalledProcessError as e: Logger.print_error(f"Error creating instance: {e}") @@ -79,34 +86,13 @@ class Moonraker(BaseInstance): try: command = ["sudo", "rm", "-f", service_file_path] run(command, check=True) - self._delete_logfiles() + self.delete_logfiles(MOONRAKER_LOG_NAME) Logger.print_ok("Instance successfully removed!") except CalledProcessError as e: Logger.print_error(f"Error removing instance: {e}") raise - def _write_service_file(self) -> None: - service_file_name = self.get_service_file_name(extension=True) - service_file_target = SYSTEMD.joinpath(service_file_name) - service_content = self._prep_service_file() - command = ["sudo", "tee", service_file_target] - run( - command, - input=service_content.encode(), - stdout=DEVNULL, - check=True, - ) - Logger.print_ok(f"Service file created: {service_file_target}") - - def _write_env_file(self) -> None: - env_file_content = self._prep_env_file() - env_file_target = self.sysd_dir.joinpath(MOONRAKER_ENV_FILE_NAME) - - 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) -> str: + def _prep_service_file_content(self) -> str: template = MOONRAKER_SERVICE_TEMPLATE try: @@ -134,7 +120,7 @@ class Moonraker(BaseInstance): ) return service_content - def _prep_env_file(self) -> str: + def _prep_env_file_content(self) -> str: template = MOONRAKER_ENV_FILE_TEMPLATE try: @@ -164,12 +150,3 @@ class Moonraker(BaseInstance): port = scp.getint("server", "port", fallback=None) return port - - def _delete_logfiles(self) -> None: - from utils.fs_utils import run_remove_routines - - files = self.log_dir.iterdir() - logs = [f for f in files if f.name.startswith(MOONRAKER_LOG_NAME)] - for log in logs: - Logger.print_status(f"Remove '{log}'") - run_remove_routines(log) diff --git a/kiauh/core/instance_manager/base_instance.py b/kiauh/core/instance_manager/base_instance.py index f99db2a..da0a4b0 100644 --- a/kiauh/core/instance_manager/base_instance.py +++ b/kiauh/core/instance_manager/base_instance.py @@ -14,6 +14,7 @@ from pathlib import Path from typing import List, Optional from utils.constants import CURRENT_USER, SYSTEMD +from utils.logger import Logger class BaseInstance(ABC): @@ -159,3 +160,12 @@ class BaseInstance(ABC): return f"printer_{self._suffix}" else: return self._suffix + + def delete_logfiles(self, log_name: str) -> None: + from utils.fs_utils import run_remove_routines + + files = self.log_dir.iterdir() + logs = [f for f in files if f.name.startswith(log_name)] + for log in logs: + Logger.print_status(f"Remove '{log}'") + run_remove_routines(log) diff --git a/kiauh/utils/sys_utils.py b/kiauh/utils/sys_utils.py index d3d3b0d..707f364 100644 --- a/kiauh/utils/sys_utils.py +++ b/kiauh/utils/sys_utils.py @@ -402,3 +402,39 @@ def log_process(process: Popen) -> None: if process.poll() is not None: break + + +def create_service_file(name: str, content: str) -> None: + """ + Creates a service file at the provided path with the provided content. + :param name: the name of the service file + :param content: the content of the service file + :return: None + """ + try: + run( + ["sudo", "tee", SYSTEMD.joinpath(name)], + input=content.encode(), + stdout=DEVNULL, + check=True, + ) + Logger.print_ok(f"Service file created: {SYSTEMD.joinpath(name)}") + except CalledProcessError as e: + Logger.print_error(f"Error creating service file: {e}") + raise + + +def create_env_file(path: Path, content: str) -> None: + """ + Creates an env file at the provided path with the provided content. + :param path: the path of the env file + :param content: the content of the env file + :return: None + """ + try: + with open(path, "w") as env_file: + env_file.write(content) + Logger.print_ok(f"Env file created: {path}") + except OSError as e: + Logger.print_error(f"Error creating env file: {e}") + raise