h3n3/kiauh
1
0
Fork 0
Code Issues 46 Pull Requests 10 Packages Projects Releases Wiki Activity

feat(obico): add obico extension

Browse Source 
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2024-05-25 21:34:23 +02:00
parent d414be609a
commit 02bb1b34fd
6 changed files with 570 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
kiauh/extensions/obico/__init__.py Normal file
View File

1
kiauh/extensions/obico/assets/moonraker-obico.env Normal file
View File

@@ -0,0 +1 @@
OBICO_ARGS="-m moonraker_obico.app -c %CFG%"

16
kiauh/extensions/obico/assets/moonraker-obico.service Normal file
View File

@@ -0,0 +1,16 @@
#Systemd service file for moonraker-obico
[Unit]
Description=Moonraker-Obico
After=network-online.target moonraker.service
[Install]
WantedBy=multi-user.target
[Service]
Type=simple
User=%USER%
WorkingDirectory=%OBICO_DIR%
EnvironmentFile=%ENV_FILE%
ExecStart=%ENV%/bin/python3 $OBICO_ARGS
Restart=always
RestartSec=5

16
kiauh/extensions/obico/metadata.json Normal file
View File

@@ -0,0 +1,16 @@
{
"metadata": {
"index": 6,
"module": "moonraker_obico_extension",
"maintained_by": "Obico",
"display_name": "Obico for Klipper",
"description": [
"Open source 3D Printing cloud and AI",
"- AI-Powered Failure Detection",
"- Free Remote Monitoring and Access",
"- 25FPS High-Def Webcam Streaming",
"- Free 4.9-Star Mobile App"
],
"updates": true
}
}

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

