perf(extend): extends the functionalities

snigdhaos-kernel-manager.py

@@ -1,3 +1,115 @@
-#!/usr/bin/python
+#!/usr/bin/env python3
+
+import os
+import gi
+import signal
+import functions as fn
+
+from ui.ManagerGUI import ManagerGUI
+
+gi.require_version("Gtk", "4.0")
+
+from gi.repository import Gtk, Gio, GLib, Gdk
+
+base_dir = fn.os.path.dirname(fn.os.path.realpath(__file__))
 
 
+app_name = "Snigdha OS Kernel Manager"
+app_version = "${app_version}"
+app_name_dir = "snigdhaos-kernel-manager"
+app_id = "org.snigdhaos.kernelmanager"
+lock_file = "/tmp/sokm.lock"
+pid_file = "/tmp/sokm.pid"
+
+
+class Main(Gtk.Application):
+    def __init__(self):
+        super().__init__(application_id=app_id, flags=Gio.ApplicationFlags.FLAGS_NONE)
+
+    def do_activate(self):
+        default_context = GLib.MainContext.default()
+
+        win = self.props.active_window
+
+        if not win:
+            win = ManagerGUI(
+                application=self,
+                app_name=app_name,
+                default_context=default_context,
+                app_version=app_version,
+            )
+
+        display = Gtk.Widget.get_display(win)
+
+        # sourced from /usr/share/icons/hicolor/scalable/apps
+        win.set_icon_name("archlinux-kernel-manager-tux")
+        provider = Gtk.CssProvider.new()
+        css_file = Gio.file_new_for_path(base_dir + "/snigdhaos-kernel-manager.css")
+        provider.load_from_file(css_file)
+        Gtk.StyleContext.add_provider_for_display(
+            display, provider, Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION
+        )
+
+        win.present()
+
+    def do_startup(self):
+        Gtk.Application.do_startup(self)
+
+    def do_shutdown(self):
+        Gtk.Application.do_shutdown(self)
+        if os.path.exists(lock_file):
+            os.remove(lock_file)
+        if os.path.exists(pid_file):
+            os.remove(pid_file)
+
+
+def signal_handler(sig, frame):
+    Gtk.main_quit(0)
+
+
+# These should be kept as it ensures that multiple installation instances can't be run concurrently.
+if __name__ == "__main__":
+    try:
+        # signal.signal(signal.SIGINT, signal_handler)
+
+        if not fn.os.path.isfile(lock_file):
+            with open(pid_file, "w") as f:
+                f.write(str(fn.os.getpid()))
+
+            # splash = SplashScreen()
+
+            app = Main()
+            app.run(None)
+        else:
+            md = Gtk.MessageDialog(
+                parent=Main(),
+                flags=0,
+                message_type=Gtk.MessageType.INFO,
+                buttons=Gtk.ButtonsType.YES_NO,
+                text="%s Lock File Found" % app_name,
+            )
+            md.format_secondary_markup(
+                "A %s lock file has been found. This indicates there is already an instance of <b>%s</b> running.\n\
+                Click 'Yes' to remove the lock file and try running again"
+                % (lock_file, app_name)
+            )  # noqa
+
+            result = md.run()
+            md.destroy()
+
+            if result in (Gtk.ResponseType.OK, Gtk.ResponseType.YES):
+                pid = ""
+                if fn.os.path.exists(pid_file):
+                    with open(pid_file, "r") as f:
+                        line = f.read()
+                        pid = line.rstrip().lstrip()
+
+                else:
+                    # in the rare event that the lock file is present, but the pid isn't
+                    fn.os.unlink(lock_file)
+                    fn.sys.exit(1)
+            else:
+                fn.sys.exit(1)
+    except Exception as e:
+        # fn.logger.error("Exception in __main__: %s" % e)
+        print("Exception in __main__: %s" % e)

ui/ManagerGUI.py

@@ -0,0 +1,516 @@
+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")