diff --git a/src/utils/dbus_helper.py b/src/utils/dbus_helper.py deleted file mode 100644 index e56ed7c..0000000 --- a/src/utils/dbus_helper.py +++ /dev/null @@ -1,83 +0,0 @@ -from fabric.utils import Gio, GLib, logger - - -class GioDBusHelper: - """A helper class for interacting with D-Bus using the Gio library.""" - - def __init__( - self, - bus_name, - object_path, - interface_name, - bus_type=Gio.BusType.SYSTEM, - ): - self.bus = Gio.bus_get_sync(bus_type, None) - - self.bus_name = bus_name - self.object_path = object_path - self.proxy = Gio.DBusProxy.new_sync( - self.bus, - Gio.DBusProxyFlags.NONE, - None, - bus_name, - object_path, - interface_name, - None, - ) - - def call_method( - self, - bus_name, - object_path, - interface_name, - method_name, - parameters=None, - timeout=-1, - ): - if parameters is None: - parameters = GLib.Variant("()", ()) - result = self.bus.call_sync( - bus_name, - object_path, - interface_name, - method_name, - parameters, - None, - Gio.DBusCallFlags.NONE, - timeout, - None, - ) - return result.unpack() - - def listen_signal( - self, member, callback, interface_name="org.freedesktop.DBus.Properties" - ): - """Register a signal listener (conn, sender, path, iface, signal, parameters). Returns subscription id.""" - return self.bus.signal_subscribe( - self.bus_name, - interface_name, - member, - self.object_path, - arg0=None, - flags=Gio.DBusSignalFlags.NONE, - callback=callback, - ) - - def unsubscribe_signal(self, subscription_id: int): - """Unsubscribe a previously registered signal using its subscription id.""" - try: - self.bus.signal_unsubscribe(subscription_id) - except Exception as e: - logger.error(f"An error occurred: {e}") - - def set_property(self, interface_name, property_name, value_variant): - """Sets a D-Bus property using the standard D-Bus Properties interface.""" - return self.call_method( - bus_name=self.bus_name, - object_path=self.object_path, - interface_name="org.freedesktop.DBus.Properties", - method_name="Set", - parameters=GLib.Variant( - "(ssv)", (interface_name, property_name, value_variant) - ), - ) diff --git a/src/utils/debounce.py b/src/utils/debounce.py index 058febc..8be6b3b 100644 --- a/src/utils/debounce.py +++ b/src/utils/debounce.py @@ -1,36 +1,9 @@ -import asyncio import functools import typing as t from fabric.utils import GLib -def debounce(delay: float) -> t.Callable[..., t.Any]: - def decorator(func: t.Callable[..., t.Any]) -> t.Callable[..., t.Any]: - task: asyncio.Task[None] | None = None - - @functools.wraps(func) - async def wrapper(*args: t.Any, **kwargs: t.Any) -> None: - nonlocal task - - if task is not None and not task.done(): - task.cancel() - try: - await task - except asyncio.CancelledError: - pass - - async def delayed_call() -> None: - await asyncio.sleep(delay) - await func(*args, **kwargs) - - task = asyncio.create_task(delayed_call()) - - return wrapper - - return decorator - - def sync_debounce( delay: int, min_n_times: int = 0, immediate: bool = False ) -> t.Callable[..., t.Any]: diff --git a/src/utils/monitors.py b/src/utils/monitors.py deleted file mode 100644 index 6a2eb6e..0000000 --- a/src/utils/monitors.py +++ /dev/null @@ -1,61 +0,0 @@ -import json -import time -import warnings -from functools import lru_cache -from typing import Dict - -from fabric.hyprland import Hyprland -from fabric.utils import Gdk - -warnings.filterwarnings("ignore", category=DeprecationWarning) - - -def ttl_lru_cache(seconds_to_live: int, maxsize: int = 128): - def wrapper(func): - @lru_cache(maxsize) - def inner(__ttl, *args, **kwargs): - return func(*args, **kwargs) - - return lambda *args, **kwargs: inner( - time.time() // seconds_to_live, *args, **kwargs - ) - - return wrapper - - -class HyprlandWithMonitors(Hyprland): - """A Hyprland class with additional monitor common.""" - - instance = None - - @staticmethod - def get_default(): - if HyprlandWithMonitors.instance is None: - HyprlandWithMonitors.instance = HyprlandWithMonitors() - - return HyprlandWithMonitors.instance - - def __init__(self, commands_only: bool = False, **kwargs): - self.display: Gdk.Display = Gdk.Display.get_default() - super().__init__(commands_only, **kwargs) - - @ttl_lru_cache(100, 5) - def get_all_monitors(self) -> Dict: - monitors = json.loads(self.send_command("j/monitors").reply) - return {monitor["id"]: monitor["name"] for monitor in monitors} - - def get_gdk_monitor_id_from_name(self, plug_name: str) -> int | None: - for i in range(self.display.get_n_monitors()): - if self.display.get_default_screen().get_monitor_plug_name(i) == plug_name: - return i - return None - - def get_gdk_monitor_id(self, hyprland_id: int) -> int | None: - monitors = self.get_all_monitors() - if hyprland_id in monitors: - return self.get_gdk_monitor_id_from_name(monitors[hyprland_id]) - return None - - def get_current_gdk_monitor_id(self) -> int | None: - active_workspace = json.loads(self.send_command("j/activeworkspace").reply) - return self.get_gdk_monitor_id_from_name(active_workspace["monitor"]) diff --git a/src/utils/utils.py b/src/utils/utils.py index 44baf23..56c2d43 100644 --- a/src/utils/utils.py +++ b/src/utils/utils.py @@ -1,8 +1,8 @@ from pathlib import Path from typing import Literal -from fabric.utils import Gdk, bulk_connect, exec_shell_command, exec_shell_command_async -from fabric.widgets.scale import Scale, ScaleMark +from fabric.utils import Gdk, bulk_connect, exec_shell_command_async +from fabric.widgets.scale import Scale from fabric.widgets.svg import Svg from shared.widgets.animator import Animator @@ -29,10 +29,6 @@ class AnimatedScale(Scale): self.animator.play() -def is_app_running(app_name: str) -> bool: - return len(exec_shell_command(f"pidof {app_name}")) != 0 - - # Function to set up cursor hover def setup_cursor_hover( widget, cursor_name: Literal["pointer", "crosshair", "grab"] = "pointer" @@ -56,42 +52,6 @@ def setup_cursor_hover( ) -# Create a scale widget -def create_scale( - name, - marks=None, - value=0, - min_value: float = 0, - max_value: float = 100, - increments=(1, 1), - curve=(0.34, 1.56, 0.64, 1.0), - orientation="h", - h_expand=True, - h_align="center", - style_classes="", - duration=0.8, - **kwargs, -) -> AnimatedScale: - if marks is None: - marks = (ScaleMark(value=i) for i in range(1, 100, 10)) - - return AnimatedScale( - name=name, - marks=marks, - value=value, - min_value=min_value, - max_value=max_value, - increments=increments, - orientation=orientation, - curve=curve, - h_expand=h_expand, - h_align=h_align, - duration=duration, - style_classes=style_classes, - **kwargs, - ) - - # Base path to your SVG assets (resolve from repository root) BASE_SVG_PATH = (Path(__file__).resolve().parent.parent / "assets" / "icons").resolve()