jeka/kiauh
1
0
Fork 0
mirror of https://github.com/dw-0/kiauh.git synced 2025-12-25 00:33:37 +05:00
Code Issues Packages Projects Releases Wiki Activity

refactor: rework update menu, logic and typing

Browse Source 
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2024-06-30 13:45:07 +02:00
parent 8a620cdbd4
commit 956666605c
6 changed files with 163 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
kiauh/components/webui_client/client_utils.py
View File

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

22
kiauh/core/menus/main_menu.py
View File

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

161
kiauh/core/menus/update_menu.py
View File

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

10
kiauh/utils/common.py
View File

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

21
kiauh/utils/git_utils.py
View File

@@ -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 "<orga>/<name>"
:return: String in form of "<orga>/<name>" 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:

39
kiauh/utils/types.py
View File

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