mirror of
https://github.com/dw-0/kiauh.git
synced 2026-03-29 13:58:36 +05:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
757344128a | ||
|
|
7ca08f9b30 | ||
|
|
308079a821 |
@@ -113,7 +113,7 @@ def check_user_groups() -> None:
|
||||
|
||||
if not get_confirm(f"Add user '{CURRENT_USER}' to group(s) now?"):
|
||||
log = "Skipped adding user to required groups. You might encounter issues."
|
||||
Logger.warn(log)
|
||||
Logger.print_warn(log)
|
||||
return
|
||||
|
||||
try:
|
||||
|
||||
@@ -10,3 +10,4 @@
|
||||
from pathlib import Path
|
||||
|
||||
EXTENSION_ROOT = Path(__file__).resolve().parents[1].joinpath("extensions")
|
||||
GITHUB_ISSUES_URL = "https://github.com/dw-0/kiauh/issues"
|
||||
|
||||
@@ -19,7 +19,7 @@ from core.logger import Logger
|
||||
from core.menus import Option
|
||||
from core.menus.base_menu import BaseMenu
|
||||
from core.types.color import Color
|
||||
from extensions import EXTENSION_ROOT
|
||||
from extensions import EXTENSION_ROOT, GITHUB_ISSUES_URL
|
||||
from extensions.base_extension import BaseExtension
|
||||
|
||||
|
||||
@@ -56,6 +56,22 @@ class ExtensionsMenu(BaseMenu):
|
||||
with open(metadata_json, "r") as m:
|
||||
# read extension metadata from json
|
||||
metadata = json.load(m).get("metadata")
|
||||
index = str(metadata.get("index"))
|
||||
|
||||
# Prevent collisions where one extension silently overrides another.
|
||||
if index in ext_dict:
|
||||
existing_name = ext_dict[index].metadata.get("display_name")
|
||||
duplicate_name = metadata.get("display_name")
|
||||
Logger.print_warn(
|
||||
"Failed loading extension"
|
||||
f" {ext}: duplicate index '{index}'"
|
||||
f" already used by '{existing_name}'."
|
||||
f" Skipping '{duplicate_name}'."
|
||||
f" Please report this at {GITHUB_ISSUES_URL}."
|
||||
)
|
||||
continue
|
||||
|
||||
int(index)
|
||||
module_name = metadata.get("module")
|
||||
module_path = f"kiauh.extensions.{ext.name}.{module_name}"
|
||||
|
||||
@@ -73,10 +89,20 @@ class ExtensionsMenu(BaseMenu):
|
||||
|
||||
# instantiate the extension with its metadata and add to dict
|
||||
ext_instance: BaseExtension = ext_class(metadata)
|
||||
ext_dict[f"{metadata.get('index')}"] = ext_instance
|
||||
ext_dict[index] = ext_instance
|
||||
|
||||
except (IOError, json.JSONDecodeError, ImportError) as e:
|
||||
print(f"Failed loading extension {ext}: {e}")
|
||||
except (
|
||||
IOError,
|
||||
json.JSONDecodeError,
|
||||
ImportError,
|
||||
TypeError,
|
||||
ValueError,
|
||||
AttributeError,
|
||||
) as e:
|
||||
Logger.print_warn(
|
||||
f"Failed loading extension {ext}: {e}. "
|
||||
f"Please report this at {GITHUB_ISSUES_URL}."
|
||||
)
|
||||
|
||||
return dict(sorted(ext_dict.items(), key=lambda x: int(x[0])))
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ from utils.config_utils import add_config_section, remove_config_section
|
||||
from utils.fs_utils import check_file_exist, create_symlink, run_remove_routines
|
||||
from utils.git_utils import git_clone_wrapper, git_pull_wrapper
|
||||
from utils.input_utils import get_confirm
|
||||
from utils.instance_utils import get_instances
|
||||
from utils.instance_utils import get_instances, stop_klipper_instances_interactively
|
||||
from utils.sys_utils import check_python_version
|
||||
|
||||
|
||||
@@ -84,7 +84,7 @@ class TmcAutotuneExtension(BaseExtension):
|
||||
|
||||
kl_instances = get_instances(Klipper)
|
||||
|
||||
if not self._stop_klipper_instances_interactively(
|
||||
if not stop_klipper_instances_interactively(
|
||||
kl_instances, "installation of TMC Autotune"
|
||||
):
|
||||
return
|
||||
@@ -172,7 +172,7 @@ class TmcAutotuneExtension(BaseExtension):
|
||||
|
||||
kl_instances = get_instances(Klipper)
|
||||
|
||||
if not self._stop_klipper_instances_interactively(
|
||||
if not stop_klipper_instances_interactively(
|
||||
kl_instances, "update of TMC Autotune"
|
||||
):
|
||||
return
|
||||
@@ -210,7 +210,7 @@ class TmcAutotuneExtension(BaseExtension):
|
||||
|
||||
kl_instances = get_instances(Klipper)
|
||||
|
||||
if not self._stop_klipper_instances_interactively(
|
||||
if not stop_klipper_instances_interactively(
|
||||
kl_instances, "removal of TMC Autotune"
|
||||
):
|
||||
return
|
||||
@@ -345,40 +345,3 @@ class TmcAutotuneExtension(BaseExtension):
|
||||
"Klipper TMC Autotune successfully removed from Moonraker update manager(s)!"
|
||||
)
|
||||
|
||||
def _stop_klipper_instances_interactively(
|
||||
self, kl_instances: List[Klipper], operation_name: str = "operation"
|
||||
) -> bool:
|
||||
"""
|
||||
Interactively stops all active Klipper instances, warning the user that ongoing prints will be disrupted.
|
||||
|
||||
:param kl_instances: List of Klipper instances to stop.
|
||||
:param operation_name: Optional name of the operation being performed (for user messaging). Do NOT capitalize.
|
||||
:return: True if instances were stopped or no instances found, False if operation was aborted.
|
||||
"""
|
||||
|
||||
if not kl_instances:
|
||||
Logger.print_warn("No instances found, skipping instance stopping.")
|
||||
return True
|
||||
|
||||
Logger.print_dialog(
|
||||
DialogType.ATTENTION,
|
||||
[
|
||||
"Do NOT continue if there are ongoing prints running",
|
||||
f"All Klipper instances will be restarted during the {operation_name} and "
|
||||
"ongoing prints WILL FAIL.",
|
||||
],
|
||||
)
|
||||
stop_klipper = get_confirm(
|
||||
question=f"Stop Klipper now and proceed with {operation_name}?",
|
||||
default_choice=False,
|
||||
allow_go_back=True,
|
||||
)
|
||||
|
||||
if stop_klipper:
|
||||
InstanceManager.stop_all(kl_instances)
|
||||
return True
|
||||
else:
|
||||
Logger.print_warn(
|
||||
f"{operation_name.capitalize()} aborted due to user request."
|
||||
)
|
||||
return False
|
||||
|
||||
@@ -12,8 +12,12 @@ import re
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
from components.klipper.klipper import Klipper
|
||||
from core.constants import SYSTEMD
|
||||
from core.instance_manager.base_instance import SUFFIX_BLACKLIST
|
||||
from core.instance_manager.instance_manager import InstanceManager
|
||||
from core.logger import DialogType, Logger
|
||||
from utils.input_utils import get_confirm
|
||||
from utils.instance_type import InstanceType
|
||||
|
||||
|
||||
@@ -56,3 +60,40 @@ def get_instance_suffix(name: str, file_path: Path) -> str:
|
||||
# otherwise there is and hyphen left, and we return the part after the hyphen
|
||||
suffix = file_path.stem[len(name) :]
|
||||
return suffix[1:] if suffix else ""
|
||||
|
||||
|
||||
def stop_klipper_instances_interactively(
|
||||
kl_instances: List[Klipper], operation_name: str = "operation"
|
||||
) -> bool:
|
||||
"""
|
||||
Interactively stops all active Klipper instances, warning the user that ongoing prints will be disrupted.
|
||||
|
||||
:param kl_instances: List of Klipper instances to stop.
|
||||
:param operation_name: Optional name of the operation being performed (for user messaging). Do NOT capitalize.
|
||||
:return: True if instances were stopped or no instances found, False if operation was aborted.
|
||||
"""
|
||||
|
||||
if not kl_instances:
|
||||
Logger.print_warn("No instances found, skipping instance stopping.")
|
||||
return True
|
||||
|
||||
Logger.print_dialog(
|
||||
DialogType.ATTENTION,
|
||||
[
|
||||
"Do NOT continue if there are ongoing prints running",
|
||||
f"All Klipper instances will be restarted during the {operation_name} and "
|
||||
"ongoing prints WILL FAIL.",
|
||||
],
|
||||
)
|
||||
stop_klipper = get_confirm(
|
||||
question=f"Stop Klipper now and proceed with {operation_name}?",
|
||||
default_choice=False,
|
||||
allow_go_back=True,
|
||||
)
|
||||
|
||||
if stop_klipper:
|
||||
InstanceManager.stop_all(kl_instances)
|
||||
return True
|
||||
else:
|
||||
Logger.print_warn(f"{operation_name.capitalize()} aborted due to user request.")
|
||||
return False
|
||||
|
||||
Reference in New Issue
Block a user