diff --git a/kiauh/core/menus/update_menu.py b/kiauh/core/menus/update_menu.py index 8b25075..24396ec 100644 --- a/kiauh/core/menus/update_menu.py +++ b/kiauh/core/menus/update_menu.py @@ -9,7 +9,7 @@ from __future__ import annotations import textwrap -from typing import Callable, Type +from typing import Callable, List, Type from components.crowsnest.crowsnest import get_crowsnest_status, update_crowsnest from components.klipper.klipper_setup import update_klipper @@ -48,8 +48,14 @@ from utils.constants import ( COLOR_YELLOW, RESET_FORMAT, ) -from utils.logger import Logger +from utils.input_utils import get_confirm +from utils.logger import DialogType, Logger from utils.spinner import Spinner +from utils.sys_utils import ( + get_upgradable_packages, + update_system_package_lists, + upgrade_system_packages, +) from utils.types import ComponentStatus @@ -60,6 +66,9 @@ class UpdateMenu(BaseMenu): super().__init__() self.previous_menu: Type[BaseMenu] | None = previous_menu + self.packages: List[str] = [] + self.package_count: int = 0 + self.klipper_local = self.klipper_remote = "" self.moonraker_local = self.moonraker_remote = "" self.mainsail_local = self.mainsail_remote = "" @@ -108,7 +117,7 @@ class UpdateMenu(BaseMenu): } def print_menu(self) -> None: - spinner = Spinner() + spinner = Spinner("Loading update menu, please wait") spinner.start() self._fetch_update_status() @@ -118,6 +127,15 @@ class UpdateMenu(BaseMenu): header = " [ Update Menu ] " color = COLOR_GREEN count = 62 - len(color) - len(RESET_FORMAT) + + sysupgrades: str = "No upgrades available." + padding = 29 + if self.package_count > 0: + sysupgrades = ( + f"{COLOR_GREEN}{self.package_count} upgrades available!{RESET_FORMAT}" + ) + padding = 38 + menu = textwrap.dedent( f""" ╔═══════════════════════════════════════════════════════╗ @@ -143,7 +161,7 @@ class UpdateMenu(BaseMenu): ║ 9) Crowsnest │ {self.crowsnest_local:<22} │ {self.crowsnest_remote:<22} ║ ║ 10) OctoEverywhere │ {self.octoeverywhere_local:<22} │ {self.octoeverywhere_remote:<22} ║ ║ ├───────────────┴───────────────╢ - ║ 11) System │ ║ + ║ 11) System │ {sysupgrades:^{padding}} ║ ╟───────────────────────┴───────────────────────────────╢ """ )[1:] @@ -192,7 +210,8 @@ class UpdateMenu(BaseMenu): if self._check_is_installed("octoeverywhere"): update_octoeverywhere() - def upgrade_system_packages(self, **kwargs) -> None: ... + def upgrade_system_packages(self, **kwargs) -> None: + self._run_system_updates() def _fetch_update_status(self) -> None: self._set_status_data("klipper", get_klipper_status) @@ -210,6 +229,10 @@ class UpdateMenu(BaseMenu): self._set_status_data("crowsnest", get_crowsnest_status) self._set_status_data("octoeverywhere", get_octoeverywhere_status) + update_system_package_lists(silent=True) + self.packages = get_upgradable_packages() + self.package_count = len(self.packages) + def _format_local_status(self, local_version, remote_version) -> str: color = COLOR_RED if not local_version: @@ -246,3 +269,25 @@ class UpdateMenu(BaseMenu): Logger.print_info(f"{name.capitalize()} is not installed! Skipped ...") return False return True + + def _run_system_updates(self) -> None: + if not self.packages: + Logger.print_info("No system upgrades available!") + return + + try: + pkgs: str = ", ".join(self.packages) + Logger.print_dialog( + DialogType.CUSTOM, + ["The following packages will be upgraded:", "\n\n", pkgs], + custom_title="UPGRADABLE SYSTEM UPDATES", + padding_top=0, + padding_bottom=0, + ) + if not get_confirm("Continue?"): + return + Logger.print_status("Upgrading system packages ...") + upgrade_system_packages(self.packages) + except Exception as e: + Logger.print_error(f"Error upgrading system packages:\n{e}") + raise diff --git a/kiauh/utils/sys_utils.py b/kiauh/utils/sys_utils.py index 40a7b87..561f7ad 100644 --- a/kiauh/utils/sys_utils.py +++ b/kiauh/utils/sys_utils.py @@ -18,7 +18,7 @@ import time import urllib.error import urllib.request from pathlib import Path -from subprocess import DEVNULL, PIPE, CalledProcessError, Popen, run +from subprocess import DEVNULL, PIPE, CalledProcessError, Popen, check_output, run from typing import List, Literal, Set from utils.constants import SYSTEMD @@ -219,6 +219,25 @@ def update_system_package_lists(silent: bool, rls_info_change=False) -> None: raise +def get_upgradable_packages() -> List[str]: + """ + Reads all system packages that can be upgraded. + :return: A list of package names available for upgrade + """ + try: + command = ["apt", "list", "--upgradable"] + output: str = check_output(command, stderr=DEVNULL, text=True, encoding="utf-8") + pkglist = [] + for line in output.split("\n"): + if "/" not in line: + continue + pkg = line.split("/")[0] + pkglist.append(pkg) + return pkglist + except CalledProcessError as e: + raise Exception(f"Error reading upgradable packages: {e}") + + def check_package_install(packages: Set[str]) -> List[str]: """ Checks the system for installed packages | @@ -252,12 +271,29 @@ def install_system_packages(packages: List[str]) -> None: command.append(pkg) run(command, stderr=PIPE, check=True) - Logger.print_ok("Packages installed successfully.") + Logger.print_ok("Packages successfully installed.") except CalledProcessError as e: Logger.print_error(f"Error installing packages:\n{e.stderr.decode()}") raise +def upgrade_system_packages(packages: List[str]) -> None: + """ + Updates a list of system packages | + :param packages: List of system package names + :return: None + """ + try: + command = ["sudo", "apt-get", "upgrade", "-y"] + for pkg in packages: + command.append(pkg) + run(command, stderr=PIPE, check=True) + + Logger.print_ok("Packages successfully upgraded.") + except CalledProcessError as e: + raise Exception(f"Error upgrading packages:\n{e.stderr.decode()}") + + # this feels hacky and not quite right, but for now it works # see: https://stackoverflow.com/questions/166506/finding-local-ip-addresses-using-pythons-stdlib def get_ipv4_addr() -> str: