refactor: improve config creations and patching

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

kiauh/core/submodules/simple_config_parser/src/simple_config_parser/simple_config_parser.py

@@ -85,7 +85,7 @@ class DuplicateOptionError(Exception):
 class SimpleConfigParser:
     """A customized config parser targeted at handling Klipper style config files"""
 
-    _SECTION_RE = re.compile(r"\s*\[(\w+\s?[\w\-]+)]\s*([#;].*)?$")
+    _SECTION_RE = re.compile(r"\s*\[(\w+\s?.+)]\s*([#;].*)?$")
     _OPTION_RE = re.compile(r"^\s*(\w+)\s*[:=]\s*([^=:].*)\s*([#;].*)?$")
     _MLOPTION_RE = re.compile(r"^\s*(\w+)\s*[:=]\s*([#;].*)?$")
     _COMMENT_RE = re.compile(r"^\s*([#;].*)?$")
@@ -192,9 +192,9 @@ class SimpleConfigParser:
         if section not in self._all_sections:
             raise NoSectionError(section)
 
-        del self._all_sections[self._all_sections.index(section)]
-        del self._all_options[section]
-        del self._config[section]
+        self._all_sections.pop(self._all_sections.index(section))
+        self._all_options.pop(section)
+        self._config.pop(section)
 
     def options(self, section) -> List[str]:
         """Return a list of option names for the given section name"""
@@ -453,6 +453,7 @@ class SimpleConfigParser:
 
         self.section_name = section
         self._all_sections.append(section)
+        self._all_options[section] = {}
         self._config[section]: Section = {"_raw": raw_value, "body": []}
 
     def _parse_option(self, line: str) -> None:

kiauh/core/submodules/simple_config_parser/tests/features/internal_state/test_internal_state_changes.py

@@ -34,6 +34,7 @@ class TestInternalStateChanges:
         parser._store_internal_state_section(given, given)
 
         assert parser._all_sections == [given]
+        assert parser._all_options[given] == {}
         assert parser._config[given]["body"] == []
         assert parser._config[given]["_raw"] == given
 

kiauh/core/submodules/simple_config_parser/tests/features/line_parsing/data/case_parse_option.py

