refactor(KIAUH): use util functions for Klipper and Moonraker to get their status

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

kiauh/core/backup_manager/backup_manager.py

@@ -11,6 +11,7 @@
 
 import shutil
 from pathlib import Path
+from typing import List
 
 from kiauh import KIAUH_BACKUP_DIR
 from kiauh.utils.common import get_current_date
@@ -19,51 +20,53 @@ from kiauh.utils.logger import Logger
 
 # noinspection PyMethodMayBeStatic
 class BackupManager:
-    def __init__(
+    def __init__(self, backup_root_dir: Path = KIAUH_BACKUP_DIR):
-        self, backup_name: str, source: Path, backup_dir: Path = KIAUH_BACKUP_DIR
+        self._backup_root_dir = backup_root_dir
+
+    @property
+    def backup_root_dir(self) -> Path:
+        return self._backup_root_dir
+
+    @backup_root_dir.setter
+    def backup_root_dir(self, value: Path):
+        self._backup_root_dir = value
+
+    def backup_file(
+        self, files: List[Path] = None, target: Path = None, custom_filename=None
     ):
-        self._backup_name = backup_name
+        if not files:
-        self._source = source
+            raise ValueError("Parameter 'files' cannot be None or an empty List!")
-        self._backup_dir = backup_dir
 
-    @property
+        target = self.backup_root_dir if target is None else target
-    def backup_name(self) -> str:
+        for file in files:
-        return self._backup_name
+            Logger.print_status(f"Creating backup of {file} ...")
+            if Path(file).is_file():
+                date = get_current_date().get("date")
+                time = get_current_date().get("time")
+                filename = f"{file.stem}-{date}-{time}{file.suffix}"
+                filename = custom_filename if custom_filename is not None else filename
+                try:
+                    Path(target).mkdir(exist_ok=True)
+                    shutil.copyfile(file, Path(target, filename))
+                except OSError as e:
+                    Logger.print_error(f"Unable to backup '{file}':\n{e}")
+                    continue
+            else:
+                Logger.print_info(f"File '{file}' not found ...")
 
-    @backup_name.setter
+    def backup_directory(self, name: str, source: Path, target: Path = None) -> None:
-    def backup_name(self, value: str):
+        if source is None or not Path(source).exists():
-        self._backup_name = value
-
-    @property
-    def source(self) -> Path:
-        return self._source
-
-    @source.setter
-    def source(self, value: Path):
-        self._source = value
-
-    @property
-    def backup_dir(self) -> Path:
-        return self._backup_dir
-
-    @backup_dir.setter
-    def backup_dir(self, value: Path):
-        self._backup_dir = value
-
-    def backup(self) -> None:
-        if self._source is None or not Path(self._source).exists():
             raise OSError
 
+        target = self.backup_root_dir if target is None else target
         try:
-            log = f"Creating backup of {self.backup_name} in {self.backup_dir} ..."
+            log = f"Creating backup of {name} in {target} ..."
             Logger.print_status(log)
-            date = get_current_date()
+            date = get_current_date().get("date")
-            dest = Path(
+            time = get_current_date().get("time")
-                f"{self.backup_dir}/{self.backup_name}/{date.get('date')}-{date.get('time')}"
+            shutil.copytree(source, Path(target, f"{name}-{date}-{time}"))
-            )
-            shutil.copytree(src=self.source, dst=dest)
         except OSError as e:
-            Logger.print_error(f"Unable to backup source directory. Not exist.\n{e}")
+            Logger.print_error(f"Unable to backup directory '{source}':\n{e}")
             return
 
         Logger.print_ok("Backup successfull!")

kiauh/core/config_manager/config_manager.py

@@ -10,6 +10,7 @@
 # ======================================================================= #
 
 import configparser
+from pathlib import Path
 from typing import Union
 
 from kiauh.utils.logger import Logger
@@ -21,6 +22,9 @@ class ConfigManager:
         self.config_file = cfg_file
         self.config = CustomConfigParser()
 
+        if Path(cfg_file).is_file():
+            self.read_config()
+
     def read_config(self) -> None:
         if not self.config_file:
             Logger.print_error("Unable to read config file. File not found.")
@@ -32,7 +36,7 @@ class ConfigManager:
         with open(self.config_file, "w") as cfg:
             self.config.write(cfg)
 
-    def get_value(self, section: str, key: str, silent=False) -> Union[str, bool, None]:
+    def get_value(self, section: str, key: str, silent=True) -> Union[str, bool, None]:
         if not self.config.has_section(section):
             if not silent:
                 log = f"Section not defined. Unable to read section: [{section}]."

kiauh/core/menus/main_menu.py

@@ -18,7 +18,10 @@ from kiauh.core.menus.install_menu import InstallMenu
 from kiauh.core.menus.remove_menu import RemoveMenu
 from kiauh.core.menus.settings_menu import SettingsMenu
 from kiauh.core.menus.update_menu import UpdateMenu
-from kiauh.utils.constants import COLOR_MAGENTA, COLOR_CYAN, RESET_FORMAT
+from kiauh.modules.klipper.klipper_utils import get_klipper_status
+from kiauh.modules.mainsail.mainsail_utils import get_mainsail_status
+from kiauh.modules.moonraker.moonraker_utils import get_moonraker_status
+from kiauh.utils.constants import COLOR_MAGENTA, COLOR_CYAN, RESET_FORMAT, COLOR_RED
 
 
 class MainMenu(BaseMenu):
@@ -36,8 +39,38 @@ class MainMenu(BaseMenu):
             },
             footer_type=QUIT_FOOTER,
         )
