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
.vscode
.pytest_cache
.jupyter
*.ipynb
*.ipynb_checkpoints
*.tmp
__pycache__
.kiauh-env
*.code-workspace

View File

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

View File

@@ -353,10 +353,16 @@ def read_ports_from_nginx_configs() -> List[int]:
lines = cfg.readlines()
for line in lines:
line = line.replace("default_server", "")
line = re.sub(r"[;:\[\]]", "", line.strip())
if line.startswith("listen") and line.split()[-1] not in port_list:
port_list.append(line.split()[-1])
line = re.sub(
r"default_server|http://|https://|[;\[\]]",
"",
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]
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
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_ENABLED = Path("/etc/nginx/sites-enabled")
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}"
# get the class name of the extension
ext_class: Type[BaseExtension] = inspect.getmembers(
importlib.import_module(module_path),
predicate=lambda o: inspect.isclass(o)
and issubclass(o, BaseExtension)
and o != BaseExtension,
)[0][1]
module = importlib.import_module(module_path)
def predicate(o):
return (
inspect.isclass(o)
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
ext_instance: BaseExtension = ext_class(metadata)
@@ -72,7 +76,7 @@ class ExtensionsMenu(BaseMenu):
except (IOError, json.JSONDecodeError, ImportError) as 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):
ExtensionSubmenu(kwargs.get("opt_data"), self.__class__).run()

View File

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

View File

@@ -12,6 +12,7 @@ from typing import List
from components.klipper.klipper import Klipper
from components.moonraker.moonraker import Moonraker
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.submodules.simple_config_parser.src.simple_config_parser.simple_config_parser import (
SimpleConfigParser,
@@ -308,7 +309,8 @@ class ObicoExtension(BaseExtension):
def _check_and_opt_link_instances(self) -> None:
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] = [
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]
Description=Moonraker Telegram Bot SV1 %INST%
Description=Moonraker Telegram Bot SV1
Documentation=https://github.com/nlef/moonraker-telegram-bot/wiki
After=network-online.target

View File

@@ -161,10 +161,11 @@ class TelegramBotExtension(BaseExtension):
# install dependencies
script = TG_BOT_DIR.joinpath("scripts/install.sh")
package_list = parse_packages_from_file(script)
check_install_dependencies({*package_list})
# 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)
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 components.klipper.klipper import Klipper
from components.moonraker.moonraker import Moonraker
from core.constants import (
COLOR_CYAN,
GLOBAL_DEPS,
PRINTER_CFG_BACKUP_DIR,
PRINTER_DATA_BACKUP_DIR,
RESET_FORMAT,
)
from core.logger import DialogType, Logger
@@ -142,23 +143,25 @@ def backup_printer_config_dir() -> None:
instances: List[Klipper] = get_instances(Klipper)
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:
name = f"config-{instance.data_dir.name}"
bm.backup_directory(
name,
instance.data_dir.name,
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
:param name: Optional name of an installer where the check is performed
:return: True if at least one Moonraker instance exists, False otherwise
"""
from components.moonraker.moonraker import Moonraker
mr_instances: List[Moonraker] = get_instances(Moonraker)
info = (
@@ -175,8 +178,8 @@ def moonraker_exists(name: str = "") -> bool:
f"{info}. Please install Moonraker first!",
],
)
return False
return True
return []
return mr_instances
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 pathlib import Path
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.logger import Logger
@@ -70,7 +70,7 @@ def git_pull_wrapper(repo: str, target_dir: Path) -> None:
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 |
: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"]
result: str = check_output(cmd, stderr=DEVNULL).decode(encoding="utf-8")
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:
return None
return "-", "-"
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:
return False
def parse_version(v):
def parse_version(v) -> List[int]:
return list(map(int, v[1:].split(".")))
tag1_parts = parse_version(tag1)

View File

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

View File

@@ -17,7 +17,7 @@ from core.instance_manager.base_instance import SUFFIX_BLACKLIST
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
if not isinstance(instance_type, type):
@@ -30,7 +30,7 @@ def get_instances(instance_type: type) -> List[InstanceType]:
Path(SYSTEMD, service)
for service in SYSTEMD.iterdir()
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 = [

View File

@@ -91,19 +91,27 @@ def parse_packages_from_file(source_file: Path) -> List[str]:
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.
Returns True if the virtualenv was created successfully.
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 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
"""
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():
try:
cmd = ["virtualenv", "-p", "/usr/bin/python3", target.as_posix()]
run(cmd, check=True)
Logger.print_ok("Setup of virtualenv successful!")
return True