@@ -8,6 +8,11 @@ testcases = [
     ("option: value\n", "option", "value"),
     ("option: value # inline comment", "option", "value"),
     ("option: value # inline comment\n", "option", "value"),
+    (
+        "description: Helper: park toolhead used in PAUSE and CANCEL_PRINT",
+        "description",
+        "Helper: park toolhead used in PAUSE and CANCEL_PRINT",
+    ),
     ("description: homing!", "description", "homing!"),
     ("description: inline macro :-)", "description", "inline macro :-)"),
     ("path: %GCODES_DIR%", "path", "%GCODES_DIR%"),

kiauh/core/submodules/simple_config_parser/tests/features/line_parsing/data/case_parse_section.py

@@ -3,4 +3,6 @@ testcases = [
     ("[test_section two]", "test_section two"),
     ("[section1] # inline comment", "section1"),
     ("[section2] ; second comment", "section2"),
+    ("[include moonraker-obico-update.cfg]", "include moonraker-obico-update.cfg"),
+    ("[include moonraker_obico_macros.cfg]", "include moonraker_obico_macros.cfg"),
 ]

kiauh/core/submodules/simple_config_parser/tests/features/line_type_detection/data/case_line_is_section.py

@@ -1,6 +1,11 @@
 testcases = [
     ("[example_section]", True),
+    ("[gcode_macro CANCEL_PRINT]", True),
+    ("[gcode_macro SET_PAUSE_NEXT_LAYER]", True),
+    ("[gcode_macro _TOOLHEAD_PARK_PAUSE_CANCEL]", True),
     ("[update_manager moonraker-obico]", True),
+    ("[include moonraker_obico_macros.cfg]", True),
+    ("[include moonraker-obico-update.cfg]", True),
     ("[example_section two]", True),
     ("not_a_valid_section", False),
     ("section: invalid", False),

kiauh/extensions/obico/moonraker_obico.py

@@ -24,6 +24,13 @@ OBICO_DIR = Path.home().joinpath("moonraker-obico")
 OBICO_ENV = Path.home().joinpath("moonraker-obico-env")
 OBICO_REPO = "https://github.com/TheSpaghettiDetective/moonraker-obico.git"
 
+OBICO_CFG = "moonraker-obico.cfg"
+OBICO_CFG_SAMPLE = "moonraker-obico.cfg.sample"
+OBICO_LOG = "moonraker-obico.log"
+OBICO_UPDATE_CFG = "moonraker-obico-update.cfg"
+OBICO_UPDATE_CFG_SAMPLE = "moonraker-obico-update.cfg.sample"
+OBICO_MACROS_CFG = "moonraker_obico_macros.cfg"
+
 
 # noinspection PyMethodMayBeStatic
 class MoonrakerObico(BaseInstance):

kiauh/extensions/obico/moonraker_obico_extension.py

@@ -7,7 +7,6 @@
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
 import shutil
-from subprocess import run
 from typing import List
 
 from components.klipper.klipper import Klipper
@@ -18,13 +17,21 @@ from core.submodules.simple_config_parser.src.simple_config_parser.simple_config
 )
 from extensions.base_extension import BaseExtension
 from extensions.obico.moonraker_obico import (
+    OBICO_CFG_SAMPLE,
     OBICO_DIR,
     OBICO_ENV,
+    OBICO_LOG,
+    OBICO_MACROS_CFG,
     OBICO_REPO,
+    OBICO_UPDATE_CFG,
+    OBICO_UPDATE_CFG_SAMPLE,
     MoonrakerObico,
 )
 from utils.common import check_install_dependencies, moonraker_exists
-from utils.config_utils import add_config_section, remove_config_section
+from utils.config_utils import (
+    add_config_section,
+    remove_config_section,
+)
 from utils.fs_utils import remove_file
 from utils.git_utils import git_clone_wrapper, git_pull_wrapper
 from utils.input_utils import get_confirm, get_selection_input, get_string_input
@@ -102,16 +109,19 @@ class ObicoExtension(BaseExtension):
                 # create obico macros
                 self._create_obico_macros_cfg(moonraker)
 
+                # create obico update manager
+                self._create_obico_update_manager_cfg(moonraker)
+
                 obico_im.start_instance()
 
             cmd_sysctl_manage("daemon-reload")
 
             # add to klippers config
-            self._patch_obico_macros(kl_instances)
+            self._patch_printer_cfg(kl_instances)
             kl_im.restart_all_instance()
 
             # add to moonraker update manager
-            self._patch_update_manager(mr_instances)
+            self._patch_moonraker_conf(mr_instances)
             mr_im.restart_all_instance()
 
             # check linking of / ask for linking instances
@@ -154,9 +164,8 @@ class ObicoExtension(BaseExtension):
             self._remove_obico_instances(ob_im, ob_instances)
             self._remove_obico_dir()
             self._remove_obico_env()
-            remove_config_section("include moonraker_obico_macros.cfg", kl_instances)
-            remove_config_section("update_manager moonraker-obico", mr_instances)
-            remove_config_section("include moonraker-obico-update.cfg", mr_instances)
+            remove_config_section(f"include {OBICO_MACROS_CFG}", kl_instances)
+            remove_config_section(f"include {OBICO_UPDATE_CFG}", mr_instances)
             self._delete_obico_logs(ob_instances)
             Logger.print_dialog(
                 DialogType.SUCCESS,
@@ -236,19 +245,29 @@ class ObicoExtension(BaseExtension):
         install_python_requirements(OBICO_ENV, requirements)
 
     def _create_obico_macros_cfg(self, moonraker) -> None:
-        macros_cfg = OBICO_DIR.joinpath("include_cfgs/moonraker_obico_macros.cfg")
-        macros_target = moonraker.cfg_dir.joinpath("moonraker_obico_macros.cfg")
+        macros_cfg = OBICO_DIR.joinpath(f"include_cfgs/{OBICO_MACROS_CFG}")
+        macros_target = moonraker.cfg_dir.joinpath(OBICO_MACROS_CFG)
         if not macros_target.exists():
-            run(["cp", macros_cfg, macros_target], check=True)
+            shutil.copy(macros_cfg, macros_target)
         else:
             Logger.print_info(
-                f"Obico macros in {moonraker.cfg_dir} already exists! Skipped ..."
+                f"Obico's '{OBICO_MACROS_CFG}' in {moonraker.cfg_dir} already exists! Skipped ..."
+            )
+
+    def _create_obico_update_manager_cfg(self, moonraker) -> None:
+        update_cfg = OBICO_DIR.joinpath(OBICO_UPDATE_CFG_SAMPLE)
+        update_cfg_target = moonraker.cfg_dir.joinpath(OBICO_UPDATE_CFG)
+        if not update_cfg_target.exists():
+            shutil.copy(update_cfg, update_cfg_target)
+        else:
+            Logger.print_info(
+                f"Obico's '{OBICO_UPDATE_CFG}' in {moonraker.cfg_dir} already exists! Skipped ..."
             )
 
     def _create_obico_cfg(
         self, current_instance: MoonrakerObico, moonraker: Moonraker
     ) -> None:
-        cfg_template = OBICO_DIR.joinpath("moonraker-obico.cfg.sample")
+        cfg_template = OBICO_DIR.joinpath(OBICO_CFG_SAMPLE)
         cfg_target_file = current_instance.cfg_file
 
         if not cfg_template.exists():
@@ -258,7 +277,7 @@ class ObicoExtension(BaseExtension):
             return
 
         if not cfg_target_file.exists():
-            run(["cp", cfg_template, cfg_target_file], check=True)
+            shutil.copy(cfg_template, cfg_target_file)
             self._patch_obico_cfg(moonraker, current_instance)
         else:
             Logger.print_info(
@@ -273,27 +292,11 @@ class ObicoExtension(BaseExtension):
         scp.set("logging", "path", str(obico.log))
         scp.write(obico.cfg_file)
 
-    def _patch_obico_macros(self, klipper: List[Klipper]) -> None:
-        add_config_section(
-            section="include moonraker_obico_macros.cfg",
-            instances=klipper,
-        )
+    def _patch_printer_cfg(self, klipper: List[Klipper]) -> None:
+        add_config_section(section=f"include {OBICO_MACROS_CFG}", instances=klipper)
 
-    def _patch_update_manager(self, instances: List[Moonraker]) -> None:
-        env_py = f"{OBICO_ENV}/bin/python"
-        add_config_section(
-            section="update_manager moonraker-obico",
-            instances=instances,
-            options=[
-                ("type", "git_repo"),
-                ("path", str(OBICO_DIR)),
-                ("orgin", OBICO_REPO),
-                ("env", env_py),
-                ("requirements", "requirements.txt"),
-                ("install_script", "install.sh"),
-                ("managed_services", "moonraker-obico"),
-            ],
-        )
+    def _patch_moonraker_conf(self, instances: List[Moonraker]) -> None:
+        add_config_section(section=f"include {OBICO_UPDATE_CFG}", instances=instances)
 
     def _link_obico_instances(self, unlinked_instances):
         for obico in unlinked_instances:
@@ -373,7 +376,7 @@ class ObicoExtension(BaseExtension):
         Logger.print_status("Removing Obico logs ...")
         all_logfiles = []
         for instance in instances:
-            all_logfiles = list(instance.log_dir.glob("moonraker-obico.log*"))
+            all_logfiles = list(instance.log_dir.glob(f"{OBICO_LOG}*"))
         if not all_logfiles:
             Logger.print_info("No Obico logs found. Skipped ...")
             return