+        self.kl_status = None
+        self.kl_repo = None
+        self.mr_status = None
+        self.mr_repo = None
+        self.ms_status = None
+        self.fl_status = None
+        self.ks_status = None
+        self.mb_status = None
+        self.cn_status = None
+        self.tg_status = None
+        self.ob_status = None
+        self.oe_status = None
+        self.init_status()
+
+    def init_status(self) -> None:
+        status_vars = ["kl", "mr", "ms", "fl", "ks", "mb", "cn", "tg", "ob", "oe"]
+        for var in status_vars:
+            setattr(self, f"{var}_status", f"{COLOR_RED}Not installed!{RESET_FORMAT}")
+
+    def fetch_status(self) -> None:
+        # klipper
+        self.kl_status = get_klipper_status().get("status")
+        self.kl_repo = get_klipper_status().get("repo")
+        # moonraker
+        self.mr_status = get_moonraker_status().get("status")
+        self.mr_repo = get_moonraker_status().get("repo")
+        # mainsail
+        self.ms_status = get_mainsail_status()
 
     def print_menu(self):
+        self.fetch_status()
+
         header = " [ Main Menu ] "
         footer1 = "KIAUH v6.0.0"
         footer2 = f"Changelog: {COLOR_MAGENTA}https://git.io/JnmlX{RESET_FORMAT}"
@@ -48,21 +81,21 @@ class MainMenu(BaseMenu):
             /=======================================================\\
             | {color}{header:~^{count}}{RESET_FORMAT} |
             |-------------------------------------------------------|
-            |  0) [Log-Upload] |         Klipper: <TODO>            |
+            |  0) [Log-Upload] |   Klipper: {self.kl_status:<32} |
-            |                  |            Repo: <TODO>            |
+            |                  |      Repo: {self.kl_repo:<32} |
-            |  1) [Install]    |                                    |
+            |  1) [Install]    |------------------------------------|
-            |  2) [Update]     |       Moonraker: <TODO>            |
+            |  2) [Update]     | Moonraker: {self.mr_status:<32} |
-            |  3) [Remove]     |            Repo: <TODO>            |
+            |  3) [Remove]     |      Repo: {self.mr_repo:<32} |
-            |  4) [Advanced]   |                                    |
+            |  4) [Advanced]   |------------------------------------|
-            |  5) [Backup]     |        Mainsail: <TODO>            |
+            |  5) [Backup]     |        Mainsail: {self.ms_status:<26} |
-            |                  |          Fluidd: <TODO>            |
+            |                  |          Fluidd: {self.fl_status:<26} |
-            |  6) [Settings]   |   KlipperScreen: <TODO>            |
+            |  6) [Settings]   |   KlipperScreen: {self.ks_status:<26} |
-            |                  |     Mobileraker: <TODO>            |
+            |                  |     Mobileraker: {self.mb_status:<26} |
             |                  |                                    |
