Python script, ASCII text executable
1
"""
2
Izvor - modular, GTK+ desktop search and launcher
3
Copyright (C) 2024 <root@roundabout-host.com>
4
This program is free software: you can redistribute it and/or modify
6
it under the terms of the GNU General Public License as published by
7
the Free Software Foundation, either version 3 of the License, or
8
(at your option) any later version.
9
This program is distributed in the hope that it will be useful,
11
but WITHOUT ANY WARRANTY; without even the implied warranty of
12
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13
GNU General Public License for more details.
14
You should have received a copy of the GNU General Public License
16
along with this program. If not, see <https://www.gnu.org/licenses/>.
17
"""
18
import os
20
import gi
21
import asyncio
22
import importlib
23
import gbulb
24
from pathlib import Path
25
from typing import AsyncIterable, AsyncIterator, Collection, TypeVar, Iterable, Callable, Any
26
gi.require_version("Gtk", "3.0")
28
from gi.repository import Gtk, Gdk, Pango, GLib
29
_T = TypeVar("_T")
32
ICON_SIZE = 32
34
async def merge_generators_with_params(generator_funcs: list[Callable[..., AsyncIterable]],
37
*args, **kwargs):
38
iterators = [gen_func(*args, **kwargs).__aiter__() for gen_func in generator_funcs]
39
tasks = {asyncio.create_task(it.__anext__()): it for it in iterators}
40
while tasks:
42
done, _ = await asyncio.wait(tasks.keys(), return_when=asyncio.FIRST_COMPLETED)
43
for task in done:
45
iterator = tasks.pop(task)
46
try:
47
result = task.result()
48
yield result
49
tasks[asyncio.create_task(iterator.__anext__())] = iterator
50
except StopAsyncIteration:
51
pass
52
class ResultInfoWidget(Gtk.ListBoxRow):
55
def __init__(self, name: str, description: str, image: tuple[str, Any], execute: Callable[[], None]):
56
super().__init__()
57
self.name = name
58
self.description = description
59
self.image = image
60
self.execute = execute
61
self.box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
63
self.add(self.box)
64
if image[0] == "logo":
66
self.image = Gtk.Image.new_from_icon_name(self.image[1], Gtk.IconSize.LARGE_TOOLBAR)
67
self.image.set_pixel_size(ICON_SIZE)
68
self.box.pack_start(self.image, False, False, 0)
69
self.label_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
71
self.box.pack_start(self.label_box, True, True, 0)
72
self.name_label = Gtk.Label(label=self.name)
74
self.name_label.set_line_wrap(True)
75
self.name_label.set_justify(Gtk.Justification.LEFT)
76
self.name_label.set_halign(Gtk.Align.START)
77
attributes = Pango.AttrList()
78
attributes.insert(Pango.AttrSize.new(16 * Pango.SCALE))
79
self.name_label.set_attributes(attributes)
80
self.label_box.pack_start(self.name_label, True, True, 0)
81
self.description_label = Gtk.Label(label=self.description)
83
self.description_label.set_line_wrap(True)
84
self.description_label.set_justify(Gtk.Justification.LEFT)
85
self.description_label.set_halign(Gtk.Align.START)
86
self.label_box.pack_start(self.description_label, True, True, 0)
87
class Izvor(Gtk.Application):
90
USER_CONFIGS = Path(os.getenv("XDG_CONFIG_HOME", Path.home() / ".config")) / "izvor"
91
USER_DATA = Path(os.getenv("XDG_DATA_HOME", Path.home() / ".local" / "share")) / "izvor"
92
USER_PROVIDERS = USER_DATA / "providers"
93
USER_PROVIDER_CONFIGS = USER_CONFIGS / "providers"
94
SYSTEM_CONFIGS = Path("/etc/izvor")
95
SYSTEM_DATA = Path("/usr/share/izvor")
96
SYSTEM_PROVIDERS = SYSTEM_DATA / "providers"
97
SYSTEM_PROVIDER_CONFIGS = SYSTEM_CONFIGS / "providers"
98
ENABLED_PROVIDERS_FILE = USER_CONFIGS / "enabled_providers"
99
def __init__(self):
101
super().__init__(application_id="com.roundabout-host.roundabout.izvor")
102
gbulb.install(gtk=True)
103
self.loop = asyncio.get_event_loop()
104
self.builder = Gtk.Builder()
105
self.builder.add_from_file("izvor.ui")
106
self.window = self.builder.get_object("root")
107
self.window.connect("destroy", self.kill)
108
self.window.connect("key-press-event", self.check_escape)
109
self.window.set_title("Izvor")
110
if not os.getenv("XDG_DATA_HOME"):
112
print("XDG_DATA_HOME is not set. Using default path.")
113
if not os.getenv("XDG_CONFIG_HOME"):
114
print("XDG_CONFIG_HOME is not set. Using default path.")
115
if not self.USER_PROVIDER_CONFIGS.exists():
116
print("Creating user config directory.")
117
self.USER_PROVIDER_CONFIGS.mkdir(parents=True)
118
if not self.USER_PROVIDERS.exists():
119
print("Creating user data directory.")
120
self.USER_PROVIDERS.mkdir(parents=True)
121
self.available_providers = {}
123
if self.SYSTEM_PROVIDERS.exists():
125
for provider in self.SYSTEM_PROVIDERS.iterdir():
126
if not (provider / "__init__.py").exists():
127
continue
128
self.available_providers[provider.stem] = provider
129
if self.USER_PROVIDERS.exists():
130
for provider in self.USER_PROVIDERS.iterdir():
131
if not (provider / "__init__.py").exists():
132
continue
133
self.available_providers[provider.stem] = provider
134
print(f"Available providers: {list(self.available_providers.keys())}")
136
if not self.ENABLED_PROVIDERS_FILE.exists():
138
self.ENABLED_PROVIDERS_FILE.touch()
139
with open(self.ENABLED_PROVIDERS_FILE, "w") as f:
141
for provider in self.available_providers:
142
f.write(provider + "\n")
143
self.permanently_enabled_providers = {}
145
with open(self.ENABLED_PROVIDERS_FILE, "r") as f:
147
for line in f:
148
provider = line.strip()
149
if provider in self.available_providers:
150
self.permanently_enabled_providers[provider] = self.available_providers[provider]
151
self.enabled_providers = self.permanently_enabled_providers.copy()
153
providers_menu = self.builder.get_object("providers-menu")
155
self.providers = []
157
self.provider_checkboxes = {}
158
for provider in self.permanently_enabled_providers.values():
160
spec = importlib.util.spec_from_file_location(provider.stem, provider / "__init__.py")
161
module = importlib.util.module_from_spec(spec)
162
spec.loader.exec_module(module)
163
loaded_provider = module.Provider({})
164
self.providers.append(loaded_provider)
165
print(f"Loaded provider: {loaded_provider.name}")
166
print(loaded_provider.description)
167
self.provider_checkboxes[provider.stem] = Gtk.CheckMenuItem.new_with_label(loaded_provider.name)
168
providers_menu.append(self.provider_checkboxes[provider.stem])
169
self.provider_checkboxes[provider.stem].set_active(True)
170
self.provider_checkboxes[provider.stem].show()
171
self.provider_checkboxes[provider.stem].connect("toggled", self.update_enabled_providers)
172
def call_update_results(widget):
174
asyncio.create_task(self.update_results(widget))
175
def execute_result(widget, row):
177
row.execute()
178
self.kill()
179
self.builder.connect_signals(
181
{
182
"update-search": call_update_results,
183
"result-activated": execute_result,
184
"providers-menu-all-toggled": self.update_enabled_providers_all,
185
"menu-about": self.about,
186
"menu-providers": self.provider_menu,
187
}
188
)
189
GLib.idle_add(self.update_enabled_providers, None)
191
def update_enabled_providers(self, widget=None):
193
for provider, checkbox in self.provider_checkboxes.items():
194
if checkbox.get_active():
195
self.enabled_providers[provider] = self.permanently_enabled_providers[provider]
196
else:
197
self.enabled_providers.pop(provider, None)
198
asyncio.create_task(self.update_results(None))
200
def update_enabled_providers_all(self, widget):
202
for checkbox in self.provider_checkboxes.values():
203
checkbox.set_active(widget.get_active())
204
self.update_enabled_providers()
205
def kill(self, widget=None):
207
self.loop.stop()
208
self.quit()
209
def check_escape(self, widget, event):
211
results_list = self.builder.get_object("results-list")
212
search_entry = self.builder.get_object("search-query")
213
rows = results_list.get_children()
214
current_row = results_list.get_selected_row()
215
if event.keyval == Gdk.KEY_Escape:
217
self.kill()
218
return True
219
if event.keyval == Gdk.KEY_Down:
221
# Move to the next row
222
if current_row:
223
current_index = rows.index(current_row)
224
next_index = (current_index + 1) % len(rows) # Wrap to the beginning if at the end
225
else:
226
next_index = 0
227
results_list.select_row(rows[next_index])
229
search_entry.grab_focus() # Refocus the search entry
230
self.scroll_to_row(rows[next_index], results_list.get_adjustment())
231
return True
232
if event.keyval == Gdk.KEY_Up:
234
# Move to the previous row
235
if current_row:
236
current_index = rows.index(current_row)
237
prev_index = (current_index - 1) % len(rows) # Wrap to the end if at the beginning
238
else:
239
prev_index = len(rows) - 1
240
results_list.select_row(rows[prev_index])
242
search_entry.grab_focus() # Refocus the search entry
243
self.scroll_to_row(rows[prev_index], results_list.get_adjustment())
244
return True
245
if event.keyval == Gdk.KEY_Return and current_row:
247
# Execute the selected row
248
current_row.activate()
249
return True
250
return False
252
def scroll_to_row(self, row, adjustment):
254
row_geometry = row.get_allocation()
255
row_y = row_geometry.y
256
row_height = row_geometry.height
257
adjustment_value = adjustment.get_value()
258
adjustment_upper = adjustment.get_upper() - adjustment.get_page_size()
259
if row_y + row_height > adjustment_value + adjustment.get_page_size():
261
adjustment.set_value(min(row_y + row_height - adjustment.get_page_size(), adjustment_upper))
262
elif row_y < adjustment_value:
263
adjustment.set_value(max(row_y, 0))
264
def about(self, widget):
266
about_builder = Gtk.Builder()
267
about_builder.add_from_file("about.ui")
268
about_window = about_builder.get_object("about-dialog")
269
about_window.connect("destroy", lambda _: about_window.destroy())
270
about_window.connect("close", lambda _: about_window.destroy())
271
about_window.connect("response", lambda _, __: about_window.destroy())
272
about_window.show_all()
273
def provider_menu(self, widget):
275
providers_builder = Gtk.Builder()
276
providers_builder.add_from_file("providers.ui")
277
providers_window = providers_builder.get_object("providers-window")
278
providers_window.connect("destroy", lambda _: providers_window.destroy())
279
providers_table = providers_builder.get_object("providers-table") # actually Gtk.Grid
281
rows = 0
282
for provider in self.available_providers.values():
283
spec = importlib.util.spec_from_file_location(provider.stem, provider / "__init__.py")
284
module = importlib.util.module_from_spec(spec)
285
spec.loader.exec_module(module)
286
loaded_provider = module.Provider({})
287
provider_label = Gtk.Label(label=loaded_provider.name)
289
provider_label.set_halign(Gtk.Align.START)
290
provider_label.set_valign(Gtk.Align.CENTER)
291
font_desc = Pango.FontDescription()
292
font_desc.set_weight(Pango.Weight.BOLD)
293
provider_label.override_font(font_desc)
294
provider_label.set_line_wrap(True)
295
provider_label.set_justify(Gtk.Justification.LEFT)
296
provider_description = Gtk.Label(label=loaded_provider.description)
298
provider_description.set_halign(Gtk.Align.START)
299
provider_description.set_valign(Gtk.Align.CENTER)
300
provider_description.set_justify(Gtk.Justification.LEFT)
301
provider_description.set_line_wrap(True)
302
icon = Gtk.Image.new_from_icon_name(loaded_provider.icon, Gtk.IconSize.LARGE_TOOLBAR)
304
icon.set_pixel_size(ICON_SIZE)
305
switch = Gtk.Switch()
307
switch.set_valign(Gtk.Align.CENTER)
308
switch.set_active(provider.stem in self.permanently_enabled_providers)
309
switch.connect("notify::active", self.update_permanently_enabled_providers, provider.stem)
310
providers_table.attach(icon, 0, rows, 1, 1)
312
providers_table.attach(provider_label, 1, rows, 1, 1)
313
providers_table.attach(provider_description, 2, rows, 1, 1)
314
providers_table.attach(switch, 3, rows, 1, 1)
315
provider_label.show()
317
provider_description.show()
318
rows += 1
320
providers_window.show_all()
322
def update_permanently_enabled_providers(self, widget, param, provider):
325
if widget.get_active():
326
self.permanently_enabled_providers[provider] = self.available_providers[provider]
327
else:
328
self.permanently_enabled_providers.pop(provider, None)
329
with open(self.ENABLED_PROVIDERS_FILE, "w") as f:
331
f.write("\n".join(self.permanently_enabled_providers.keys()))
332
async def update_results(self, widget):
335
print("Updating results...")
336
generators = []
337
query = self.builder.get_object("search-query-buffer").get_text()
338
# print(f"Query: {query}")
339
for provider in self.enabled_providers.values():
340
spec = importlib.util.spec_from_file_location(provider.stem, provider / "__init__.py")
341
module = importlib.util.module_from_spec(spec)
342
spec.loader.exec_module(module)
343
loaded_provider = module.Provider({})
344
# print(f"Searching with provider: {loaded_provider.name}")
345
generators.append(loaded_provider.search)
347
# Clear the results list
349
results_list = self.builder.get_object("results-list")
350
for row in results_list.get_children():
351
results_list.remove(row)
352
async for result in merge_generators_with_params(generators, query):
354
# print(result)
355
result_box = ResultInfoWidget(result["name"], result["description"], result["image"], result["execute"])
356
results_list.add(result_box)
357
result_box.show_all()
358
results_list.show_all()
360
if __name__ == "__main__":
363
izvor = Izvor()
364
izvor.window.show_all()
365
try:
366
izvor.loop.run_forever()
367
finally:
368
izvor.loop.close()
369