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
This commit is contained in:
6 files changed
+461
-30
No files matched your search
@@ -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)
|
||||
),
|
||||
)
|
||||
+117
-23
@@ -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}")
|
||||
@@ -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()
|
||||
+5
-4
@@ -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)
|
||||
|
||||
|
||||
+7
-3
@@ -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:
|
||||
|
||||
+147
@@ -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
|
||||
Reference in new issue
Block a user