This commit is contained in:
sankalp committed 2024-11-05 14:09:20 +05:45
1 parent 55c823562f
commit b94984d0e3
12 files changed
+335 -737

No files matched your search

+6 -15
View File
@@ -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",
+37 -121
View File
@@ -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()
+33 -41
View File
@@ -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,
@@ -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
@@ -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(
@@ -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):
+62 -28
View File
@@ -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)
@@ -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}")
-82
View File
@@ -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)
-55
View File
@@ -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()
-55
View File
@@ -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}")
-249
View File
@@ -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()