mirror of
https://github.com/dw-0/kiauh.git
synced 2025-12-14 19:14:27 +05:00
287 lines
12 KiB
Python
287 lines
12 KiB
Python
# ======================================================================= #
|
|
# Copyright (C) 2020 - 2025 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
|
|
|
|
import re
|
|
from typing import Dict, List, Optional, Set
|
|
|
|
from components.klipper.klipper import Klipper
|
|
from core.instance_manager.instance_manager import InstanceManager
|
|
from core.logger import DialogType, Logger
|
|
from core.types.color import Color
|
|
from core.menus.base_menu import print_back_footer
|
|
from extensions.base_extension import BaseExtension
|
|
from extensions.octoprint import (
|
|
OP_SUDOERS_FILE, OP_DEFAULT_PORT,
|
|
)
|
|
from extensions.octoprint.octoprint import Octoprint
|
|
from utils.common import check_install_dependencies
|
|
from utils.fs_utils import run_remove_routines, remove_with_sudo
|
|
from utils.input_utils import get_selection_input, get_confirm
|
|
from utils.instance_utils import get_instances
|
|
from utils.sys_utils import (
|
|
create_python_venv,
|
|
get_ipv4_addr,
|
|
install_python_packages,
|
|
)
|
|
|
|
|
|
# noinspection PyMethodMayBeStatic
|
|
class OctoprintExtension(BaseExtension):
|
|
def install_extension(self, **kwargs) -> None:
|
|
Logger.print_status("Installing OctoPrint ...")
|
|
|
|
klipper_instances: List[Klipper] = get_instances(Klipper)
|
|
if not klipper_instances:
|
|
Logger.print_dialog(
|
|
DialogType.WARNING,
|
|
[
|
|
"Klipper not found! Please install Klipper first.",
|
|
],
|
|
)
|
|
return
|
|
|
|
existing_ops: List[Octoprint] = get_instances(Octoprint)
|
|
existing_by_suffix: Dict[str, Octoprint] = {op.suffix: op for op in existing_ops}
|
|
candidates: List[Klipper] = [k for k in klipper_instances if k.suffix not in existing_by_suffix]
|
|
|
|
chosen: List[Klipper] = []
|
|
|
|
if len(klipper_instances) == 1:
|
|
k = klipper_instances[0]
|
|
if k.suffix in existing_by_suffix:
|
|
if not get_confirm(
|
|
f"OctoPrint already exists for '{k.service_file_path.stem}'. Reinstall?",
|
|
default_choice=True,
|
|
allow_go_back=True,
|
|
):
|
|
Logger.print_info("Aborted OctoPrint installation.")
|
|
return
|
|
chosen = [k]
|
|
else:
|
|
while True:
|
|
dialog = "╔═══════════════════════════════════════════════════════╗\n"
|
|
headline = Color.apply(
|
|
"The following Klipper instances were found:", Color.GREEN
|
|
)
|
|
dialog += f"║{headline:^64}║\n"
|
|
dialog += "╟───────────────────────────────────────────────────────╢\n"
|
|
|
|
if candidates:
|
|
line_all = Color.apply("a) Select all (install for all missing)", Color.YELLOW)
|
|
dialog += f"║ {line_all:<63}║\n"
|
|
dialog += "║ ║\n"
|
|
|
|
index_map: Dict[str, Klipper] = {}
|
|
for i, k in enumerate(klipper_instances, start=1):
|
|
mapping = existing_by_suffix.get(k.suffix)
|
|
suffix = f" <-> {mapping.service_file_path.stem}" if mapping else ""
|
|
line = Color.apply(f"{i}) {k.service_file_path.stem}{suffix}", Color.CYAN)
|
|
dialog += f"║ {line:<63}║\n"
|
|
index_map[str(i)] = k
|
|
|
|
dialog += "╟───────────────────────────────────────────────────────╢\n"
|
|
print(dialog, end="")
|
|
print_back_footer()
|
|
|
|
allowed = list(index_map.keys()) + ["b"] + (["a"] if candidates else [])
|
|
choice = get_selection_input("Choose instance to install OctoPrint for", allowed)
|
|
|
|
if choice == "b":
|
|
Logger.print_info("Aborted OctoPrint installation.")
|
|
return
|
|
if choice == "a":
|
|
chosen = candidates
|
|
break
|
|
|
|
selected = index_map[choice]
|
|
if selected.suffix in existing_by_suffix:
|
|
confirm = get_confirm(
|
|
f"OctoPrint already exists for '{selected.service_file_path.stem}'. Reinstall?",
|
|
default_choice=True,
|
|
allow_go_back=True,
|
|
)
|
|
if not confirm:
|
|
# back to menu
|
|
continue
|
|
chosen = [selected]
|
|
break
|
|
|
|
deps = {
|
|
"git",
|
|
"wget",
|
|
"python3-pip",
|
|
"python3-dev",
|
|
"libyaml-dev",
|
|
"build-essential",
|
|
"python3-setuptools",
|
|
"python3-virtualenv",
|
|
}
|
|
check_install_dependencies(deps)
|
|
|
|
# Determine used ports from existing OctoPrint services and prepare regex
|
|
used_ports: Set[int] = set()
|
|
port_re = re.compile(r"--port=(\d+)")
|
|
for op in existing_ops:
|
|
try:
|
|
content = op.service_file_path.read_text()
|
|
m = port_re.search(content)
|
|
if m:
|
|
used_ports.add(int(m.group(1)))
|
|
except OSError:
|
|
pass
|
|
|
|
# noinspection PyShadowingNames
|
|
def read_existing_port(suffix: str) -> Optional[int]:
|
|
op = existing_by_suffix.get(suffix)
|
|
if not op:
|
|
return None
|
|
try:
|
|
content = op.service_file_path.read_text()
|
|
m = port_re.search(content)
|
|
return int(m.group(1)) if m else None
|
|
except OSError:
|
|
return None
|
|
|
|
def next_free_port(start: int, used: Set[int]) -> int:
|
|
p = start
|
|
while p in used:
|
|
p += 1
|
|
used.add(p)
|
|
return p
|
|
|
|
created_ops: List[Octoprint] = []
|
|
for k in chosen:
|
|
# Keep existing port on reinstall, otherwise assign next free one
|
|
existing_port = read_existing_port(k.suffix)
|
|
port = existing_port if existing_port is not None else next_free_port(OP_DEFAULT_PORT, used_ports)
|
|
|
|
instance = Octoprint(suffix=k.suffix)
|
|
|
|
if create_python_venv(instance.env_dir, force=False):
|
|
Logger.print_ok(
|
|
f"Virtualenv created: {instance.env_dir}", prefix=False
|
|
)
|
|
else:
|
|
Logger.print_info(
|
|
f"Virtualenv exists: {instance.env_dir}. Skipping creation ..."
|
|
)
|
|
|
|
install_python_packages(instance.env_dir, ["octoprint"])
|
|
|
|
instance.create(port=port)
|
|
created_ops.append(instance)
|
|
|
|
for inst in created_ops:
|
|
try:
|
|
InstanceManager.enable(inst)
|
|
InstanceManager.start(inst)
|
|
except Exception as e:
|
|
Logger.print_error(
|
|
f"Failed to enable/start {inst.service_file_path.name}: {e}"
|
|
)
|
|
|
|
ip = get_ipv4_addr()
|
|
lines = ["Access your new OctoPrint instance(s) at:"]
|
|
for inst in created_ops:
|
|
try:
|
|
content = inst.service_file_path.read_text()
|
|
m = port_re.search(content)
|
|
if m:
|
|
# noinspection HttpUrlsUsage
|
|
lines.append(f"● {inst.service_file_path.stem}: http://{ip}:{m.group(1)}")
|
|
except OSError:
|
|
pass
|
|
|
|
Logger.print_dialog(DialogType.SUCCESS, lines, center_content=False)
|
|
|
|
def remove_extension(self, **kwargs) -> None:
|
|
Logger.print_status("Removing OctoPrint ...")
|
|
|
|
try:
|
|
op_instances: List[Octoprint] = get_instances(Octoprint)
|
|
if not op_instances:
|
|
Logger.print_info("No OctoPrint instances found. Skipped ...")
|
|
return
|
|
|
|
remove_all = False
|
|
if len(op_instances) == 1:
|
|
to_remove = op_instances
|
|
else:
|
|
dialog = "╔═══════════════════════════════════════════════════════╗\n"
|
|
headline = Color.apply(
|
|
"The following OctoPrint instances were found:", Color.GREEN
|
|
)
|
|
dialog += f"║{headline:^64}║\n"
|
|
dialog += "╟───────────────────────────────────────────────────────╢\n"
|
|
select_all = Color.apply("a) Select all", Color.YELLOW)
|
|
dialog += f"║ {select_all:<63}║\n"
|
|
dialog += "║ ║\n"
|
|
|
|
for i, inst in enumerate(op_instances, start=1):
|
|
line = Color.apply(
|
|
f"{i}) {inst.service_file_path.stem}", Color.CYAN
|
|
)
|
|
dialog += f"║ {line:<63}║\n"
|
|
dialog += "╟───────────────────────────────────────────────────────╢\n"
|
|
print(dialog, end="")
|
|
print_back_footer()
|
|
|
|
allowed = [str(i) for i in range(1, len(op_instances) + 1)]
|
|
allowed.extend(["a", "b"])
|
|
choice = get_selection_input("Choose instance to remove", allowed)
|
|
|
|
if choice == "a":
|
|
remove_all = True
|
|
to_remove = op_instances
|
|
elif choice == "b":
|
|
Logger.print_info("Aborted OctoPrint removal.")
|
|
return
|
|
else:
|
|
idx = int(choice) - 1
|
|
to_remove = [op_instances[idx]]
|
|
|
|
for inst in to_remove:
|
|
Logger.print_status(
|
|
f"Removing instance {inst.service_file_path.stem} ..."
|
|
)
|
|
try:
|
|
InstanceManager.remove(inst)
|
|
except Exception as e:
|
|
Logger.print_error(
|
|
f"Failed to remove service {inst.service_file_path.name}: {e}"
|
|
)
|
|
|
|
# Remove only this instance's env and basedir
|
|
if inst.env_dir.exists():
|
|
Logger.print_status(f"Removing {inst.env_dir} ...")
|
|
run_remove_routines(inst.env_dir)
|
|
if inst.basedir.exists():
|
|
Logger.print_status(f"Removing {inst.basedir} ...")
|
|
run_remove_routines(inst.basedir)
|
|
|
|
# Remove sudoers file only if no instances remain
|
|
remaining = get_instances(Octoprint)
|
|
if not remaining and OP_SUDOERS_FILE.exists():
|
|
Logger.print_status(f"Removing {OP_SUDOERS_FILE} ...")
|
|
remove_with_sudo(OP_SUDOERS_FILE)
|
|
|
|
Logger.print_dialog(
|
|
DialogType.SUCCESS,
|
|
[
|
|
"Selected OctoPrint instance(s) successfully removed!"
|
|
if not remove_all
|
|
else "All OctoPrint instances successfully removed!",
|
|
],
|
|
center_content=True,
|
|
)
|
|
|
|
except Exception as e:
|
|
Logger.print_error(f"Error during OctoPrint removal: {e}")
|