Compare commits

...

3 Commits

Author SHA1 Message Date
Théo Gaillard
757344128a fix(extensions_menu): prevent extension index collisions during loading (#783)
* fix(extensions_menu): prevent extension index collisions during loading

* feat(extensions_menu): add GITHUB_ISSUES_URL for reporting extension loading issues
2026-03-22 13:43:20 +01:00
Théo Gaillard
7ca08f9b30 feat(instance_manager): add interactive stopping of Klipper instances (#784)
* feat(instance_manager): add interactive stopping of Klipper instances with user confirmation

* fix(instance_utils): remove unnecessary 'self' parameter from stop_klipper_instances_interactively function

* refactor(tmc_autotune): replace internal stop function with direct call to stop_klipper_instances_interactively
2026-03-22 12:01:15 +01:00
David Rios
308079a821 fix: incorrect Logger.warn (#778)
Fix method name
2026-02-27 22:45:58 +01:00
5 changed files with 77 additions and 46 deletions

View File

@@ -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:

View File

@@ -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"

View File

@@ -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])))

View File

@@ -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

View File

@@ -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