jeka/kiauh
1
0
Fork 0
mirror of https://github.com/dw-0/kiauh.git synced 2025-12-25 08:43:36 +05:00
Code Issues Packages Projects Releases Wiki Activity

refactor: refactor Obico for Klipper to dataclass

Browse Source 
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2024-08-04 19:29:29 +02:00
parent 1fc50848b0
commit 88f784348b
3 changed files with 137 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
kiauh/extensions/obico/__init__.py
View File

@@ -0,0 +1,33 @@
# ======================================================================= #
# 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
MODULE_PATH = Path(__file__).resolve().parent
# repo
OBICO_REPO = "https://github.com/TheSpaghettiDetective/moonraker-obico.git"
# names
OBICO_SERVICE_NAME = "moonraker-obico.service"
OBICO_ENV_FILE_NAME = "moonraker-obico.env"
OBICO_CFG_NAME = "moonraker-obico.cfg"
OBICO_CFG_SAMPLE_NAME = "moonraker-obico.cfg.sample"
OBICO_LOG_NAME = "moonraker-obico.log"
OBICO_UPDATE_CFG_NAME = "moonraker-obico-update.cfg"
OBICO_UPDATE_CFG_SAMPLE_NAME = "moonraker-obico-update.cfg.sample"
OBICO_MACROS_CFG_NAME = "moonraker_obico_macros.cfg"
# directories
OBICO_DIR = Path.home().joinpath("moonraker-obico")
OBICO_ENV_DIR = Path.home().joinpath("moonraker-obico-env")
# files
OBICO_SERVICE_TEMPLATE = MODULE_PATH.joinpath(f"assets/{OBICO_SERVICE_NAME}")
OBICO_ENV_FILE_TEMPLATE = MODULE_PATH.joinpath(f"assets/{OBICO_ENV_FILE_NAME}")
OBICO_LINK_SCRIPT = OBICO_DIR.joinpath("scripts/link.sh")

167
kiauh/extensions/obico/moonraker_obico.py
View File

