refactor: add force flag to git_clone_wrapper

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

kiauh/components/octoeverywhere/__init__.py

@@ -20,6 +20,7 @@ OE_STORE_DIR = OE_DIR.joinpath("octoeverywhere-store")
 OE_REQ_FILE = OE_DIR.joinpath("requirements.txt")
 OE_DEPS_JSON_FILE = OE_DIR.joinpath("moonraker-system-dependencies.json")
 OE_INSTALL_SCRIPT = OE_DIR.joinpath("install.sh")
+OE_UPDATE_SCRIPT = OE_DIR.joinpath("update.sh")
 
 # filenames
 OE_CFG_NAME = "octoeverywhere.conf"

kiauh/components/octoeverywhere/octoeverywhere.py

@@ -19,6 +19,7 @@ from components.octoeverywhere import (
     OE_LOG_NAME,
     OE_STORE_DIR,
     OE_SYS_CFG_NAME,
+    OE_UPDATE_SCRIPT,
 )
 from core.instance_manager.base_instance import BaseInstance
 from utils.logger import Logger
@@ -61,6 +62,15 @@ class Octoeverywhere(BaseInstance):
             Logger.print_error(f"Error creating instance: {e}")
             raise
 
+    @staticmethod
+    def update():
+        try:
+            run(str(OE_UPDATE_SCRIPT), check=True, shell=True, cwd=OE_DIR)
+
+        except CalledProcessError as e:
+            Logger.print_error(f"Error updating OctoEverywhere for Klipper: {e}")
+            raise
+
     def delete(self) -> None:
         service_file = self.get_service_file_name(extension=True)
         service_file_path = self.get_service_file_path()

kiauh/components/octoeverywhere/octoeverywhere_setup.py

@@ -32,7 +32,7 @@ from utils.config_utils import (
     remove_config_section,
 )
 from utils.fs_utils import run_remove_routines
-from utils.git_utils import git_clone_wrapper, git_pull_wrapper
+from utils.git_utils import git_clone_wrapper
 from utils.input_utils import get_confirm
 from utils.logger import DialogType, Logger
 from utils.sys_utils import (
@@ -54,6 +54,7 @@ def install_octoeverywhere() -> None:
     if not moonraker_exists():
         return
 
+    force_clone = False
     oe_im = InstanceManager(Octoeverywhere)
     oe_instances: List[Octoeverywhere] = oe_im.instances
     if oe_instances:
@@ -61,7 +62,7 @@ def install_octoeverywhere() -> None:
             DialogType.INFO,
             [
                 "OctoEverywhere is already installed!",
-                "It is save to run the installer again to link your "
+                "It is safe to run the installer again to link your "
                 "printer or repair any issues.",
             ],
             padding_top=0,
@@ -72,6 +73,7 @@ def install_octoeverywhere() -> None:
             return
         else:
             Logger.print_status("Re-Installing OctoEverywhere for Klipper ...")
+            force_clone = True
 
     mr_im = InstanceManager(Moonraker)
     mr_instances: List[Moonraker] = mr_im.instances
@@ -99,7 +101,7 @@ def install_octoeverywhere() -> None:
         return
 
     try:
-        git_clone_wrapper(OE_REPO, OE_DIR)
+        git_clone_wrapper(OE_REPO, OE_DIR, force=force_clone)
 
         for moonraker in mr_instances:
             oe_im.current_instance = Octoeverywhere(suffix=moonraker.suffix)
@@ -122,14 +124,12 @@ def install_octoeverywhere() -> None:
 def update_octoeverywhere() -> None:
     Logger.print_status("Updating OctoEverywhere for Klipper ...")
     try:
-        oe_im = InstanceManager(Octoeverywhere)
-        oe_im.stop_all_instance()
-
-        git_pull_wrapper(OE_REPO, OE_DIR)
-        install_oe_dependencies()
-
-        oe_im.start_all_instance()
-        Logger.print_ok("OctoEverywhere for Klipper successfully updated!")
+        Octoeverywhere.update()
+        Logger.print_dialog(
+            DialogType.SUCCESS,
+            ["OctoEverywhere for Klipper successfully updated!"],
+            center_content=True,
+        )
 
     except Exception as e:
         Logger.print_error(f"Error during OctoEverywhere for Klipper update:\n{e}")

kiauh/extensions/obico/moonraker_obico_extension.py

@@ -57,6 +57,7 @@ class ObicoExtension(BaseExtension):
 
         # if obico is already installed, ask if the user wants to repair an
         # incomplete installation or link to the obico server
+        force_clone = False
         obico_im = InstanceManager(MoonrakerObico)
         obico_instances: List[MoonrakerObico] = obico_im.instances
         if obico_instances:
@@ -74,6 +75,7 @@ class ObicoExtension(BaseExtension):
                 return
             else:
                 Logger.print_status("Re-Installing Obico for Klipper ...")
+                force_clone = True
 
         # let the user confirm installation
         kl_im = InstanceManager(Klipper)
@@ -89,7 +91,7 @@ class ObicoExtension(BaseExtension):
             return
 
         try:
-            git_clone_wrapper(OBICO_REPO, OBICO_DIR)
+            git_clone_wrapper(OBICO_REPO, OBICO_DIR, force=force_clone)
             self._install_dependencies()
 
             # ask the user for the obico server url
@@ -212,7 +214,7 @@ class ObicoExtension(BaseExtension):
             DialogType.INFO,
             [
                 "Obico is already installed!",
-                "It is save to run the installer again to link your "
+                "It is safe to run the installer again to link your "
                 "printer or repair any issues.",
                 "\n\n",
                 "You can perform the following actions:",

kiauh/utils/git_utils.py

@@ -14,7 +14,7 @@ from utils.logger import Logger
 
 
 def git_clone_wrapper(
-    repo: str, target_dir: Path, branch: Optional[str] = None
+    repo: str, target_dir: Path, branch: Optional[str] = None, force: bool = False
 ) -> None:
     """
     Clones a repository from the given URL and checks out the specified branch if given.
@@ -22,6 +22,7 @@ def git_clone_wrapper(
     :param repo: The URL of the repository to clone.
     :param branch: The branch to check out. If None, the default branch will be checked out.
     :param target_dir: The directory where the repository will be cloned.
+    :param force: Force the cloning of the repository even if it already exists.
     :return: None
     """
     log = f"Cloning repository from '{repo}'"
@@ -29,7 +30,7 @@ def git_clone_wrapper(
     try:
         if Path(target_dir).exists():
             question = f"'{target_dir}' already exists. Overwrite?"
-            if not get_confirm(question, default_choice=False):
+            if not force and not get_confirm(question, default_choice=False):
                 Logger.print_info("Skip cloning of repository ...")
                 return
             shutil.rmtree(target_dir)