@@ -0,0 +1,167 @@
# ======================================================================= #
# 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
from subprocess import DEVNULL, CalledProcessError, run
from typing import List
from core.instance_manager.base_instance import BaseInstance
from utils.constants import SYSTEMD
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"
# noinspection PyMethodMayBeStatic
class MoonrakerObico(BaseInstance):
@classmethod
def blacklist(cls) -> List[str]:
return ["None", "mcu"]
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")
@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 create(self) -> None:
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
)
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}"
)
raise
except OSError as e:
Logger.print_error(f"Error creating env file {env_file_target}: {e}")
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"Deleting Obico for Klipper Instance: {service_file}")
try:
command = ["sudo", "rm", "-f", service_file_path]
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 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]
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}"]
if self.suffix:
cmd.append(f"-n {self.suffix}")
run(cmd, check=True, shell=True)
except CalledProcessError as e:
Logger.print_error(f"Error during Obico linking: {e}")
raise
def _prep_service_file(
self, service_template_path: Path, env_file_path: Path
) -> str:
try:
with open(service_template_path, "r") as template_file:
template_content = template_file.read()
except FileNotFoundError:
Logger.print_error(
f"Unable to open {service_template_path} - 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))
return service_content
def _prep_env_file(self, env_template_file_path: Path) -> str:
try:
with open(env_template_file_path, "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"
)
raise
env_file_content = env_template_file_content.replace(
"%CFG%",
f"{self.cfg_dir}/{self.cfg_file}",
)
return env_file_content
def _check_link_status(self) -> bool:
from core.config_manager.config_manager import ConfigManager
cm = ConfigManager(self.cfg_file)
return cm.get_value("server", "auth_token") is not None

370
kiauh/extensions/obico/moonraker_obico_extension.py Normal file
View File

@@ -0,0 +1,370 @@
# ======================================================================= #
# 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 #
# ======================================================================= #
import shutil
from subprocess import run
from typing import List
from components.klipper.klipper import Klipper
from components.moonraker.moonraker import Moonraker
from core.config_manager.config_manager import ConfigManager
from core.instance_manager.instance_manager import InstanceManager
from extensions.base_extension import BaseExtension
from extensions.obico.moonraker_obico import (
OBICO_DIR,
OBICO_ENV,
OBICO_REPO,
MoonrakerObico,
)
from utils.common import check_install_dependencies, moonraker_exists
from utils.config_utils import add_config_section, remove_config_section
from utils.fs_utils import remove_file
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
from utils.sys_utils import (
cmd_sysctl_manage,
create_python_venv,
install_python_requirements,
parse_packages_from_file,
)
# noinspection PyMethodMayBeStatic
class ObicoExtension(BaseExtension):
server_url: str
def install_extension(self, **kwargs) -> None:
Logger.print_status("Installing Obico for Klipper ...")
# check if moonraker is installed. if not, notify the user and exit
if not moonraker_exists():
return
# if obico is already installed, ask if the user wants to repair an
# incomplete installation or link to the obico server
obico_im = InstanceManager(MoonrakerObico)
obico_instances: List[MoonrakerObico] = obico_im.instances
if obico_instances:
self._print_is_already_installed()
options = ["l", "L", "r", "R", "b", "B"]
action = get_selection_input("Perform action", option_list=options)
if action.lower() == "b":
Logger.print_info("Exiting Obico for Klipper installation ...")
return
elif action.lower() == "l":
unlinked_instances: List[MoonrakerObico] = [
obico for obico in obico_instances if not obico.is_linked
]
self._link_obico_instances(unlinked_instances)
return
else:
Logger.print_status("Re-Installing Obico for Klipper ...")
# let the user confirm installation
kl_im = InstanceManager(Klipper)
kl_instances: List[Klipper] = kl_im.instances
mr_im = InstanceManager(Moonraker)
mr_instances: List[Moonraker] = mr_im.instances
self._print_moonraker_instances(mr_instances)
if not get_confirm(
"Continue Obico for Klipper installation?",
default_choice=True,
allow_go_back=True,
):
return
try:
git_clone_wrapper(OBICO_REPO, OBICO_DIR)
self._install_dependencies()
# ask the user for the obico server url
self._get_server_url()
# create obico instances
for moonraker in mr_instances:
current_instance = MoonrakerObico(suffix=moonraker.suffix)
obico_im.current_instance = current_instance
obico_im.create_instance()
obico_im.enable_instance()
# create obico config
self._create_obico_cfg(current_instance, moonraker)
# create obico macros
self._create_obico_macros_cfg(moonraker)
obico_im.start_instance()
cmd_sysctl_manage("daemon-reload")
# add to klippers config
self._patch_obico_macros(kl_instances)
kl_im.restart_all_instance()
# add to moonraker update manager
self._patch_update_manager(mr_instances)
mr_im.restart_all_instance()
# check linking of / ask for linking instances
self._check_and_opt_link_instances()
Logger.print_dialog(
DialogType.SUCCESS,
["Obico for Klipper successfully installed!"],
center_content=True,
)
except Exception as e:
Logger.print_error(f"Error during Obico for Klipper installation: {e}")
def update_extension(self, **kwargs) -> None:
Logger.print_status("Updating Obico for Klipper ...")
try:
tb_im = InstanceManager(MoonrakerObico)
tb_im.stop_all_instance()
git_pull_wrapper(OBICO_REPO, OBICO_DIR)
self._install_dependencies()
tb_im.start_all_instance()
Logger.print_ok("Obico for Klipper successfully updated!")
except Exception as e:
Logger.print_error(f"Error during Obico for Klipper update: {e}")
def remove_extension(self, **kwargs) -> None:
Logger.print_status("Removing Obico for Klipper ...")
kl_im = InstanceManager(Klipper)
kl_instances: List[Klipper] = kl_im.instances
mr_im = InstanceManager(Moonraker)
mr_instances: List[Moonraker] = mr_im.instances
ob_im = InstanceManager(MoonrakerObico)
ob_instances: List[MoonrakerObico] = ob_im.instances
try:
self._remove_obico_instances(ob_im, ob_instances)
self._remove_obico_dir()
self._remove_obico_env()
remove_config_section("include moonraker_obico_macros.cfg", kl_instances)
remove_config_section("update_manager moonraker-obico", mr_instances)
remove_config_section("include moonraker-obico-update.cfg", mr_instances)
self._delete_obico_logs(ob_instances)
Logger.print_dialog(
DialogType.SUCCESS,
["Obico for Klipper successfully removed!"],
center_content=True,
)
except Exception as e:
Logger.print_error(f"Error during Obico for Klipper removal: {e}")
def _obico_server_url_prompt(self) -> None:
Logger.print_dialog(
DialogType.CUSTOM,
custom_title="Obico Server URL",
content=[
"You can use a self-hosted Obico Server or the Obico Cloud. "
"For more information, please visit:",
"https://obico.io.",
"\n\n",
"For the Obico Cloud, leave it as the default:",
"https://app.obico.io.",
"\n\n",
"For self-hosted server, specify:",
"http://server_ip:port",
"For instance, 'http://192.168.0.5:3334'.",
],
end="",
)
def _print_moonraker_instances(self, mr_instances) -> None:
mr_names = [f"● {moonraker.data_dir_name}" for moonraker in mr_instances]
if len(mr_names) > 1:
Logger.print_dialog(
DialogType.INFO,
[
"The following Moonraker instances were found:",
*mr_names,
"\n\n",
"The setup will apply the same names to Obico!",
],
end="",
)
def _print_is_already_installed(self) -> None:
Logger.print_dialog(
DialogType.INFO,
[
"Obico is already installed!",
"It is save to run the installer again to link your "
"printer or repair any issues.",
"\n\n",
"You can perform the following actions:",
"L) Link printer to the Obico server",
"R) Repair installation",
],
end="",
)
def _get_server_url(self) -> None:
self._obico_server_url_prompt()
pattern = r"^(http|https)://[a-zA-Z0-9./?=_%:-]*$"
self.server_url = get_string_input(
"Obico Server URL",
regex=pattern,
default="https://app.obico.io",
)
def _install_dependencies(self) -> None:
# install dependencies
script = OBICO_DIR.joinpath("install.sh")
package_list = parse_packages_from_file(script)
check_install_dependencies(package_list)
# create virtualenv
create_python_venv(OBICO_ENV)
requirements = OBICO_DIR.joinpath("requirements.txt")
install_python_requirements(OBICO_ENV, requirements)
def _create_obico_macros_cfg(self, moonraker) -> None:
macros_cfg = OBICO_DIR.joinpath("include_cfgs/moonraker_obico_macros.cfg")
macros_target = moonraker.cfg_dir.joinpath("moonraker_obico_macros.cfg")
if not macros_target.exists():
run(["cp", macros_cfg, macros_target], check=True)
else:
Logger.print_info(
f"Obico macros in {moonraker.cfg_dir} already exists! Skipped ..."
)
def _create_obico_cfg(self, current_instance, moonraker) -> None:
cfg_template = OBICO_DIR.joinpath("moonraker-obico.cfg.sample")
cfg_target_file = current_instance.cfg_file
if not cfg_target_file.exists():
run(["cp", cfg_template, cfg_target_file], check=True)
self._patch_obico_cfg(moonraker, current_instance)
else:
Logger.print_info(
f"Obico config in {current_instance.cfg_dir} already exists! Skipped ..."
)
def _patch_obico_cfg(self, moonraker: Moonraker, obico: MoonrakerObico) -> None:
cm = ConfigManager(obico.cfg_file)
cm.set_value("server", "url", self.server_url)
cm.set_value("moonraker", "port", str(moonraker.port))
cm.set_value("logging", "path", str(obico.log))
cm.write_config()
def _patch_obico_macros(self, klipper: List[Klipper]) -> None:
add_config_section(
section="include moonraker_obico_macros.cfg",
instances=klipper,
)
def _patch_update_manager(self, instances: List[Moonraker]) -> None:
env_py = f"{OBICO_ENV}/bin/python"
add_config_section(
section="update_manager moonraker-obico",
instances=instances,
options=[
("type", "git_repo"),
("path", str(OBICO_DIR)),
("orgin", OBICO_REPO),
("env", env_py),
("requirements", "requirements.txt"),
("install_script", "install.sh"),
("managed_services", "moonraker-obico"),
],
)
def _link_obico_instances(self, unlinked_instances):
for obico in unlinked_instances:
obico.link()
def _check_and_opt_link_instances(self):
Logger.print_status("Checking link status of Obico instances ...")
ob_im = InstanceManager(MoonrakerObico)
ob_instances: List[MoonrakerObico] = ob_im.instances
unlinked_instances: List[MoonrakerObico] = [
obico for obico in ob_instances if not obico.is_linked
]
if unlinked_instances:
Logger.print_dialog(
DialogType.INFO,
[
"The Obico instances for the following printers are not "
"linked to the server:",
*[f"● {obico.data_dir_name}" for obico in unlinked_instances],
"\n\n",
"It will take only 10 seconds to link the printer to the Obico server.",
"For more information visit:",
"https://www.obico.io/docs/user-guides/klipper-setup/",
"\n\n",
"If you don't want to link the printer now, you can restart the "
"linking process later by running this installer again.",
],
end="",
)
if not get_confirm("Do you want to link the printers now?"):
self._link_obico_instances(unlinked_instances)
else:
Logger.print_info("Linking to server skipped ...")
def _remove_obico_instances(
self,
instance_manager: InstanceManager,
instance_list: List[MoonrakerObico],
) -> None:
if not instance_list:
Logger.print_info("No Obico instances found. Skipped ...")
return
for instance in instance_list:
Logger.print_status(
f"Removing instance {instance.get_service_file_name()} ..."
)
instance_manager.current_instance = instance
instance_manager.stop_instance()
instance_manager.disable_instance()
instance_manager.delete_instance()
cmd_sysctl_manage("daemon-reload")
def _remove_obico_dir(self) -> None:
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}")
def _remove_obico_env(self) -> None:
if not OBICO_ENV.exists():
Logger.print_info(f"'{OBICO_ENV}' 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("moonraker-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)