-            |                  |       Crowsnest: <TODO>            |
+            |                  |       Crowsnest: {self.cn_status:<26} |
-            |                  |    Telegram Bot: <TODO>            |
+            |                  |    Telegram Bot: {self.tg_status:<26} |
-            |                  |           Obico: <TODO>            |
+            |                  |           Obico: {self.ob_status:<26} |
-            |                  |  OctoEverywhere: <TODO>            |
+            |                  |  OctoEverywhere: {self.oe_status:<26} |
             |-------------------------------------------------------|
             | {COLOR_CYAN}{footer1:^16}{RESET_FORMAT} | {footer2:^43} |
             """

kiauh/modules/klipper/klipper_setup.py

@@ -145,8 +145,6 @@ def install_klipper(
 
 def setup_klipper_prerequesites() -> None:
     cm = ConfigManager(cfg_file=KIAUH_CFG)
-    cm.read_config()
-
     repo = str(cm.get_value("klipper", "repository_url") or DEFAULT_KLIPPER_REPO_URL)
     branch = str(cm.get_value("klipper", "branch") or "master")
 
@@ -265,14 +263,10 @@ def update_klipper() -> None:
         return
 
     cm = ConfigManager(cfg_file=KIAUH_CFG)
-    cm.read_config()
-
     if cm.get_value("kiauh", "backup_before_update"):
-        backup_manager = BackupManager(source=KLIPPER_DIR, backup_name="klipper")
+        bm = BackupManager()
-        backup_manager.backup()
+        bm.backup_directory("klipper", KLIPPER_DIR)
-        backup_manager.backup_name = "klippy-env"
+        bm.backup_directory("klippy-env", KLIPPER_ENV_DIR)
-        backup_manager.source = KLIPPER_ENV_DIR
-        backup_manager.backup()
 
     instance_manager = InstanceManager(Klipper)
     instance_manager.stop_all_instance()

kiauh/modules/klipper/klipper_utils.py

@@ -16,22 +16,30 @@ import shutil
 import subprocess
 import textwrap
 
-from typing import List, Union
+from typing import List, Union, Literal, Dict
 
 from kiauh.core.config_manager.config_manager import ConfigManager
 from kiauh.core.instance_manager.instance_manager import InstanceManager
-from kiauh.modules.klipper import MODULE_PATH
+from kiauh.modules.klipper import MODULE_PATH, KLIPPER_DIR, KLIPPER_ENV_DIR
 from kiauh.modules.klipper.klipper import Klipper
 from kiauh.modules.klipper.klipper_dialogs import (
     print_missing_usergroup_dialog,
     print_select_custom_name_dialog,
 )
+from kiauh.utils.common import get_install_status_common, get_repo_name
 from kiauh.utils.constants import CURRENT_USER
 from kiauh.utils.input_utils import get_confirm, get_string_input
 from kiauh.utils.logger import Logger
 from kiauh.utils.system_utils import mask_system_service
 
 
+def get_klipper_status() -> Dict[Literal["status", "repo"], str]:
+    return {
+        "status": get_install_status_common(Klipper, KLIPPER_DIR, KLIPPER_ENV_DIR),
+        "repo": get_repo_name(KLIPPER_DIR),
+    }
+
+
 def assign_custom_names(
     instance_count: int, install_count: int, instance_list: List[Klipper] = None
 ) -> List[str]:
@@ -209,7 +217,6 @@ def create_example_printer_cfg(instance: Klipper) -> None:
         return
 
     cm = ConfigManager(target)
-    cm.read_config()
     cm.set_value("virtual_sdcard", "path", instance.gcodes_dir)
     cm.write_config()
     Logger.print_ok(f"Example printer.cfg created in '{instance.cfg_dir}'")

kiauh/modules/mainsail/mainsail_remove.py

@@ -106,7 +106,6 @@ def remove_updater_section(name: str) -> None:
             continue
 
         cm = ConfigManager(instance.cfg_file)
-        cm.read_config()
         if not cm.config.has_section(name):
             Logger.print_info("Section not present. Skipped ...")
             continue
@@ -153,7 +152,6 @@ def remove_printer_cfg_include() -> None:
             continue
 
         cm = ConfigManager(instance.cfg_file)
-        cm.read_config()
         if not cm.config.has_section("include mainsail.cfg"):
             Logger.print_info("Section not present. Skipped ...")
             continue

kiauh/modules/mainsail/mainsail_setup.py

@@ -78,7 +78,7 @@ def run_mainsail_installation() -> None:
         print_mainsail_already_installed_dialog()
         do_reinstall = get_confirm("Re-install Mainsail?", allow_go_back=True)
         if do_reinstall:
-            backup_config_json()
+            backup_config_json(is_temp=True)
         else:
             return
 
@@ -91,7 +91,6 @@ def run_mainsail_installation() -> None:
         install_ms_config = get_confirm(question, allow_go_back=False)
 
     cm = ConfigManager(cfg_file=KIAUH_CFG)
-    cm.read_config()
     default_port = cm.get_value("mainsail", "default_port")
     mainsail_port = default_port if default_port else 80
     if not default_port:
@@ -214,7 +213,6 @@ def patch_moonraker_conf(
             return
 
         cm = ConfigManager(cfg_file)
-        cm.read_config()
         if cm.config.has_section(section_name):
             Logger.print_info("Section already exist. Skipped ...")
             return
@@ -238,7 +236,6 @@ def patch_printer_config(klipper_instances: List[Klipper]) -> None:
             return
 
         cm = ConfigManager(cfg_file)
-        cm.read_config()
         if cm.config.has_section("include mainsail.cfg"):
             Logger.print_info("Section already exist. Skipped ...")
             return

kiauh/modules/mainsail/mainsail_utils.py

@@ -15,20 +15,31 @@ import shutil
 from pathlib import Path
 from typing import List
 
+from kiauh.core.backup_manager.backup_manager import BackupManager
 from kiauh.modules.klipper.klipper import Klipper
-from kiauh.modules.mainsail import MAINSAIL_CONFIG_JSON
+from kiauh.modules.mainsail import MAINSAIL_CONFIG_JSON, MAINSAIL_DIR
+from kiauh.utils import NGINX_SITES_AVAILABLE, NGINX_CONFD
+from kiauh.utils.common import get_install_status_webui
 from kiauh.utils.logger import Logger
 
 
-# TODO: give this method an optional target dir to backup to
+def get_mainsail_status() -> str:
-#  alteratively use the BackupManager for handling this task (probably best)
+    return get_install_status_webui(
-def backup_config_json() -> None:
+        MAINSAIL_DIR,
-    try:
+        Path(NGINX_SITES_AVAILABLE, "mainsail"),
-        Logger.print_status(f"Backup '{MAINSAIL_CONFIG_JSON}' ...")
+        Path(NGINX_CONFD, "upstreams.conf"),
-        target = os.path.join(Path.home(), "config.json.kiauh.bak")
+        Path(NGINX_CONFD, "common_vars.conf"),
-        shutil.copy(MAINSAIL_CONFIG_JSON, target)
+    )
-    except OSError:
+
-        Logger.print_info(f"Unable to backup config.json. Skipped ...")
+
+def backup_config_json(is_temp=False) -> None:
+    Logger.print_status(f"Backup '{MAINSAIL_CONFIG_JSON}' ...")
+    bm = BackupManager()
+    if is_temp:
+        fn = Path(Path.home(), "config.json.kiauh.bak")
+        bm.backup_file([MAINSAIL_CONFIG_JSON], custom_filename=fn)
+    else:
+        bm.backup_file([MAINSAIL_CONFIG_JSON])
 
 
 def restore_config_json() -> None:
@@ -37,7 +48,7 @@ def restore_config_json() -> None:
         source = os.path.join(Path.home(), "config.json.kiauh.bak")
         shutil.copy(source, MAINSAIL_CONFIG_JSON)
     except OSError:
-        Logger.print_info(f"Unable to restore config.json. Skipped ...")
+        Logger.print_info("Unable to restore config.json. Skipped ...")
 
 
 def enable_mainsail_remotemode() -> None:

kiauh/modules/moonraker/moonraker.py

@@ -156,7 +156,6 @@ class Moonraker(BaseInstance):
             return None
 
         cm = ConfigManager(cfg_file=self.cfg_file)
-        cm.read_config()
         port = cm.get_value("server", "port")
 
         return int(port) if port is not None else port

kiauh/modules/moonraker/moonraker_setup.py

@@ -150,8 +150,6 @@ def install_moonraker(
 
 def setup_moonraker_prerequesites() -> None:
     cm = ConfigManager(cfg_file=KIAUH_CFG)
-    cm.read_config()
-
     repo = str(
         cm.get_value("moonraker", "repository_url") or DEFAULT_MOONRAKER_REPO_URL
     )
@@ -291,14 +289,10 @@ def update_moonraker() -> None:
         return
 
     cm = ConfigManager(cfg_file=KIAUH_CFG)
-    cm.read_config()
-
     if cm.get_value("kiauh", "backup_before_update"):
-        backup_manager = BackupManager(source=MOONRAKER_DIR, backup_name="moonraker")
+        bm = BackupManager()
-        backup_manager.backup()
+        bm.backup_directory("moonraker", MOONRAKER_DIR)
-        backup_manager.backup_name = "moonraker-env"
+        bm.backup_directory("moonraker-env", MOONRAKER_ENV_DIR)
-        backup_manager.source = MOONRAKER_ENV_DIR
-        backup_manager.backup()
 
     instance_manager = InstanceManager(Moonraker)
     instance_manager.stop_all_instance()

kiauh/modules/moonraker/moonraker_utils.py

@@ -11,20 +11,32 @@
 
 import os
 import shutil
-from typing import List, Dict
+from typing import Dict, Literal
 
 from kiauh.core.config_manager.config_manager import ConfigManager
 from kiauh.modules.moonraker import (
     DEFAULT_MOONRAKER_PORT,
     MODULE_PATH,
+    MOONRAKER_DIR,
+    MOONRAKER_ENV_DIR,
 )
 from kiauh.modules.moonraker.moonraker import Moonraker
+from kiauh.utils.common import get_install_status_common, get_repo_name
 from kiauh.utils.logger import Logger
 from kiauh.utils.system_utils import (
     get_ipv4_addr,
 )
 
 
+def get_moonraker_status() -> Dict[Literal["status", "repo"], str]:
+    return {
+        "status": get_install_status_common(
+            Moonraker, MOONRAKER_DIR, MOONRAKER_ENV_DIR
+        ),
+        "repo": get_repo_name(MOONRAKER_DIR),
+    }
+
+
 def create_example_moonraker_conf(
     instance: Moonraker, ports_map: Dict[str, int]
 ) -> None:
@@ -41,9 +53,6 @@ def create_example_moonraker_conf(
         Logger.print_error(f"Unable to create example moonraker.conf:\n{e}")
         return
 
-    cm = ConfigManager(target)
-    cm.read_config()
-
     ports = [
         ports_map.get(instance)
         for instance in ports_map
@@ -61,9 +70,11 @@ def create_example_moonraker_conf(
 
     ports_map[instance.suffix] = port
 
-    uds = f"{instance.comms_dir}/klippy.sock"
     ip = get_ipv4_addr().split(".")[:2]
     ip.extend(["0", "0/16"])
+    uds = f"{instance.comms_dir}/klippy.sock"
+
+    cm = ConfigManager(target)
     trusted_clients = f"\n{'.'.join(ip)}"
     trusted_clients += cm.get_value("authorization", "trusted_clients")
 

kiauh/utils/common.py

@@ -9,10 +9,21 @@
 #  This file may be distributed under the terms of the GNU GPLv3 license  #
 # ======================================================================= #
 
+import subprocess
 from datetime import datetime
-from typing import Dict, Literal, List
+from pathlib import Path
+from typing import Dict, Literal, List, Type
 
-from kiauh.utils.constants import COLOR_CYAN, RESET_FORMAT
+from kiauh.core.instance_manager.base_instance import BaseInstance
+from kiauh.core.instance_manager.instance_manager import InstanceManager
+from kiauh.utils.constants import (
+    COLOR_CYAN,
+    RESET_FORMAT,
+    COLOR_YELLOW,
+    COLOR_GREEN,
+    COLOR_RED,
+)
+from kiauh.utils.filesystem_utils import check_file_exist
 from kiauh.utils.logger import Logger
 from kiauh.utils.system_utils import check_package_install, install_system_packages
 
@@ -23,8 +34,8 @@ def get_current_date() -> Dict[Literal["date", "time"], str]:
     :return: Dict holding a date and time key:value pair
     """
     now: datetime = datetime.today()
