chore(src/utils): drop unused dbus_helper and monitors modules and cleanup utils code
This commit is contained in:
4 files changed
+2
-213
No files matched your search
@@ -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)
|
||||
),
|
||||
)
|
||||
@@ -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]:
|
||||
|
||||
@@ -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"])
|
||||
+2
-42
@@ -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()
|
||||
|
||||
|
||||
Reference in new issue
Block a user