@@ -6,131 +6,91 @@
# #
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from subprocess import DEVNULL, CalledProcessError, run
from typing import List
from subprocess import CalledProcessError, run
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 extensions.obico import (
OBICO_CFG_NAME,
OBICO_DIR,
OBICO_ENV_DIR,
OBICO_ENV_FILE_NAME,
OBICO_ENV_FILE_TEMPLATE,
OBICO_LINK_SCRIPT,
OBICO_LOG_NAME,
OBICO_SERVICE_TEMPLATE,
)
from utils.logger import Logger
MODULE_PATH = Path(__file__).resolve().parent
OBICO_DIR = Path.home().joinpath("moonraker-obico")
OBICO_ENV = Path.home().joinpath("moonraker-obico-env")
OBICO_REPO = "https://github.com/TheSpaghettiDetective/moonraker-obico.git"
OBICO_CFG = "moonraker-obico.cfg"
OBICO_CFG_SAMPLE = "moonraker-obico.cfg.sample"
OBICO_LOG = "moonraker-obico.log"
OBICO_UPDATE_CFG = "moonraker-obico-update.cfg"
OBICO_UPDATE_CFG_SAMPLE = "moonraker-obico-update.cfg.sample"
OBICO_MACROS_CFG = "moonraker_obico_macros.cfg"
# noinspection PyMethodMayBeStatic
# todo: make this to a dataclass
@dataclass
class MoonrakerObico(BaseInstance):
@classmethod
def blacklist(cls) -> List[str]:
return ["None", "mcu"]
dir: Path = OBICO_DIR
env_dir: Path = OBICO_ENV_DIR
cfg_file: Path | None = None
log: Path | None = None
is_linked: bool = False
def __init__(self, suffix: str = ""):
super().__init__(instance_type=self, suffix=suffix)
self.dir: Path = OBICO_DIR
self.env_dir: Path = OBICO_ENV
self._cfg_file = self.cfg_dir.joinpath("moonraker-obico.cfg")
self._log = self.log_dir.joinpath("moonraker-obico.log")
self._is_linked: bool = self._check_link_status()
self._assets_dir = MODULE_PATH.joinpath("assets")
super().__init__(suffix=suffix)
@property
def cfg_file(self) -> Path:
return self._cfg_file
@property
def log(self) -> Path:
return self._log
@property
def is_linked(self) -> bool:
return self._is_linked
def __post_init__(self):
super().__post_init__()
self.cfg_file = self.cfg_dir.joinpath(OBICO_CFG_NAME)
self.log = self.log_dir.joinpath(OBICO_LOG_NAME)
self.is_linked: bool = self._check_link_status()
def create(self) -> None:
from utils.sys_utils import create_env_file, create_service_file
Logger.print_status("Creating new Obico for Klipper Instance ...")
service_template_path = MODULE_PATH.joinpath("assets/moonraker-obico.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-obico.env")
env_file_target = self.sysd_dir.joinpath("moonraker-obico.env")
try:
self.create_folders()
self.write_service_file(
service_template_path, service_file_target, env_file_target
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(OBICO_ENV_FILE_NAME),
content=self._prep_env_file_content(),
)
self.write_env_file(env_template_file_path, env_file_target)
except CalledProcessError as e:
Logger.print_error(
f"Error creating service file {service_file_target}: {e}"
)
Logger.print_error(f"Error creating instance: {e}")
raise
except OSError as e:
Logger.print_error(f"Error creating env file {env_file_target}: {e}")
Logger.print_error(f"Error creating env file: {e}")
raise
def delete(self) -> None:
service_file = self.get_service_file_name(extension=True)
service_file_path = self.get_service_file_path()
service_file: str = self.get_service_file_name(extension=True)
service_file_path: Path = self.get_service_file_path()
Logger.print_status(f"Deleting Obico for Klipper Instance: {service_file}")
try:
command = ["sudo", "rm", "-f", service_file_path]
command = ["sudo", "rm", "-f", service_file_path.as_posix()]
run(command, check=True)
self.delete_logfiles(OBICO_LOG_NAME)
Logger.print_ok(f"Service file deleted: {service_file_path}")
except 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.as_posix()]
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, 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 link(self) -> None:
Logger.print_status(
f"Linking instance for printer {self.data_dir_name} to the Obico server ..."
)
try:
script = OBICO_DIR.joinpath("scripts/link.sh")
cmd = [f"{script} -q -c {self.cfg_file}"]
cmd = [f"{OBICO_LINK_SCRIPT} -q -c {self.cfg_file}"]
if self.suffix:
cmd.append(f"-n {self.suffix}")
run(cmd, check=True, shell=True)
@@ -138,31 +98,42 @@ class MoonrakerObico(BaseInstance):
Logger.print_error(f"Error during Obico linking: {e}")
raise
def _prep_service_file(
self, service_template_path: Path, env_file_path: Path
) -> str:
def _prep_service_file_content(self) -> str:
template = OBICO_SERVICE_TEMPLATE
try:
with open(service_template_path, "r") as template_file:
with open(template, "r") as template_file:
template_content = template_file.read()
except FileNotFoundError:
Logger.print_error(
f"Unable to open {service_template_path} - File not found"
)
Logger.print_error(f"Unable to open {template} - File not found")
raise
service_content = template_content.replace("%USER%", self.user)
service_content = service_content.replace("%OBICO_DIR%", str(self.dir))
service_content = service_content.replace("%ENV%", str(self.env_dir))
service_content = service_content.replace("%ENV_FILE%", str(env_file_path))
service_content = template_content.replace(
"%USER%",
self.user,
)
service_content = service_content.replace(
"%OBICO_DIR%",
self.dir.as_posix(),
)
service_content = service_content.replace(
"%ENV%",
self.env_dir.as_posix(),
)
service_content = service_content.replace(
"%ENV_FILE%",
self.sysd_dir.joinpath(OBICO_ENV_FILE_NAME).as_posix(),
)
return service_content
def _prep_env_file(self, env_template_file_path: Path) -> str:
def _prep_env_file_content(self) -> str:
template = OBICO_ENV_FILE_TEMPLATE
try:
with open(env_template_file_path, "r") as env_file:
with open(template, "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"
)
Logger.print_error(f"Unable to open {template} - File not found")
raise
env_file_content = env_template_file_content.replace(
"%CFG%",
@@ -171,7 +142,7 @@ class MoonrakerObico(BaseInstance):
return env_file_content
def _check_link_status(self) -> bool:
if not self.cfg_file.exists():
if not self.cfg_file or not self.cfg_file.exists():
return False
scp = SimpleConfigParser()

81
kiauh/extensions/obico/moonraker_obico_extension.py
View File

@@ -16,15 +16,16 @@ from core.submodules.simple_config_parser.src.simple_config_parser.simple_config
SimpleConfigParser,
)
from extensions.base_extension import BaseExtension
from extensions.obico.moonraker_obico import (
OBICO_CFG_SAMPLE,
from extensions.obico import (
OBICO_CFG_SAMPLE_NAME,
OBICO_DIR,
OBICO_ENV,
OBICO_LOG,
OBICO_MACROS_CFG,
OBICO_ENV_DIR,
OBICO_MACROS_CFG_NAME,
OBICO_REPO,
OBICO_UPDATE_CFG,
OBICO_UPDATE_CFG_SAMPLE,
OBICO_UPDATE_CFG_NAME,
OBICO_UPDATE_CFG_SAMPLE_NAME,
)
from extensions.obico.moonraker_obico import (
MoonrakerObico,
)
from utils.common import check_install_dependencies, moonraker_exists
@@ -32,7 +33,7 @@ from utils.config_utils import (
add_config_section,
remove_config_section,
)
from utils.fs_utils import remove_file
from utils.fs_utils import run_remove_routines
from utils.git_utils import git_clone_wrapper, git_pull_wrapper
from utils.input_utils import get_confirm, get_selection_input, get_string_input
from utils.logger import DialogType, Logger
@@ -166,9 +167,8 @@ class ObicoExtension(BaseExtension):
self._remove_obico_instances(ob_im, ob_instances)
self._remove_obico_dir()
self._remove_obico_env()
remove_config_section(f"include {OBICO_MACROS_CFG}", kl_instances)
remove_config_section(f"include {OBICO_UPDATE_CFG}", mr_instances)
self._delete_obico_logs(ob_instances)
remove_config_section(f"include {OBICO_MACROS_CFG_NAME}", kl_instances)
remove_config_section(f"include {OBICO_UPDATE_CFG_NAME}", mr_instances)
Logger.print_dialog(
DialogType.SUCCESS,
["Obico for Klipper successfully removed!"],
@@ -239,34 +239,34 @@ class ObicoExtension(BaseExtension):
check_install_dependencies(package_list)
# create virtualenv
create_python_venv(OBICO_ENV)
create_python_venv(OBICO_ENV_DIR)
requirements = OBICO_DIR.joinpath("requirements.txt")
install_python_requirements(OBICO_ENV, requirements)
install_python_requirements(OBICO_ENV_DIR, requirements)
def _create_obico_macros_cfg(self, moonraker) -> None:
macros_cfg = OBICO_DIR.joinpath(f"include_cfgs/{OBICO_MACROS_CFG}")
macros_target = moonraker.cfg_dir.joinpath(OBICO_MACROS_CFG)
macros_cfg = OBICO_DIR.joinpath(f"include_cfgs/{OBICO_MACROS_CFG_NAME}")
macros_target = moonraker.cfg_dir.joinpath(OBICO_MACROS_CFG_NAME)
if not macros_target.exists():
shutil.copy(macros_cfg, macros_target)
else:
Logger.print_info(
f"Obico's '{OBICO_MACROS_CFG}' in {moonraker.cfg_dir} already exists! Skipped ..."
f"Obico's '{OBICO_MACROS_CFG_NAME}' in {moonraker.cfg_dir} already exists! Skipped ..."
)
def _create_obico_update_manager_cfg(self, moonraker) -> None:
update_cfg = OBICO_DIR.joinpath(OBICO_UPDATE_CFG_SAMPLE)
update_cfg_target = moonraker.cfg_dir.joinpath(OBICO_UPDATE_CFG)
update_cfg = OBICO_DIR.joinpath(OBICO_UPDATE_CFG_SAMPLE_NAME)
update_cfg_target = moonraker.cfg_dir.joinpath(OBICO_UPDATE_CFG_NAME)
if not update_cfg_target.exists():
shutil.copy(update_cfg, update_cfg_target)
else:
Logger.print_info(
f"Obico's '{OBICO_UPDATE_CFG}' in {moonraker.cfg_dir} already exists! Skipped ..."
f"Obico's '{OBICO_UPDATE_CFG_NAME}' in {moonraker.cfg_dir} already exists! Skipped ..."
)
def _create_obico_cfg(
self, current_instance: MoonrakerObico, moonraker: Moonraker
) -> None:
cfg_template = OBICO_DIR.joinpath(OBICO_CFG_SAMPLE)
cfg_template = OBICO_DIR.joinpath(OBICO_CFG_SAMPLE_NAME)
cfg_target_file = current_instance.cfg_file
if not cfg_template.exists():
@@ -288,14 +288,18 @@ class ObicoExtension(BaseExtension):
scp.read(obico.cfg_file)
scp.set("server", "url", self.server_url)
scp.set("moonraker", "port", str(moonraker.port))
scp.set("logging", "path", str(obico.log))
scp.set("logging", "path", obico.log.as_posix())
scp.write(obico.cfg_file)
def _patch_printer_cfg(self, klipper: List[Klipper]) -> None:
add_config_section(section=f"include {OBICO_MACROS_CFG}", instances=klipper)
add_config_section(
section=f"include {OBICO_MACROS_CFG_NAME}", instances=klipper
)
def _patch_moonraker_conf(self, instances: List[Moonraker]) -> None:
add_config_section(section=f"include {OBICO_UPDATE_CFG}", instances=instances)
add_config_section(
section=f"include {OBICO_UPDATE_CFG_NAME}", instances=instances
)
def _link_obico_instances(self, unlinked_instances) -> None:
for obico in unlinked_instances:
@@ -351,34 +355,19 @@ class ObicoExtension(BaseExtension):
cmd_sysctl_manage("daemon-reload")
def _remove_obico_dir(self) -> None:
Logger.print_status("Removing Obico for Klipper directory ...")
if not OBICO_DIR.exists():
Logger.print_info(f"'{OBICO_DIR}' does not exist. Skipped ...")
return
try:
shutil.rmtree(OBICO_DIR)
except OSError as e:
Logger.print_error(f"Unable to delete '{OBICO_DIR}':\n{e}")
run_remove_routines(OBICO_DIR)
def _remove_obico_env(self) -> None:
if not OBICO_ENV.exists():
Logger.print_info(f"'{OBICO_ENV}' does not exist. Skipped ...")
Logger.print_status("Removing Obico for Klipper environment ...")
if not OBICO_ENV_DIR.exists():
Logger.print_info(f"'{OBICO_ENV_DIR}' does not exist. Skipped ...")
return
try:
shutil.rmtree(OBICO_ENV)
except OSError as e:
Logger.print_error(f"Unable to delete '{OBICO_ENV}':\n{e}")
def _delete_obico_logs(self, instances: List[MoonrakerObico]) -> None:
Logger.print_status("Removing Obico logs ...")
all_logfiles = []
for instance in instances:
all_logfiles = list(instance.log_dir.glob(f"{OBICO_LOG}*"))
if not all_logfiles:
Logger.print_info("No Obico logs found. Skipped ...")
return
for log in all_logfiles:
Logger.print_status(f"Remove '{log}'")
remove_file(log)
run_remove_routines(OBICO_ENV_DIR)