""" Izvor - modular, GTK+ desktop search and launcher Copyright (C) 2024 This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program. If not, see . """ import os import gi import asyncio import importlib import gbulb from pathlib import Path from typing import AsyncIterable, AsyncIterator, Collection, TypeVar, Iterable, Callable, Any from twisted.conch.insults.text import attributes gi.require_version("Gtk", "3.0") from gi.repository import Gtk, Gdk, Pango, GLib _T = TypeVar("_T") ICON_SIZE = 32 async def merge_generators_with_params(generator_funcs: list[Callable[..., AsyncIterable]], *args, **kwargs): iterators = [gen_func(*args, **kwargs).__aiter__() for gen_func in generator_funcs] tasks = {asyncio.create_task(it.__anext__()): it for it in iterators} while tasks: done, _ = await asyncio.wait(tasks.keys(), return_when=asyncio.FIRST_COMPLETED) for task in done: iterator = tasks.pop(task) try: result = task.result() yield result tasks[asyncio.create_task(iterator.__anext__())] = iterator except StopAsyncIteration: pass class ResultInfoWidget(Gtk.ListBoxRow): def __init__(self, name: str, description: str, image: tuple[str, Any], execute: Callable[[], None]): super().__init__() self.name = name self.description = description self.image = image self.execute = execute self.box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6) self.add(self.box) if image[0] == "logo": self.image = Gtk.Image.new_from_icon_name(self.image[1], Gtk.IconSize.LARGE_TOOLBAR) self.image.set_pixel_size(ICON_SIZE) self.box.pack_start(self.image, False, False, 0) self.label_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) self.box.pack_start(self.label_box, True, True, 0) self.name_label = Gtk.Label(label=self.name) self.name_label.set_line_wrap(True) self.name_label.set_justify(Gtk.Justification.LEFT) self.name_label.set_halign(Gtk.Align.START) attributes = Pango.AttrList() attributes.insert(Pango.AttrSize.new(16 * Pango.SCALE)) self.name_label.set_attributes(attributes) self.label_box.pack_start(self.name_label, True, True, 0) self.description_label = Gtk.Label(label=self.description) self.description_label.set_line_wrap(True) self.description_label.set_justify(Gtk.Justification.LEFT) self.description_label.set_halign(Gtk.Align.START) self.label_box.pack_start(self.description_label, True, True, 0) class Izvor(Gtk.Application): USER_CONFIGS = Path(os.getenv("XDG_CONFIG_HOME", Path.home() / ".config")) / "izvor" USER_DATA = Path(os.getenv("XDG_DATA_HOME", Path.home() / ".local" / "share")) / "izvor" USER_PROVIDERS = USER_DATA / "providers" USER_PROVIDER_CONFIGS = USER_CONFIGS / "providers" SYSTEM_CONFIGS = Path("/etc/izvor") SYSTEM_DATA = Path("/usr/share/izvor") SYSTEM_PROVIDERS = SYSTEM_DATA / "providers" SYSTEM_PROVIDER_CONFIGS = SYSTEM_CONFIGS / "providers" ENABLED_PROVIDERS_FILE = USER_CONFIGS / "enabled_providers" def __init__(self): super().__init__(application_id="com.roundabout-host.roundabout.izvor") gbulb.install(gtk=True) self.loop = asyncio.get_event_loop() self.builder = Gtk.Builder() self.builder.add_from_file("izvor.ui") self.window = self.builder.get_object("root") self.window.connect("destroy", self.kill) self.window.connect("key-press-event", self.check_escape) self.window.set_title("Izvor") if not os.getenv("XDG_DATA_HOME"): print("XDG_DATA_HOME is not set. Using default path.") if not os.getenv("XDG_CONFIG_HOME"): print("XDG_CONFIG_HOME is not set. Using default path.") if not self.USER_PROVIDER_CONFIGS.exists(): print("Creating user config directory.") self.USER_PROVIDER_CONFIGS.mkdir(parents=True) if not self.USER_PROVIDERS.exists(): print("Creating user data directory.") self.USER_PROVIDERS.mkdir(parents=True) self.available_providers = {} if self.SYSTEM_PROVIDERS.exists(): for provider in self.SYSTEM_PROVIDERS.iterdir(): if not (provider / "__init__.py").exists(): continue self.available_providers[provider.stem] = provider if self.USER_PROVIDERS.exists(): for provider in self.USER_PROVIDERS.iterdir(): if not (provider / "__init__.py").exists(): continue self.available_providers[provider.stem] = provider print(f"Available providers: {list(self.available_providers.keys())}") if not self.ENABLED_PROVIDERS_FILE.exists(): self.ENABLED_PROVIDERS_FILE.touch() with open(self.ENABLED_PROVIDERS_FILE, "w") as f: for provider in self.available_providers: f.write(provider + "\n") self.permanently_enabled_providers = {} with open(self.ENABLED_PROVIDERS_FILE, "r") as f: for line in f: provider = line.strip() if provider in self.available_providers: self.permanently_enabled_providers[provider] = self.available_providers[provider] self.enabled_providers = self.permanently_enabled_providers.copy() providers_menu = self.builder.get_object("providers-menu") self.providers = [] self.provider_checkboxes = {} for provider in self.permanently_enabled_providers.values(): spec = importlib.util.spec_from_file_location(provider.stem, provider / "__init__.py") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) loaded_provider = module.Provider({}) self.providers.append(loaded_provider) print(f"Loaded provider: {loaded_provider.name}") print(loaded_provider.description) self.provider_checkboxes[provider.stem] = Gtk.CheckMenuItem.new_with_label(loaded_provider.name) providers_menu.append(self.provider_checkboxes[provider.stem]) self.provider_checkboxes[provider.stem].set_active(True) self.provider_checkboxes[provider.stem].show() self.provider_checkboxes[provider.stem].connect("toggled", self.update_enabled_providers) def call_update_results(widget): asyncio.create_task(self.update_results(widget)) def execute_result(widget, row): row.execute() self.kill() self.builder.connect_signals( { "update-search": call_update_results, "result-activated": execute_result, "providers-menu-all-toggled": self.update_enabled_providers_all, "menu-about": self.about, "menu-providers": self.provider_menu, } ) GLib.idle_add(self.update_enabled_providers, None) def update_enabled_providers(self, widget=None): for provider, checkbox in self.provider_checkboxes.items(): if checkbox.get_active(): self.enabled_providers[provider] = self.permanently_enabled_providers[provider] else: self.enabled_providers.pop(provider, None) asyncio.create_task(self.update_results(None)) def update_enabled_providers_all(self, widget): for checkbox in self.provider_checkboxes.values(): checkbox.set_active(widget.get_active()) self.update_enabled_providers() def kill(self, widget=None): self.loop.stop() self.quit() def check_escape(self, widget, event): if event.keyval == 65307: self.kill() def about(self, widget): about_builder = Gtk.Builder() about_builder.add_from_file("about.ui") about_window = about_builder.get_object("about-dialog") about_window.connect("destroy", lambda _: about_window.destroy()) about_window.connect("close", lambda _: about_window.destroy()) about_window.connect("response", lambda _, __: about_window.destroy()) about_window.show_all() def provider_menu(self, widget): providers_builder = Gtk.Builder() providers_builder.add_from_file("providers.ui") providers_window = providers_builder.get_object("providers-window") providers_window.connect("destroy", lambda _: providers_window.destroy()) providers_table = providers_builder.get_object("providers-table") # actually Gtk.Grid rows = 0 for provider in self.available_providers.values(): spec = importlib.util.spec_from_file_location(provider.stem, provider / "__init__.py") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) loaded_provider = module.Provider({}) provider_label = Gtk.Label(label=loaded_provider.name) provider_label.set_halign(Gtk.Align.START) provider_label.set_valign(Gtk.Align.CENTER) font_desc = Pango.FontDescription() font_desc.set_weight(Pango.Weight.BOLD) provider_label.override_font(font_desc) provider_label.set_line_wrap(True) provider_label.set_justify(Gtk.Justification.LEFT) provider_description = Gtk.Label(label=loaded_provider.description) provider_description.set_halign(Gtk.Align.START) provider_description.set_valign(Gtk.Align.CENTER) provider_description.set_justify(Gtk.Justification.LEFT) provider_description.set_line_wrap(True) icon = Gtk.Image.new_from_icon_name(loaded_provider.icon, Gtk.IconSize.LARGE_TOOLBAR) icon.set_pixel_size(ICON_SIZE) switch = Gtk.Switch() switch.set_valign(Gtk.Align.CENTER) switch.set_active(provider.stem in self.permanently_enabled_providers) switch.connect("notify::active", self.update_permanently_enabled_providers, provider.stem) providers_table.attach(icon, 0, rows, 1, 1) providers_table.attach(provider_label, 1, rows, 1, 1) providers_table.attach(provider_description, 2, rows, 1, 1) providers_table.attach(switch, 3, rows, 1, 1) provider_label.show() provider_description.show() rows += 1 providers_window.show_all() def update_permanently_enabled_providers(self, widget, param, provider): if widget.get_active(): self.permanently_enabled_providers[provider] = self.available_providers[provider] else: self.permanently_enabled_providers.pop(provider, None) with open(self.ENABLED_PROVIDERS_FILE, "w") as f: f.write("\n".join(self.permanently_enabled_providers.keys())) async def update_results(self, widget): print("Updating results...") generators = [] query = self.builder.get_object("search-query-buffer").get_text() # print(f"Query: {query}") for provider in self.enabled_providers.values(): spec = importlib.util.spec_from_file_location(provider.stem, provider / "__init__.py") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) loaded_provider = module.Provider({}) # print(f"Searching with provider: {loaded_provider.name}") generators.append(loaded_provider.search) # Clear the results list results_list = self.builder.get_object("results-list") for row in results_list.get_children(): results_list.remove(row) async for result in merge_generators_with_params(generators, query): # print(result) result_box = ResultInfoWidget(result["name"], result["description"], result["image"], result["execute"]) results_list.add(result_box) result_box.show_all() results_list.show_all() if __name__ == "__main__": izvor = Izvor() izvor.window.show_all() try: izvor.loop.run_forever() finally: izvor.loop.close()