From b94984d0e366f07c2ab414e2356fe5b3d5c47685 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Tue, 5 Nov 2024 14:09:20 +0545 Subject: [PATCH] idk --- config/fabric/modules/bar/bar.py | 21 +- config/fabric/modules/launcher/launcher.py | 158 +++-------- config/fabric/modules/launcher/widgets/app.py | 74 +++--- .../modules/launcher/widgets/clipboard.py | 126 +++++++-- .../modules/launcher/widgets/control.py | 11 - .../fabric/modules/launcher/widgets/emoji.py | 48 +++- .../fabric/modules/launcher/widgets/shell.py | 90 +++++-- .../modules/launcher/widgets/wallpaper.py | 103 ++++---- config/fabric/services/cliphist.py | 82 ------ config/fabric/services/emoji.py | 55 ---- config/fabric/services/sh.py | 55 ---- config/fabric/services/wifi.py | 249 ------------------ 12 files changed, 335 insertions(+), 737 deletions(-) delete mode 100644 config/fabric/services/cliphist.py delete mode 100644 config/fabric/services/emoji.py delete mode 100644 config/fabric/services/sh.py delete mode 100644 config/fabric/services/wifi.py diff --git a/config/fabric/modules/bar/bar.py b/config/fabric/modules/bar/bar.py index 80401209..a36e8f68 100644 --- a/config/fabric/modules/bar/bar.py +++ b/config/fabric/modules/bar/bar.py @@ -6,6 +6,7 @@ from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox from fabric.widgets.datetime import DateTime from fabric.widgets.wayland import WaylandWindow as Window +from icon import MaterialIcon from modules.bar.widgets import ( BatteryLabel, Bluetooth, @@ -17,7 +18,6 @@ from modules.bar.widgets import ( VolumeIndicator, workspace, ) -from icon import MaterialIcon class Bar(Window): @@ -44,7 +44,6 @@ class Bar(Window): ) self.date_time = DateTime(formatters=["%-I:%M 󰧞 %a %d %b"], name="datetime") self.system_tray = SystemTray(name="tray", spacing=4, icon_size=18) - self.TaskBar = TaskBar(icon_size=18) self.volume = VolumeIndicator() self.network = Network() @@ -52,18 +51,7 @@ class Bar(Window): self.battery = BatteryLabel(name="battery") self.microphone = MicrophoneIndicator() self.power = PowerProfile() - self.workspaces = Button( - child=workspace, - name="workspaces", - on_clicked=lambda _: exec_shell_command("wofi --show drun --fork"), - ) - self.logo = Button( - child=MaterialIcon("search", size="16px"), - name="logo", - on_clicked=lambda _: exec_shell_command( - "/home/sankalp/fabric/fabric_send 'launcher.toggle()'" - ), - ) + self.workspaces = Button(child=workspace, name="workspaces") self.info = HardwareUsage() self.applets = Box( name="applets", @@ -83,7 +71,10 @@ class Bar(Window): name="start-container", spacing=8, orientation="h", - children=[self.logo, self.workspaces, self.info], + children=[ + self.workspaces, + self.info, + ], ), center_children=Box( name="center-container", diff --git a/config/fabric/modules/launcher/launcher.py b/config/fabric/modules/launcher/launcher.py index e6ca9b79..904d2f70 100644 --- a/config/fabric/modules/launcher/launcher.py +++ b/config/fabric/modules/launcher/launcher.py @@ -1,8 +1,6 @@ from fabric.utils import get_desktop_applications, idle_add, remove_handler from fabric.widgets.box import Box -from fabric.widgets.button import Button from fabric.widgets.entry import Entry -from fabric.widgets.label import Label from fabric.widgets.revealer import Revealer from fabric.widgets.scrolledwindow import ScrolledWindow from fabric.widgets.wayland import WaylandWindow as Window @@ -12,26 +10,17 @@ from modules.launcher.widgets import ( EmojiManager, ShellCommandManager, WallpaperManager, - bake_favorite_slot, handle_application_search, ) class Launcher(Window): - FAVORITE_APPS = [ - "Firefox", - "Discord", - "materialgram", - "Spotify", - "kitty", - "Neovim", - "Thunar File Manager", - ] - - CONTENT_SIZES = { - "emoji": (180, 180), - "help": (220, 220), - "default": (437, 437), + COMMAND_ICONS = { + "/e": "face-smile-symbolic", + "/c": "edit-copy-symbolic", + "/d": "applications-system-symbolic", + "/s": "utilities-terminal-symbolic", + "/w": "image-symbolic", } def __init__(self, **kwargs): @@ -50,7 +39,7 @@ class Launcher(Window): self.connect("key-press-event", self.on_key_press) def initialize_managers(self): - self.clipboard_manager = ClipboardManager(self) + self.clipboard_manager = ClipboardManager(self, self.get_viewport()) self.emoji_manager = EmojiManager(self) self.shell_command_manager = ShellCommandManager(self) self.wallpaper_manager = WallpaperManager() @@ -67,33 +56,25 @@ class Launcher(Window): name="launcher", children=[ Box(spacing=10, orientation="h", children=[self.search_entry]), - self.get_favorites_box(), self.scrolled_revealer, ], ) self.add(launcher_box) - self.show_favorites() self.show_all() def create_search_entry(self): - return Entry( + self.entry = Entry( placeholder="Search.....", h_expand=True, notify_text=lambda entry, *_: self.debounce_search(entry.get_text()), - icon_name="preferences-system-search-symbolic", ) - - def get_favorites_box(self): - if not hasattr(self, "fav_apps_box"): - self.fav_apps_box = Box(spacing=10, orientation="h") - return self.fav_apps_box + return self.entry def create_scrolled_window(self): return ScrolledWindow( - name="launcher-scroll", child=self.get_viewport(), h_scrollbar_policy="never", - # v_scrollbar_policy="never", + v_scrollbar_policy="never", ) def create_revealer(self): @@ -121,128 +102,63 @@ class Launcher(Window): remove_handler(self._arranger_handler) self._arranger_handler = idle_add(self.arrange_viewport, query, pin=True) - def show_favorites(self): - for app_name in self.FAVORITE_APPS: - app = next((a for a in self.all_apps if a.name == app_name), None) - if app: - self.get_favorites_box().add(bake_favorite_slot(self, app)) - def arrange_viewport(self, query: str = ""): - self.get_viewport().children = [] - self.scrolled_window.min_content_size = (-1, -1) - self.scrolled_window.max_content_size = (-1, -1) - + self.clear_viewport() command_handlers = { - ":em": self.show_emojis, - ":ch": self.show_clipboard_history, - ":db": self.show_desktop_buttons, - ":sh": self.show_shell_commands, - ":wl": self.show_wallpapers, + "/e": self.show_emojis, + "/c": self.show_clipboard_history, + "/d": self.show_desktop_buttons, + "/s": self.show_shell_commands, + "/w": self.show_wallpapers, } for command, handler in command_handlers.items(): if query.startswith(command): - handler(query[len(command) :].strip()) + self.handle_command(command, query, handler) return - if query.startswith(":"): - self.get_favorites_box().set_visible(False) - self.show_help() - self.adjust_scrolled_window_size("help") - self.scrolled_revealer.reveal() - return - + self.reset_search_icon() handle_application_search(self, query) - def show_desktop_buttons(self, query: str = None): - self.desktop_button_manager.show_desktop_buttons(query) - self.adjust_scrolled_window_size("default") + def clear_viewport(self): + # Clear the children of the viewport + self.get_viewport().children = [] + + def handle_command(self, command: str, query: str, handler): + icon_name = self.get_command_icon(command) + self.entry.set_icon_from_icon_name(0, icon_name) + handler(query[len(command) :].strip()) self.scrolled_revealer.reveal() + def get_command_icon(self, command: str) -> str: + return self.COMMAND_ICONS.get(command, "") + + def reset_search_icon(self): + self.entry.set_icon_from_icon_name(0, "preferences-system-search-symbolic") + + def show_desktop_buttons(self, query: str): + self.desktop_button_manager.show_desktop_buttons(query) + def show_emojis(self, query: str): self.emoji_manager.show_emojis(self.get_viewport(), query) - self.adjust_scrolled_window_size("emoji") - self.scrolled_revealer.reveal() def show_clipboard_history(self, query: str): self.clipboard_manager.show_clipboard_history(self.get_viewport(), query) - self.adjust_scrolled_window_size("default") - self.scrolled_revealer.reveal() def show_shell_commands(self, query: str): if not query: self.scrolled_revealer.unreveal() return self.shell_command_manager.show_shell_commands(self.get_viewport(), query) - self.adjust_scrolled_window_size("default") - self.scrolled_revealer.reveal() def show_wallpapers(self, query: str): if query == "random": self.wallpaper_manager.apply_wallpaper_random() else: self.wallpaper_manager.show_wallpaper_thumbnails(self.get_viewport(), query) - self.adjust_scrolled_window_size("emoji") - self.scrolled_revealer.reveal() - - def show_help(self): - help_commands = [ - (":ch", "Copy a clipboard history entry", self.reveal_clipboard_history), - (":db", "Control System", lambda: self.show_desktop_buttons()), - (":em", "Copy an emoji", self.reveal_emojis), - (":sh", "Run a binary", self.reveal_shell_commands), - (":wl", "Change or show wallpapers", self.reveal_wallpapers), - ] - - for command, description, action in help_commands: - button = Button( - child=self.create_help_button_content(command, description), - on_clicked=lambda _, cmd=command, act=action: ( - self.search_entry.set_text(cmd), - act(), - ), - name="help-item", - ) - self.get_viewport().add(button) - - def create_help_button_content(self, command, description): - return Box( - orientation="h", - spacing=10, - children=[ - Label( - label=command, h_expand=True, name="command-label", h_align="start" - ), - Label( - label=description, - h_expand=True, - name="description-label", - h_align="end", - ), - ], - ) - - def reveal_clipboard_history(self): - self.clipboard_manager.show_clipboard_history(self.get_viewport(), "") - - def reveal_emojis(self): - self.emoji_manager.show_emojis(self.get_viewport(), "") - - def reveal_shell_commands(self): - self.shell_command_manager.show_shell_commands(self.get_viewport(), "") - - def reveal_wallpapers(self): - self.wallpaper_manager.show_wallpaper_thumbnails(self.get_viewport(), "") - - def adjust_scrolled_window_size(self, content_type: str): - if content_type in self.CONTENT_SIZES: - min_height, max_height = self.CONTENT_SIZES[content_type] - self.scrolled_window.min_content_size = (-1, min_height) - self.scrolled_window.max_content_size = (-1, max_height) def toggle(self): + self.set_visible(not self.is_visible()) if self.is_visible(): - self.set_visible(False) - else: self.search_entry.set_text("") - self.set_visible(True) + self.search_entry.grab_focus() diff --git a/config/fabric/modules/launcher/widgets/app.py b/config/fabric/modules/launcher/widgets/app.py index b5060043..6869eb63 100644 --- a/config/fabric/modules/launcher/widgets/app.py +++ b/config/fabric/modules/launcher/widgets/app.py @@ -1,3 +1,4 @@ +from fabric.utils import DesktopApp from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.image import Image @@ -6,11 +7,9 @@ from fabric.widgets.label import Label def handle_application_search(launcher, query: str): if not query.strip(): - launcher.fav_apps_box.set_visible(True) launcher.scrolled_revealer.unreveal() return - launcher.fav_apps_box.set_visible(False) filtered_apps = [ app for app in launcher.all_apps @@ -25,55 +24,48 @@ def handle_application_search(launcher, query: str): ] if filtered_apps: + # Limit to the first 6 applications + limited_apps = filtered_apps[:6] launcher.viewport.children = [ - bake_application_slot(launcher, app) for app in filtered_apps + bake_application_slot(launcher, app) for app in limited_apps ] launcher.scrolled_revealer.reveal() - launcher.adjust_scrolled_window_size(content_type="default") else: launcher.scrolled_revealer.unreveal() -def bake_favorite_slot(launcher, app, **kwargs) -> Button: - return Button( - child=Image(pixbuf=app.get_icon_pixbuf(), size=40), - name="fav-item", - tooltip_text=app.description, - on_clicked=lambda *_: (app.launch(), launcher.set_visible(False)), - **kwargs, +def bake_application_slot(launcher, app: DesktopApp, **kwargs) -> Button: + app_slot = Box( + orientation="h", + spacing=10, + children=[ + Image(pixbuf=app.get_icon_pixbuf(), h_align="start", size=62), + Box( + orientation="v", + spacing=5, + children=[ + Label( + label=app.display_name or "Unknown", + h_expand=True, + name="title", + v_align="center", + h_align="start", + ), + Label( + label=app.description or "", + name="description", + h_expand=True, + v_align="center", + justification="left", + line_wrap="char", + ), + ], + ), + ], ) - -def bake_application_slot(launcher, app, **kwargs) -> Button: return Button( - child=Box( - orientation="h", - spacing=10, - children=[ - Image(pixbuf=app.get_icon_pixbuf(), h_align="start", size=62), - Box( - orientation="v", - spacing=10, - children=[ - Label( - label=app.display_name or "Unknown", - h_expand=True, - name="title", - v_align="center", - h_align="start", - ), - Label( - label=app.description or "", - name="description", - h_expand=True, - v_align="center", - justification="left", - line_wrap="char", - ), - ], - ), - ], - ), + child=app_slot, name="app-item", on_clicked=lambda *_: (app.launch(), launcher.set_visible(False)), **kwargs, diff --git a/config/fabric/modules/launcher/widgets/clipboard.py b/config/fabric/modules/launcher/widgets/clipboard.py index 65b14ca1..c317f414 100644 --- a/config/fabric/modules/launcher/widgets/clipboard.py +++ b/config/fabric/modules/launcher/widgets/clipboard.py @@ -1,54 +1,126 @@ +import os import subprocess + +from gi.repository import GLib + from fabric.widgets.button import Button +from fabric.widgets.image import Image from fabric.widgets.label import Label -from fabric.widgets.box import Box class ClipboardManager: - def __init__(self, launcher): + def __init__(self, launcher, viewport): self.launcher = launcher - self.clipboard_history = self.get_clip_history() + self.viewport = viewport + self.clipboard_history = [] + self.update_clipboard_history() + self.refresh_ui() + GLib.timeout_add(10000, self.update_clipboard_history) def get_clip_history(self): try: result = subprocess.run( ["cliphist", "list"], capture_output=True, text=True, check=True ) - clips = [ + return [ {"text": line.strip(), "id": line.split()[0]} for line in result.stdout.splitlines() if line.strip() - ] - return clips - except subprocess.CalledProcessError as e: - print(f"Error fetching clipboard history: {e}") + ][:10] + except subprocess.CalledProcessError: return [] - def copy_to_clipboard(self, clip_id: str): - command = f"cliphist decode {clip_id} | wl-copy" + def update_clipboard_history(self): + new_history = self.get_clip_history() + if new_history != self.clipboard_history: + self.clipboard_history = new_history + self.refresh_ui() + return True + + def refresh_ui(self): + for child in self.viewport.children: + self.viewport.remove(child) + self.show_clipboard_history(self.viewport, "") + + def run_command(self, command, input_data=None): try: - subprocess.run(command, shell=True, check=True) - print(f"Copied Clip ID {clip_id} to clipboard.") - except subprocess.CalledProcessError as e: - print(f"Error copying clip: {e}") + if input_data is not None: + input_data = input_data.encode() + return subprocess.run( + command, input=input_data, capture_output=True, check=True + ) + except subprocess.CalledProcessError: + return None + + def copy_by_id(self, clip_id: str): + result = self.run_command(["cliphist", "decode", clip_id]) + if result: + self.run_command(["wl-copy"], input=result.stdout) + + def clear_clipboard(self): + self.run_command(["cliphist", "wipe"]) + + def save_cache_file(self, clip_id: str): + output_file = f"/tmp/cliphist/{clip_id}.png" + os.makedirs("/tmp/cliphist/", exist_ok=True) + + result = self.run_command(["cliphist", "decode", clip_id]) + if result: + with open(output_file, "wb") as f: + f.write(result.stdout) + return output_file + return None + + def clear_tmp(self): + self.run_command(["rm", "-rf", "/tmp/cliphist/*"]) def show_clipboard_history(self, viewport, query: str): filtered_history = [ item for item in self.clipboard_history if query in item["text"] ] - for item in filtered_history: - button = Button( - child=Label( - label=item["text"][:55], - h_expand=True, - v_align="center", - h_align="start", + self.add_clipboard_item(viewport, item) + + def add_clipboard_item(self, viewport, item): + if "png" in item["text"]: + self.add_image_button(viewport, item) + else: + self.add_label_button(viewport, item) + + def add_image_button(self, viewport, item): + output_file = self.save_cache_file(item["id"]) + if output_file and os.path.exists(output_file): + image_button = Button( + h_align="start", + child=Image(file=output_file), + on_clicked=lambda _, clip_id=item["id"]: self.handle_clip_click( + clip_id ), - on_clicked=lambda _, clip_id=item["id"]: ( - self.copy_to_clipboard(clip_id), - self.launcher.set_visible(False), - ), - name="cliphist-item", + name="cliphist-img-item", ) - viewport.add(button) + image_button.set_style(f"background-image: url('{output_file}');") + viewport.add(image_button) + # GLib.timeout_add(3000, lambda: self.cleanup_file(output_file)) + + def add_label_button(self, viewport, item): + label_button = Button( + child=Label( + label=item["text"][:55], + h_expand=True, + v_align="center", + h_align="start", + ), + on_clicked=lambda _, clip_id=item["id"]: self.handle_clip_click(clip_id), + name="cliphist-item", + ) + viewport.add(label_button) + + def handle_clip_click(self, clip_id): + self.copy_by_id(clip_id) + self.launcher.set_visible(False) + + def cleanup_file(self, file_path): + try: + os.remove(file_path) + except FileNotFoundError: + pass diff --git a/config/fabric/modules/launcher/widgets/control.py b/config/fabric/modules/launcher/widgets/control.py index a706d75e..e3633ec1 100644 --- a/config/fabric/modules/launcher/widgets/control.py +++ b/config/fabric/modules/launcher/widgets/control.py @@ -4,7 +4,6 @@ from fabric.utils import exec_shell_command from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.label import Label -from loguru import logger from services.inhibit import Inhibit @@ -133,7 +132,6 @@ class DesktopButtonManager: exec_shell_command(command) self.show_desktop_buttons() - logger.info("Toggled Power Profile to:", command.split()[-1]) def check_power_profile_state(self): result = exec_shell_command("powerprofilesctl get") @@ -145,7 +143,6 @@ class DesktopButtonManager: command = "nmcli radio wifi on" if not state else "nmcli radio wifi off" exec_shell_command(command) self.show_desktop_buttons() - logger.info("Toggled Wi-Fi.") def check_wifi_state(self): result = exec_shell_command("nmcli radio wifi") @@ -156,7 +153,6 @@ class DesktopButtonManager: command = "bluetoothctl power on" if not state else "bluetoothctl power off" exec_shell_command(command) self.show_desktop_buttons() - logger.info("Toggled Bluetooth.") def check_bluetooth_state(self): result = exec_shell_command("bluetoothctl show") @@ -166,7 +162,6 @@ class DesktopButtonManager: command = os.path.expanduser("~/fabric/assets/scripts/dark-theme.sh --toggle") exec_shell_command(command) self.show_desktop_buttons() - logger.info("Toggled Dark Mode.") def check_dark_mode_state(self): result = exec_shell_command( @@ -178,11 +173,6 @@ class DesktopButtonManager: def toggle_idle_mode(self, *_): self.inhibit_service.toggle() self.show_desktop_buttons() - logger.info( - "Idle mode activated." - if self.inhibit_service.is_inhibit - else "Idle mode deactivated." - ) def toggle_dnd(self, *_): current_state = self.check_dnd_state() @@ -193,7 +183,6 @@ class DesktopButtonManager: ) exec_shell_command(command) self.show_desktop_buttons() - logger.info("Toggled Do Not Disturb.") def check_dnd_state(self): result = exec_shell_command( diff --git a/config/fabric/modules/launcher/widgets/emoji.py b/config/fabric/modules/launcher/widgets/emoji.py index 51c200b2..1ded6210 100644 --- a/config/fabric/modules/launcher/widgets/emoji.py +++ b/config/fabric/modules/launcher/widgets/emoji.py @@ -1,19 +1,59 @@ +import json +import os +import subprocess + from fabric.widgets.box import Box from fabric.widgets.button import Button from fabric.widgets.label import Label -from services.emoji import EmojiItem, EmojiService + + +class EmojiItem: + def __init__(self, str: str, name: str, slug: str, group: str): + self.str = str + self.name = name + self.slug = slug + self.group = group class EmojiManager: def __init__(self, launcher): - self.emoji_service = EmojiService() self.launcher = launcher + self.emojis = None # Delay loading until first use + self.query_cache = {} + + def load_emojis(self): + if self.emojis is None: + emoji_path = os.path.expanduser("~/fabric/assets/emoji.json") + try: + with open(emoji_path, "r") as f: + data = json.load(f) + self.emojis = [ + EmojiItem(str, item["name"], item["slug"], item["group"]) + for str, item in data.items() + ] + except Exception as e: + print(f"Error loading emojis: {e}") + self.emojis = [] def query_emojis(self, query: str): - return self.emoji_service.query(query) + # Return cached results if available + if query in self.query_cache: + return self.query_cache[query] + + self.load_emojis() # Ensure emojis are loaded + + filtered_emojis = [ + e + for e in self.emojis + if query in e.name or query in e.slug or query in e.group + ][:44] + + # Cache the results + self.query_cache[query] = filtered_emojis + return filtered_emojis def copy_emoji(self, emoji: EmojiItem): - self.emoji_service.copy(emoji) + subprocess.run(["wl-copy"], input=emoji.str.encode(), check=True) self.launcher.set_visible(False) def show_emojis(self, viewport, query: str): diff --git a/config/fabric/modules/launcher/widgets/shell.py b/config/fabric/modules/launcher/widgets/shell.py index f81058a1..32f0caca 100644 --- a/config/fabric/modules/launcher/widgets/shell.py +++ b/config/fabric/modules/launcher/widgets/shell.py @@ -1,44 +1,78 @@ +import os +import subprocess + from fabric.widgets.button import Button from fabric.widgets.label import Label -from services.sh import Sh + +CACHE_DIR = os.getenv( + "XDG_CACHE_HOME", os.path.join(os.path.expanduser("~"), ".cache", "fabric") +) +BINS_PATH = os.path.join(CACHE_DIR, "binaries") class ShellCommandManager: def __init__(self, launcher): - self.sh_instance = Sh() - self.launcher = launcher # Store reference to Launcher + self.launcher = launcher + self.reload_binaries() + + def reload_binaries(self): + """Reload the binaries from the PATH environment variable.""" + paths = os.getenv("PATH", "").split(":") + unique_bins = set() + + for path in paths: + unique_bins.update(self.list_binaries_in_path(path)) + + self.save_binaries_to_cache(unique_bins) + + def list_binaries_in_path(self, path): + if not os.path.exists(path): + return [] + try: + return ( + subprocess.check_output(f"ls {path}", shell=True, text=True) + .strip() + .splitlines() + ) + except subprocess.CalledProcessError: + return [] + + def save_binaries_to_cache(self, binaries): + with open(BINS_PATH, "w") as f: + f.write("\n".join(binaries)) def show_shell_commands(self, viewport, search_query: str): if not search_query: return - def on_query_done(results): - if not results: - return + results = self.query_binaries(search_query) + self.display_results(viewport, results) - for bin in results: - button = Button( - child=Label( - label=bin, - h_expand=True, - v_align="center", - h_align="start", - ), - on_clicked=lambda _, cmd=bin: ( - self.execute_command(cmd), # Execute command and close launcher - ), - name="sh-item", - ) - viewport.add(button) + def query_binaries(self, filter_str): + try: + with open(BINS_PATH) as f: + return [line.strip() for line in f if filter_str in line][:16] + except FileNotFoundError: + return [] - # Reload shell command list and query - self.sh_instance.reload() - results = Sh.query(search_query) - on_query_done(results) + def display_results(self, viewport, results): + for command in results: + button = Button( + child=Label( + label=command, + v_align="center", + h_align="start", + ), + on_clicked=lambda _, cmd=command: self.execute_command(cmd), + name="sh-item", + ) + viewport.add(button) def execute_command(self, cmd: str): try: - Sh.run(cmd) # Run the shell command - self.launcher.set_visible(False) # Close the launcher after execution - except Exception as e: - logger.error(f"Error executing command: {e}") + output = subprocess.check_output(cmd, shell=True, text=True) + print(output) + except subprocess.CalledProcessError as e: + print(e.output) + finally: + self.launcher.set_visible(False) diff --git a/config/fabric/modules/launcher/widgets/wallpaper.py b/config/fabric/modules/launcher/widgets/wallpaper.py index 8f7978a3..c285af90 100644 --- a/config/fabric/modules/launcher/widgets/wallpaper.py +++ b/config/fabric/modules/launcher/widgets/wallpaper.py @@ -1,11 +1,11 @@ import os from pathlib import Path +from loguru import logger + from fabric.utils import exec_shell_command_async from fabric.widgets.box import Box from fabric.widgets.button import Button -from fabric.widgets.label import Label -from loguru import logger class WallpaperManager: @@ -17,25 +17,25 @@ class WallpaperManager: self.thumbnail_dir.mkdir(parents=True, exist_ok=True) def show_wallpaper_thumbnails(self, viewport, search_term: str = ""): - wallpapers = [ + wallpapers = self._get_wallpapers(search_term)[:24] + self._populate_viewport(viewport, wallpapers) + + def _get_wallpapers(self, search_term: str): + return [ f for f in self.wallpaper_dir.iterdir() - if f.is_file() and f.suffix.lower() in [".png", ".jpg", ".jpeg", ".webp"] + if f.is_file() + and f.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"} + and (not search_term or search_term.lower() in f.name.lower()) ] - if search_term: - wallpapers = [ - wallpaper - for wallpaper in wallpapers - if search_term.lower() in wallpaper.name.lower() - ] - - viewport.children = [] + def _populate_viewport(self, viewport, wallpapers): + viewport.children.clear() row = Box(orientation="h", spacing=10) for i, wallpaper in enumerate(wallpapers): - thumbnail_path = self.generate_thumbnail(wallpaper) - thumbnail_button = self.create_wallpaper_thumbnail( + thumbnail_path = self._generate_thumbnail(wallpaper) + thumbnail_button = self._create_wallpaper_thumbnail( thumbnail_path, wallpaper ) row.add(thumbnail_button) @@ -47,63 +47,68 @@ class WallpaperManager: if row.children: viewport.add(row) - def generate_thumbnail(self, wallpaper_path): + def _generate_thumbnail(self, wallpaper_path): thumbnail_path = self.thumbnail_dir / wallpaper_path.name if not thumbnail_path.exists(): - exec_shell_command_async( - [ - "magick", - str(wallpaper_path), - "-resize", - "x36", - "-gravity", - "Center", - "-extent", - "1:1", - str(thumbnail_path), - ], - lambda output: logger.info("Thumbnail created successfully!"), - ) + self._create_thumbnail(wallpaper_path, thumbnail_path) return thumbnail_path - def create_wallpaper_thumbnail(self, thumbnail_path, wallpaper_path): + def _create_thumbnail(self, wallpaper_path, thumbnail_path): + exec_shell_command_async( + [ + "magick", + str(wallpaper_path), + "-resize", + "x32", + "-gravity", + "Center", + "-extent", + "1:1", + str(thumbnail_path), + ], + lambda *args: logger.debug(args), + ) + + def _create_wallpaper_thumbnail(self, thumbnail_path, wallpaper_path): thumbnail_box = Box( orientation="h", h_align="center", v_align="center", - style=f"background-image: url('{thumbnail_path}'); min-width: 36px; min-height: 36px; background-repeat: no-repeat; background-size: cover; background-position: center;", + style=self._thumbnail_style(thumbnail_path), ) - button = Button( + return Button( child=thumbnail_box, h_align="start", v_align="center", name="wall-item", - on_clicked=lambda _: self.select_wallpaper(wallpaper_path), + on_clicked=lambda _: self._select_wallpaper(wallpaper_path), ) - return button - def select_wallpaper(self, wallpaper_path): + def _thumbnail_style(self, thumbnail_path): + return ( + f"background-image: url('{thumbnail_path}'); " + "min-width: 32px; min-height: 32px; " + "background-repeat: no-repeat; " + "background-size: cover; background-position: center;" + ) + + def _select_wallpaper(self, wallpaper_path): logger.info(f"Selected wallpaper: {wallpaper_path}") - self.apply_wallpaper(wallpaper_path) + self._apply_wallpaper(wallpaper_path) - def apply_wallpaper(self, wallpaper_path): - script_path = os.path.expanduser("~/dotfiles/hypr/scripts/wallpaper.py") - try: - exec_shell_command_async( - ["python3", script_path, "--image", str(wallpaper_path)], - lambda output: logger.info(f"{output}"), - ) - logger.info(f"Wallpaper applied: {wallpaper_path}") - except Exception as e: - logger.error(f"Error applying wallpaper: {e}") + def _apply_wallpaper(self, wallpaper_path): + self._execute_wallpaper_script("--image", str(wallpaper_path)) def apply_wallpaper_random(self): + self._execute_wallpaper_script("-R") + + def _execute_wallpaper_script(self, *args): script_path = os.path.expanduser("~/dotfiles/hypr/scripts/wallpaper.py") try: exec_shell_command_async( - ["python3", script_path, "-R"], + ["python3", script_path, *args], lambda output: logger.info(f"{output}"), ) - logger.info("Random wallpaper applied.") + logger.info(f"Wallpaper applied with args: {args}") except Exception as e: - logger.error(f"Error applying random wallpaper: {e}") + logger.error(f"Error applying wallpaper: {e}") diff --git a/config/fabric/services/cliphist.py b/config/fabric/services/cliphist.py deleted file mode 100644 index aed3207a..00000000 --- a/config/fabric/services/cliphist.py +++ /dev/null @@ -1,82 +0,0 @@ -import subprocess -from gi.repository import GLib -from fabric.core.service import Service -from fabric.utils import exec_shell_command_async - - -class CliphistItem: - def __init__(self, id: int, content: str, content_type: str): - self.id = id - self.content = content - self.content_type = content_type - - -async def query(filter: str): - fzf_args = '-f ""' if filter == "" else f"-f {filter}" - - try: - result = exec_shell_command_async( - ["bash", "-c", f"cliphist list | fzf {fzf_args} | head -n 16"] - ) - items = set() # Using a set to ensure uniqueness - - for line in result.split("\n"): - if line: - match = GLib.regex_match(r"(\d+)\s+(.*)", line) - if match: - id = int(match.group(1)) - content = match.group(2) - content_type = "text" - if content.startswith("[["): - binary_match = GLib.regex_match( - r"\[\[\s+binary data.*(jpg|png|webp|jpeg|bmp)", content - ) - if binary_match: - content_type = binary_match.group(1) - - items.add(CliphistItem(id, content, content_type)) - - return list(items) - - except Exception as err: - print(err) - return [] - - -async def run(args: CliphistItem): - try: - output = exec_shell_command_async( - ["bash", "-c", f"cliphist decode {args.id} | wl-copy"] - ) - print(f":cliphist {args.id}:") - print(output) - except Exception as err: - print("hello") - - -async def get_file(item: CliphistItem): - filepath = f"/tmp/cliphist-{item.id}.{item.content_type}" - if not GLib.file_test(filepath, GLib.FileTest.EXISTS): - exec_shell_command_async( - ["bash", "-c", f"cliphist decode {item.id} > {filepath}"] - ) - return filepath - - -class Cliphist(Service): - def __init__(self): - super().__init__() - - async def get_file(self, item: CliphistItem): - return await get_file(item) - - async def query(self, filter: str): - return await query(filter) - - async def run(self, args: CliphistItem): - await run(args) - - -if __name__ == "__main__": - cliphist = Cliphist() - Service.register(cliphist) diff --git a/config/fabric/services/emoji.py b/config/fabric/services/emoji.py deleted file mode 100644 index 3b5f19b0..00000000 --- a/config/fabric/services/emoji.py +++ /dev/null @@ -1,55 +0,0 @@ -import json -import os -from loguru import logger -from fabric.utils import exec_shell_command_async -import subprocess - - -class EmojiItem: - def __init__(self, str: str, name: str, slug: str, group: str): - self.str = str - self.name = name - self.slug = slug - self.group = group - - -class EmojiService: - def __init__(self): - self.emoji = None - self.load_emojis() - - def load_emojis(self): - try: - # Construct the path using os.path.expanduser to handle the '~' correctly - emoji_path = os.path.expanduser("~/fabric/assets/emoji.json") - with open(emoji_path, "r") as f: - data = json.load(f) - self.emoji = [ - EmojiItem(str, item["name"], item["slug"], item["group"]) - for str, item in data.items() - ] - except Exception as e: - logger.error(f"Failed to load emoji data: {e}") - self.emoji = [] - - def get_emojis(self): - return self.emoji - - def query(self, filter: str): - if self.emoji is None: - return [] - return [ - e - for e in self.emoji - if filter in e.name or filter in e.slug or filter in e.group - ][:44] - - def copy(self, emoji_item: EmojiItem): - try: - # Use subprocess to run the wl-copy command to copy emoji - subprocess.run(["wl-copy"], input=emoji_item.str.encode(), check=True) - except Exception as e: - logger.error(f"Failed to copy emoji: {e}") - - -emoji_service = EmojiService() diff --git a/config/fabric/services/sh.py b/config/fabric/services/sh.py deleted file mode 100644 index ddddae87..00000000 --- a/config/fabric/services/sh.py +++ /dev/null @@ -1,55 +0,0 @@ -from __init__ import * - -CACHE_DIR = os.getenv("XDG_CACHE_HOME", os.path.join(os.path.expanduser("~"), ".cache")) -BINS = os.path.join(CACHE_DIR, "binaries") - - -class Sh: - def __init__(self): - self.reload() # Load binaries on initialization - - def ls(self, path): - if not os.path.exists(path): - return [] - try: - return ( - subprocess.check_output(f"ls {path}", shell=True, text=True) - .strip() - .splitlines() - ) - except subprocess.CalledProcessError: - return [] - - def reload(self): - """Reload the binaries from the PATH environment variable.""" - paths = os.getenv("PATH", "").split(":") - bins = [] - - for path in paths: - bins.extend(self.ls(path)) - - # Remove duplicates and save to file - unique_bins = set(bins) - with open(BINS, "w") as f: - f.write("\n".join(unique_bins)) - - @staticmethod - def query(filter_str): - """Filter binaries using fzf and return the results.""" - try: - result = subprocess.check_output( - f"cat {BINS} | fzf -f {filter_str} | head -n 16", shell=True, text=True - ) - return list(dict.fromkeys(result.splitlines())) # Deduplicate results - except subprocess.CalledProcessError: - return [] - - @staticmethod - def run(args: str): - """Execute a command and print its output.""" - try: - output = subprocess.check_output(args, shell=True, text=True) - print(f":sh {args.strip()}:") - print(output) - except subprocess.CalledProcessError as err: - print(f"Error running command: {err}") diff --git a/config/fabric/services/wifi.py b/config/fabric/services/wifi.py deleted file mode 100644 index 261a415d..00000000 --- a/config/fabric/services/wifi.py +++ /dev/null @@ -1,249 +0,0 @@ -from typing import List, Literal -import gi -from fabric.core.service import Property, Service, Signal -from fabric.utils import bulk_connect -from gi.repository import Gio -from loguru import logger - -from fabric.utils import exec_shell_command_async - -try: - gi.require_version("NM", "1.0") - from gi.repository import NM -except ValueError: - logger.error("Failed to start network manager") - - -class Wifi(Service): - @Signal - def changed(self) -> None: - ... - - @Signal - def enabled(self) -> bool: - ... - - def __init__(self, client: NM.Client, device: NM.DeviceWifi, **kwargs): - self._client: NM.Client = client - self._device: NM.DeviceWifi = device - self._ap: NM.AccessPoint | None = None - self._ap_signal: int | None = None - super().__init__(**kwargs) - - self._client.connect( - "notify::wireless-enabled", - lambda *args: self.notifier("enabled", args), - ) - if self._device: - bulk_connect( - self._device, - { - "notify::active-access-point": lambda *args: self._activate_ap(), - "access-point-added": lambda *args: self.emit("changed"), - "access-point-removed": lambda *args: self.emit("changed"), - "state-changed": lambda *args: self.ap_update(), - }, - ) - self._activate_ap() - - def ap_update(self): - self.emit("changed") - for sn in [ - "enabled", - "internet", - "strength", - "frequency", - "access-points", - "ssid", - "state", - "icon-name", - ]: - self.notify(sn) - - def _activate_ap(self): - if self._ap: - self._ap.disconnect(self._ap_signal) - self._ap = self._device.get_active_access_point() - if not self._ap: - return - - self._ap_signal = self._ap.connect( - "notify::strength", lambda *args: self.ap_update() - ) # type: ignore - - def toggle_wifi(self): - self._client.wireless_set_enabled(not self._client.wireless_get_enabled()) - - # def set_active_ap(self, ap): - # self._device.access - - def scan(self): - self._device.request_scan_async( - None, - lambda device, result: [ - device.request_scan_finish(result), - self.emit("changed"), - ], - ) - - def notifier(self, name: str, *args): - self.notify(name) - self.emit("changed") - return - - @Property(bool, "read-write", default_value=False) - def enabled(self) -> bool: # type: ignore - return bool(self._client.wireless_get_enabled()) - - @enabled.setter - def enabled(self, value: bool): - self._client.wireless_set_enabled(value) - - @Property(int, "readable") - def strength(self): - return self._ap.get_strength() if self._ap else -1 - - @Property(str, "readable") - def icon_name(self): - if not self._ap: - return "network-wireless-disabled-symbolic" - - if self.internet == "activated": - return { - 80: "network-wireless-signal-excellent-symbolic", - 60: "network-wireless-signal-good-symbolic", - 40: "network-wireless-signal-ok-symbolic", - 20: "network-wireless-signal-weak-symbolic", - 00: "network-wireless-signal-none-symbolic", - }.get( - min(80, 20 * round(self._ap.get_strength() / 20)), - "network-wireless-no-route-symbolic", - ) - if self.internet == "activating": - return "network-wireless-acquiring-symbolic" - - return "network-wireless-offline-symbolic" - - @Property(int, "readable") - def frequency(self): - return self._ap.get_frequency() if self._ap else -1 - - @Property(int, "readable") - def internet(self): - return { - NM.ActiveConnectionState.ACTIVATED: "activated", - NM.ActiveConnectionState.ACTIVATING: "activating", - NM.ActiveConnectionState.DEACTIVATING: "deactivating", - NM.ActiveConnectionState.DEACTIVATED: "deactivated", - }.get( - self._device.get_active_connection().get_state(), - "unknown", - ) - - @Property(object, "readable") - def access_points(self) -> List[object]: - points: list[NM.AccessPoint] = self._device.get_access_points() - - def make_ap_dict(ap: NM.AccessPoint): - return { - "bssid": ap.get_bssid(), - # "address": ap.get_ - "last_seen": ap.get_last_seen(), - "ssid": NM.utils_ssid_to_utf8(ap.get_ssid().get_data()) - if ap.get_ssid() - else "Unknown", - "active-ap": self._ap, - "strength": ap.get_strength(), - "frequency": ap.get_frequency(), - "icon-name": { - 80: "network-wireless-signal-excellent-symbolic", - 60: "network-wireless-signal-good-symbolic", - 40: "network-wireless-signal-ok-symbolic", - 20: "network-wireless-signal-weak-symbolic", - 00: "network-wireless-signal-none-symbolic", - }.get( - min(80, 20 * round(ap.get_strength() / 20)), - "network-wireless-no-route-symbolic", - ), - } - - return list(map(make_ap_dict, points)) - - @Property(str, "readable") - def ssid(self): - if not self._ap: - return "Disconnected" - ssid = self._ap.get_ssid().get_data() - return NM.utils_ssid_to_utf8(ssid) if ssid else "Unknown" - - @Property(int, "readable") - def state(self): - return { - NM.DeviceState.UNMANAGED: "unmanaged", - NM.DeviceState.UNAVAILABLE: "unavailable", - NM.DeviceState.DISCONNECTED: "disconnected", - NM.DeviceState.PREPARE: "prepare", - NM.DeviceState.CONFIG: "config", - NM.DeviceState.NEED_AUTH: "need_auth", - NM.DeviceState.IP_CONFIG: "ip_config", - NM.DeviceState.IP_CHECK: "ip_check", - NM.DeviceState.SECONDARIES: "secondaries", - NM.DeviceState.ACTIVATED: "activated", - NM.DeviceState.DEACTIVATING: "deactivating", - NM.DeviceState.FAILED: "failed", - }.get(self._device.get_state(), "unknown") - - -class NetworkClient(Service): - @Signal - def device_ready(self) -> None: - ... - - def __init__(self, **kwargs): - self._client: NM.Client | None = None - self.wifi_device: Wifi | None = None - super().__init__(**kwargs) - NM.Client.new_async( - cancellable=None, - callback=self._init_network_client, - **kwargs, - ) - - def _init_network_client(self, client: NM.Client, task: Gio.Task, **kwargs): - self._client = client - wifi_device: NM.DeviceWifi = self._get_device(NM.DeviceType.WIFI) # type: ignore - # eth_device: NM.DeviceEthernet = self._get_device(NM.DeviceType.ETHERNET) # type: ignore - - if wifi_device: - self.wifi_device = Wifi(self._client, wifi_device) - self.emit("device-ready") - - # if eth_device: - # self.eth_device = Ethernet(self._client, eth_device) - - def _get_device(self, device_type) -> NM.Device | None: - devices: List[NM.Device] = self._client.get_devices() # type: ignore - return next((x for x in devices if x.get_device_type() == device_type), None) - - def get_primary_device(self) -> Literal["wifi", "wired"] | None: - if not self._client: - return None - return ( - "wifi" - if "wireless" - in str(self._client.get_primary_connection().get_connection_type()) - else "wired" - if "ethernet" - in str(self._client.get_primary_connection().get_connection_type()) - else None - ) - - def connect_wifi_bssid(self, bssid): - # We are using nmcli here, idk im lazy - exec_shell_command_async( - f"nmcli device wifi connect {bssid}", lambda *args: print(args) - ) - - @Property(str, "readable") - def primary_device(self) -> Literal["wifi", "wired"] | None: - return self.get_primary_device()