From 1ed576f222ce140ca7c27fb6b1791573f4ba8c3a Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Thu, 26 Feb 2026 06:24:22 +0545 Subject: [PATCH] feat(utils): add dbus helper and extend utility functions The changes introduce a new GioDBusHelper class for D-Bus interactions and expand utils/functions.py with utilities for timeout parsing, cursor hover handling, animated scale widgets, and SVG loading, improving functionality and maintainability. - New Features - Add GioDBusHelper for method calls and signal subscription. - Enhancements - Add parse_timeout_string and replace placeholder set_process_name implementation. - Implement AnimatedScale widget and related utilities. - Provide svg_file helper with dynamic style updates. - Bug Fixes - None --- utils/dbus_helper.py | 83 ++++++++++++++++++++++++ utils/functions.py | 140 ++++++++++++++++++++++++++++++++++------- utils/gamemode.py | 102 ++++++++++++++++++++++++++++++ utils/monitors.py | 9 +-- utils/roam.py | 10 ++- utils/utils.py | 147 +++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 461 insertions(+), 30 deletions(-) create mode 100644 utils/dbus_helper.py create mode 100644 utils/gamemode.py create mode 100644 utils/utils.py diff --git a/utils/dbus_helper.py b/utils/dbus_helper.py new file mode 100644 index 0000000..60ca14e --- /dev/null +++ b/utils/dbus_helper.py @@ -0,0 +1,83 @@ +from fabric.utils import Gio, GLib + + +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: + pass + + 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/utils/functions.py b/utils/functions.py index ac1b847..c322fea 100644 --- a/utils/functions.py +++ b/utils/functions.py @@ -1,21 +1,50 @@ import ctypes import html import json +import os +import subprocess import threading from typing import Dict, List, Optional -from fabric.utils import logger, os +from fabric.utils import exec_shell_command, exec_shell_command_async, logger -# Function to set the process name def set_process_name(name: str): libc = ctypes.CDLL("libc.so.6") - libc.prctl(15, name.encode("utf-8"), 0, 0, 0) + libc.prctl(15, name.encode("utf-8"), 0, 0, 0) # 15 = PR_SET_NAME + + +def parse_timeout_string(timeout_str): + """ + Parse timeout string in format like '5s', '10m', '30s' etc. + Returns timeout in milliseconds. + """ + if not timeout_str or not isinstance(timeout_str, str): + return 5000 + + timeout_str = timeout_str.strip().lower() + + if timeout_str.endswith("s"): + try: + seconds = int(timeout_str[:-1]) + return seconds * 1000 + except ValueError: + return 5000 + elif timeout_str.endswith("m"): + try: + minutes = int(timeout_str[:-1]) + return minutes * 60 * 1000 + except ValueError: + return 5000 + else: + try: + seconds = int(timeout_str) + return seconds * 1000 + except ValueError: + return 5000 # Threading helper functions - - def thread(target, *args, **kwargs) -> threading.Thread: """ Simply run the given function in a thread. @@ -37,7 +66,7 @@ def run_in_thread(func): return wrapper -@run_in_thread +# @run_in_thread def write_json_file(data: Dict, path: str): try: with open(path, "w") as f: @@ -87,7 +116,7 @@ def get_wifi_icon_for_strength(strength: int) -> str: else: icon_name = "network-wireless-0.svg" - return os.path.join(project_root, "config", "assets", "icons", "wifi", icon_name) + return os.path.join(project_root, "assets", "icons", "wifi", icon_name) def get_wifi_connecting_icon() -> str: @@ -107,8 +136,16 @@ def get_wifi_connecting_icon() -> str: ) -# Function to check if a workspace ID is special def is_special_workspace_id(ws_id) -> bool: + """ + Check if a workspace ID represents a special workspace. + + Args: + ws_id: Workspace ID (can be int, string, or other types) + + Returns: + True if the workspace is special, False otherwise + """ try: # Convert to int if it's a string workspace_id = int(ws_id) @@ -121,8 +158,16 @@ def is_special_workspace_id(ws_id) -> bool: return False -# Function to check if a client is on a special workspace def is_special_workspace(client: dict) -> bool: + """ + Check if a client is in a special workspace. + + Args: + client: Client data dictionary from Hyprland + + Returns: + True if the client is in a special workspace, False otherwise + """ if "workspace" not in client: return False @@ -130,32 +175,81 @@ def is_special_workspace(client: dict) -> bool: # Check workspace name first if "name" in workspace: - workspace_name = workspace["name"] - if is_special_workspace_id(workspace_name): + workspace_name = str(workspace["name"]) + # Special workspaces typically start with "special:" or have negative IDs + if workspace_name.startswith("special:"): return True # Check workspace ID if "id" in workspace: workspace_id = workspace["id"] - if is_special_workspace_id(workspace_id): + # Special workspaces have negative IDs + if workspace_id < 0: return True return False def escape_markup_text(text: str) -> str: + return html.escape(text.replace("\n", " ")) + + +# Function to toggle a shell command +def toggle_command(command: str, full_command: str): + if is_app_running(command): + kill_process(command) + else: + subprocess.Popen( + full_command.split(" "), + stdin=subprocess.DEVNULL, # No input stream + stdout=subprocess.DEVNULL, # Optionally discard the output + stderr=subprocess.DEVNULL, # Optionally discard the error output + start_new_session=True, # This prevents the process from being killed + ) + + +# Function to execute a shell command asynchronously +def kill_process(process_name: str): + exec_shell_command_async(f"pkill {process_name}", lambda *_: None) + + +# Function to check if an app is running +def is_app_running(app_name: str) -> bool: + return len(exec_shell_command(f"pidof {app_name}")) != 0 + + +# General utilities +def format_duration(seconds: int) -> str: """ - Escape special characters in text to make it safe for Pango markup. + Convert a duration in seconds to a compact human-friendly string. - Args: - text: Raw text that may contain special characters - - Returns: - Escaped text safe for use in Pango markup + Examples: + 0 -> "N/A" + 59 -> "0m" + 61 -> "1m" + 3600 -> "1h 0m" """ - if not text or not isinstance(text, str): - return "" + try: + total = int(seconds) + except (TypeError, ValueError): + return "N/A" - # Use html.escape to escape XML/HTML special characters - # This handles &, <, >, and quotes - return html.escape(text) + if total <= 0: + return "N/A" + + hours = total // 3600 + minutes = (total % 3600) // 60 + if hours > 0: + return f"{hours}h {minutes}m" + return f"{minutes}m" + + +def clear_children(container) -> None: + """ + Destroy all children of a GTK container-like widget which exposes get_children(). + """ + try: + for child in list(container.get_children()): + child.destroy() + except Exception as e: + logger.warning(f"clear_children failed: {e}") diff --git a/utils/gamemode.py b/utils/gamemode.py new file mode 100644 index 0000000..449d81c --- /dev/null +++ b/utils/gamemode.py @@ -0,0 +1,102 @@ +""" +Hyprland Game Mode Toggle Script + +This script toggles game mode in Hyprland by disabling/enabling animations +and other visual effects for better performance during gaming. + +Uses a marker file in /tmp to track state. +""" + +import subprocess +import sys +from datetime import datetime +from typing import Literal + +from utils.functions import read_json_file, write_json_file + +STATE_FILE = "/tmp/hyprland_gamemode" + + +def run_hyprctl(command: str) -> str: + """Run a hyprctl command and return the output (raises on error).""" + result = subprocess.run( + ["hyprctl"] + command.split(), capture_output=True, text=True, check=True + ) + return result.stdout.strip() + + +def check_gamemode() -> Literal["t", "f"]: + """Check if game mode is active.""" + state = read_json_file(STATE_FILE) + return "t" if state and state.get("enabled") else "f" + + +def enable_gamemode(): + """Enable game mode by disabling visual effects.""" + batch_commands = [ + "keyword animations:enabled 0", + "keyword decoration:shadow:enabled 0", + "keyword decoration:blur:enabled 0", + "keyword general:gaps_in 0", + "keyword general:gaps_out 0", + "keyword general:border_size 1", + "keyword decoration:rounding 0", + ] + run_hyprctl(f'--batch "{"; ".join(batch_commands)}"') + + state = { + "enabled": True, + "last_toggled": datetime.now().isoformat(timespec="seconds"), + } + write_json_file(state, STATE_FILE) + print("Game mode enabled - visual effects disabled for better performance") + + +def disable_gamemode(): + """Disable game mode by reloading Hyprland configuration.""" + run_hyprctl("reload") + + state = { + "enabled": False, + "last_toggled": datetime.now().isoformat(timespec="seconds"), + } + write_json_file(state, STATE_FILE) + print("Game mode disabled - visual effects restored") + + +def toggle_gamemode(): + """Toggle game mode state using JSON file.""" + state = read_json_file(STATE_FILE) or {} + if state.get("enabled"): + disable_gamemode() + else: + enable_gamemode() + + +def print_usage(): + print("Usage:") + print(" gamemode check - Check if game mode is active (t/f)") + print(" gamemode toggle - Toggle game mode") + + +def main(): + if len(sys.argv) < 2: + toggle_gamemode() + return + + command = sys.argv[1].lower() + + if command == "check": + print(check_gamemode()) + elif command == "toggle": + toggle_gamemode() + elif command in ["help", "-h", "--help"]: + print_usage() + else: + print(f"Unknown command: {command}", file=sys.stderr) + print_usage() + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/utils/monitors.py b/utils/monitors.py index 01b706e..6a2eb6e 100644 --- a/utils/monitors.py +++ b/utils/monitors.py @@ -1,10 +1,11 @@ import json -import warnings -from typing import Dict import time -from fabric.hyprland import Hyprland -from gi.repository import Gdk +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) diff --git a/utils/roam.py b/utils/roam.py index 4127d98..6fe0e1e 100644 --- a/utils/roam.py +++ b/utils/roam.py @@ -1,8 +1,12 @@ -from loguru import logger - from fabric.audio import Audio +from fabric.utils import logger -from services.modus import ModusService, notification_service as notification_service_instance +from services.modus import ( + ModusService, +) +from services.modus import ( + notification_service as notification_service_instance, +) global modus_service try: diff --git a/utils/utils.py b/utils/utils.py new file mode 100644 index 0000000..bc33a1b --- /dev/null +++ b/utils/utils.py @@ -0,0 +1,147 @@ +from pathlib import Path +from typing import Literal + +from fabric.utils import Gdk, bulk_connect +from fabric.widgets.scale import Scale, ScaleMark +from fabric.widgets.svg import Svg + +from utils.animator import Animator + + +class AnimatedScale(Scale): + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.animator = None + + def animate_value(self, value: float): + if not self.animator: + self.animator = Animator( + bezier_curve=(0.34, 1.56, 0.64, 1.0), + duration=0.8, + min_value=self.min_value, + max_value=self.value, + tick_widget=self, + notify_value=lambda p, *_: self.set_value(p.value), + ) + self.animator.pause() + self.animator.min_value = self.value + self.animator.max_value = value + self.animator.play() + + +# Function to set up cursor hover +def setup_cursor_hover( + widget, cursor_name: Literal["pointer", "crosshair", "grab"] = "pointer" +): + display = Gdk.Display.get_default() + cursor = Gdk.Cursor.new_from_name(display, cursor_name) + + def on_enter_notify_event(widget, _): + widget.get_window().set_cursor(cursor) + + def on_leave_notify_event(widget, _): + # Restore default cursor when leaving the widget's area + widget.get_window().set_cursor(None) + + bulk_connect( + widget, + { + "enter-notify-event": on_enter_notify_event, + "leave-notify-event": on_leave_notify_event, + }, + ) + + +# 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() + + +def svg_file(relative_path: str, **kwargs) -> Svg: + """ + Return a fresh Svg widget for a given relative path. + + Avoids reusing the same widget instance across multiple containers, which + can cause GTK warnings. Example: + svg_file("misc/logo.svg", size=16) + """ + + def _resolve_path(path_like): + s = str(path_like) + marker = "/config/assets/icons/" + if marker in s: + rel = s.split(marker, 1)[1] + return (BASE_SVG_PATH / rel).resolve() + p = Path(s) + if p.is_absolute(): + return p + return (BASE_SVG_PATH / p).resolve() + + full_path = _resolve_path(relative_path) + svg = Svg(svg_file=str(full_path), **kwargs) + + original_set_from_file = svg.set_from_file + + def _set_from_file_resolved(file_path): + return original_set_from_file(str(_resolve_path(file_path))) + + svg.set_from_file = _set_from_file_resolved # type: ignore[attr-defined] + + # Provide a convenience alias for dynamic updates + def _dynamic_file(file_path): + _set_from_file_resolved(file_path) + return svg + + svg.dynamic_file = _dynamic_file # type: ignore[attr-defined] + + # Convenience method to update style and return the same instance + def _dynamic_style( + style: str, + *, + compiled: bool = True, + append: bool = False, + add_brackets: bool = True, + ): + svg.set_style( + style, compiled=compiled, append=append, add_brackets=add_brackets + ) + return svg + + svg.dynamic_style = _dynamic_style # type: ignore[attr-defined] + + return svg