516 lines
17 KiB
Python
516 lines
17 KiB
Python
import gi
|
|
import os
|
|
from ui.MenuButton import MenuButton
|
|
from ui.Stack import Stack
|
|
from ui.KernelStack import KernelStack
|
|
from ui.FlowBox import FlowBox, FlowBoxInstalled
|
|
from ui.AboutDialog import AboutDialog
|
|
from ui.SplashScreen import SplashScreen
|
|
from ui.MessageWindow import MessageWindow
|
|
from ui.SettingsWindow import SettingsWindow
|
|
import libs.functions as fn
|
|
|
|
gi.require_version("Gtk", "4.0")
|
|
from gi.repository import Gtk, Gio, Gdk, GLib
|
|
|
|
|
|
base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
|
|
class ManagerGUI(Gtk.ApplicationWindow):
|
|
def __init__(self, app_name, default_context, app_version, **kwargs):
|
|
super().__init__(**kwargs)
|
|
|
|
self.default_context = default_context
|
|
|
|
self.app_version = app_version
|
|
|
|
if self.app_version == "${app_version}":
|
|
self.app_version = "dev"
|
|
|
|
fn.logger.info("Version = %s" % self.app_version)
|
|
|
|
self.set_title(app_name)
|
|
self.set_resizable(True)
|
|
self.set_default_size(950, 650)
|
|
|
|
# get list of kernels from the arch archive website, aur, then cache
|
|
self.official_kernels = []
|
|
self.community_kernels = []
|
|
|
|
# splashscreen queue for threading
|
|
self.queue_load_progress = fn.Queue()
|
|
|
|
# official kernels queue for threading
|
|
self.queue_kernels = fn.Queue()
|
|
|
|
# community kernels queue for threading
|
|
self.queue_community_kernels = fn.Queue()
|
|
|
|
hbox_notify_revealer = Gtk.Box(
|
|
orientation=Gtk.Orientation.HORIZONTAL, spacing=20
|
|
)
|
|
hbox_notify_revealer.set_name("hbox_notify_revealer")
|
|
hbox_notify_revealer.set_halign(Gtk.Align.CENTER)
|
|
|
|
self.notify_revealer = Gtk.Revealer()
|
|
self.notify_revealer.set_reveal_child(False)
|
|
self.label_notify_revealer = Gtk.Label(xalign=0, yalign=0)
|
|
self.label_notify_revealer.set_name("label_notify_revealer")
|
|
|
|
self.notify_revealer.set_child(hbox_notify_revealer)
|
|
|
|
hbox_notify_revealer.append(self.label_notify_revealer)
|
|
|
|
self.splash_screen = SplashScreen(app_name)
|
|
|
|
try:
|
|
fn.Thread(
|
|
target=self.wait_for_gui_load,
|
|
daemon=True,
|
|
).start()
|
|
except Exception as e:
|
|
fn.logger.error(e)
|
|
|
|
while self.default_context.pending():
|
|
fn.time.sleep(0.1)
|
|
self.default_context.iteration(True)
|
|
|
|
self.bootloader = None
|
|
self.bootloader_grub_cfg = None
|
|
|
|
# self.bootloader = fn.get_boot_loader()
|
|
|
|
config_data = fn.setup_config(self)
|
|
|
|
if "bootloader" in config_data.keys():
|
|
if config_data["bootloader"]["name"] is not None:
|
|
self.bootloader = config_data["bootloader"]["name"].lower()
|
|
|
|
if self.bootloader == "grub":
|
|
if config_data["bootloader"]["grub_config"] is not None:
|
|
self.bootloader_grub_cfg = config_data["bootloader"][
|
|
"grub_config"
|
|
]
|
|
elif self.bootloader != "systemd-boot" or self.bootloader != "grub":
|
|
fn.logger.warning(
|
|
"Invalid bootloader config found it should only be systemd-boot or grub"
|
|
)
|
|
|
|
fn.logger.warning("Using bootctl to determine current bootloader")
|
|
self.bootloader = None
|
|
|
|
if self.bootloader is not None or self.bootloader_grub_cfg is not None:
|
|
fn.logger.info("User provided bootloader options read from config file")
|
|
fn.logger.info("User bootloader option = %s " % self.bootloader)
|
|
if self.bootloader_grub_cfg is not None:
|
|
fn.logger.info(
|
|
"User bootloader Grub config = %s " % self.bootloader_grub_cfg
|
|
)
|
|
else:
|
|
# no config setting found for bootloader use default method
|
|
self.bootloader = fn.get_boot_loader()
|
|
if self.bootloader == "grub":
|
|
self.bootloader_grub_cfg = "/boot/grub/grub.cfg"
|
|
|
|
if self.bootloader is not None:
|
|
fn.create_cache_dir()
|
|
fn.create_log_dir()
|
|
fn.get_pacman_repos()
|
|
|
|
self.stack = Stack(transition_type="OVER_DOWN")
|
|
self.kernel_stack = KernelStack(self)
|
|
|
|
header_bar = Gtk.HeaderBar()
|
|
|
|
label_title = Gtk.Label(xalign=0.5, yalign=0.5)
|
|
label_title.set_markup("<b>%s</b>" % app_name)
|
|
|
|
header_bar.set_title_widget(label_title)
|
|
header_bar.set_show_title_buttons(True)
|
|
|
|
self.set_titlebar(header_bar)
|
|
|
|
menu_outerbox = Gtk.Box(spacing=6, orientation=Gtk.Orientation.VERTICAL)
|
|
header_bar.pack_end(menu_outerbox)
|
|
|
|
menu_outerbox.show()
|
|
|
|
menubutton = MenuButton()
|
|
|
|
menu_outerbox.append(menubutton)
|
|
|
|
menubutton.show()
|
|
|
|
action_about = Gio.SimpleAction(name="about")
|
|
action_about.connect("activate", self.on_about)
|
|
|
|
action_settings = Gio.SimpleAction(name="settings")
|
|
action_settings.connect("activate", self.on_settings, fn)
|
|
|
|
self.add_action(action_settings)
|
|
|
|
self.add_action(action_about)
|
|
|
|
action_refresh = Gio.SimpleAction(name="refresh")
|
|
action_refresh.connect("activate", self.on_refresh)
|
|
|
|
self.add_action(action_refresh)
|
|
|
|
action_quit = Gio.SimpleAction(name="quit")
|
|
action_quit.connect("activate", self.on_quit)
|
|
|
|
self.add_action(action_quit)
|
|
|
|
# add shortcut keys
|
|
|
|
event_controller_key = Gtk.EventControllerKey.new()
|
|
event_controller_key.connect("key-pressed", self.key_pressed)
|
|
|
|
self.add_controller(event_controller_key)
|
|
|
|
# overlay = Gtk.Overlay()
|
|
# self.set_child(child=overlay)
|
|
|
|
self.vbox = Gtk.Box.new(orientation=Gtk.Orientation.VERTICAL, spacing=10)
|
|
self.vbox.set_name("main")
|
|
|
|
self.set_child(child=self.vbox)
|
|
|
|
self.vbox.append(self.notify_revealer)
|
|
|
|
self.installed_kernels = fn.get_installed_kernels()
|
|
|
|
self.active_kernel = fn.get_active_kernel()
|
|
|
|
fn.logger.info("Installed kernels = %s" % len(self.installed_kernels))
|
|
|
|
self.refresh_cache = False
|
|
|
|
self.refresh_cache = fn.get_latest_kernel_updates(self)
|
|
|
|
self.start_get_kernels_threads()
|
|
|
|
self.load_kernels_gui()
|
|
|
|
# validate bootloader
|
|
if self.bootloader_grub_cfg and not os.path.exists(
|
|
self.bootloader_grub_cfg
|
|
):
|
|
mw = MessageWindow(
|
|
title="Grub config file not found",
|
|
message=f"The specified Grub config file: {self.bootloader_grub_cfg} does not exist\n"
|
|
f"This will cause an issue when updating the bootloader\n"
|
|
f"Update the configuration file/use the Advanced Settings to change this\n",
|
|
image_path="images/48x48/akm-error.png",
|
|
detailed_message=False,
|
|
transient_for=self,
|
|
)
|
|
|
|
mw.present()
|
|
if self.bootloader == "systemd-boot":
|
|
if not os.path.exists(
|
|
"/sys/firmware/efi/fw_platform_size"
|
|
) or not os.path.exists("/sys/firmware/efi/efivars"):
|
|
mw = MessageWindow(
|
|
title="Legacy boot detected",
|
|
message=f"Cannot select systemd-boot, UEFI boot mode is not available\n"
|
|
f"Update the configuration file\n"
|
|
f"Or use the Advanced Settings to change this\n",
|
|
image_path="images/48x48/akm-warning.png",
|
|
detailed_message=False,
|
|
transient_for=self,
|
|
)
|
|
|
|
mw.present()
|
|
|
|
else:
|
|
fn.logger.error("Failed to set bootloader, application closing")
|
|
fn.sys.exit(1)
|
|
|
|
def key_pressed(self, keyval, keycode, state, userdata):
|
|
shortcut = Gtk.accelerator_get_label(
|
|
keycode, keyval.get_current_event().get_modifier_state()
|
|
)
|
|
|
|
# quit application
|
|
if shortcut in ("Ctrl+Q", "Ctrl+Mod2+Q"):
|
|
self.destroy()
|
|
|
|
def open_settings(self, fn):
|
|
settings_win = SettingsWindow(fn, self)
|
|
settings_win.present()
|
|
|
|
def timeout(self):
|
|
self.hide_notify()
|
|
|
|
def hide_notify(self):
|
|
self.notify_revealer.set_reveal_child(False)
|
|
if self.timeout_id is not None:
|
|
GLib.source_remove(self.timeout_id)
|
|
self.timeout_id = None
|
|
|
|
def reveal_notify(self):
|
|
# reveal = self.notify_revealer.get_reveal_child()
|
|
self.notify_revealer.set_reveal_child(True)
|
|
self.timeout_id = GLib.timeout_add(3000, self.timeout)
|
|
|
|
def start_get_kernels_threads(self):
|
|
if self.refresh_cache is False:
|
|
fn.logger.info("Starting get official Linux kernels thread")
|
|
try:
|
|
fn.Thread(
|
|
name=fn.thread_get_kernels,
|
|
target=fn.get_official_kernels,
|
|
daemon=True,
|
|
args=(self,),
|
|
).start()
|
|
|
|
except Exception as e:
|
|
fn.logger.error("Exception in thread fn.get_official_kernels(): %s" % e)
|
|
finally:
|
|
self.official_kernels = self.queue_kernels.get()
|
|
self.queue_kernels.task_done()
|
|
|
|
else:
|
|
self.official_kernels = self.queue_kernels.get()
|
|
self.queue_kernels.task_done()
|
|
|
|
fn.logger.info("Starting pacman db synchronization thread")
|
|
self.queue_load_progress.put("Starting pacman db synchronization")
|
|
|
|
self.pacman_db_sync()
|
|
|
|
fn.logger.info("Starting get community kernels thread")
|
|
self.queue_load_progress.put("Getting community based Linux kernels")
|
|
|
|
try:
|
|
thread_get_community_kernels = fn.Thread(
|
|
name=fn.thread_get_community_kernels,
|
|
target=fn.get_community_kernels,
|
|
daemon=True,
|
|
args=(self,),
|
|
)
|
|
|
|
thread_get_community_kernels.start()
|
|
|
|
except Exception as e:
|
|
fn.logger.error("Exception in thread_get_community_kernels: %s" % e)
|
|
finally:
|
|
self.community_kernels = self.queue_community_kernels.get()
|
|
self.queue_community_kernels.task_done()
|
|
|
|
# =====================================================
|
|
# PACMAN DB SYNC
|
|
# =====================================================
|
|
|
|
def pacman_db_sync(self):
|
|
sync_err = fn.sync_package_db()
|
|
|
|
if sync_err is not None:
|
|
fn.logger.error("Pacman db synchronization failed")
|
|
|
|
print(
|
|
"---------------------------------------------------------------------------"
|
|
)
|
|
|
|
GLib.idle_add(
|
|
self.show_sync_db_message_dialog,
|
|
sync_err,
|
|
priority=GLib.PRIORITY_DEFAULT,
|
|
)
|
|
|
|
else:
|
|
fn.logger.info("Pacman DB synchronization completed")
|
|
|
|
def show_sync_db_message_dialog(self, sync_err):
|
|
mw = MessageWindow(
|
|
title="Error - Pacman db synchronization",
|
|
message=f"Pacman db synchronization failed\n"
|
|
f"Failed to run 'pacman -Syu'\n"
|
|
f"{sync_err}\n",
|
|
image_path="images/48x48/akm-warning.png",
|
|
transient_for=self,
|
|
detailed_message=True,
|
|
)
|
|
|
|
mw.present()
|
|
|
|
# keep splash screen open, until main gui is loaded
|
|
def wait_for_gui_load(self):
|
|
while True:
|
|
fn.time.sleep(0.2)
|
|
status = self.queue_load_progress.get()
|
|
if status == 1:
|
|
GLib.idle_add(
|
|
self.splash_screen.destroy,
|
|
priority=GLib.PRIORITY_DEFAULT,
|
|
)
|
|
break
|
|
|
|
def on_settings(self, action, param, fn):
|
|
self.open_settings(fn)
|
|
|
|
def on_about(self, action, param):
|
|
about_dialog = AboutDialog(self)
|
|
about_dialog.present()
|
|
|
|
def on_refresh(self, action, param):
|
|
if not fn.is_thread_alive(fn.thread_refresh_ui):
|
|
fn.Thread(
|
|
name=fn.thread_refresh_ui,
|
|
target=self.refresh_ui,
|
|
daemon=True,
|
|
).start()
|
|
|
|
def refresh_ui(self):
|
|
fn.logger.debug("Refreshing UI")
|
|
|
|
self.label_notify_revealer.set_text("Refreshing UI started")
|
|
GLib.idle_add(
|
|
self.reveal_notify,
|
|
priority=GLib.PRIORITY_DEFAULT,
|
|
)
|
|
fn.pacman_repos_list = []
|
|
fn.get_pacman_repos()
|
|
|
|
fn.cached_kernels_list = []
|
|
fn.community_kernels_list = []
|
|
|
|
self.official_kernels = None
|
|
|
|
self.community_kernels = None
|
|
|
|
self.installed_kernels = None
|
|
|
|
self.start_get_kernels_threads()
|
|
|
|
self.installed_kernels = fn.get_installed_kernels()
|
|
|
|
self.label_notify_revealer.set_text("Refreshing official kernels")
|
|
GLib.idle_add(
|
|
self.reveal_notify,
|
|
priority=GLib.PRIORITY_DEFAULT,
|
|
)
|
|
|
|
GLib.idle_add(
|
|
self.kernel_stack.add_official_kernels_to_stack,
|
|
True,
|
|
priority=GLib.PRIORITY_DEFAULT,
|
|
)
|
|
|
|
self.label_notify_revealer.set_text("Refreshing community kernels")
|
|
GLib.idle_add(
|
|
self.reveal_notify,
|
|
priority=GLib.PRIORITY_DEFAULT,
|
|
)
|
|
|
|
GLib.idle_add(
|
|
self.kernel_stack.add_community_kernels_to_stack,
|
|
True,
|
|
priority=GLib.PRIORITY_DEFAULT,
|
|
)
|
|
|
|
self.label_notify_revealer.set_text("Refreshing installed kernels")
|
|
GLib.idle_add(
|
|
self.reveal_notify,
|
|
priority=GLib.PRIORITY_DEFAULT,
|
|
)
|
|
|
|
GLib.idle_add(
|
|
self.kernel_stack.add_installed_kernels_to_stack,
|
|
True,
|
|
priority=GLib.PRIORITY_DEFAULT,
|
|
)
|
|
|
|
while self.default_context.pending():
|
|
fn.time.sleep(0.3)
|
|
self.default_context.iteration(False)
|
|
|
|
# fn.time.sleep(0.5)
|
|
|
|
fn.logger.debug("Refresh UI completed")
|
|
|
|
self.label_notify_revealer.set_text("Refreshing UI completed")
|
|
GLib.idle_add(
|
|
self.reveal_notify,
|
|
priority=GLib.PRIORITY_DEFAULT,
|
|
)
|
|
|
|
def on_quit(self, action, param):
|
|
self.destroy()
|
|
fn.logger.info("Application quit")
|
|
|
|
def on_button_quit_response(self, widget):
|
|
self.destroy()
|
|
fn.logger.info("Application quit")
|
|
|
|
def load_kernels_gui(self):
|
|
hbox_sep = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
|
|
hsep = Gtk.Separator(orientation=Gtk.Orientation.VERTICAL)
|
|
hbox_sep.append(hsep)
|
|
|
|
# handle error here with message
|
|
if self.official_kernels is None:
|
|
fn.logger.error("Failed to retrieve kernel list")
|
|
|
|
stack_sidebar = Gtk.StackSidebar()
|
|
stack_sidebar.set_name("stack_sidebar")
|
|
stack_sidebar.set_stack(self.stack)
|
|
|
|
hbox_stack_sidebar = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10)
|
|
hbox_stack_sidebar.set_name("hbox_stack_sidebar")
|
|
hbox_stack_sidebar.append(stack_sidebar)
|
|
hbox_stack_sidebar.append(self.stack)
|
|
|
|
self.vbox.append(hbox_stack_sidebar)
|
|
|
|
button_quit = Gtk.Button.new_with_label("Quit")
|
|
# button_quit.set_size_request(100, 30)
|
|
button_quit.connect(
|
|
"clicked",
|
|
self.on_button_quit_response,
|
|
)
|
|
|
|
btn_context = button_quit.get_style_context()
|
|
btn_context.add_class("destructive-action")
|
|
|
|
grid_bottom_panel = Gtk.Grid()
|
|
grid_bottom_panel.set_halign(Gtk.Align.END)
|
|
grid_bottom_panel.set_row_homogeneous(True)
|
|
|
|
grid_bottom_panel.attach(button_quit, 0, 1, 1, 1)
|
|
|
|
self.vbox.append(grid_bottom_panel)
|
|
|
|
self.textbuffer = Gtk.TextBuffer()
|
|
|
|
self.textview = Gtk.TextView()
|
|
self.textview.set_property("editable", False)
|
|
self.textview.set_property("monospace", True)
|
|
|
|
self.textview.set_vexpand(True)
|
|
self.textview.set_hexpand(True)
|
|
|
|
self.textview.set_buffer(self.textbuffer)
|
|
|
|
fn.logger.info("Creating kernel UI")
|
|
|
|
# add official kernel flowbox
|
|
|
|
fn.logger.debug("Adding official kernels to UI")
|
|
self.kernel_stack.add_official_kernels_to_stack(reload=False)
|
|
|
|
fn.logger.debug("Adding community kernels to UI")
|
|
self.kernel_stack.add_community_kernels_to_stack(reload=False)
|
|
|
|
fn.logger.debug("Adding installed kernels to UI")
|
|
self.kernel_stack.add_installed_kernels_to_stack(reload=False)
|
|
|
|
while self.default_context.pending():
|
|
self.default_context.iteration(True)
|
|
|
|
fn.time.sleep(0.3)
|
|
|
|
self.queue_load_progress.put(1)
|
|
fn.logger.info("Kernel manager UI loaded")
|