diff --git a/kiauh/components/webui_client/client_utils.py b/kiauh/components/webui_client/client_utils.py index b64b7fb..1548876 100644 --- a/kiauh/components/webui_client/client_utils.py +++ b/kiauh/components/webui_client/client_utils.py @@ -6,6 +6,7 @@ # # # This file may be distributed under the terms of the GNU GPLv3 license # # ======================================================================= # +from __future__ import annotations import json # noqa: I001 import shutil @@ -29,7 +30,7 @@ from utils.git_utils import ( get_latest_unstable_tag, ) from utils.logger import Logger -from utils.types import ComponentStatus, InstallStatus +from utils.types import ComponentStatus def get_client_status( @@ -40,16 +41,16 @@ def get_client_status( NGINX_CONFD.joinpath("upstreams.conf"), NGINX_CONFD.joinpath("common_vars.conf"), ] - status = get_install_status(client.client_dir, files=files) + comp_status: ComponentStatus = get_install_status(client.client_dir, files=files) # if the client dir does not exist, set the status to not # installed even if the other files are present if not client.client_dir.exists(): - status["status"] = InstallStatus.NOT_INSTALLED + comp_status.status = 0 - status["local"] = get_local_client_version(client) - status["remote"] = get_remote_client_version(client) if fetch_remote else None - return status + comp_status.local = get_local_client_version(client) + comp_status.remote = get_remote_client_version(client) if fetch_remote else None + return comp_status def get_client_config_status(client: BaseWebClient) -> ComponentStatus: @@ -125,12 +126,12 @@ def symlink_webui_nginx_log(klipper_instances: List[Klipper]) -> None: desti_error.symlink_to(error_log) -def get_local_client_version(client: BaseWebClient) -> str: +def get_local_client_version(client: BaseWebClient) -> str | None: relinfo_file = client.client_dir.joinpath("release_info.json") version_file = client.client_dir.joinpath(".version") if not client.client_dir.exists(): - return "-" + return None if not relinfo_file.is_file() and not version_file.is_file(): return "n/a" @@ -142,13 +143,13 @@ def get_local_client_version(client: BaseWebClient) -> str: return f.readlines()[0] -def get_remote_client_version(client: BaseWebClient) -> str: +def get_remote_client_version(client: BaseWebClient) -> str | None: try: if (tag := get_latest_tag(client.repo_path)) != "": return tag - return "ERROR" + return None except Exception: - return "ERROR" + return None def backup_client_data(client: BaseWebClient) -> None: diff --git a/kiauh/core/menus/main_menu.py b/kiauh/core/menus/main_menu.py index 2a6497b..4000457 100644 --- a/kiauh/core/menus/main_menu.py +++ b/kiauh/core/menus/main_menu.py @@ -41,7 +41,7 @@ from utils.constants import ( RESET_FORMAT, ) from utils.logger import Logger -from utils.types import ComponentStatus +from utils.types import ComponentStatus, StatusMap, StatusText # noinspection PyUnusedLocal @@ -84,6 +84,7 @@ class MainMenu(BaseMenu): ) def fetch_status(self) -> None: + pass self._get_component_status("kl", get_klipper_status) self._get_component_status("mr", get_moonraker_status) self._get_component_status("ms", get_client_status, MainsailData()) @@ -96,10 +97,10 @@ class MainMenu(BaseMenu): def _get_component_status(self, name: str, status_fn: callable, *args) -> None: status_data: ComponentStatus = status_fn(*args) - code: int = status_data.get("status").value.code - status: str = status_data.get("status").value.txt - repo: str = status_data.get("repo") - instance_count: int = status_data.get("instances") + code: int = status_data.status + status: StatusText = StatusMap[code] + repo: str = status_data.repo + instance_count: int = status_data.instances count_txt: str = "" if instance_count > 0 and code == 1: @@ -109,12 +110,15 @@ class MainMenu(BaseMenu): setattr(self, f"{name}_repo", f"{COLOR_CYAN}{repo}{RESET_FORMAT}") def _format_by_code(self, code: int, status: str, count: str) -> str: - if code == 1: - return f"{COLOR_GREEN}{status}{count}{RESET_FORMAT}" + color = COLOR_RED + if code == 0: + color = COLOR_RED + elif code == 1: + color = COLOR_YELLOW elif code == 2: - return f"{COLOR_RED}{status}{count}{RESET_FORMAT}" + color = COLOR_GREEN - return f"{COLOR_YELLOW}{status}{count}{RESET_FORMAT}" + return f"{color}{status}{count}{RESET_FORMAT}" def print_menu(self): self.fetch_status() diff --git a/kiauh/core/menus/update_menu.py b/kiauh/core/menus/update_menu.py index 5e4f3c8..5ea217e 100644 --- a/kiauh/core/menus/update_menu.py +++ b/kiauh/core/menus/update_menu.py @@ -47,6 +47,7 @@ from utils.constants import ( COLOR_YELLOW, RESET_FORMAT, ) +from utils.logger import Logger from utils.types import ComponentStatus @@ -57,14 +58,32 @@ class UpdateMenu(BaseMenu): super().__init__() self.previous_menu = previous_menu - self.kl_local = self.kl_remote = self.mr_local = self.mr_remote = "" - self.ms_local = self.ms_remote = self.fl_local = self.fl_remote = "" - self.mc_local = self.mc_remote = self.fc_local = self.fc_remote = "" - self.ks_local = self.ks_remote = self.mb_local = self.mb_remote = "" - self.cn_local = self.cn_remote = self.oe_local = self.oe_remote = "" + self.klipper_local = self.klipper_remote = None + self.moonraker_local = self.moonraker_remote = None + self.mainsail_local = self.mainsail_remote = None + self.mainsail_config_local = self.mainsail_config_remote = None + self.fluidd_local = self.fluidd_remote = None + self.fluidd_config_local = self.fluidd_config_remote = None + self.klipperscreen_local = self.klipperscreen_remote = None + self.mobileraker_local = self.mobileraker_remote = None + self.crowsnest_local = self.crowsnest_remote = None + self.octoeverywhere_local = self.octoeverywhere_remote = None self.mainsail_data = MainsailData() self.fluidd_data = FluiddData() + self.status_data = { + "klipper": {"installed": False, "local": None, "remote": None}, + "moonraker": {"installed": False, "local": None, "remote": None}, + "mainsail": {"installed": False, "local": None, "remote": None}, + "mainsail_config": {"installed": False, "local": None, "remote": None}, + "fluidd": {"installed": False, "local": None, "remote": None}, + "fluidd_config": {"installed": False, "local": None, "remote": None}, + "mobileraker": {"installed": False, "local": None, "remote": None}, + "klipperscreen": {"installed": False, "local": None, "remote": None}, + "crowsnest": {"installed": False, "local": None, "remote": None}, + "octoeverywhere": {"installed": False, "local": None, "remote": None}, + } + self._fetch_update_status() def set_previous_menu(self, previous_menu: Optional[Type[BaseMenu]]) -> None: @@ -76,7 +95,7 @@ class UpdateMenu(BaseMenu): def set_options(self) -> None: self.options = { - "0": Option(self.update_all, menu=False), + "a": Option(self.update_all, menu=False), "1": Option(self.update_klipper, menu=False), "2": Option(self.update_moonraker, menu=False), "3": Option(self.update_mainsail, menu=False), @@ -101,25 +120,25 @@ class UpdateMenu(BaseMenu): ╔═══════════════════════════════════════════════════════╗ ║ {color}{header:~^{count}}{RESET_FORMAT} ║ ╟───────────────────────┬───────────────┬───────────────╢ - ║ 0) Update all │ │ ║ + ║ a) Update all │ │ ║ ║ │ Current: │ Latest: ║ ║ Klipper & API: ├───────────────┼───────────────╢ - ║ 1) Klipper │ {self.kl_local:<22} │ {self.kl_remote:<22} ║ - ║ 2) Moonraker │ {self.mr_local:<22} │ {self.mr_remote:<22} ║ + ║ 1) Klipper │ {self.klipper_local:<22} │ {self.klipper_remote:<22} ║ + ║ 2) Moonraker │ {self.moonraker_local:<22} │ {self.moonraker_remote:<22} ║ ║ │ │ ║ ║ Webinterface: ├───────────────┼───────────────╢ - ║ 3) Mainsail │ {self.ms_local:<22} │ {self.ms_remote:<22} ║ - ║ 4) Fluidd │ {self.fl_local:<22} │ {self.fl_remote:<22} ║ + ║ 3) Mainsail │ {self.mainsail_local:<22} │ {self.mainsail_remote:<22} ║ + ║ 4) Fluidd │ {self.fluidd_local:<22} │ {self.fluidd_remote:<22} ║ ║ │ │ ║ ║ Client-Config: ├───────────────┼───────────────╢ - ║ 5) Mainsail-Config │ {self.mc_local:<22} │ {self.mc_remote:<22} ║ - ║ 6) Fluidd-Config │ {self.fc_local:<22} │ {self.fc_remote:<22} ║ + ║ 5) Mainsail-Config │ {self.mainsail_config_local:<22} │ {self.mainsail_config_remote:<22} ║ + ║ 6) Fluidd-Config │ {self.fluidd_config_local:<22} │ {self.fluidd_config_remote:<22} ║ ║ │ │ ║ ║ Other: ├───────────────┼───────────────╢ - ║ 7) KlipperScreen │ {self.ks_local:<22} │ {self.ks_remote:<22} ║ - ║ 8) Mobileraker │ {self.mb_local:<22} │ {self.mb_remote:<22} ║ - ║ 9) Crowsnest │ {self.cn_local:<22} │ {self.cn_remote:<22} ║ - ║ 10) OctoEverywhere │ {self.oe_local:<22} │ {self.oe_remote:<22} ║ + ║ 7) KlipperScreen │ {self.klipperscreen_local:<22} │ {self.klipperscreen_remote:<22} ║ + ║ 8) Mobileraker │ {self.mobileraker_local:<22} │ {self.mobileraker_remote:<22} ║ + ║ 9) Crowsnest │ {self.crowsnest_local:<22} │ {self.crowsnest_remote:<22} ║ + ║ 10) OctoEverywhere │ {self.octoeverywhere_local:<22} │ {self.octoeverywhere_remote:<22} ║ ║ ├───────────────┴───────────────╢ ║ 11) System │ ║ ╟───────────────────────┴───────────────────────────────╢ @@ -131,68 +150,96 @@ class UpdateMenu(BaseMenu): print("update_all") def update_klipper(self, **kwargs): - update_klipper() + if self._check_is_installed("klipper"): + update_klipper() def update_moonraker(self, **kwargs): - update_moonraker() + if self._check_is_installed("moonraker"): + update_moonraker() def update_mainsail(self, **kwargs): - update_client(self.mainsail_data) + if self._check_is_installed("mainsail"): + update_client(self.mainsail_data) def update_mainsail_config(self, **kwargs): - update_client_config(self.mainsail_data) + if self._check_is_installed("mainsail_config"): + update_client_config(self.mainsail_data) def update_fluidd(self, **kwargs): - update_client(self.fluidd_data) + if self._check_is_installed("fluidd"): + update_client(self.fluidd_data) def update_fluidd_config(self, **kwargs): - update_client_config(self.fluidd_data) + if self._check_is_installed("fluidd_config"): + update_client_config(self.fluidd_data) def update_klipperscreen(self, **kwargs): - update_klipperscreen() + if self._check_is_installed("klipperscreen"): + update_klipperscreen() def update_mobileraker(self, **kwargs): - update_mobileraker() + if self._check_is_installed("mobileraker"): + update_mobileraker() def update_crowsnest(self, **kwargs): - update_crowsnest() + if self._check_is_installed("crowsnest"): + update_crowsnest() def update_octoeverywhere(self, **kwargs): - update_octoeverywhere() + if self._check_is_installed("octoeverywhere"): + update_octoeverywhere() def upgrade_system_packages(self, **kwargs): ... def _fetch_update_status(self): - # klipper - self._get_update_status("kl", get_klipper_status) - # moonraker - self._get_update_status("mr", get_moonraker_status) - # mainsail - self._get_update_status("ms", get_client_status, self.mainsail_data, True) - # mainsail-config - self._get_update_status("mc", get_client_config_status, self.mainsail_data) - # fluidd - self._get_update_status("fl", get_client_status, self.fluidd_data, True) - # fluidd-config - self._get_update_status("fc", get_client_config_status, self.fluidd_data) - # klipperscreen - self._get_update_status("ks", get_klipperscreen_status) - # mobileraker - self._get_update_status("mb", get_mobileraker_status) - # crowsnest - self._get_update_status("cn", get_crowsnest_status) - # octoeverywhere - self._get_update_status("oe", get_octoeverywhere_status) + self._set_status_data("klipper", get_klipper_status) + self._set_status_data("moonraker", get_moonraker_status) + self._set_status_data("mainsail", get_client_status, self.mainsail_data, True) + self._set_status_data( + "mainsail_config", get_client_config_status, self.mainsail_data + ) + self._set_status_data("fluidd", get_client_status, self.fluidd_data, True) + self._set_status_data( + "fluidd_config", get_client_config_status, self.fluidd_data + ) + self._set_status_data("klipperscreen", get_klipperscreen_status) + self._set_status_data("mobileraker", get_mobileraker_status) + self._set_status_data("crowsnest", get_crowsnest_status) + self._set_status_data("octoeverywhere", get_octoeverywhere_status) def _format_local_status(self, local_version, remote_version) -> str: - if local_version == remote_version: - return f"{COLOR_GREEN}{local_version}{RESET_FORMAT}" - return f"{COLOR_YELLOW}{local_version}{RESET_FORMAT}" + color = COLOR_RED + if not local_version: + color = COLOR_RED + elif local_version == remote_version: + color = COLOR_GREEN + elif local_version != remote_version: + color = COLOR_YELLOW - def _get_update_status(self, name: str, status_fn: callable, *args) -> None: - status_data: ComponentStatus = status_fn(*args) - local_ver = status_data.get("local") - remote_ver = status_data.get("remote") - color = COLOR_GREEN if remote_ver != "ERROR" else COLOR_RED - setattr(self, f"{name}_local", self._format_local_status(local_ver, remote_ver)) - setattr(self, f"{name}_remote", f"{color}{remote_ver}{RESET_FORMAT}") + return f"{color}{local_version or '-'}{RESET_FORMAT}" + + def _set_status_data(self, name: str, status_fn: callable, *args) -> None: + comp_status: ComponentStatus = status_fn(*args) + + self.status_data[name]["installed"] = True if comp_status.status == 2 else False + self.status_data[name]["local"] = comp_status.local + self.status_data[name]["remote"] = comp_status.remote + + self._set_status_string(name) + + def _set_status_string(self, name: str) -> None: + local_status = self.status_data[name].get("local", None) + remote_status = self.status_data[name].get("remote", None) + + color = COLOR_GREEN if remote_status else COLOR_RED + local_txt = self._format_local_status(local_status, remote_status) + remote_txt = f"{color}{remote_status or '-'}{RESET_FORMAT}" + + setattr(self, f"{name}_local", local_txt) + setattr(self, f"{name}_remote", remote_txt) + + def _check_is_installed(self, name: str) -> bool: + if not self.status_data[name]["installed"]: + Logger.print_info(f"{name.capitalize()} is not installed! Skipped ...") + return False + return True diff --git a/kiauh/utils/common.py b/kiauh/utils/common.py index 98d982b..c38464e 100644 --- a/kiauh/utils/common.py +++ b/kiauh/utils/common.py @@ -26,7 +26,7 @@ from utils.sys_utils import ( install_system_packages, update_system_package_lists, ) -from utils.types import ComponentStatus, InstallStatus +from utils.types import ComponentStatus, StatusCode def convert_camelcase_to_kebabcase(name: str) -> str: @@ -92,13 +92,11 @@ def get_install_status( checks.append(f.exists()) if all(checks): - status = InstallStatus.INSTALLED - + status: StatusCode = 2 # installed elif not any(checks): - status = InstallStatus.NOT_INSTALLED - + status: StatusCode = 0 # not installed else: - status = InstallStatus.INCOMPLETE + status: StatusCode = 1 # incomplete return ComponentStatus( status=status, diff --git a/kiauh/utils/git_utils.py b/kiauh/utils/git_utils.py index 98dead9..8952887 100644 --- a/kiauh/utils/git_utils.py +++ b/kiauh/utils/git_utils.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import json import shutil import urllib.request @@ -63,11 +65,11 @@ def git_pull_wrapper(repo: str, target_dir: Path) -> None: return -def get_repo_name(repo: Path) -> str: +def get_repo_name(repo: Path) -> str | None: """ Helper method to extract the organisation and name of a repository | :param repo: repository to extract the values from - :return: String in form of "/" + :return: String in form of "/" or None """ if not repo.exists() or not repo.joinpath(".git").exists(): return "-" @@ -77,7 +79,7 @@ def get_repo_name(repo: Path) -> str: result = check_output(cmd, stderr=DEVNULL) return "/".join(result.decode().strip().split("/")[-2:]) except CalledProcessError: - return "-" + return None def get_tags(repo_path: str) -> List[str]: @@ -110,7 +112,6 @@ def get_latest_tag(repo_path: str) -> str: else: return "" except Exception: - Logger.print_error("Error while getting the latest tag") raise @@ -130,20 +131,20 @@ def get_latest_unstable_tag(repo_path: str) -> str: raise -def get_local_commit(repo: Path) -> str: +def get_local_commit(repo: Path) -> str | None: if not repo.exists() or not repo.joinpath(".git").exists(): - return "-" + return None try: cmd = f"cd {repo} && git describe HEAD --always --tags | cut -d '-' -f 1,2" return check_output(cmd, shell=True, text=True).strip() except CalledProcessError: - return "-" + return None -def get_remote_commit(repo: Path) -> str: +def get_remote_commit(repo: Path) -> str | None: if not repo.exists() or not repo.joinpath(".git").exists(): - return "-" + return None try: # get locally checked out branch @@ -153,7 +154,7 @@ def get_remote_commit(repo: Path) -> str: cmd = f"cd {repo} && git describe 'origin/{branch}' --always --tags | cut -d '-' -f 1,2" return check_output(cmd, shell=True, text=True).strip() except CalledProcessError: - return "-" + return None def git_cmd_clone(repo: str, target_dir: Path) -> None: diff --git a/kiauh/utils/types.py b/kiauh/utils/types.py index c49a1c9..274ef9d 100644 --- a/kiauh/utils/types.py +++ b/kiauh/utils/types.py @@ -6,25 +6,24 @@ # # # This file may be distributed under the terms of the GNU GPLv3 license # # ======================================================================= # -from enum import Enum -from typing import Optional, TypedDict +from __future__ import annotations + +from dataclasses import dataclass +from typing import Dict, Literal + +StatusText = Literal["Installed", "Not installed", "Incomplete"] +StatusCode = Literal[0, 1, 2] +StatusMap: Dict[StatusCode, StatusText] = { + 0: "Not installed", + 1: "Incomplete", + 2: "Installed", +} -class StatusInfo: - def __init__(self, txt: str, code: int): - self.txt: str = txt - self.code: int = code - - -class InstallStatus(Enum): - INSTALLED = StatusInfo("Installed", 1) - NOT_INSTALLED = StatusInfo("Not installed", 2) - INCOMPLETE = StatusInfo("Incomplete", 3) - - -class ComponentStatus(TypedDict): - status: InstallStatus - repo: Optional[str] - local: Optional[str] - remote: Optional[str] - instances: Optional[int] +@dataclass +class ComponentStatus: + status: StatusCode + repo: str | None = None + local: str | None = None + remote: str | None = None + instances: int | None = None