-    date: str = now.strftime("%Y-%m-%d")
+    date: str = now.strftime("%Y%m%d")
-    time: str = now.strftime("%H-%M-%S")
+    time: str = now.strftime("%H%M%S")
 
     return {"date": date, "time": time}
 
@@ -43,3 +54,69 @@ def check_install_dependencies(deps: List[str]) -> None:
         for _ in requirements:
             print(f"{COLOR_CYAN}● {_}{RESET_FORMAT}")
         install_system_packages(requirements)
+
+
+def get_repo_name(repo_dir: str) -> str:
+    """
+    Helper method to extract the organisation and name of a repository |
+    :param repo_dir: repository to extract the values from
+    :return: String in form of "<orga>/<name>"
+    """
+    try:
+        cmd = ["git", "-C", repo_dir, "config", "--get", "remote.origin.url"]
+        result = subprocess.check_output(cmd, stderr=subprocess.DEVNULL)
+        result = "/".join(result.decode().strip().split("/")[-2:])
+        return f"{COLOR_CYAN}{result}{RESET_FORMAT}"
+    except subprocess.CalledProcessError:
+        return f"{COLOR_YELLOW}Unknown{RESET_FORMAT}"
+
+
+def get_install_status_common(
+    instance_type: Type[BaseInstance], repo_dir: str, env_dir: str
+) -> str:
+    """
+    Helper method to get the installation status of software components,
+    which only consist of 3 major parts and if those parts exist, the
+    component can be considered as "installed". Typically, Klipper or
+    Moonraker match that criteria.
+    :param instance_type: The component type
+    :param repo_dir: the repository directory
+    :param env_dir: the python environment directory
+    :return: formatted string, containing the status
+    """
+    im = InstanceManager(instance_type)
+    dir_exist = Path(repo_dir).exists()
+    env_dir_exist = Path(env_dir).exists()
+    instances_exist = len(im.instances) > 0
+    status = [dir_exist, env_dir_exist, instances_exist]
+    if all(status):
+        return f"{COLOR_GREEN}Installed: {len(im.instances)}{RESET_FORMAT}"
+    elif not any(status):
+        return f"{COLOR_RED}Not installed!{RESET_FORMAT}"
+    else:
+        return f"{COLOR_YELLOW}Incomplete!{RESET_FORMAT}"
+
+
+def get_install_status_webui(
+    install_dir: Path, nginx_cfg: Path, upstreams_cfg: Path, common_cfg: Path
+) -> str:
+    """
+    Helper method to get the installation status of webuis
+    like Mainsail or Fluidd |
+    :param install_dir: folder of the static webui files
+    :param nginx_cfg: the webuis NGINX config
+    :param upstreams_cfg: the required upstreams.conf
+    :param common_cfg: the required common_vars.conf
+    :return: formatted string, containing the status
+    """
+    dir_exist = Path(install_dir).exists()
+    nginx_cfg_exist = check_file_exist(nginx_cfg)
+    upstreams_cfg_exist = check_file_exist(upstreams_cfg)
+    common_cfg_exist = check_file_exist(common_cfg)
+    status = [dir_exist, nginx_cfg_exist, upstreams_cfg_exist, common_cfg_exist]
+    if all(status):
+        return f"{COLOR_GREEN}Installed!{RESET_FORMAT}"
+    elif not any(status):
+        return f"{COLOR_RED}Not installed!{RESET_FORMAT}"
+    else:
+        return f"{COLOR_YELLOW}Incomplete!{RESET_FORMAT}"