import hashlib import json import os from PIL import Image from fabric.widgets.box import Box from fabric.widgets.centerbox import CenterBox from fabric.widgets.entry import Entry from fabric.widgets.scrolledwindow import ScrolledWindow from gi.repository import GdkPixbuf, GLib, Gtk, Gio, Gdk from fabric.utils import get_relative_path import concurrent.futures from concurrent.futures import ThreadPoolExecutor from snippets import WALLPAPERS_DIR class WallpaperSelector(Box): CACHE_DIR = os.path.expanduser("~/.cache/modus/wallpapers") SETTINGS_FILE = get_relative_path("../../../config/assets/settings.json") SCHEMES = { "TonalSpot": "tonalSpot", "Expressive": "expressive", "FruitSalad": "fruitSalad", "Monochrome": "monochrome", "Rainbow": "rainbow", "Vibrant": "vibrant", "Neutral": "neutral", "Fidelity": "fidelity", "Content": "content", } def __init__(self, **kwargs): super().__init__( name="wallpapers", spacing=4, orientation="v", h_expand=False, v_expand=False, **kwargs, ) self.launcher = kwargs["launcher"] os.makedirs(self.CACHE_DIR, exist_ok=True) self.files = [f for f in os.listdir(WALLPAPERS_DIR) if self._is_image(f)] self.thumbnails = [] self.thumbnail_queue = [] self.executor = ThreadPoolExecutor(max_workers=4) # Shared executor self.selected_index = -1 # IconView for displaying thumbnails. self.viewport = Gtk.IconView(name="wallpaper-icons") self.viewport.set_model(Gtk.ListStore(GdkPixbuf.Pixbuf, str)) self.viewport.set_pixbuf_column(0) self.viewport.set_text_column(-1) self.viewport.set_item_width(0) self.viewport.connect("item-activated", self.on_wallpaper_selected) self.scrolled_window = ScrolledWindow( name="scrolled-window", spacing=10, h_expand=True, v_expand=True, child=self.viewport, ) self.search_entry = Entry( name="search-entry-walls", h_expand=True, notify_text=lambda entry, *_: self.arrange_viewport(entry.get_text()), on_key_press_event=self.on_search_entry_key_press, ) self.search_entry.connect("focus-out-event", self.on_search_entry_focus_out) self.scheme_dropdown = Gtk.ComboBoxText() self.scheme_dropdown.set_name("scheme-dropdown") self.scheme_dropdown.set_tooltip_text("Select color scheme") for display_name, scheme_id in sorted(self.SCHEMES.items()): self.scheme_dropdown.append(scheme_id, display_name) self.scheme_dropdown.set_active_id("tonalSpot") self.scheme_dropdown.connect("changed", self.on_scheme_changed) self.dropdown_box = Box( name="dropdown-box", children=[ self.scheme_dropdown, ], ) self.header_box = CenterBox( name="header-box", spacing=10, orientation="h", start_children=[self.search_entry], end_children=[self.dropdown_box], ) self.add(self.header_box) self.add(self.scrolled_window) self._start_thumbnail_thread() self.setup_file_monitor() self.show_all() self.search_entry.grab_focus() def setup_file_monitor(self): gfile = Gio.File.new_for_path(WALLPAPERS_DIR) self.file_monitor = gfile.monitor_directory(Gio.FileMonitorFlags.NONE, None) self.file_monitor.connect("changed", self.on_directory_changed) def on_directory_changed(self, monitor, file, other_file, event_type): file_name = file.get_basename() if event_type == Gio.FileMonitorEvent.DELETED: if file_name in self.files: self.files.remove(file_name) cache_path = self._get_cache_path(file_name) if os.path.exists(cache_path): try: os.remove(cache_path) except Exception as e: print(f"Error deleting cache {cache_path}: {e}") self.thumbnails = [(p, n) for p, n in self.thumbnails if n != file_name] GLib.idle_add(self.arrange_viewport, self.search_entry.get_text()) elif event_type == Gio.FileMonitorEvent.CREATED: if self._is_image(file_name) and file_name not in self.files: self.files.append(file_name) self.files.sort() self.executor.submit(self._process_file, file_name) elif event_type == Gio.FileMonitorEvent.CHANGED: if self._is_image(file_name) and file_name in self.files: cache_path = self._get_cache_path(file_name) if os.path.exists(cache_path): try: os.remove(cache_path) except Exception as e: print(f"Error deleting cache for changed file {file_name}: {e}") self.executor.submit(self._process_file, file_name) def arrange_viewport(self, query: str = ""): model = self.viewport.get_model() model.clear() filtered_thumbnails = [ (thumb, name) for thumb, name in self.thumbnails if query.casefold() in name.casefold() ] filtered_thumbnails.sort(key=lambda x: x[1].lower()) for pixbuf, file_name in filtered_thumbnails: model.append([pixbuf, file_name]) if query.strip() == "": self.viewport.unselect_all() self.selected_index = -1 elif len(model) > 0: self.update_selection(0) def on_wallpaper_selected(self, iconview, path): model = iconview.get_model() file_name = model[path][1] full_path = os.path.join(WALLPAPERS_DIR, file_name) selected_scheme = self.scheme_dropdown.get_active_id() wallpaper = get_relative_path("../../../config/scripts/wallpaper.py") command = f"python -O {wallpaper} -I {full_path}" GLib.spawn_command_line_async(command) self.update_scheme(selected_scheme) def on_scheme_changed(self, combo): scheme_id = combo.get_active_id() display_name = next( name for name, id in self.SCHEMES.items() if id == scheme_id ) print(f"Color scheme selected: {display_name} ({scheme_id})") self.update_scheme(scheme_id) color_generator = get_relative_path( "../../../config/material-colors/generate.py" ) try: command = f'python -O {color_generator} -R --scheme "{scheme_id}"' GLib.spawn_command_line_async(command) print(f"Applied color scheme: {display_name}") except Exception as e: print(f"Failed to apply color scheme: {e}") def update_scheme(self, scheme: str): try: with open(self.SETTINGS_FILE, "r") as f: settings = json.loads(f.read()) settings["generation-scheme"] = scheme with open(self.SETTINGS_FILE, "w") as f: json.dump(settings, f, indent=2) except Exception as error: print(f"Failed to update generation-scheme in settings.json: {error}") def on_search_entry_key_press(self, widget, event): if event.state & Gdk.ModifierType.SHIFT_MASK: if event.keyval in (Gdk.KEY_Up, Gdk.KEY_Down): schemes_list = list(self.SCHEMES.keys()) current_id = self.scheme_dropdown.get_active_id() current_index = ( schemes_list.index(current_id) if current_id in schemes_list else 0 ) if event.keyval == Gdk.KEY_Up: new_index = (current_index - 1) % len(schemes_list) else: new_index = (current_index + 1) % len(schemes_list) self.scheme_dropdown.set_active(new_index) return True elif event.keyval == Gdk.KEY_Right: self.scheme_dropdown.popup() return True if event.keyval in (Gdk.KEY_Up, Gdk.KEY_Down, Gdk.KEY_Left, Gdk.KEY_Right): self.move_selection_2d(event.keyval) return True elif event.keyval in (Gdk.KEY_Return, Gdk.KEY_KP_Enter): if self.selected_index != -1: path = Gtk.TreePath.new_from_indices([self.selected_index]) self.on_wallpaper_selected(self.viewport, path) return True return False def move_selection_2d(self, keyval): model = self.viewport.get_model() total_items = len(model) if total_items == 0: return if self.selected_index == -1: new_index = ( 0 if keyval in (Gdk.KEY_Down, Gdk.KEY_Right) else total_items - 1 ) else: current_index = self.selected_index # Calculate columns based on the allocation width. allocation = self.viewport.get_allocation() item_width = 108 columns = max(1, allocation.width // item_width) if keyval == Gdk.KEY_Right: new_index = current_index + 1 elif keyval == Gdk.KEY_Left: new_index = current_index - 1 elif keyval == Gdk.KEY_Down: new_index = current_index + columns elif keyval == Gdk.KEY_Up: new_index = current_index - columns # Ensure new_index remains within limits. if new_index < 0: new_index = 0 if new_index >= total_items: new_index = total_items - 1 self.update_selection(new_index) def update_selection(self, new_index: int): self.viewport.unselect_all() path = Gtk.TreePath.new_from_indices([new_index]) self.viewport.select_path(path) self.viewport.scroll_to_path(path, False, 0.5, 0.5) self.selected_index = new_index def _start_thumbnail_thread(self): thread = GLib.Thread.new("thumbnail-loader", self._preload_thumbnails, None) def _preload_thumbnails(self, _data): futures = [ self.executor.submit(self._process_file, file_name) for file_name in self.files ] concurrent.futures.wait(futures) GLib.idle_add(self._process_batch) def _process_file(self, file_name): full_path = os.path.join(WALLPAPERS_DIR, file_name) cache_path = self._get_cache_path(file_name) if not os.path.exists(cache_path): try: with Image.open(full_path) as img: width, height = img.size side = min(width, height) left = (width - side) // 2 top = (height - side) // 2 right = left + side bottom = top + side img_cropped = img.crop((left, top, right, bottom)) img_cropped.thumbnail((96, 96), Image.Resampling.LANCZOS) img_cropped.save(cache_path, "PNG") except Exception as e: print(f"Error processing {file_name}: {e}") return self.thumbnail_queue.append((cache_path, file_name)) GLib.idle_add(self._process_batch) def _process_batch(self): batch = self.thumbnail_queue[:10] del self.thumbnail_queue[:10] for cache_path, file_name in batch: try: pixbuf = GdkPixbuf.Pixbuf.new_from_file(cache_path) self.thumbnails.append((pixbuf, file_name)) self.viewport.get_model().append([pixbuf, file_name]) except Exception as e: print(f"Error loading thumbnail {cache_path}: {e}") if self.thumbnail_queue: GLib.idle_add(self._process_batch) return False def _get_cache_path(self, file_name: str) -> str: file_hash = hashlib.md5(file_name.encode("utf-8")).hexdigest() return os.path.join(self.CACHE_DIR, f"{file_hash}.png") @staticmethod def _is_image(file_name: str) -> bool: return file_name.lower().endswith( (".png", ".jpg", ".jpeg", ".bmp", ".gif", ".webp") ) def on_search_entry_focus_out(self, widget, event): if self.get_mapped(): widget.grab_focus() return False