Compare commits

..

12 Commits

Author SHA1 Message Date
dw-0
1b5691f2f5 Release v6.0.0-alpha.7
Merge develop into master (v6.0.0-alpha.7)

fixes #561
fixes #564
fixes #565
2024-10-13 11:51:19 +02:00
dw-0
e7eae5a0d1 fix: correctly handle IPs in nginx config files when parsing ports (#568)
* chore: add jupyter files to .gitignore

Signed-off-by: Dominik Willner <th33xitus@gmail.com>

* fix: correctly handle IPs in nginx config files when parsing ports

Signed-off-by: Dominik Willner <th33xitus@gmail.com>

---------

Signed-off-by: Dominik Willner <th33xitus@gmail.com>
2024-10-13 11:21:18 +02:00
dw-0
dc561a562c fix: always return string tuple from get_repo_name() (#567)
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
2024-10-13 10:27:31 +02:00
dw-0
55cfe124b2 feat: add SimplyPrint extension (#566)
* refactor: correctly sort extensions in extension menu

Signed-off-by: Dominik Willner <th33xitus@gmail.com>

* refactor: use different name for printer_data backup dir

Signed-off-by: Dominik Willner <th33xitus@gmail.com>

* refactor: change return type to List for moonraker_exists function

Signed-off-by: Dominik Willner <th33xitus@gmail.com>

* feat: add SimplyPrint extension

Signed-off-by: Dominik Willner <th33xitus@gmail.com>

---------

Signed-off-by: Dominik Willner <th33xitus@gmail.com>
2024-10-13 09:35:15 +02:00
Christian Würthner
43d6598be6 fix: remove octoapp_store dir when uninstalling (#562)
Co-authored-by: dw-0 <th33xitus@gmail.com>
2024-10-05 12:35:14 +02:00
dw-0
dc026a7a2b Release v6.0.0-alpha.6
Merge develop into master (v6.0.0-alpha.6)

fixes #545
fixes #553
fixes #557
2024-10-05 08:29:40 +02:00
dw-0
ac54d04b40 fix: correctly find connected UART devices (#559)
fixes #557

Signed-off-by: Dominik Willner <th33xitus@gmail.com>
2024-10-05 08:21:39 +02:00
dw-0
c19364360c fix: correctly find connected USB DFU devices (#555)
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
2024-10-05 08:07:47 +02:00
dw-0
2e6c66e524 fix: allow moonraker-telegram-bot-env access to systems site-packages dir (#556)
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
2024-10-03 10:22:32 +02:00
Christian Würthner
cd8003add9 feat(extension): add OctoApp (#554)
* Add OctoApp to v6

* fix: set correct index to new extension

Signed-off-by: Dominik Willner <th33xitus@gmail.com>

---------

Signed-off-by: Dominik Willner <th33xitus@gmail.com>
Co-authored-by: dw-0 <th33xitus@gmail.com>
2024-10-03 08:51:38 +02:00
Kenneth Jiang
1f75395063 fix: self.cfg_file is already a full path (#552)
Signed-off-by: Kenneth Jiang <kenneth.jiang@gmail.com>
2024-09-29 20:33:54 +02:00
Kenneth Jiang
6e1bffa975 fix: remove "obico" from the suffix_blacklist so that it can discover its own instances. (#551)
Signed-off-by: Kenneth Jiang <kenneth.jiang@gmail.com>
2024-09-29 16:41:20 +02:00
21 changed files with 572 additions and 48 deletions

4
.gitignore vendored
View File

@@ -1,6 +1,10 @@
.idea .idea
.vscode .vscode
.pytest_cache .pytest_cache
.jupyter
*.ipynb
*.ipynb_checkpoints
*.tmp
__pycache__ __pycache__
.kiauh-env .kiauh-env
*.code-workspace *.code-workspace

View File

@@ -6,8 +6,16 @@
# # # #
# This file may be distributed under the terms of the GNU GPLv3 license # # This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= # # ======================================================================= #
import re
from subprocess import PIPE, STDOUT, CalledProcessError, Popen, check_output, run from subprocess import (
DEVNULL,
PIPE,
STDOUT,
CalledProcessError,
Popen,
check_output,
run,
)
from typing import List from typing import List
from components.klipper import KLIPPER_DIR from components.klipper import KLIPPER_DIR
@@ -32,16 +40,18 @@ def find_firmware_file() -> bool:
f3 = "klipper.bin" f3 = "klipper.bin"
f4 = "klipper.uf2" f4 = "klipper.uf2"
fw_file_exists: bool = ( fw_file_exists: bool = (
target.joinpath(f1).exists() and target.joinpath(f2).exists() (target.joinpath(f1).exists() and target.joinpath(f2).exists())
) or target.joinpath(f3).exists() or target.joinpath(f4).exists() or target.joinpath(f3).exists()
or target.joinpath(f4).exists()
)
return target_exists and fw_file_exists return target_exists and fw_file_exists
def find_usb_device_by_id() -> List[str]: def find_usb_device_by_id() -> List[str]:
try: try:
command = "find /dev/serial/by-id/* 2>/dev/null" command = "find /dev/serial/by-id/*"
output = check_output(command, shell=True, text=True) output = check_output(command, shell=True, text=True, stderr=DEVNULL)
return output.splitlines() return output.splitlines()
except CalledProcessError as e: except CalledProcessError as e:
Logger.print_error("Unable to find a USB device!") Logger.print_error("Unable to find a USB device!")
@@ -51,9 +61,14 @@ def find_usb_device_by_id() -> List[str]:
def find_uart_device() -> List[str]: def find_uart_device() -> List[str]:
try: try:
command = '"find /dev -maxdepth 1 -regextype posix-extended -regex "^\/dev\/tty(AMA0|S0)$" 2>/dev/null"' cmd = "find /dev -maxdepth 1"
output = check_output(command, shell=True, text=True) output = check_output(cmd, shell=True, text=True, stderr=DEVNULL)
return output.splitlines() device_list = []
if output:
pattern = r"^/dev/tty(AMA0|S0)$"
devices = output.splitlines()
device_list = [d for d in devices if re.search(pattern, d)]
return device_list
except CalledProcessError as e: except CalledProcessError as e:
Logger.print_error("Unable to find a UART device!") Logger.print_error("Unable to find a UART device!")
Logger.print_error(e, prefix=False) Logger.print_error(e, prefix=False)
@@ -62,9 +77,13 @@ def find_uart_device() -> List[str]:
def find_usb_dfu_device() -> List[str]: def find_usb_dfu_device() -> List[str]:
try: try:
command = '"lsusb | grep "DFU" | cut -d " " -f 6 2>/dev/null"' output = check_output("lsusb", shell=True, text=True, stderr=DEVNULL)
output = check_output(command, shell=True, text=True) device_list = []
return output.splitlines() if output:
devices = output.splitlines()
device_list = [d.split(" ")[5] for d in devices if "DFU" in d]
return device_list
except CalledProcessError as e: except CalledProcessError as e:
Logger.print_error("Unable to find a USB DFU device!") Logger.print_error("Unable to find a USB DFU device!")
Logger.print_error(e, prefix=False) Logger.print_error(e, prefix=False)

View File

@@ -353,10 +353,16 @@ def read_ports_from_nginx_configs() -> List[int]:
lines = cfg.readlines() lines = cfg.readlines()
for line in lines: for line in lines:
line = line.replace("default_server", "") line = re.sub(
line = re.sub(r"[;:\[\]]", "", line.strip()) r"default_server|http://|https://|[;\[\]]",
if line.startswith("listen") and line.split()[-1] not in port_list: "",
port_list.append(line.split()[-1]) line.strip(),
)
if line.startswith("listen"):
if ":" not in line:
port_list.append(line.split()[-1])
else:
port_list.append(line.split(":")[-1])
ports_to_ints_list = [int(port) for port in port_list] ports_to_ints_list = [int(port) for port in port_list]
return sorted(ports_to_ints_list, key=lambda x: int(x)) return sorted(ports_to_ints_list, key=lambda x: int(x))

View File

@@ -33,7 +33,7 @@ CURRENT_USER = pwd.getpwuid(os.getuid())[0]
# dirs # dirs
SYSTEMD = Path("/etc/systemd/system") SYSTEMD = Path("/etc/systemd/system")
PRINTER_CFG_BACKUP_DIR = BACKUP_ROOT_DIR.joinpath("printer-cfg-backups") PRINTER_DATA_BACKUP_DIR = BACKUP_ROOT_DIR.joinpath("printer-data-backups")
NGINX_SITES_AVAILABLE = Path("/etc/nginx/sites-available") NGINX_SITES_AVAILABLE = Path("/etc/nginx/sites-available")
NGINX_SITES_ENABLED = Path("/etc/nginx/sites-enabled") NGINX_SITES_ENABLED = Path("/etc/nginx/sites-enabled")
NGINX_CONFD = Path("/etc/nginx/conf.d") NGINX_CONFD = Path("/etc/nginx/conf.d")

View File

@@ -58,12 +58,16 @@ class ExtensionsMenu(BaseMenu):
module_path = f"kiauh.extensions.{ext.name}.{module_name}" module_path = f"kiauh.extensions.{ext.name}.{module_name}"
# get the class name of the extension # get the class name of the extension
ext_class: Type[BaseExtension] = inspect.getmembers( module = importlib.import_module(module_path)
importlib.import_module(module_path),
predicate=lambda o: inspect.isclass(o) def predicate(o):
and issubclass(o, BaseExtension) return (
and o != BaseExtension, inspect.isclass(o)
)[0][1] and issubclass(o, BaseExtension)
and o != BaseExtension
)
ext_class: type = inspect.getmembers(module, predicate)[0][1]
# instantiate the extension with its metadata and add to dict # instantiate the extension with its metadata and add to dict
ext_instance: BaseExtension = ext_class(metadata) ext_instance: BaseExtension = ext_class(metadata)
@@ -72,7 +76,7 @@ class ExtensionsMenu(BaseMenu):
except (IOError, json.JSONDecodeError, ImportError) as e: except (IOError, json.JSONDecodeError, ImportError) as e:
print(f"Failed loading extension {ext}: {e}") print(f"Failed loading extension {ext}: {e}")
return dict(sorted(ext_dict.items())) return dict(sorted(ext_dict.items(), key=lambda x: int(x[0])))
def extension_submenu(self, **kwargs): def extension_submenu(self, **kwargs):
ExtensionSubmenu(kwargs.get("opt_data"), self.__class__).run() ExtensionSubmenu(kwargs.get("opt_data"), self.__class__).run()

View File

@@ -132,7 +132,7 @@ class MoonrakerObico:
raise raise
env_file_content = env_template_file_content.replace( env_file_content = env_template_file_content.replace(
"%CFG%", "%CFG%",
f"{self.base.cfg_dir}/{self.cfg_file}", f"{self.cfg_file}",
) )
return env_file_content return env_file_content

View File

@@ -12,6 +12,7 @@ from typing import List
from components.klipper.klipper import Klipper from components.klipper.klipper import Klipper
from components.moonraker.moonraker import Moonraker from components.moonraker.moonraker import Moonraker
from core.instance_manager.instance_manager import InstanceManager from core.instance_manager.instance_manager import InstanceManager
from core.instance_manager.base_instance import SUFFIX_BLACKLIST
from core.logger import DialogType, Logger from core.logger import DialogType, Logger
from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import ( from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
SimpleConfigParser, SimpleConfigParser,
@@ -308,7 +309,8 @@ class ObicoExtension(BaseExtension):
def _check_and_opt_link_instances(self) -> None: def _check_and_opt_link_instances(self) -> None:
Logger.print_status("Checking link status of Obico instances ...") Logger.print_status("Checking link status of Obico instances ...")
ob_instances: List[MoonrakerObico] = get_instances(MoonrakerObico) suffix_blacklist: List[str] = [suffix for suffix in SUFFIX_BLACKLIST if suffix != 'obico']
ob_instances: List[MoonrakerObico] = get_instances(MoonrakerObico, suffix_blacklist=suffix_blacklist)
unlinked_instances: List[MoonrakerObico] = [ unlinked_instances: List[MoonrakerObico] = [
obico for obico in ob_instances if not obico.is_linked obico for obico in ob_instances if not obico.is_linked
] ]

View File

@@ -0,0 +1,28 @@
# ======================================================================= #
# 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
# repo
OA_REPO = "https://github.com/crysxd/OctoApp-Plugin.git"
# directories
OA_DIR = Path.home().joinpath("octoapp")
OA_ENV_DIR = Path.home().joinpath("octoapp-env")
# files
OA_REQ_FILE = OA_DIR.joinpath("requirements.txt")
OA_DEPS_JSON_FILE = OA_DIR.joinpath("moonraker-system-dependencies.json")
OA_INSTALL_SCRIPT = OA_DIR.joinpath("install.sh")
OA_UPDATE_SCRIPT = OA_DIR.joinpath("update.sh")
OA_INSTALLER_LOG_FILE = Path.home().joinpath("octoapp-installer.log")
# filenames
OA_CFG_NAME = "octoapp.conf"
OA_LOG_NAME = "octoapp.log"
OA_SYS_CFG_NAME = "octoapp-system.cfg"

View File

@@ -0,0 +1,17 @@
{
"metadata": {
"index": 9,
"module": "octoapp_extension",
"maintained_by": "crysxd",
"display_name": "OctoApp for Klipper",
"description": [
"Your favorite 3D printing app for iOS & Android",
"- Print notifications on your phone & watch",
"- Control and start prints from your phone",
"- Live webcam view",
"- Live Gcode preview",
"- And much much more!"
],
"updates": true
}
}

View File

@@ -0,0 +1,75 @@
# ======================================================================= #
# 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 __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from subprocess import CalledProcessError, run
from components.moonraker import MOONRAKER_CFG_NAME
from components.moonraker.moonraker import Moonraker
from core.instance_manager.base_instance import BaseInstance
from core.logger import Logger
from extensions.octoapp import (
OA_CFG_NAME,
OA_DIR,
OA_ENV_DIR,
OA_INSTALL_SCRIPT,
OA_LOG_NAME,
OA_SYS_CFG_NAME,
OA_UPDATE_SCRIPT,
)
from utils.sys_utils import get_service_file_path
@dataclass
class Octoapp:
suffix: str
base: BaseInstance = field(init=False, repr=False)
service_file_path: Path = field(init=False)
log_file_name = OA_LOG_NAME
dir: Path = OA_DIR
env_dir: Path = OA_ENV_DIR
data_dir: Path = field(init=False)
store_dir: Path = field(init=False)
cfg_file: Path = field(init=False)
sys_cfg_file: Path = field(init=False)
def __post_init__(self):
self.base: BaseInstance = BaseInstance(Moonraker, self.suffix)
self.base.log_file_name = self.log_file_name
self.service_file_path: Path = get_service_file_path(
Octoapp, self.suffix
)
self.store_dir = self.base.data_dir.joinpath("store")
self.cfg_file = self.base.cfg_dir.joinpath(OA_CFG_NAME)
self.sys_cfg_file = self.base.cfg_dir.joinpath(OA_SYS_CFG_NAME)
self.data_dir = self.base.data_dir
self.sys_cfg_file = self.base.cfg_dir.joinpath(OA_SYS_CFG_NAME)
def create(self) -> None:
Logger.print_status("Creating OctoApp for Klipper Instance ...")
try:
cmd = f"{OA_INSTALL_SCRIPT} {self.base.cfg_dir}/{MOONRAKER_CFG_NAME}"
run(cmd, check=True, shell=True)
except CalledProcessError as e:
Logger.print_error(f"Error creating instance: {e}")
raise
@staticmethod
def update() -> None:
try:
run(OA_UPDATE_SCRIPT.as_posix(), check=True, shell=True, cwd=OA_DIR)
except CalledProcessError as e:
Logger.print_error(f"Error updating OctoApp for Klipper: {e}")
raise

View File

@@ -0,0 +1,208 @@
# ======================================================================= #
# 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 json
from typing import List
from components.moonraker.moonraker import Moonraker
from components.klipper.klipper import Klipper
from core.instance_manager.instance_manager import InstanceManager
from core.logger import DialogType, Logger
from extensions.base_extension import BaseExtension
from extensions.octoapp import (
OA_DEPS_JSON_FILE,
OA_DIR,
OA_ENV_DIR,
OA_INSTALL_SCRIPT,
OA_INSTALLER_LOG_FILE,
OA_REPO,
OA_REQ_FILE,
OA_SYS_CFG_NAME,
)
from extensions.octoapp.octoapp import Octoapp
from utils.common import (
check_install_dependencies,
moonraker_exists,
)
from utils.config_utils import (
remove_config_section,
)
from utils.fs_utils import run_remove_routines
from utils.git_utils import git_clone_wrapper
from utils.input_utils import get_confirm
from utils.instance_utils import get_instances
from utils.sys_utils import (
install_python_requirements,
parse_packages_from_file,
)
# noinspection PyMethodMayBeStatic
class OctoappExtension(BaseExtension):
def install_extension(self, **kwargs) -> None:
Logger.print_status("Installing OctoApp for Klipper ...")
# check if moonraker is installed. if not, notify the user and exit
if not moonraker_exists():
return
force_clone = False
OA_instances: List[Octoapp] = get_instances(Octoapp)
if OA_instances:
Logger.print_dialog(
DialogType.INFO,
[
"OctoApp is already installed!",
"It is safe to run the installer again to link your "
"printer or repair any issues.",
],
)
if not get_confirm("Re-run OctoApp installation?"):
Logger.print_info("Exiting OctoApp for Klipper installation ...")
return
else:
Logger.print_status("Re-Installing OctoApp for Klipper ...")
force_clone = True
mr_instances: List[Moonraker] = get_instances(Moonraker)
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 OctoApp!",
],
)
if not get_confirm(
"Continue OctoApp for Klipper installation?",
default_choice=True,
allow_go_back=True,
):
Logger.print_info("Exiting OctoApp for Klipper installation ...")
return
try:
git_clone_wrapper(OA_REPO, OA_DIR, force=force_clone)
for moonraker in mr_instances:
instance = Octoapp(suffix=moonraker.suffix)
instance.create()
InstanceManager.restart_all(mr_instances)
Logger.print_dialog(
DialogType.SUCCESS,
["OctoApp for Klipper successfully installed!"],
center_content=True,
)
except Exception as e:
Logger.print_error(
f"Error during OctoApp for Klipper installation:\n{e}"
)
def update_extension(self, **kwargs) -> None:
Logger.print_status("Updating OctoApp for Klipper ...")
try:
Octoapp.update()
Logger.print_dialog(
DialogType.SUCCESS,
["OctoApp for Klipper successfully updated!"],
center_content=True,
)
except Exception as e:
Logger.print_error(f"Error during OctoApp for Klipper update:\n{e}")
def remove_extension(self, **kwargs) -> None:
Logger.print_status("Removing OctoApp for Klipper ...")
mr_instances: List[Moonraker] = get_instances(Moonraker)
ob_instances: List[Octoapp] = get_instances(Octoapp)
try:
self._remove_OA_instances(ob_instances)
self._remove_OA_store_dirs()
self._remove_OA_dir()
self._remove_OA_env()
remove_config_section(f"include {OA_SYS_CFG_NAME}", mr_instances)
run_remove_routines(OA_INSTALLER_LOG_FILE)
Logger.print_dialog(
DialogType.SUCCESS,
["OctoApp for Klipper successfully removed!"],
center_content=True,
)
except Exception as e:
Logger.print_error(f"Error during OctoApp for Klipper removal:\n{e}")
def _install_OA_dependencies(self) -> None:
OA_deps = []
if OA_DEPS_JSON_FILE.exists():
with open(OA_DEPS_JSON_FILE, "r") as deps:
OA_deps = json.load(deps).get("debian", [])
elif OA_INSTALL_SCRIPT.exists():
OA_deps = parse_packages_from_file(OA_INSTALL_SCRIPT)
if not OA_deps:
raise ValueError("Error reading OctoApp dependencies!")
check_install_dependencies({*OA_deps})
install_python_requirements(OA_ENV_DIR, OA_REQ_FILE)
def _remove_OA_instances(
self,
instance_list: List[Octoapp],
) -> None:
if not instance_list:
Logger.print_info("No OctoApp instances found. Skipped ...")
return
for instance in instance_list:
Logger.print_status(
f"Removing instance {instance.service_file_path.stem} ..."
)
InstanceManager.remove(instance)
def _remove_OA_dir(self) -> None:
Logger.print_status("Removing OctoApp for Klipper directory ...")
if not OA_DIR.exists():
Logger.print_info(f"'{OA_DIR}' does not exist. Skipped ...")
return
run_remove_routines(OA_DIR)
def _remove_OA_store_dirs(self) -> None:
Logger.print_status("Removing OctoApp for Klipper store directory ...")
klipper_instances: List[Moonraker] = get_instances(Klipper)
for instance in klipper_instances:
store_dir = instance.data_dir.joinpath("octoapp-store")
if not store_dir.exists():
Logger.print_info(f"'{store_dir}' does not exist. Skipped ...")
return
run_remove_routines(store_dir)
def _remove_OA_env(self) -> None:
Logger.print_status("Removing OctoApp for Klipper environment ...")
if not OA_ENV_DIR.exists():
Logger.print_info(f"'{OA_ENV_DIR}' does not exist. Skipped ...")
return
run_remove_routines(OA_ENV_DIR)

View File

@@ -0,0 +1,13 @@
{
"metadata": {
"index": 10,
"module": "simply_print_extension",
"maintained_by": "dw-0",
"display_name": "SimplyPrint",
"description": [
"3D Printer Cloud Management Software.",
"\n\n",
"3D printing doesn't have to be a complicated, analog, SD card-filled experience; step into the future of modern 3D printing"
]
}
}

View File

@@ -0,0 +1,131 @@
# ======================================================================= #
# 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 typing import List
from components.moonraker.moonraker import Moonraker
from core.instance_manager.instance_manager import InstanceManager
from core.logger import DialogType, Logger
from core.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
SimpleConfigParser,
)
from extensions.base_extension import BaseExtension
from utils.common import backup_printer_config_dir, moonraker_exists
from utils.input_utils import get_confirm
# noinspection PyMethodMayBeStatic
class SimplyPrintExtension(BaseExtension):
def install_extension(self, **kwargs) -> None:
Logger.print_status("Installing SimplyPrint ...")
if not (mr_instances := moonraker_exists("SimplyPrint Installer")):
return
Logger.print_dialog(
DialogType.INFO,
self._construct_dialog(mr_instances, True),
)
if not get_confirm(
"Continue SimplyPrint installation?",
default_choice=True,
allow_go_back=True,
):
Logger.print_info("Exiting SimplyPrint installation ...")
return
try:
self._patch_moonraker_confs(mr_instances, True)
except Exception as e:
Logger.print_error(f"Error during SimplyPrint installation:\n{e}")
def remove_extension(self, **kwargs) -> None:
Logger.print_status("Removing SimplyPrint ...")
if not (mr_instances := moonraker_exists("SimplyPrint Uninstaller")):
return
Logger.print_dialog(
DialogType.INFO,
self._construct_dialog(mr_instances, False),
)
if not get_confirm(
"Do you really want to uninstall SimplyPrint?",
default_choice=True,
allow_go_back=True,
):
Logger.print_info("Exiting SimplyPrint uninstallation ...")
return
try:
self._patch_moonraker_confs(mr_instances, False)
except Exception as e:
Logger.print_error(f"Error during SimplyPrint installation:\n{e}")
def _construct_dialog(
self, mr_instances: List[Moonraker], is_install: bool
) -> List[str]:
mr_names = [f"{m.service_file_path.name}" for m in mr_instances]
_type = "install" if is_install else "uninstall"
return [
"The following Moonraker instances were found:",
*mr_names,
"\n\n",
f"The setup will {_type} SimplyPrint for all Moonraker instances. "
f"After {_type}ation, all Moonraker services will be restarted!",
]
def _patch_moonraker_confs(
self, mr_instances: List[Moonraker], is_install: bool
) -> None:
section = "simplyprint"
_type, _ft = ("Adding", "to") if is_install else ("Removing", "from")
patched_files = []
for moonraker in mr_instances:
Logger.print_status(
f"{_type} section 'simplyprint' {_ft} {moonraker.cfg_file} ..."
)
scp = SimpleConfigParser()
scp.read_file(moonraker.cfg_file)
install_and_has_section = is_install and scp.has_section(section)
uninstall_and_has_no_section = not is_install and not scp.has_section(
section
)
if install_and_has_section or uninstall_and_has_no_section:
status = "already" if is_install else "does not"
Logger.print_info(
f"Section 'simplyprint' {status} exists! Skipping ..."
)
continue
if is_install and not scp.has_section("simplyprint"):
backup_printer_config_dir()
scp.add_section(section)
elif not is_install and scp.has_section("simplyprint"):
backup_printer_config_dir()
scp.remove_section(section)
scp.write_file(moonraker.cfg_file)
patched_files.append(moonraker.cfg_file)
if patched_files:
InstanceManager.restart_all(mr_instances)
install_state = "successfully" if patched_files else "was already"
Logger.print_dialog(
DialogType.SUCCESS,
[f"SimplyPrint {install_state} {'' if is_install else 'un'}installed!"],
center_content=True,
)

View File

@@ -1,5 +1,5 @@
[Unit] [Unit]
Description=Moonraker Telegram Bot SV1 %INST% Description=Moonraker Telegram Bot SV1
Documentation=https://github.com/nlef/moonraker-telegram-bot/wiki Documentation=https://github.com/nlef/moonraker-telegram-bot/wiki
After=network-online.target After=network-online.target

View File

@@ -161,10 +161,11 @@ class TelegramBotExtension(BaseExtension):
# install dependencies # install dependencies
script = TG_BOT_DIR.joinpath("scripts/install.sh") script = TG_BOT_DIR.joinpath("scripts/install.sh")
package_list = parse_packages_from_file(script) package_list = parse_packages_from_file(script)
check_install_dependencies({*package_list}) check_install_dependencies({*package_list})
# create virtualenv # create virtualenv
if create_python_venv(TG_BOT_ENV): if create_python_venv(TG_BOT_ENV, allow_access_to_system_site_packages=True):
install_python_requirements(TG_BOT_ENV, TG_BOT_REQ_FILE) install_python_requirements(TG_BOT_ENV, TG_BOT_REQ_FILE)
def _patch_bot_update_manager(self, instances: List[Moonraker]) -> None: def _patch_bot_update_manager(self, instances: List[Moonraker]) -> None:

View File

@@ -14,10 +14,11 @@ from pathlib import Path
from typing import Dict, List, Literal, Optional, Set from typing import Dict, List, Literal, Optional, Set
from components.klipper.klipper import Klipper from components.klipper.klipper import Klipper
from components.moonraker.moonraker import Moonraker
from core.constants import ( from core.constants import (
COLOR_CYAN, COLOR_CYAN,
GLOBAL_DEPS, GLOBAL_DEPS,
PRINTER_CFG_BACKUP_DIR, PRINTER_DATA_BACKUP_DIR,
RESET_FORMAT, RESET_FORMAT,
) )
from core.logger import DialogType, Logger from core.logger import DialogType, Logger
@@ -142,23 +143,25 @@ def backup_printer_config_dir() -> None:
instances: List[Klipper] = get_instances(Klipper) instances: List[Klipper] = get_instances(Klipper)
bm = BackupManager() bm = BackupManager()
if not instances:
Logger.print_info("Unable to find directory to backup!")
Logger.print_info("Are there no Klipper instances installed?")
return
for instance in instances: for instance in instances:
name = f"config-{instance.data_dir.name}"
bm.backup_directory( bm.backup_directory(
name, instance.data_dir.name,
source=instance.base.cfg_dir, source=instance.base.cfg_dir,
target=PRINTER_CFG_BACKUP_DIR, target=PRINTER_DATA_BACKUP_DIR,
) )
def moonraker_exists(name: str = "") -> bool: def moonraker_exists(name: str = "") -> List[Moonraker]:
""" """
Helper method to check if a Moonraker instance exists Helper method to check if a Moonraker instance exists
:param name: Optional name of an installer where the check is performed :param name: Optional name of an installer where the check is performed
:return: True if at least one Moonraker instance exists, False otherwise :return: True if at least one Moonraker instance exists, False otherwise
""" """
from components.moonraker.moonraker import Moonraker
mr_instances: List[Moonraker] = get_instances(Moonraker) mr_instances: List[Moonraker] = get_instances(Moonraker)
info = ( info = (
@@ -175,8 +178,8 @@ def moonraker_exists(name: str = "") -> bool:
f"{info}. Please install Moonraker first!", f"{info}. Please install Moonraker first!",
], ],
) )
return False return []
return True return mr_instances
def trunc_string(input_str: str, length: int) -> str: def trunc_string(input_str: str, length: int) -> str:

View File

@@ -7,7 +7,7 @@ from http.client import HTTPResponse
from json import JSONDecodeError from json import JSONDecodeError
from pathlib import Path from pathlib import Path
from subprocess import DEVNULL, PIPE, CalledProcessError, check_output, run from subprocess import DEVNULL, PIPE, CalledProcessError, check_output, run
from typing import List, Type from typing import List, Tuple, Type
from core.instance_manager.instance_manager import InstanceManager from core.instance_manager.instance_manager import InstanceManager
from core.logger import Logger from core.logger import Logger
@@ -70,7 +70,7 @@ def git_pull_wrapper(repo: str, target_dir: Path) -> None:
return return
def get_repo_name(repo: Path) -> tuple[str, str] | None: def get_repo_name(repo: Path) -> Tuple[str, str]:
""" """
Helper method to extract the organisation and name of a repository | Helper method to extract the organisation and name of a repository |
:param repo: repository to extract the values from :param repo: repository to extract the values from
@@ -83,11 +83,14 @@ def get_repo_name(repo: Path) -> tuple[str, str] | None:
cmd = ["git", "-C", repo.as_posix(), "config", "--get", "remote.origin.url"] cmd = ["git", "-C", repo.as_posix(), "config", "--get", "remote.origin.url"]
result: str = check_output(cmd, stderr=DEVNULL).decode(encoding="utf-8") result: str = check_output(cmd, stderr=DEVNULL).decode(encoding="utf-8")
substrings: List[str] = result.strip().split("/")[-2:] substrings: List[str] = result.strip().split("/")[-2:]
return substrings[0], substrings[1]
# return "/".join(substrings).replace(".git", "") orga: str = substrings[0] if substrings[0] else "-"
name: str = substrings[1] if substrings[1] else "-"
return orga, name
except CalledProcessError: except CalledProcessError:
return None return "-", "-"
def get_local_tags(repo_path: Path, _filter: str | None = None) -> List[str]: def get_local_tags(repo_path: Path, _filter: str | None = None) -> List[str]:
@@ -184,7 +187,7 @@ def compare_semver_tags(tag1: str, tag2: str) -> bool:
if tag1 == tag2: if tag1 == tag2:
return False return False
def parse_version(v): def parse_version(v) -> List[int]:
return list(map(int, v[1:].split("."))) return list(map(int, v[1:].split(".")))
tag1_parts = parse_version(tag1) tag1_parts = parse_version(tag1)

View File

@@ -13,6 +13,7 @@ from components.klipper.klipper import Klipper
from components.moonraker.moonraker import Moonraker from components.moonraker.moonraker import Moonraker
from extensions.obico.moonraker_obico import MoonrakerObico from extensions.obico.moonraker_obico import MoonrakerObico
from extensions.octoeverywhere.octoeverywhere import Octoeverywhere from extensions.octoeverywhere.octoeverywhere import Octoeverywhere
from extensions.octoapp.octoapp import Octoapp
from extensions.telegram_bot.moonraker_telegram_bot import MoonrakerTelegramBot from extensions.telegram_bot.moonraker_telegram_bot import MoonrakerTelegramBot
InstanceType = TypeVar( InstanceType = TypeVar(
@@ -22,4 +23,5 @@ InstanceType = TypeVar(
MoonrakerTelegramBot, MoonrakerTelegramBot,
MoonrakerObico, MoonrakerObico,
Octoeverywhere, Octoeverywhere,
Octoapp,
) )

View File

@@ -17,7 +17,7 @@ from core.instance_manager.base_instance import SUFFIX_BLACKLIST
from utils.instance_type import InstanceType from utils.instance_type import InstanceType
def get_instances(instance_type: type) -> List[InstanceType]: def get_instances(instance_type: type, suffix_blacklist: List[str] = SUFFIX_BLACKLIST) -> List[InstanceType]:
from utils.common import convert_camelcase_to_kebabcase from utils.common import convert_camelcase_to_kebabcase
if not isinstance(instance_type, type): if not isinstance(instance_type, type):
@@ -30,7 +30,7 @@ def get_instances(instance_type: type) -> List[InstanceType]:
Path(SYSTEMD, service) Path(SYSTEMD, service)
for service in SYSTEMD.iterdir() for service in SYSTEMD.iterdir()
if pattern.search(service.name) if pattern.search(service.name)
and not any(s in service.name for s in SUFFIX_BLACKLIST) and not any(s in service.name for s in suffix_blacklist)
] ]
instance_list = [ instance_list = [

View File

@@ -91,19 +91,27 @@ def parse_packages_from_file(source_file: Path) -> List[str]:
return packages return packages
def create_python_venv(target: Path, force: bool = False) -> bool: def create_python_venv(
target: Path,
force: bool = False,
allow_access_to_system_site_packages: bool = False,
) -> bool:
""" """
Create a python 3 virtualenv at the provided target destination. Create a python 3 virtualenv at the provided target destination.
Returns True if the virtualenv was created successfully. Returns True if the virtualenv was created successfully.
Returns False if the virtualenv already exists, recreation was declined or creation failed. Returns False if the virtualenv already exists, recreation was declined or creation failed.
:param force: Force recreation of the virtualenv
:param target: Path where to create the virtualenv at :param target: Path where to create the virtualenv at
:param force: Force recreation of the virtualenv
:param allow_access_to_system_site_packages: give the virtual environment access to the system site-packages dir
:return: bool :return: bool
""" """
Logger.print_status("Set up Python virtual environment ...") Logger.print_status("Set up Python virtual environment ...")
cmd = ["virtualenv", "-p", "/usr/bin/python3", target.as_posix()]
cmd.append(
"--system-site-packages"
) if allow_access_to_system_site_packages else None
if not target.exists(): if not target.exists():
try: try:
cmd = ["virtualenv", "-p", "/usr/bin/python3", target.as_posix()]
run(cmd, check=True) run(cmd, check=True)
Logger.print_ok("Setup of virtualenv successful!") Logger.print_ok("Setup of virtualenv successful!")
return True return True