refactor: create a remove base method and use it in all subclasses

replace get_service_file_path with service_file_path and remove get_service_file_name

Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2024-08-19 21:14:19 +02:00
parent fdfdf18dd2
commit fe0bfc5376
15 changed files with 89 additions and 193 deletions

View File

@@ -10,7 +10,7 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from subprocess import CalledProcessError, run from subprocess import CalledProcessError
from components.klipper import ( from components.klipper import (
KLIPPER_CFG_NAME, KLIPPER_CFG_NAME,
@@ -32,8 +32,8 @@ from core.logger import Logger
class Klipper(BaseInstance): class Klipper(BaseInstance):
klipper_dir: Path = KLIPPER_DIR klipper_dir: Path = KLIPPER_DIR
env_dir: Path = KLIPPER_ENV_DIR env_dir: Path = KLIPPER_ENV_DIR
log_file_name = KLIPPER_LOG_NAME
cfg_file: Path | None = None cfg_file: Path | None = None
log: Path | None = None
serial: Path | None = None serial: Path | None = None
uds: Path | None = None uds: Path | None = None
@@ -43,7 +43,6 @@ class Klipper(BaseInstance):
def __post_init__(self) -> None: def __post_init__(self) -> None:
super().__post_init__() super().__post_init__()
self.cfg_file = self.cfg_dir.joinpath(KLIPPER_CFG_NAME) self.cfg_file = self.cfg_dir.joinpath(KLIPPER_CFG_NAME)
self.log = self.log_dir.joinpath(KLIPPER_LOG_NAME)
self.serial = self.comms_dir.joinpath(KLIPPER_SERIAL_NAME) self.serial = self.comms_dir.joinpath(KLIPPER_SERIAL_NAME)
self.uds = self.comms_dir.joinpath(KLIPPER_UDS_NAME) self.uds = self.comms_dir.joinpath(KLIPPER_UDS_NAME)
@@ -56,7 +55,7 @@ class Klipper(BaseInstance):
self.create_folders() self.create_folders()
create_service_file( create_service_file(
name=self.get_service_file_name(extension=True), name=self.service_file_path.name,
content=self._prep_service_file_content(), content=self._prep_service_file_content(),
) )
@@ -72,21 +71,6 @@ class Klipper(BaseInstance):
Logger.print_error(f"Error creating env file: {e}") Logger.print_error(f"Error creating env file: {e}")
raise 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"Removing Klipper Instance: {service_file}")
try:
command = ["sudo", "rm", "-f", service_file_path]
run(command, check=True)
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 _prep_service_file_content(self) -> str: def _prep_service_file_content(self) -> str:
template = KLIPPER_SERVICE_TEMPLATE template = KLIPPER_SERVICE_TEMPLATE
@@ -138,7 +122,7 @@ class Klipper(BaseInstance):
) )
env_file_content = env_file_content.replace( env_file_content = env_file_content.replace(
"%LOG%", "%LOG%",
self.log.as_posix() if self.log else "", self.log_dir.joinpath(self.log_file_name).as_posix(),
) )
env_file_content = env_file_content.replace( env_file_content = env_file_content.replace(
"%UDS%", "%UDS%",

View File

@@ -53,7 +53,7 @@ def print_instance_overview(
for i, s in enumerate(instances): for i, s in enumerate(instances):
if display_type is DisplayType.SERVICE_NAME: if display_type is DisplayType.SERVICE_NAME:
name = s.get_service_file_name() name = s.service_file_path.stem
else: else:
name = s.data_dir name = s.data_dir
line = f"{COLOR_CYAN}{f'{i + start_index})' if show_index else ''} {name}{RESET_FORMAT}" line = f"{COLOR_CYAN}{f'{i + start_index})' if show_index else ''} {name}{RESET_FORMAT}"

View File

@@ -17,7 +17,6 @@ from core.instance_manager.instance_manager import InstanceManager
from core.logger import Logger from core.logger import Logger
from utils.fs_utils import run_remove_routines from utils.fs_utils import run_remove_routines
from utils.input_utils import get_selection_input from utils.input_utils import get_selection_input
from utils.sys_utils import cmd_sysctl_manage
def run_klipper_removal( def run_klipper_removal(
@@ -25,17 +24,17 @@ def run_klipper_removal(
remove_dir: bool, remove_dir: bool,
remove_env: bool, remove_env: bool,
) -> None: ) -> None:
im = InstanceManager(Klipper) klipper_instances = InstanceManager(Klipper).instances
if remove_service: if remove_service:
Logger.print_status("Removing Klipper instances ...") Logger.print_status("Removing Klipper instances ...")
if im.instances: if klipper_instances:
instances_to_remove = select_instances_to_remove(im.instances) instances_to_remove = select_instances_to_remove(klipper_instances)
remove_instances(im, instances_to_remove) remove_instances(instances_to_remove)
else: else:
Logger.print_info("No Klipper Services installed! Skipped ...") Logger.print_info("No Klipper Services installed! Skipped ...")
if (remove_dir or remove_env) and im.instances: if (remove_dir or remove_env) and klipper_instances:
Logger.print_info("There are still other Klipper services installed:") Logger.print_info("There are still other Klipper services installed:")
Logger.print_info(f"'{KLIPPER_DIR}' was not removed.", prefix=False) Logger.print_info(f"'{KLIPPER_DIR}' was not removed.", prefix=False)
Logger.print_info(f"'{KLIPPER_ENV_DIR}' was not removed.", prefix=False) Logger.print_info(f"'{KLIPPER_ENV_DIR}' was not removed.", prefix=False)
@@ -74,20 +73,14 @@ def select_instances_to_remove(instances: List[Klipper]) -> List[Klipper] | None
def remove_instances( def remove_instances(
instance_manager: InstanceManager,
instance_list: List[Klipper] | None, instance_list: List[Klipper] | None,
) -> None: ) -> None:
if not instance_list: if not instance_list:
return return
for instance in instance_list: for instance in instance_list:
Logger.print_status(f"Removing instance {instance.get_service_file_name()} ...") Logger.print_status(f"Removing instance {instance.service_file_path.stem} ...")
instance_manager.current_instance = instance instance.remove()
instance_manager.stop_instance()
instance_manager.disable_instance()
instance_manager.delete_instance()
cmd_sysctl_manage("daemon-reload")
def delete_klipper_logs(instances: List[Klipper]) -> None: def delete_klipper_logs(instances: List[Klipper]) -> None:

View File

@@ -231,7 +231,7 @@ def display_moonraker_info(moonraker_list: List[Moonraker]) -> bool:
DialogType.INFO, DialogType.INFO,
[ [
"Existing Moonraker instances detected:", "Existing Moonraker instances detected:",
*[f"{m.get_service_file_name()}" for m in moonraker_list], *[f"{m.service_file_path.stem}" for m in moonraker_list],
"\n\n", "\n\n",
"The following Klipper instances will be installed:", "The following Klipper instances will be installed:",
*[f"● klipper-{m.suffix}" for m in moonraker_list], *[f"● klipper-{m.suffix}" for m in moonraker_list],

View File

@@ -10,7 +10,7 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from subprocess import CalledProcessError, run from subprocess import CalledProcessError
from components.moonraker import ( from components.moonraker import (
MOONRAKER_CFG_NAME, MOONRAKER_CFG_NAME,
@@ -33,12 +33,12 @@ from core.submodules.simple_config_parser.src.simple_config_parser.simple_config
class Moonraker(BaseInstance): class Moonraker(BaseInstance):
moonraker_dir: Path = MOONRAKER_DIR moonraker_dir: Path = MOONRAKER_DIR
env_dir: Path = MOONRAKER_ENV_DIR env_dir: Path = MOONRAKER_ENV_DIR
log_file_name = MOONRAKER_LOG_NAME
cfg_file: Path | None = None cfg_file: Path | None = None
port: int | None = None port: int | None = None
backup_dir: Path | None = None backup_dir: Path | None = None
certs_dir: Path | None = None certs_dir: Path | None = None
db_dir: Path | None = None db_dir: Path | None = None
log: Path | None = None
def __init__(self, suffix: str = ""): def __init__(self, suffix: str = ""):
super().__init__(suffix=suffix) super().__init__(suffix=suffix)
@@ -50,7 +50,6 @@ class Moonraker(BaseInstance):
self.backup_dir = self.data_dir.joinpath("backup") self.backup_dir = self.data_dir.joinpath("backup")
self.certs_dir = self.data_dir.joinpath("certs") self.certs_dir = self.data_dir.joinpath("certs")
self.db_dir = self.data_dir.joinpath("database") self.db_dir = self.data_dir.joinpath("database")
self.log = self.log_dir.joinpath(MOONRAKER_LOG_NAME)
def create(self, create_example_cfg: bool = False) -> None: def create(self, create_example_cfg: bool = False) -> None:
from utils.sys_utils import create_env_file, create_service_file from utils.sys_utils import create_env_file, create_service_file
@@ -60,7 +59,7 @@ class Moonraker(BaseInstance):
try: try:
self.create_folders([self.backup_dir, self.certs_dir, self.db_dir]) self.create_folders([self.backup_dir, self.certs_dir, self.db_dir])
create_service_file( create_service_file(
name=self.get_service_file_name(extension=True), name=self.service_file_path.name,
content=self._prep_service_file_content(), content=self._prep_service_file_content(),
) )
create_env_file( create_env_file(
@@ -75,21 +74,6 @@ class Moonraker(BaseInstance):
Logger.print_error(f"Error creating env file: {e}") Logger.print_error(f"Error creating env file: {e}")
raise 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"Removing Moonraker Instance: {service_file}")
try:
command = ["sudo", "rm", "-f", service_file_path]
run(command, check=True)
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 _prep_service_file_content(self) -> str: def _prep_service_file_content(self) -> str:
template = MOONRAKER_SERVICE_TEMPLATE template = MOONRAKER_SERVICE_TEMPLATE

View File

@@ -37,8 +37,8 @@ def print_moonraker_overview(
dialog += "║ ║\n" dialog += "║ ║\n"
instance_map = { instance_map = {
k.get_service_file_name(): ( k.service_file_path.stem: (
k.get_service_file_name().replace("klipper", "moonraker") k.service_file_path.stem.replace("klipper", "moonraker")
if k.suffix in [m.suffix for m in moonraker_instances] if k.suffix in [m.suffix for m in moonraker_instances]
else "" else ""
) )

View File

@@ -18,7 +18,6 @@ from core.instance_manager.instance_manager import InstanceManager
from core.logger import Logger from core.logger import Logger
from utils.fs_utils import run_remove_routines from utils.fs_utils import run_remove_routines
from utils.input_utils import get_selection_input from utils.input_utils import get_selection_input
from utils.sys_utils import cmd_sysctl_manage
def run_moonraker_removal( def run_moonraker_removal(
@@ -27,17 +26,17 @@ def run_moonraker_removal(
remove_env: bool, remove_env: bool,
remove_polkit: bool, remove_polkit: bool,
) -> None: ) -> None:
im = InstanceManager(Moonraker) moonraker_instances = InstanceManager(Moonraker).instances
if remove_service: if remove_service:
Logger.print_status("Removing Moonraker instances ...") Logger.print_status("Removing Moonraker instances ...")
if im.instances: if moonraker_instances:
instances_to_remove = select_instances_to_remove(im.instances) instances_to_remove = select_instances_to_remove(moonraker_instances)
remove_instances(im, instances_to_remove) remove_instances(instances_to_remove)
else: else:
Logger.print_info("No Moonraker Services installed! Skipped ...") Logger.print_info("No Moonraker Services installed! Skipped ...")
if (remove_polkit or remove_dir or remove_env) and im.instances: if (remove_polkit or remove_dir or remove_env) and moonraker_instances:
Logger.print_info("There are still other Moonraker services installed") Logger.print_info("There are still other Moonraker services installed")
Logger.print_info( Logger.print_info(
"● Moonraker PolicyKit rules were not removed.", prefix=False "● Moonraker PolicyKit rules were not removed.", prefix=False
@@ -84,20 +83,14 @@ def select_instances_to_remove(
def remove_instances( def remove_instances(
instance_manager: InstanceManager,
instance_list: List[Moonraker] | None, instance_list: List[Moonraker] | None,
) -> None: ) -> None:
if not instance_list: if not instance_list:
Logger.print_info("No Moonraker instances found. Skipped ...") Logger.print_info("No Moonraker instances found. Skipped ...")
return return
for instance in instance_list: for instance in instance_list:
Logger.print_status(f"Removing instance {instance.get_service_file_name()} ...") Logger.print_status(f"Removing instance {instance.service_file_path.stem} ...")
instance_manager.current_instance = instance instance.remove()
instance_manager.stop_instance()
instance_manager.disable_instance()
instance_manager.delete_instance()
cmd_sysctl_manage("daemon-reload")
def remove_polkit_rules() -> None: def remove_polkit_rules() -> None:

View File

@@ -31,10 +31,10 @@ from core.logger import Logger
class Octoeverywhere(BaseInstance): class Octoeverywhere(BaseInstance):
dir: Path = OE_DIR dir: Path = OE_DIR
env_dir: Path = OE_ENV_DIR env_dir: Path = OE_ENV_DIR
log_file_name = OE_LOG_NAME
store_dir: Path = OE_STORE_DIR store_dir: Path = OE_STORE_DIR
cfg_file: Path | None = None cfg_file: Path | None = None
sys_cfg_file: Path | None = None sys_cfg_file: Path | None = None
log: Path | None = None
def __init__(self, suffix: str = ""): def __init__(self, suffix: str = ""):
super().__init__(suffix=suffix) super().__init__(suffix=suffix)
@@ -43,7 +43,6 @@ class Octoeverywhere(BaseInstance):
super().__post_init__() super().__post_init__()
self.cfg_file = self.cfg_dir.joinpath(OE_CFG_NAME) self.cfg_file = self.cfg_dir.joinpath(OE_CFG_NAME)
self.sys_cfg_file = self.cfg_dir.joinpath(OE_SYS_CFG_NAME) self.sys_cfg_file = self.cfg_dir.joinpath(OE_SYS_CFG_NAME)
self.log = self.log_dir.joinpath(OE_LOG_NAME)
def create(self) -> None: def create(self) -> None:
Logger.print_status("Creating OctoEverywhere for Klipper Instance ...") Logger.print_status("Creating OctoEverywhere for Klipper Instance ...")
@@ -64,20 +63,3 @@ class Octoeverywhere(BaseInstance):
except CalledProcessError as e: except CalledProcessError as e:
Logger.print_error(f"Error updating OctoEverywhere for Klipper: {e}") Logger.print_error(f"Error updating OctoEverywhere for Klipper: {e}")
raise raise
def delete(self) -> None:
service_file: str = self.get_service_file_name(extension=True)
service_file_path: Path = self.get_service_file_path()
Logger.print_status(
f"Deleting OctoEverywhere for Klipper Instance: {service_file}"
)
try:
command = ["sudo", "rm", "-f", service_file_path.as_posix()]
run(command, check=True)
self.delete_logfiles(OE_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

View File

@@ -23,6 +23,7 @@ from components.octoeverywhere import (
from components.octoeverywhere.octoeverywhere import Octoeverywhere from components.octoeverywhere.octoeverywhere import Octoeverywhere
from core.instance_manager.instance_manager import InstanceManager from core.instance_manager.instance_manager import InstanceManager
from core.logger import DialogType, Logger from core.logger import DialogType, Logger
from core.types import ComponentStatus
from utils.common import ( from utils.common import (
check_install_dependencies, check_install_dependencies,
get_install_status, get_install_status,
@@ -35,11 +36,9 @@ from utils.fs_utils import run_remove_routines
from utils.git_utils import git_clone_wrapper from utils.git_utils import git_clone_wrapper
from utils.input_utils import get_confirm from utils.input_utils import get_confirm
from utils.sys_utils import ( from utils.sys_utils import (
cmd_sysctl_manage,
install_python_requirements, install_python_requirements,
parse_packages_from_file, parse_packages_from_file,
) )
from core.types import ComponentStatus
def get_octoeverywhere_status() -> ComponentStatus: def get_octoeverywhere_status() -> ComponentStatus:
@@ -142,7 +141,7 @@ def remove_octoeverywhere() -> None:
ob_instances: List[Octoeverywhere] = ob_im.instances ob_instances: List[Octoeverywhere] = ob_im.instances
try: try:
remove_oe_instances(ob_im, ob_instances) remove_oe_instances(ob_instances)
remove_oe_dir() remove_oe_dir()
remove_oe_env() remove_oe_env()
remove_config_section(f"include {OE_SYS_CFG_NAME}", mr_instances) remove_config_section(f"include {OE_SYS_CFG_NAME}", mr_instances)
@@ -173,7 +172,6 @@ def install_oe_dependencies() -> None:
def remove_oe_instances( def remove_oe_instances(
instance_manager: InstanceManager,
instance_list: List[Octoeverywhere], instance_list: List[Octoeverywhere],
) -> None: ) -> None:
if not instance_list: if not instance_list:
@@ -181,13 +179,8 @@ def remove_oe_instances(
return return
for instance in instance_list: for instance in instance_list:
Logger.print_status(f"Removing instance {instance.get_service_file_name()} ...") Logger.print_status(f"Removing instance {instance.service_file_path.stem} ...")
instance_manager.current_instance = instance instance.remove()
instance_manager.stop_instance()
instance_manager.disable_instance()
instance_manager.delete_instance()
cmd_sysctl_manage("daemon-reload")
def remove_oe_dir() -> None: def remove_oe_dir() -> None:

View File

@@ -23,18 +23,21 @@ from core.logger import Logger
class BaseInstance(ABC): class BaseInstance(ABC):
suffix: str suffix: str
user: str = field(default=CURRENT_USER, init=False) user: str = field(default=CURRENT_USER, init=False)
service_file_path: Path | None = None
is_legacy_instance: bool = False
data_dir: Path | None = None data_dir: Path | None = None
data_dir_name: str = "" data_dir_name: str = ""
is_legacy_instance: bool = False
cfg_dir: Path | None = None cfg_dir: Path | None = None
log_dir: Path | None = None sysd_dir: Path | None = None # NOT to be confused with /etc/systemd/system
comms_dir: Path | None = None comms_dir: Path | None = None
sysd_dir: Path | None = None
gcodes_dir: Path | None = None gcodes_dir: Path | None = None
log_dir: Path | None = None
log_file_name: str = ""
def __post_init__(self) -> None: def __post_init__(self) -> None:
self._set_data_dir() self._set_data_dir()
self._set_is_legacy_instance() self._set_is_legacy_instance()
self._set_service_file_path()
if self.data_dir is not None: if self.data_dir is not None:
self.cfg_dir = self.data_dir.joinpath("config") self.cfg_dir = self.data_dir.joinpath("config")
self.log_dir = self.data_dir.joinpath("logs") self.log_dir = self.data_dir.joinpath("logs")
@@ -50,9 +53,28 @@ class BaseInstance(ABC):
def create(self) -> None: def create(self) -> None:
raise NotImplementedError("Subclasses must implement the create method") raise NotImplementedError("Subclasses must implement the create method")
@abstractmethod def remove(self) -> None:
def delete(self) -> None: from utils.fs_utils import run_remove_routines
raise NotImplementedError("Subclasses must implement the delete method") from utils.sys_utils import remove_system_service
try:
# remove the service file
if self.service_file_path is not None:
remove_system_service(self.service_file_path.name)
# then remove all the log files
if not self.log_file_name or not self.log_dir or not self.log_dir.exists():
return
files = self.log_dir.iterdir()
logs = [f for f in files if f.name.startswith(self.log_file_name)]
for log in logs:
Logger.print_status(f"Remove '{log}'")
run_remove_routines(log)
except Exception as e:
Logger.print_error(f"Error removing service: {e}")
raise
def create_folders(self, add_dirs: List[Path] | None = None) -> None: def create_folders(self, add_dirs: List[Path] | None = None) -> None:
dirs: List[Path | None] = [ dirs: List[Path | None] = [
@@ -72,22 +94,7 @@ class BaseInstance(ABC):
continue continue
_dir.mkdir(exist_ok=True) _dir.mkdir(exist_ok=True)
# todo: refactor into a set method and access the value by accessing the property def remove_logfiles(self, log_name: str) -> None:
def get_service_file_name(self, extension: bool = False) -> str:
from utils.common import convert_camelcase_to_kebabcase
name: str = convert_camelcase_to_kebabcase(self.__class__.__name__)
if self.suffix != "":
name += f"-{self.suffix}"
return name if not extension else f"{name}.service"
# todo: refactor into a set method and access the value by accessing the property
def get_service_file_path(self) -> Path:
path: Path = SYSTEMD.joinpath(self.get_service_file_name(extension=True))
return path
def delete_logfiles(self, log_name: str) -> None:
from utils.fs_utils import run_remove_routines from utils.fs_utils import run_remove_routines
if not self.log_dir or not self.log_dir.exists(): if not self.log_dir or not self.log_dir.exists():
@@ -105,14 +112,23 @@ class BaseInstance(ABC):
else: else:
self.data_dir = Path.home().joinpath(f"printer_{self.suffix}_data") self.data_dir = Path.home().joinpath(f"printer_{self.suffix}_data")
if self.get_service_file_path().exists(): if self.service_file_path is not None and self.service_file_path.exists():
with open(self.get_service_file_path(), "r") as service_file: with open(self.service_file_path, "r") as service_file:
service_content = service_file.read() service_content = service_file.read()
pattern = re.compile("^EnvironmentFile=(.+)(/systemd/.+\.env)") pattern = re.compile("^EnvironmentFile=(.+)(/systemd/.+\.env)")
match = re.search(pattern, service_content) match = re.search(pattern, service_content)
if match: if match:
self.data_dir = Path(match.group(1)) self.data_dir = Path(match.group(1))
def _set_service_file_path(self) -> None:
from utils.common import convert_camelcase_to_kebabcase
name: str = convert_camelcase_to_kebabcase(self.__class__.__name__)
if self.suffix != "":
name += f"-{self.suffix}"
self.service_file_path = SYSTEMD.joinpath(f"{name}.service")
def _set_is_legacy_instance(self) -> None: def _set_is_legacy_instance(self) -> None:
if ( if (
self.suffix != "" self.suffix != ""

View File

@@ -49,8 +49,8 @@ class InstanceManager:
self._current_instance = value self._current_instance = value
if value is not None: if value is not None:
self.instance_suffix = value.suffix self.instance_suffix = value.suffix
self.instance_service = value.get_service_file_name() self.instance_service = value.service_file_path.stem
self.instance_service_path = value.get_service_file_path() self.instance_service_path = value.service_file_path
@property @property
def instance_suffix(self) -> str | None: def instance_suffix(self) -> str | None:
@@ -98,16 +98,6 @@ class InstanceManager:
else: else:
raise ValueError("current_instance cannot be None") raise ValueError("current_instance cannot be None")
def delete_instance(self) -> None:
if self.current_instance is not None:
try:
self.current_instance.delete()
except (OSError, subprocess.CalledProcessError) as e:
Logger.print_error(f"Removing instance failed: {e}")
raise
else:
raise ValueError("current_instance cannot be None")
def enable_instance(self) -> None: def enable_instance(self) -> None:
try: try:
cmd_sysctl_service(self.instance_service_full, "enable") cmd_sysctl_service(self.instance_service_full, "enable")

View File

@@ -34,8 +34,8 @@ from extensions.obico import (
class MoonrakerObico(BaseInstance): class MoonrakerObico(BaseInstance):
dir: Path = OBICO_DIR dir: Path = OBICO_DIR
env_dir: Path = OBICO_ENV_DIR env_dir: Path = OBICO_ENV_DIR
log_file_name = OBICO_LOG_NAME
cfg_file: Path | None = None cfg_file: Path | None = None
log: Path | None = None
is_linked: bool = False is_linked: bool = False
def __init__(self, suffix: str = ""): def __init__(self, suffix: str = ""):
@@ -44,7 +44,6 @@ class MoonrakerObico(BaseInstance):
def __post_init__(self): def __post_init__(self):
super().__post_init__() super().__post_init__()
self.cfg_file = self.cfg_dir.joinpath(OBICO_CFG_NAME) 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() self.is_linked: bool = self._check_link_status()
def create(self) -> None: def create(self) -> None:
@@ -55,7 +54,7 @@ class MoonrakerObico(BaseInstance):
try: try:
self.create_folders() self.create_folders()
create_service_file( create_service_file(
name=self.get_service_file_name(extension=True), name=self.service_file_path.name,
content=self._prep_service_file_content(), content=self._prep_service_file_content(),
) )
create_env_file( create_env_file(
@@ -70,21 +69,6 @@ class MoonrakerObico(BaseInstance):
Logger.print_error(f"Error creating env file: {e}") Logger.print_error(f"Error creating env file: {e}")
raise raise
def delete(self) -> None:
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.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 link(self) -> None: def link(self) -> None:
Logger.print_status( Logger.print_status(
f"Linking instance for printer {self.data_dir_name} to the Obico server ..." f"Linking instance for printer {self.data_dir_name} to the Obico server ..."

View File

@@ -165,7 +165,7 @@ class ObicoExtension(BaseExtension):
ob_instances: List[MoonrakerObico] = ob_im.instances ob_instances: List[MoonrakerObico] = ob_im.instances
try: try:
self._remove_obico_instances(ob_im, ob_instances) self._remove_obico_instances(ob_instances)
self._remove_obico_dir() self._remove_obico_dir()
self._remove_obico_env() self._remove_obico_env()
remove_config_section(f"include {OBICO_MACROS_CFG_NAME}", kl_instances) remove_config_section(f"include {OBICO_MACROS_CFG_NAME}", kl_instances)
@@ -288,7 +288,11 @@ class ObicoExtension(BaseExtension):
scp.read(obico.cfg_file) scp.read(obico.cfg_file)
scp.set("server", "url", self.server_url) scp.set("server", "url", self.server_url)
scp.set("moonraker", "port", str(moonraker.port)) scp.set("moonraker", "port", str(moonraker.port))
scp.set("logging", "path", obico.log.as_posix()) scp.set(
"logging",
"path",
obico.log_dir.joinpath(obico.log_file_name).as_posix(),
)
scp.write(obico.cfg_file) scp.write(obico.cfg_file)
def _patch_printer_cfg(self, klipper: List[Klipper]) -> None: def _patch_printer_cfg(self, klipper: List[Klipper]) -> None:
@@ -336,7 +340,6 @@ class ObicoExtension(BaseExtension):
def _remove_obico_instances( def _remove_obico_instances(
self, self,
instance_manager: InstanceManager,
instance_list: List[MoonrakerObico], instance_list: List[MoonrakerObico],
) -> None: ) -> None:
if not instance_list: if not instance_list:
@@ -345,14 +348,9 @@ class ObicoExtension(BaseExtension):
for instance in instance_list: for instance in instance_list:
Logger.print_status( Logger.print_status(
f"Removing instance {instance.get_service_file_name()} ..." f"Removing instance {instance.service_file_path.stem} ..."
) )
instance_manager.current_instance = instance instance.remove()
instance_manager.stop_instance()
instance_manager.disable_instance()
instance_manager.delete_instance()
cmd_sysctl_manage("daemon-reload")
def _remove_obico_dir(self) -> None: def _remove_obico_dir(self) -> None:
Logger.print_status("Removing Obico for Klipper directory ...") Logger.print_status("Removing Obico for Klipper directory ...")

View File

@@ -10,7 +10,7 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from subprocess import CalledProcessError, run from subprocess import CalledProcessError
from core.instance_manager.base_instance import BaseInstance from core.instance_manager.base_instance import BaseInstance
from core.logger import Logger from core.logger import Logger
@@ -30,8 +30,8 @@ from extensions.telegram_bot import (
class MoonrakerTelegramBot(BaseInstance): class MoonrakerTelegramBot(BaseInstance):
bot_dir: Path = TG_BOT_DIR bot_dir: Path = TG_BOT_DIR
env_dir: Path = TG_BOT_ENV env_dir: Path = TG_BOT_ENV
log_file_name = TG_BOT_LOG_NAME
cfg_file: Path | None = None cfg_file: Path | None = None
log: Path | None = None
def __init__(self, suffix: str = ""): def __init__(self, suffix: str = ""):
super().__init__(suffix=suffix) super().__init__(suffix=suffix)
@@ -39,7 +39,6 @@ class MoonrakerTelegramBot(BaseInstance):
def __post_init__(self): def __post_init__(self):
super().__post_init__() super().__post_init__()
self.cfg_file = self.cfg_dir.joinpath(TG_BOT_CFG_NAME) self.cfg_file = self.cfg_dir.joinpath(TG_BOT_CFG_NAME)
self.log = self.log_dir.joinpath(TG_BOT_LOG_NAME)
def create(self) -> None: def create(self) -> None:
from utils.sys_utils import create_env_file, create_service_file from utils.sys_utils import create_env_file, create_service_file
@@ -49,7 +48,7 @@ class MoonrakerTelegramBot(BaseInstance):
try: try:
self.create_folders() self.create_folders()
create_service_file( create_service_file(
name=self.get_service_file_name(extension=True), name=self.service_file_path.name,
content=self._prep_service_file_content(), content=self._prep_service_file_content(),
) )
create_env_file( create_env_file(
@@ -64,20 +63,6 @@ class MoonrakerTelegramBot(BaseInstance):
Logger.print_error(f"Error creating env file: {e}") Logger.print_error(f"Error creating env file: {e}")
raise raise
def delete(self) -> None:
service_file: str = self.get_service_file_name(extension=True)
service_file_path: 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.as_posix()]
run(command, check=True)
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 _prep_service_file_content(self) -> str: def _prep_service_file_content(self) -> str:
template = TG_BOT_SERVICE_TEMPLATE template = TG_BOT_SERVICE_TEMPLATE
@@ -126,6 +111,6 @@ class MoonrakerTelegramBot(BaseInstance):
) )
env_file_content = env_file_content.replace( env_file_content = env_file_content.replace(
"%LOG%", "%LOG%",
self.log.as_posix() if self.log else "", self.log_dir.joinpath(self.log_file_name).as_posix(),
) )
return env_file_content return env_file_content

View File

@@ -144,7 +144,7 @@ class TelegramBotExtension(BaseExtension):
tb_instances: List[MoonrakerTelegramBot] = tb_im.instances tb_instances: List[MoonrakerTelegramBot] = tb_im.instances
try: try:
self._remove_bot_instances(tb_im, tb_instances) self._remove_bot_instances(tb_instances)
self._remove_bot_dir() self._remove_bot_dir()
self._remove_bot_env() self._remove_bot_env()
remove_config_section("update_manager moonraker-telegram-bot", mr_instances) remove_config_section("update_manager moonraker-telegram-bot", mr_instances)
@@ -181,19 +181,13 @@ class TelegramBotExtension(BaseExtension):
def _remove_bot_instances( def _remove_bot_instances(
self, self,
instance_manager: InstanceManager,
instance_list: List[MoonrakerTelegramBot], instance_list: List[MoonrakerTelegramBot],
) -> None: ) -> None:
for instance in instance_list: for instance in instance_list:
Logger.print_status( Logger.print_status(
f"Removing instance {instance.get_service_file_name()} ..." f"Removing instance {instance.service_file_path.stem} ..."
) )
instance_manager.current_instance = instance instance.remove()
instance_manager.stop_instance()
instance_manager.disable_instance()
instance_manager.delete_instance()
cmd_sysctl_manage("daemon-reload")
def _remove_bot_dir(self) -> None: def _remove_bot_dir(self) -> None:
if not TG_BOT_DIR.exists(): if not TG_BOT_DIR.exists():