From 06be28535a19bb636b660bd9024057f8254e3045 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Wed, 8 Jul 2026 12:24:39 +0545 Subject: [PATCH] refactor: clean up service implementations and remove dead code --- src/main.py | 10 + src/services/brightness.py | 16 +- src/services/custom_notification.py | 28 +- src/services/environment.py | 72 ++ src/services/gamemode.py | 9 +- src/services/globalmenu.py | 1028 +++++++++++++++++---------- src/services/modus.py | 500 ++++++++++--- src/services/screencapture.py | 7 +- src/shared/data.py | 15 +- src/shared/dialogs/about.py | 16 +- src/shared/widgets/clipping_box.py | 2 +- src/shared/widgets/custom_image.py | 3 +- src/shared/widgets/flat_scale.py | 3 +- src/shared/widgets/smooth_switch.py | 3 +- src/shared/window/dropdown.py | 3 +- src/utils/dbusmenu.py | 160 ++++- src/utils/functions.py | 67 +- src/utils/gtkmenu.py | 400 ++++++++++- src/utils/icon_resolver.py | 6 + src/utils/memory_tracker.py | 280 ++++++++ src/utils/occlusion.py | 140 +--- src/utils/roam.py | 55 +- 22 files changed, 2067 insertions(+), 756 deletions(-) create mode 100644 src/services/environment.py create mode 100644 src/utils/memory_tracker.py diff --git a/src/main.py b/src/main.py index a841501..4eacdb1 100644 --- a/src/main.py +++ b/src/main.py @@ -1,3 +1,4 @@ +import atexit import os from fabric import Application @@ -88,6 +89,15 @@ def main(): app.set_css = set_css app.set_css() + def cleanup_css_monitors(): + for m in css_monitors: + try: + m.cancel() + except Exception: + pass + + atexit.register(cleanup_css_monitors) + # Inject into the executing module's namespace (__main__) # to emulate what happened when this file was run directly. # This allows fabric-cli exec to execute commands flawlessly. diff --git a/src/services/brightness.py b/src/services/brightness.py index 5e72763..3444bb6 100644 --- a/src/services/brightness.py +++ b/src/services/brightness.py @@ -1,10 +1,10 @@ -import shutil -import subprocess import time from fabric.core.service import Property, Service, Signal from fabric.utils import GLib, exec_shell_command_async, logger, os, re +from utils.functions import find_binary, run_command + class Brightness(Service): """Service for controlling screen brightness using ddcutil or brightnessctl backends. @@ -103,7 +103,7 @@ class Brightness(Service): logger.info(f"Using forced backend: {backend}") return backend - if shutil.which("brightnessctl"): + if find_binary("brightnessctl"): device = self._get_screen_device() if device: logger.info(f"Using brightnessctl backend with device: {device}") @@ -113,7 +113,7 @@ class Brightness(Service): "brightnessctl is available but no backlight devices found in /sys/class/backlight/" ) - if shutil.which("ddcutil"): + if find_binary("ddcutil"): bus = self._detect_ddcutil_bus() if bus != -1: self.ddcutil_bus = bus @@ -139,9 +139,7 @@ class Brightness(Service): def _detect_ddcutil_bus(self): """Detect I2C bus number for ddcutil.""" try: - process = subprocess.run( - ["ddcutil", "detect"], text=True, capture_output=True, timeout=2 - ) + process = run_command(["ddcutil", "detect"], timeout=2) if process.returncode == 0: match = re.search(r"I2C bus:\s*/dev/i2c-(\d+)", process.stdout) return int(match.group(1)) if match else -1 @@ -154,7 +152,7 @@ class Brightness(Service): if self.backend: if self.backend == "ddcutil": try: - process = subprocess.run( + process = run_command( [ "ddcutil", "--bus", @@ -163,8 +161,6 @@ class Brightness(Service): "getvcp", "10", ], - text=True, - capture_output=True, timeout=2, ) diff --git a/src/services/custom_notification.py b/src/services/custom_notification.py index 4933d4b..fb2414e 100644 --- a/src/services/custom_notification.py +++ b/src/services/custom_notification.py @@ -30,6 +30,7 @@ class CachedNotification(Service): # Store cache metadata for cleanup self.cache_metadata = data.get("cache_metadata", {}) self.timestamp = data.get("timestamp", int(time.time())) + self._pixbuf_cache = {} return self @@ -82,18 +83,23 @@ class CachedNotification(Service): @Property(object, "readable") def image_pixbuf(self) -> GdkPixbuf.Pixbuf | None: + if hasattr(self, "_cached_image_pixbuf"): + return self._cached_image_pixbuf try: if self.image_pixmap: - return self.image_pixmap.as_pixbuf() + self._cached_image_pixbuf = self.image_pixmap.as_pixbuf() + return self._cached_image_pixbuf if self.image_file and os.path.exists(self.image_file): try: - return GdkPixbuf.Pixbuf.new_from_file(self.image_file) + self._cached_image_pixbuf = GdkPixbuf.Pixbuf.new_from_file( + self.image_file + ) + return self._cached_image_pixbuf except Exception: - # If file can't be loaded, return None pass except Exception: - # If any error occurs (including temp file gone), return None safely pass + self._cached_image_pixbuf = None return None @Property(dict, "readable") @@ -169,6 +175,12 @@ class CachedNotification(Service): self._cache_id = cache_id self.cache_metadata = {} self.timestamp = int(time.time()) + self._pixbuf_cache: dict[str, GdkPixbuf.Pixbuf | None] = {} + + def get_or_cache_pixbuf(self, cache_key: str, loader) -> GdkPixbuf.Pixbuf | None: + if cache_key not in self._pixbuf_cache: + self._pixbuf_cache[cache_key] = loader() + return self._pixbuf_cache[cache_key] def remove_from_cache(self): self.removed_from_cache.emit() @@ -280,8 +292,8 @@ class CachedNotifications(Notifications): handler_id = cached_notification.connect( "removed-from-cache", - lambda *args: self.remove_cached_notification( - notification_id=cache_id + lambda *args, cid=cache_id: self.remove_cached_notification( + notification_id=cid ), ) self._signal_handlers[cache_id] = handler_id @@ -435,7 +447,9 @@ class CachedNotifications(Notifications): # IMMEDIATELY store to history before attempting any caching operations handler_id = cached_notification.connect( "removed-from-cache", - lambda *args: self.remove_cached_notification(notification_id=cache_id), + lambda *args, cid=cache_id: self.remove_cached_notification( + notification_id=cid + ), ) self._signal_handlers[cache_id] = handler_id self._cached_notifications[cache_id] = cached_notification diff --git a/src/services/environment.py b/src/services/environment.py new file mode 100644 index 0000000..b130b89 --- /dev/null +++ b/src/services/environment.py @@ -0,0 +1,72 @@ +"""Global menu environment setup — GTK_MODULES, UBUNTU_MENUPROXY, settings.ini, Flatpak.""" + +from fabric.utils import exec_shell_command, logger, os + + +ENV_VARS = { + "GTK_MODULES": "appmenu-gtk-module", + "UBUNTU_MENUPROXY": "1", +} + + +def setup_global_menu_environment(): + """Write env config files so GTK apps register with AppMenu at next login. + + Does NOT set env via hyprctl keyword env — that would cause Fabric's own + GTK windows to load appmenu-gtk-module and trigger Gdk-CRITICAL assertions. + Uses session-level config files that take effect at login. + """ + try: + # exec_shell_command("hyprctl keyword env GTK_MODULES,appmenu-gtk-module") + # exec_shell_command("hyprctl keyword env UBUNTU_MENUPROXY,1") + exec_shell_command( + "dbus-update-activation-environment --systemd GTK_MODULES=appmenu-gtk-module UBUNTU_MENUPROXY=1" + ) + exec_shell_command( + "flatpak override --user --talk-name=com.canonical.AppMenu.Registrar" + ) + + env_dir = os.path.expanduser("~/.config/environment.d") + try: + os.makedirs(env_dir, exist_ok=True) + env_path = os.path.join(env_dir, "appmenu.conf") + with open(env_path, "w") as f: + for k, v in ENV_VARS.items(): + f.write(f"{k}={v}\n") + except OSError: + pass + + settings_path = os.path.expanduser("~/.config/gtk-3.0/settings.ini") + try: + if not os.path.exists(os.path.dirname(settings_path)): + os.makedirs(os.path.dirname(settings_path), exist_ok=True) + content = "" + if os.path.exists(settings_path): + with open(settings_path, "r") as f: + content = f.read() + if "gtk-shell-shows-menubar" not in content: + with open(settings_path, "a") as f: + if content and not content.endswith("\n"): + f.write("\n") + f.write("gtk-shell-shows-menubar=1\n") + except OSError: + pass + + pam_path = os.path.expanduser("~/.pam_environment") + try: + existing = "" + if os.path.exists(pam_path): + with open(pam_path, "r") as f: + existing = f.read() + lines = existing.splitlines(True) + for k, v in ENV_VARS.items(): + line = f"{k} DEFAULT={v}\n" + if not any(ln.startswith(f"{k} ") for ln in lines): + with open(pam_path, "a") as f: + f.write(line) + except OSError: + pass + + logger.info("[GlobalMenu] GTK environment injected") + except Exception as e: + logger.warning(f"[GlobalMenu] Environment setup failed: {e}") diff --git a/src/services/gamemode.py b/src/services/gamemode.py index 449d81c..7553748 100644 --- a/src/services/gamemode.py +++ b/src/services/gamemode.py @@ -7,21 +7,20 @@ 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 +from utils.functions import read_json_file, run_command, 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 - ) + result = run_command(["hyprctl"] + command.split(), timeout=5) + if result.returncode != 0: + raise RuntimeError(f"hyprctl failed: {result.stderr}") return result.stdout.strip() diff --git a/src/services/globalmenu.py b/src/services/globalmenu.py index 63cdeba..c067e89 100644 --- a/src/services/globalmenu.py +++ b/src/services/globalmenu.py @@ -1,9 +1,10 @@ -"""Global Menu Service - extracts and manages application menus.""" +"""Global Menu Service — event-driven, persistent importer architecture.""" import json import re import threading -from typing import List, Optional, Dict, Union +from dataclasses import dataclass +from typing import Dict, List, Optional, Set, Tuple, Union from fabric.core.service import Property, Service, Signal from fabric.utils import ( @@ -15,15 +16,13 @@ from fabric.utils import ( os, ) - +from services.environment import setup_global_menu_environment from utils.dbusmenu import DBusMenuClient, DBusMenuItem, _get_bus from utils.gtkmenu import ActionMenuClient, GtkMenuClient -_MAX_CACHE_SIZE = 5 -_EMPTY_CACHE_TTL = 30.0 # re-check apps with empty menus after 30s -_DBUS_TIMEOUT_INTROSPECT = 500 # ms for introspection calls -_DBUS_TIMEOUT_PID = 200 # ms for PID lookups -_DBUS_TIMEOUT_FAST = 100 # ms for expected-fast calls +_DBUS_TIMEOUT_INTROSPECT = 500 +_DBUS_TIMEOUT_PID = 200 +_DBUS_TIMEOUT_FAST = 100 _DE_NAMESPACES = [ "xfce", @@ -38,7 +37,56 @@ _DE_NAMESPACES = [ ] +class _IntrospectionCache: + """Caches DBus Introspect XML per service+path to avoid redundant round-trips.""" + + def __init__(self): + self._cache: Dict[str, str] = {} + self._lock = threading.Lock() + + def _key(self, service: str, path: str) -> str: + return f"{service}:{path}" + + def get(self, service: str, path: str, bus, timeout: int) -> str: + if not path or not GLib.Variant.is_object_path(path): + return "" + key = self._key(service, path) + with self._lock: + cached = self._cache.get(key) + if cached is not None: + return cached + xml = _dbus_introspect(bus, service, path, timeout) + with self._lock: + self._cache[key] = xml + return xml + + def invalidate(self, service: str): + with self._lock: + keys = [k for k in self._cache if k.startswith(f"{service}:")] + for k in keys: + self._cache.pop(k, None) + + +_exe_to_service: Dict[str, str] = {} +"""Maps executable path to last-known DBus service name. Populated during discovery.""" + + +@dataclass +class AppServiceInfo: + service_name: str + object_path: str = "" + pid: int = 0 + executable: str = "" + wm_class: str = "" + app_id: str = "" + importer: Optional[Union[DBusMenuClient, GtkMenuClient, ActionMenuClient]] = None + revision: int = 0 + connected_signals: bool = False + + def _dbus_introspect(bus, service: str, path: str, timeout: int) -> str: + if not path or not GLib.Variant.is_object_path(path): + return "" try: res = bus.call_sync( service, @@ -96,10 +144,7 @@ REGISTRAR_XML = """ """ - -_EMPTY_SENTINEL: List[ - DBusMenuItem -] = [] # distinct sentinel for "discovery returned empty" +_EMPTY_SENTINEL: List[DBusMenuItem] = [] class GlobalMenuService(Service): @@ -130,33 +175,132 @@ class GlobalMenuService(Service): self._current_wm_class: str = "" self._current_pid: int = 0 - # Cached dbus menu clients mapping: wm_class -> Client - self._client_cache: Dict[str, Union[GtkMenuClient, DBusMenuClient]] = {} - # Cached layout for fast repaints - self._menu_cache: Dict[str, List[DBusMenuItem]] = {} + # Service registry: service_name -> AppServiceInfo + # This is the core event-driven data structure (single source of truth). + self._service_registry: Dict[str, AppServiceInfo] = {} + self._pid_to_service: Dict[int, str] = {} - # Registrar mapping: sender -> (service_name, object_path) - self._registered_menus: dict[str, tuple[str, str]] = {} + self._introspection_cache = _IntrospectionCache() self._seq_lock = threading.Lock() self._extraction_seq = 0 - self._extracting_in_flight: set[str] = set() + self._extracting_in_flight: Set[str] = set() self._lock = threading.Lock() self._registry_lock = threading.Lock() self._state_lock = threading.Lock() - self._setup_registrar() - threading.Thread(target=self._setup_environment, daemon=True).start() + self._registered_menus: dict[str, tuple[str, str]] = {} - logger.info( - "[GlobalMenuService] Initialized native Python DBusMenu parser successfully" + self._setup_registrar() + self._subscribe_name_owner_changed() + threading.Thread(target=self._setup_environment, daemon=True).start() + threading.Thread(target=self._scan_initial_services, daemon=True).start() + + logger.info("[GlobalMenuService] Initialized event-driven registry") + + def _subscribe_name_owner_changed(self): + try: + bus = _get_bus() + bus.signal_subscribe( + "org.freedesktop.DBus", + "org.freedesktop.DBus", + "NameOwnerChanged", + "/org/freedesktop/DBus", + None, + Gio.DBusCallFlags.NONE, + self._on_name_owner_changed, + ) + logger.debug("[GlobalMenuService] Subscribed to NameOwnerChanged") + except Exception as e: + logger.error( + f"[GlobalMenuService] Failed to subscribe NameOwnerChanged: {e}" + ) + + def _on_name_owner_changed( + self, + _connection, + _sender_name, + _object_path, + _interface_name, + _signal_name, + parameters, + ): + try: + name, old_owner, new_owner = parameters.unpack() + except Exception: + return + + if not name or name.startswith(":") or name == "org.freedesktop.DBus": + return + + if new_owner and not old_owner: + threading.Thread( + target=self._on_service_appeared, + args=(name,), + daemon=True, + ).start() + elif not new_owner: + self._on_service_disappeared(name) + + def _on_service_appeared(self, service_name: str): + """Discover a newly appeared DBus service and populate the registry.""" + try: + bus = _get_bus() + pid = _dbus_get_pid(bus, service_name, _DBUS_TIMEOUT_FAST) + if pid <= 0: + return + executable = "" + try: + executable = os.path.realpath(f"/proc/{pid}/exe") + except Exception: + pass + with self._registry_lock: + if service_name not in self._service_registry: + self._service_registry[service_name] = AppServiceInfo( + service_name=service_name, + pid=pid, + executable=executable, + ) + if pid: + self._pid_to_service[pid] = service_name + if executable: + _exe_to_service[executable] = service_name + self._introspection_cache.invalidate(service_name) + logger.debug( + f"[GlobalMenuService] Registered new service: {service_name} pid={pid}" + ) + except Exception as e: + logger.debug( + f"[GlobalMenuService] Failed to register new service {service_name}: {e}" + ) + + def _on_service_disappeared(self, service_name: str): + with self._registry_lock: + entry = self._service_registry.pop(service_name, None) + if entry is None: + return + if entry.pid: + self._pid_to_service.pop(entry.pid, None) + if entry.importer and isinstance(entry.importer, DBusMenuClient): + try: + entry.importer.disconnect_signals() + except Exception: + pass + + self._introspection_cache.invalidate(service_name) + + with self._state_lock: + if entry and entry.wm_class and entry.wm_class == self._current_wm_class: + self._current_menu = None + + logger.debug( + f"[GlobalMenuService] Cleaned up disappeared service: {service_name}" ) def _setup_registrar(self): - """Host the AppMenu Registrar on DBus to allow GTK/KDE apps to register menus.""" try: - self._bus = _get_bus() # Shared connection — no extra allocation + self._bus = _get_bus() self._node = Gio.DBusNodeInfo.new_for_xml(REGISTRAR_XML) self._reg_id = self._bus.register_object( "/com/canonical/AppMenu/Registrar", @@ -177,39 +321,7 @@ class GlobalMenuService(Service): logger.error(f"[GlobalMenuService] Failed to start Registrar: {e}") def _setup_environment(self): - """Automatically setup environment variables and system overrides for global menus.""" - try: - # 1. Inject GTK Modules for standard GTK3 apps - # NOTE: We do NOT set os.environ here — that would affect Modus's own GTK windows, - # causing gdk_wayland_window_set_dbus_properties_libgtk_only assertion failures. - # The hyprctl/dbus-update-activation-environment calls set it for future apps instead. - exec_shell_command("hyprctl keyword env GTK_MODULES,appmenu-gtk-module") - exec_shell_command("hyprctl keyword env UBUNTU_MENUPROXY,1") - exec_shell_command( - "dbus-update-activation-environment --systemd GTK_MODULES=appmenu-gtk-module UBUNTU_MENUPROXY=1" - ) - - # 2. Automatically grant Flatpak apps permission to talk to the DBus Registrar - exec_shell_command( - "flatpak override --user --talk-name=com.canonical.AppMenu.Registrar" - ) - - # 3. Force XFCE / GTK3 apps to show menubars globally by writing to settings.ini - settings_path = os.path.expanduser("~/.config/gtk-3.0/settings.ini") - if os.path.exists(settings_path): - with open(settings_path, "r") as f: - content = f.read() - if "gtk-shell-shows-menubar" not in content: - with open(settings_path, "a") as f: - f.write("\ngtk-shell-shows-menubar=1\n") - - logger.info( - "[GlobalMenuService] Injected GTK environment and Flatpak overrides automatically" - ) - except Exception as e: - logger.warning( - f"[GlobalMenuService] Global menu environment setup failed: {e}" - ) + setup_global_menu_environment() def _handle_registrar_method( self, @@ -221,7 +333,6 @@ class GlobalMenuService(Service): parameters, invocation, ): - """Handles incoming DBus requests from apps registering their menus.""" try: if method_name == "RegisterWindow": window_id, menu_path = parameters.unpack() @@ -229,7 +340,6 @@ class GlobalMenuService(Service): f"[GlobalMenuService] App {sender} registered menu {menu_path} for window {window_id}" ) with self._registry_lock: - # the real bus name instead of echoing back the caller's address. self._registered_menus[sender] = (sender, menu_path) invocation.return_value(None) @@ -246,7 +356,6 @@ class GlobalMenuService(Service): svc_name, svc_path = entry invocation.return_value(GLib.Variant("(so)", (svc_name, svc_path))) else: - # Fallback — nothing registered for this sender invocation.return_value( GLib.Variant("(so)", (sender, "/com/canonical/menu/0")) ) @@ -254,15 +363,13 @@ class GlobalMenuService(Service): elif method_name == "GetMenus": invocation.return_value(GLib.Variant("(a(uso))", ([],))) except Exception as e: - logger.error( - f"[GlobalMenuService] Error handling Registrar method {method_name}: {e}" - ) + logger.error(f"[GlobalMenuService] Registrar error {method_name}: {e}") invocation.return_error_literal(Gio.DBusError.FAILED, "FAILED", str(e)) - def update_active_window(self, app_name: str, wm_class: str): - """Called when the active window changes. Triggers menu extraction.""" + def _uid(self, wm_class: str, pid: int) -> str: + return f"{wm_class}:{pid}" - # Capture the PID right now at focus time — don't re-query later in the thread + def update_active_window(self, app_name: str, wm_class: str): target_pid = 0 try: out = exec_shell_command("hyprctl activewindow -j") @@ -272,29 +379,21 @@ class GlobalMenuService(Service): except Exception: pass - # Snapshot current state under lock for race-free comparison with self._state_lock: same_class = wm_class == self._current_wm_class same_pid = target_pid != 0 and target_pid == self._current_pid - # Re-focus of EXACT same window instance → use cached menu, not stale _current_menu if same_class and same_pid: - with self._lock: - cached = self._menu_cache.get(wm_class) - if cached: - with self._state_lock: - self._current_menu = cached - self.menu_changed(cached) - else: - with self._state_lock: - current_menu = self._current_menu - if current_menu is not None: - self.menu_changed(current_menu) + with self._state_lock: + current_menu = self._current_menu + if current_menu is not None: + self.menu_changed(current_menu) return logger.info( f"[GlobalMenuService] Active window changed: {app_name} ({wm_class}) pid={target_pid}" ) + with self._state_lock: self._current_app = app_name self._current_wm_class = wm_class @@ -306,40 +405,43 @@ class GlobalMenuService(Service): self.active_app_changed(app_name, wm_class) - # Check cache — snapshot under lock, serve non-empty results - with self._lock: - cached_menu = self._menu_cache.get(wm_class) - has_good_cache = wm_class in self._client_cache and cached_menu is not None + # PRIORITY 1: Registry lookup by PID (persistent importers) + with self._registry_lock: + svc_name = self._pid_to_service.get(target_pid) if target_pid else None + registered_entry = ( + self._service_registry.get(svc_name) if svc_name else None + ) - if has_good_cache: - with self._state_lock: - self._current_menu = cached_menu - self._current_app = app_name - self._current_wm_class = wm_class - self._current_pid = target_pid - logger.info(f"[GlobalMenuService] Cache hit for '{wm_class}'") - self.menu_changed(cached_menu) + if registered_entry and registered_entry.importer: + cached_menu = getattr(registered_entry.importer, "_cached_items", None) + if cached_menu is not None and len(cached_menu) > 0: + with self._state_lock: + self._current_menu = cached_menu + logger.info(f"[GlobalMenuService] Registry cache hit for '{wm_class}'") + self.menu_changed(cached_menu) + return + thread = threading.Thread( + target=self._extract_from_importer, + args=(wm_class, seq, registered_entry.importer, ""), + daemon=True, + ) + thread.start() return - # Clear menu IMMEDIATELY for the new app (before in_flight check below). - # This ensures the old app's menu doesn't persist if an extraction is - # already running for this wm_class (e.g. user focused Kitty, then Zen, - # then Kitty again before the first extraction finished). + # No importer found in registry — must discover with self._state_lock: self._current_menu = None self.menu_changed([]) - # Dedup in-flight extractions — atomic check+add under lock - inflight_key = f"{wm_class}:{target_pid}" + inflight_key = self._uid(wm_class, target_pid) with self._lock: if inflight_key in self._extracting_in_flight: logger.debug( - f"[GlobalMenuService] Extraction already in-flight for '{inflight_key}' — skipping" + f"[GlobalMenuService] Extraction in-flight for '{inflight_key}' — skipping" ) return self._extracting_in_flight.add(inflight_key) - # Extract menu in background thread, passing captured pid logger.info( f"[GlobalMenuService] Starting extraction for '{wm_class}' pid={target_pid}" ) @@ -350,59 +452,110 @@ class GlobalMenuService(Service): ) thread.start() - def _discover_dbus_client( - self, wm_class: str, target_pid: int = 0 - ) -> Optional[DBusMenuClient]: - """Fast and deterministic DBusMenu discovery (Wayland-safe).""" + def _connect_importer_signals(self, entry: AppServiceInfo): + if entry.connected_signals or not isinstance(entry.importer, DBusMenuClient): + return + + svc_name = entry.service_name + + def on_layout_updated(parent_id, revision, _properties): + threading.Thread( + target=self._handle_layout_updated, + args=(svc_name, parent_id, revision), + daemon=True, + ).start() + + def on_items_updated(updated_props, removed_ids): + threading.Thread( + target=self._handle_items_updated, + args=(svc_name, updated_props, removed_ids), + daemon=True, + ).start() + + entry.importer.connect_signals( + on_layout_updated=on_layout_updated, + on_items_updated=on_items_updated, + on_item_activated=lambda item_id, ts: logger.debug( + f"[GlobalMenuService] Item activated: {item_id}" + ), + ) + entry.connected_signals = True + logger.debug(f"[GlobalMenuService] Connected signals for {entry.service_name}") + + def _handle_layout_updated(self, svc_name: str, parent_id: int, revision: int): + with self._registry_lock: + entry = self._service_registry.get(svc_name) + if not entry or not entry.importer: + return + entry.revision = revision + subtree = entry.importer.update_layout(parent_id, revision) + if subtree is None: + return + if parent_id == 0: + with self._state_lock: + if entry.wm_class == self._current_wm_class: + self._current_menu = subtree + with self._seq_lock: + self._extraction_seq += 1 + seq = self._extraction_seq + idle_add(self._emit_menu_changed, subtree, seq) + + def _handle_items_updated( + self, + svc_name: str, + updated_props: List[Tuple[int, dict]], + removed_ids: List[int], + ): + with self._registry_lock: + entry = self._service_registry.get(svc_name) + if ( + not entry + or not entry.importer + or not isinstance(entry.importer, DBusMenuClient) + ): + return + entry.importer.apply_items_update(updated_props, removed_ids) + with self._state_lock: + if entry.wm_class == self._current_wm_class: + self._current_menu = entry.importer.get_layout() + + def _extract_from_importer( + self, + wm_class: str, + seq: int, + client: Union[DBusMenuClient, GtkMenuClient, ActionMenuClient], + inflight_key: str, + ): try: - if target_pid <= 0: - out = exec_shell_command("hyprctl activewindow -j") - if not out: - logger.warning("[GlobalMenuService] hyprctl returned no output") - return None - target_pid = json.loads(out).get("pid", 0) + with self._seq_lock: + if seq != self._extraction_seq: + return - if target_pid <= 0: - return None + items = client.get_layout() or [] + with self._seq_lock: + current_seq = self._extraction_seq + + if seq != current_seq: + return + + with self._state_lock: + if wm_class == self._current_wm_class: + self._current_menu = items + + idle_add(self._emit_menu_changed, items, current_seq) + except Exception as e: + logger.error( + f"[GlobalMenuService] Error extracting from importer for '{wm_class}': {e}" + ) + finally: + if inflight_key: + with self._lock: + self._extracting_in_flight.discard(inflight_key) + + def _scan_initial_services(self): + try: bus = _get_bus() - - # 1. FAST PATH: Registrar cache - with self._registry_lock: - registered = dict(self._registered_menus) - - for sender, (service, path) in registered.items(): - try: - pid = _dbus_get_pid(bus, service, _DBUS_TIMEOUT_FAST) - if pid == target_pid or self._is_same_app(pid, target_pid): - logger.info( - f"[GlobalMenuService] Registrar match: {service} {path}" - ) - return DBusMenuClient(service, path) - except Exception as e: - if "NameHasNoOwner" in str(e): - with self._registry_lock: - self._registered_menus.pop(sender, None) - - # 2. FALLBACK: Scan DBus services matching the target PID - static_paths = [ - "/com/canonical/menu/0", - "/com/canonical/AppMenu/Registrar/0", - "/com/canonical/Unity/Panel/Service", - "/appmenu", - "/org/gtk/Application/menus/appmenu", - "/org/gtk/Application/menus/menubar", - "/org/gtk/Application/menus/appmenu/0", - "/org/gtk/Application/menus/menubar/0", - "/org/appmenu/gtk/window/menus/menubar", - "/org/appmenu/gtk/window/menus/appmenu", - "/org/appmenu/gtk/window/menus/menubar/0", - "/org/appmenu/gtk/window/menus/appmenu/0", - "/MenuBar", - "/KDEAppMenu", - "/org/xfce/Thunar/menus/menubar/0", - ] - res = bus.call_sync( "org.freedesktop.DBus", "/org/freedesktop/DBus", @@ -414,177 +567,345 @@ class GlobalMenuService(Service): -1, None, ) + if not res: + return names = res.get_child_value(0).unpack() - services = [ - n - for n in names - if n != "org.freedesktop.DBus" - and not n.startswith("org.freedesktop.DBus") + for name in names: + if name.startswith(":"): + continue + pid = _dbus_get_pid(bus, name, _DBUS_TIMEOUT_FAST) + if pid <= 0: + continue + executable = "" + try: + executable = os.path.realpath(f"/proc/{pid}/exe") + except Exception: + pass + with self._registry_lock: + if name not in self._service_registry: + self._service_registry[name] = AppServiceInfo( + service_name=name, + pid=pid, + executable=executable, + ) + if pid: + self._pid_to_service[pid] = name + if executable: + _exe_to_service[executable] = name + logger.debug( + f"[GlobalMenuService] Initial scan: registered {len(names)} services" + ) + except Exception as e: + logger.debug(f"[GlobalMenuService] Initial scan error: {e}") + + def _discover_dbus_client( + self, wm_class: str, target_pid: int = 0 + ) -> Optional[Union[DBusMenuClient, GtkMenuClient, ActionMenuClient]]: + try: + if target_pid <= 0: + out = exec_shell_command("hyprctl activewindow -j") + if not out: + return None + target_pid = json.loads(out).get("pid", 0) + if target_pid <= 0: + return None + + strategies = [ + self._find_registry, + self._find_registrar, + self._find_fallback_scan, ] - for svc in services: - try: - pid = _dbus_get_pid(bus, svc, _DBUS_TIMEOUT_PID) - if not (pid == target_pid or self._is_same_app(pid, target_pid)): - continue - - with self._registry_lock: - if reg_entry := self._registered_menus.get(svc): - logger.info( - f"[GlobalMenuService] Found Registrar path {reg_entry[1]} for {svc}" - ) - return DBusMenuClient(reg_entry[0], reg_entry[1]) - - _safe_cls = ( - re.sub(r"[^A-Za-z0-9_]", "", wm_class) if wm_class else "" - ) - _safe_cap = _safe_cls.capitalize() if _safe_cls else "" - - dynamic_bases = [ - "/MenuBar", - "/com/canonical/menu", - "/org/appmenu/gtk/window/menus/menubar", - "/org/appmenu/gtk/window/menus/appmenu", - "/org/appmenu/gtk/window", - "/org/gtk/Application/menus", - "/org/gtk/Application", - "/org/libreoffice", - "/org/xfce", - "/org/mate", - "/org/kde", - "/org/gnome", - "/org/enlightenment", - "/org/pantheon", - "/org/elementary", - "/com/deepin", - "/com/solus-project", - "/com/canonical", - ] - - if _safe_cls: - cls_lower = _safe_cls.lower() - for ns in _DE_NAMESPACES + [cls_lower]: - for app_part in (_safe_cap, _safe_cls): - for suffix in ( - "menus/menubar", - "menus/appmenu", - "menus", - ): - p = f"/org/{ns}/{app_part}/{suffix}" - if GLib.Variant.is_object_path(p): - dynamic_bases.append(p) - - fallback_paths = list(static_paths) - seen_fallback = set(fallback_paths) - - # Helper for safe path addition - def _add_paths(parent_path, xml_str): - for child in re.findall(r' Optional[Union[DBusMenuClient, GtkMenuClient, ActionMenuClient]]: + with self._registry_lock: + svc_name = self._pid_to_service.get(target_pid) + entry = self._service_registry.get(svc_name) if svc_name else None + if entry and entry.importer: + logger.info(f"[GlobalMenuService] Registry match: {svc_name}") + return entry.importer + return None + + def _find_registrar( + self, wm_class: str, target_pid: int + ) -> Optional[Union[DBusMenuClient, GtkMenuClient, ActionMenuClient]]: + bus = _get_bus() + with self._registry_lock: + registered = dict(self._registered_menus) + for sender, (service, path) in registered.items(): + try: + pid = _dbus_get_pid(bus, service, _DBUS_TIMEOUT_FAST) + if pid == target_pid or self._is_same_app(pid, target_pid): + logger.info( + f"[GlobalMenuService] Registrar match: {service} {path}" + ) + client = DBusMenuClient(service, path) + if client.get_layout(): + self._cache_discovery(service, path, pid, wm_class, client) + return client + gtk_client = GtkMenuClient(service, path) + if gtk_client.get_layout(): + self._cache_discovery(service, path, pid, wm_class, gtk_client) + return gtk_client + self._cache_discovery(service, path, pid, wm_class, client) + return client + except Exception as e: + if "NameHasNoOwner" in str(e): + with self._registry_lock: + self._registered_menus.pop(sender, None) + return None + + def _find_fallback_scan( + self, wm_class: str, target_pid: int + ) -> Optional[Union[DBusMenuClient, GtkMenuClient, ActionMenuClient]]: + bus = _get_bus() + try: + res = bus.call_sync( + "org.freedesktop.DBus", + "/org/freedesktop/DBus", + "org.freedesktop.DBus", + "ListNames", + None, + GLib.VariantType("(as)"), + Gio.DBusCallFlags.NONE, + -1, + None, + ) + if not res: + return None + names = res.get_child_value(0).unpack() + except Exception: + return None + + services = [ + n + for n in names + if n != "org.freedesktop.DBus" and not n.startswith("org.freedesktop.DBus") + ] + + for svc in services: + try: + pid = _dbus_get_pid(bus, svc, _DBUS_TIMEOUT_PID) + if not (pid == target_pid or self._is_same_app(pid, target_pid)): + continue + + with self._registry_lock: + if reg_entry := self._registered_menus.get(svc): + logger.info( + f"[GlobalMenuService] Found Registrar path {reg_entry[1]} for {svc}" + ) + client = DBusMenuClient(reg_entry[0], reg_entry[1]) + self._cache_discovery(svc, reg_entry[1], pid, wm_class, client) + return client + + _safe_cls = re.sub(r"[^A-Za-z0-9_]", "", wm_class) if wm_class else "" + _safe_cap = _safe_cls.capitalize() if _safe_cls else "" + + dynamic_bases = self._build_dynamic_bases(svc, _safe_cls, _safe_cap) + fallback_paths, seen_fallback = self._introspect_paths( + bus, svc, dynamic_bases + ) + + # PRIORITY A: DBusMenu (com.canonical.dbusmenu) + for path in fallback_paths: + xml = self._introspection_cache.get( + svc, path, bus, _DBUS_TIMEOUT_INTROSPECT + ) + if "com.canonical.dbusmenu" in xml: + logger.info(f"[GlobalMenuService] DBusMenu found: {svc} {path}") + client = DBusMenuClient(svc, path) + self._cache_discovery(svc, path, pid, wm_class, client) + return client + + # PRIORITY B: org.gtk.Menus + gtk_candidates = [] + for path in fallback_paths: + xml = self._introspection_cache.get( + svc, path, bus, _DBUS_TIMEOUT_INTROSPECT + ) + if "org.gtk.Menus" in xml: + gtk_candidates.append((svc, path)) + + for dfs_path in self._dfs_find_interface(svc, "org.gtk.Menus"): + if dfs_path not in {p for _, p in gtk_candidates}: + gtk_candidates.append((svc, dfs_path)) + + for gsvc, gpath in gtk_candidates: + try: + test_client = GtkMenuClient(gsvc, gpath) + test_items = test_client.get_layout() + if not test_items: + import time + + for _ in range(4): + time.sleep(0.05) + if test_items := test_client.get_layout(): + break + if test_items: + logger.info( + f"[GlobalMenuService] GTK menu found: {gsvc} {gpath} ({len(test_items)} items)" + ) + self._cache_discovery( + gsvc, gpath, pid, wm_class, test_client + ) + return test_client + except Exception: + pass + + # PRIORITY C: org.gtk.Actions (last resort) + action_candidates = ["/"] + name_path = ( + "/" + svc.replace(".", "/") if not svc.startswith(":") else "" + ) + if name_path and GLib.Variant.is_object_path(name_path): + action_candidates.append(name_path) + action_candidates.append("/org/gtk/Application") + + if _safe_cls: + cls_lower = _safe_cls.lower() + action_candidates.extend( + [f"/org/{cls_lower}/{_safe_cap}", f"/org/{cls_lower}"] + ) + for ns in _DE_NAMESPACES + [cls_lower]: + for app_part in (_safe_cap, _safe_cls): + p = f"/org/{ns}/{app_part}" + if GLib.Variant.is_object_path(p): + action_candidates.append(p) + + action_candidates.extend( + self._dfs_find_interface(svc, "org.gtk.Actions") + ) + + for act_base in action_candidates: + xml = self._introspection_cache.get( + svc, act_base, bus, _DBUS_TIMEOUT_FAST + ) + if 'interface name="org.gtk.Actions"' in xml: + test_client = ActionMenuClient(svc, act_base) + if test_items := test_client.get_layout(): + logger.info( + f"[GlobalMenuService] Action menu: {svc} {act_base} ({len(test_items)} items)" + ) + self._cache_discovery( + svc, act_base, pid, wm_class, test_client + ) + return test_client + + logger.info( + f"[GlobalMenuService] No menu found for {svc} wm={wm_class}" + ) + except Exception: + continue + return None + + def _build_dynamic_bases(self, svc: str, _safe_cls: str, _safe_cap: str) -> list: + static_paths = [ + "/MenuBar", + "/com/canonical/menu", + "/org/appmenu/gtk/window/menus/menubar", + "/org/appmenu/gtk/window/menus/appmenu", + "/org/appmenu/gtk/window", + "/org/gtk/Application/menus", + "/org/gtk/Application", + "/org/xfce", + "/org/cinnamon", + "/org/libreoffice", + "/org/mate", + "/org/kde", + "/org/gnome", + "/org/enlightenment", + "/org/pantheon", + "/org/elementary", + "/com/deepin", + "/com/solus-project", + "/com/canonical", + ] + + svc_path = "/" + svc.replace(".", "/") if not svc.startswith(":") else "" + if svc_path and GLib.Variant.is_object_path(svc_path): + static_paths.append(svc_path) + + if _safe_cls: + cls_lower = _safe_cls.lower() + for ns in _DE_NAMESPACES + [cls_lower]: + for app_part in (_safe_cap, _safe_cls): + for suffix in ("menus/menubar", "menus/appmenu", "menus"): + p = f"/org/{ns}/{app_part}/{suffix}" + if GLib.Variant.is_object_path(p): + static_paths.append(p) + return static_paths + + def _introspect_paths(self, bus, svc: str, bases: list) -> Tuple[list, set]: + fallback_paths = list(bases) + seen_fallback = set(fallback_paths) + + def _add_paths(parent_path, xml_str): + for child in re.findall(r' list[str]: - """Bounded DFS returning ALL paths containing a target interface in a service's object tree.""" if _node_count is None: _node_count = [0] if ( @@ -609,7 +929,9 @@ class GlobalMenuService(Service): _node_count[0] += 1 results: list[str] = [] try: - xml = _dbus_introspect(_get_bus(), service, path, _DBUS_TIMEOUT_INTROSPECT) + xml = self._introspection_cache.get( + service, path, _get_bus(), _DBUS_TIMEOUT_INTROSPECT + ) if not xml: return results @@ -640,14 +962,10 @@ class GlobalMenuService(Service): if pid1 == pid2: return True try: - # Compare resolved executable paths exe1 = os.path.realpath(f"/proc/{pid1}/exe") exe2 = os.path.realpath(f"/proc/{pid2}/exe") if exe1 and exe2 and exe1 == exe2: return True - - # Compare process names (comm) — handles cases where exe differs - # but the app is the same (e.g., launcher wrappers) try: with open(f"/proc/{pid1}/comm") as f: comm1 = f.read().strip() @@ -658,8 +976,6 @@ class GlobalMenuService(Service): except OSError: pass - # Also allow: one pid is the direct parent of the other - # (Qt apps register DBus under a child thread pid) def _ppid(p: int) -> int: try: with open(f"/proc/{p}/stat") as f: @@ -674,25 +990,10 @@ class GlobalMenuService(Service): pass return False - def _evict_cache_if_needed(self, key: str): - """Evict oldest cache entries when over the size limit. Must be called under _lock.""" - if not hasattr(self, "_cache_order"): - self._cache_order: list[str] = [] - if key not in self._cache_order: - self._cache_order.append(key) - - while len(self._cache_order) > _MAX_CACHE_SIZE: - oldest = self._cache_order.pop(0) - self._client_cache.pop(oldest, None) - self._menu_cache.pop(oldest, None) - logger.debug(f"[GlobalMenuService] Evicted cache for '{oldest}'") - def _extract_menu( self, wm_class: str, seq: int, target_pid: int = 0, inflight_key: str = "" ): - """Extract menu for a given WM class in a background thread.""" try: - # Bail early if a newer extraction has already superseded this one with self._seq_lock: if seq != self._extraction_seq: return @@ -705,9 +1006,6 @@ class GlobalMenuService(Service): if items: with self._lock: - self._client_cache[wm_class] = client - self._menu_cache[wm_class] = items - self._evict_cache_if_needed(wm_class) logger.info( f"[GlobalMenuService] Found menu for '{wm_class}': {len(items)} items" ) @@ -720,9 +1018,11 @@ class GlobalMenuService(Service): if seq != current_seq: return - # Capture seq for race-safe emission in idle callback - emit_seq = current_seq - idle_add(self._emit_menu_changed, items, emit_seq) + with self._state_lock: + if wm_class == self._current_wm_class: + self._current_menu = items + + idle_add(self._emit_menu_changed, items, current_seq) except Exception as e: logger.error( @@ -734,7 +1034,6 @@ class GlobalMenuService(Service): self._extracting_in_flight.discard(inflight_key) def _emit_menu_changed(self, items: List[DBusMenuItem], seq: int = 0): - """Emit menu_changed signal on the main thread. Checks seq to avoid stale emissions.""" if seq: with self._seq_lock: if seq != self._extraction_seq: @@ -742,39 +1041,40 @@ class GlobalMenuService(Service): with self._state_lock: self._current_menu = items self.menu_changed(items) - return False # Remove from idle queue + return False def click_item(self, item_id: int) -> bool: - """Click a menu item by ID.""" with self._state_lock: - wm_class = self._current_wm_class - with self._lock: - client = self._client_cache.get(wm_class) - if client: - client.click_item(item_id) + target_pid = self._current_pid + with self._registry_lock: + svc = self._pid_to_service.get(target_pid) + entry = self._service_registry.get(svc) if svc else None + if entry and entry.importer: + entry.importer.click_item(item_id) return True return False def about_to_show(self, item_id: int) -> bool: - """Triggers the AboutToShow event on a DBusMenu item.""" with self._state_lock: - wm_class = self._current_wm_class - with self._lock: - client = self._client_cache.get(wm_class) - if client: - return client.about_to_show(item_id) + target_pid = self._current_pid + with self._registry_lock: + svc = self._pid_to_service.get(target_pid) + entry = self._service_registry.get(svc) if svc else None + if entry and entry.importer: + return entry.importer.about_to_show(item_id) return False def refresh_menu_sync(self) -> List[DBusMenuItem]: - """Synchronously refetches the menu for the current app.""" with self._state_lock: - wm_class = self._current_wm_class - with self._lock: - client = self._client_cache.get(wm_class) - if client: - items = client.get_layout() - with self._lock: - self._menu_cache[wm_class] = items + target_pid = self._current_pid + with self._registry_lock: + svc = self._pid_to_service.get(target_pid) + entry = self._service_registry.get(svc) if svc else None + if entry and entry.importer: + items = entry.importer.get_layout(force_refresh=True) + with self._state_lock: + if target_pid == self._current_pid: + self._current_menu = items return items return [] diff --git a/src/services/modus.py b/src/services/modus.py index f06919a..536da1f 100644 --- a/src/services/modus.py +++ b/src/services/modus.py @@ -1,12 +1,54 @@ import json +import os +import subprocess +import sys from fabric.core.service import Property, Service, Signal from fabric.hyprland.service import Hyprland from fabric.utils import logger -from services.custom_notification import CachedNotifications +from utils.functions import ( + find_binary, + find_process_pid, + kill_process as _kill_process, + run_command, + spawn_detached, +) -notification_service = CachedNotifications() + +def __getattr__(name): + mod = sys.modules[__name__] + if name == "notification_service": + from services.custom_notification import CachedNotifications + + _inst = CachedNotifications() + setattr(mod, name, _inst) + return _inst + if name == "modus_service": + return get_modus_service() + if name == "audio_service": + try: + from fabric.audio import Audio + + _inst = Audio() + except Exception as e: + logger.error(f"[Main] Failed to create AudioService: {e}") + _inst = None + setattr(mod, name, _inst) + return _inst + if name == "screen_capture_service": + from services.screencapture import ScreenCapture + + _inst = ScreenCapture() + setattr(mod, name, _inst) + return _inst + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + + +def get_notification_service(): + from services.custom_notification import CachedNotifications + + return CachedNotifications() class ModusService(Service): @@ -146,7 +188,7 @@ class ModusService(Service): @volume.setter def volume(self, value: int): if value != self._volume: - self._name = value + self._volume = value self.volume_changed(value) @wlan.setter @@ -223,7 +265,6 @@ class ModusService(Service): def sc(self, signal_name: str, callback: callable, def_value="..."): self.connect(signal_name, callback) - # Return current property value instead of default if signal_name == "bluetooth-changed": return self.bluetooth if self.bluetooth else "Off" elif signal_name == "wlan-changed": @@ -243,7 +284,7 @@ class ModusService(Service): self._bluetooth = "" self._dock_apps = "" self._dont_disturb = False - self._current_active_app_name = "Finder" # Changed from "Hyprland" to "Finder" + self._current_active_app_name = "Finder" self._current_active_wm_class = "" self._current_workspace = "1" self._music = "" @@ -251,119 +292,350 @@ class ModusService(Service): self._dropdowns_hide = False self._dock_hidden = False self._show_notificationcenter = False - self._dock_width = 0 self._dock_height = 0 - # Initialize Hyprland connection for workspace and window monitoring - self._setup_workspace_monitoring() - self._setup_active_window_monitoring() - - def _setup_workspace_monitoring(self): - """Setup Hyprland connection and workspace monitoring""" - try: - self._hyprland_connection = Hyprland() - - # Get initial workspace - workspace_data = self._hyprland_connection.send_command( - "j/activeworkspace" - ).reply - active_workspace = json.loads(workspace_data.decode("utf-8"))["name"] - self._current_workspace = str(active_workspace) - - # Connect to workspace change events - self._hyprland_connection.connect( - "event::workspace", self._on_workspace_changed - ) - - except Exception as e: - logger.error(f"[ModusService] Failed to setup workspace monitoring: {e}") - self._current_workspace = "1" - - def _setup_active_window_monitoring(self): - """Setup active window monitoring""" - try: - if ( - not hasattr(self, "_hyprland_connection") - or not self._hyprland_connection - ): - return - - self._update_active_window() - except Exception as e: - logger.error( - f"[ModusService] Failed to setup active window monitoring: {e}" - ) - - def _update_active_window(self): - try: - if ( - not hasattr(self, "_hyprland_connection") - or not self._hyprland_connection - ): - return - - window_data = self._hyprland_connection.send_command("j/activewindow").reply - if not window_data: - self.current_active_app_name = "Finder" - return - - window_info = json.loads(window_data.decode("utf-8")) - wmclass = window_info.get("class", "") - title = window_info.get("title", "") - - if not title and not wmclass: - self.current_active_app_name = "Finder" - return - - # Delegate name resolution to app_name_resolver (lazy import to avoid circular) - from utils.app_name_resolver import app_name_resolver - - name = app_name_resolver.format_app_name(title, wmclass) - - # Set wm_class FIRST so that signal handlers reading it get the new value - self.current_active_wm_class = wmclass - self.current_active_app_name = name - - except Exception as e: - logger.error(f"[ModusService] Error updating active window: {e}") - self.current_active_app_name = "Finder" - - def _on_workspace_changed(self, obj, signal): - try: - workspace_name = json.loads(signal.data[0]) - self.current_workspace = str(workspace_name) - except Exception as e: - logger.error(f"[ModusService] Error processing workspace change: {e}") - - def remove_notification(self, id: int): - notification_service.remove_cached_notification(id) - self.notification_count_changed(self.notification_count) - - def clear_all_notifications(self): - notification_service.clear_all_cached_notifications() - self.notification_count_changed(self.notification_count) - - def get_cached_notifications(self): - return notification_service.cached_notifications - - def get_deserialized_with_ids(self): - return [ - (notif._notification, notif.cache_id) - for notif in notification_service.cached_notifications - ] - @property def notification_count(self) -> int: - return notification_service.count - - def toggle_dnd(self): - notification_service.toggle_dnd() - self.dont_disturb_changed(notification_service.dont_disturb) + return get_notification_service().count -global modus_service -try: - modus_service = ModusService() -except Exception as e: - logger.error("[Main] Failed to create ModusShellService:", e) +_modus_instance: ModusService | None = None +_hyprland_connection: Hyprland | None = None + + +def get_modus_service() -> ModusService: + global _modus_instance + if _modus_instance is None: + try: + _modus_instance = ModusService() + _setup_workspace_monitoring() + _setup_active_window_monitoring() + except Exception as e: + logger.error("[ModusService] Failed to create instance:", e) + raise + return _modus_instance + + +def _setup_workspace_monitoring(): + global _hyprland_connection + service = get_modus_service() + try: + _hyprland_connection = Hyprland() + service._hyprland_connection = _hyprland_connection + workspace_data = _hyprland_connection.send_command("j/activeworkspace").reply + active_workspace = json.loads(workspace_data.decode("utf-8"))["name"] + service._current_workspace = str(active_workspace) + _hyprland_connection.connect("event::workspace", _on_workspace_changed) + except Exception as e: + logger.error(f"[ModusService] Failed to setup workspace monitoring: {e}") + service._current_workspace = "1" + + +def _setup_active_window_monitoring(): + global _hyprland_connection + get_modus_service() + try: + if not _hyprland_connection: + return + _update_active_window() + except Exception as e: + logger.error(f"[ModusService] Failed to setup active window monitoring: {e}") + + +def _update_active_window(): + global _hyprland_connection + service = get_modus_service() + try: + if not _hyprland_connection: + return + window_data = _hyprland_connection.send_command("j/activewindow").reply + if not window_data: + service.current_active_app_name = "Finder" + return + window_info = json.loads(window_data.decode("utf-8")) + wmclass = window_info.get("class", "") + title = window_info.get("title", "") + if not title and not wmclass: + service.current_active_app_name = "Finder" + return + from utils.app_name_resolver import app_name_resolver + + name = app_name_resolver.format_app_name(title, wmclass) + service.current_active_wm_class = wmclass + service.current_active_app_name = name + except Exception as e: + logger.error(f"[ModusService] Error updating active window: {e}") + service.current_active_app_name = "Finder" + + +def _on_workspace_changed(obj, signal): + service = get_modus_service() + try: + workspace_name = json.loads(signal.data[0]) + service.current_workspace = str(workspace_name) + except Exception as e: + logger.error(f"[ModusService] Error processing workspace change: {e}") + + +def remove_notification(id: int): + get_notification_service().remove_cached_notification(id) + get_modus_service().notification_count_changed(get_notification_service().count) + + +def clear_all_notifications(): + get_notification_service().clear_all_cached_notifications() + get_modus_service().notification_count_changed(get_notification_service().count) + + +def get_cached_notifications(): + return get_notification_service().cached_notifications + + +def get_deserialized_with_ids(): + ns = get_notification_service() + return [(notif._notification, notif.cache_id) for notif in ns.cached_notifications] + + +def toggle_dnd(): + ns = get_notification_service() + ns.toggle_dnd() + get_modus_service().dont_disturb_changed(ns.dont_disturb) + + +# --- Night light orchestration --- +_night_light_active = False + + +def toggle_night_light() -> bool: + global _night_light_active + try: + if _night_light_active: + _kill_process("hyprsunset") + _night_light_active = False + else: + spawn_detached(["hyprsunset", "-t", "4500"]) + _night_light_active = True + return True + except Exception as e: + logger.error(f"[Modus] Failed to toggle night light: {e}") + return False + + +def set_night_light_temperature(temperature: int) -> bool: + global _night_light_active + try: + _kill_process("hyprsunset") + spawn_detached(["hyprsunset", "-t", str(temperature)]) + _night_light_active = True + return True + except Exception as e: + logger.error(f"[Modus] Failed to set night light temperature: {e}") + return False + + +def is_night_light_active() -> bool: + global _night_light_active + if _night_light_active: + return True + try: + _night_light_active = bool(find_process_pid("hyprsunset")) + return _night_light_active + except Exception: + return False + + +# --- Caffeine orchestration --- +_caffeine_process: subprocess.Popen | None = None + + +def is_caffeine_active() -> bool: + if _caffeine_process is not None: + return True + try: + return bool(find_process_pid("modus-inhibit", timeout=0.5)) + except Exception: + return False + + +def toggle_caffeine() -> bool: + global _caffeine_process + try: + if is_caffeine_active(): + from fabric.utils import get_relative_path + + inhibit_script = get_relative_path("utils/inhibit.py") + run_command(["python3", inhibit_script, "off"]) + if _caffeine_process: + _caffeine_process.terminate() + _caffeine_process = None + return False + else: + from fabric.utils import get_relative_path + + inhibit_script = get_relative_path("utils/inhibit.py") + _caffeine_process = spawn_detached(["python3", inhibit_script, "on"]) + return True + except Exception as e: + logger.error(f"[Modus] Failed to toggle caffeine: {e}") + return False + + +# --- Hyprland window management API --- + + +def _ensure_hyprland(): + """Ensure hyprland_connection is initialized.""" + global _hyprland_connection + if _hyprland_connection is None: + get_modus_service() + return _hyprland_connection + + +def _hyprctl_send(cmd: str) -> bytes | None: + conn = _ensure_hyprland() + if not conn: + return None + try: + reply = conn.send_command(cmd).reply + return reply + except Exception as e: + logger.error(f"[Modus] hyprctl send error ({cmd}): {e}") + return None + + +def _hyprctl_json(cmd: str) -> list | dict | None: + reply = _hyprctl_send(cmd) + if not reply: + return None + try: + return json.loads(reply.decode("utf-8")) + except Exception as e: + logger.error(f"[Modus] hyprctl JSON parse error ({cmd}): {e}") + return None + + +def get_clients() -> list: + data = _hyprctl_json("j/clients") + return data if isinstance(data, list) else [] + + +def get_active_window() -> dict: + data = _hyprctl_json("j/activewindow") + return data if isinstance(data, dict) else {} + + +def get_monitors() -> list: + data = _hyprctl_json("j/monitors") + return data if isinstance(data, list) else [] + + +def get_active_workspace_id() -> int: + data = _hyprctl_json("j/activeworkspace") + if isinstance(data, dict): + return data.get("id", -1) + return -1 + + +def get_monitor_for_workspace(workspace_id: int | None = None) -> dict | None: + if workspace_id is None: + workspace_id = get_active_workspace_id() + for monitor in get_monitors(): + if monitor.get("activeWorkspace", {}).get("id") == workspace_id: + return monitor + monitors = get_monitors() + return monitors[0] if monitors else None + + +def get_screen_dimensions() -> tuple[int, int]: + monitor = get_monitor_for_workspace() + if monitor: + return monitor.get("width", 1920), monitor.get("height", 1080) + return 1920, 1080 + + +def focus_window(address: str): + conn = _ensure_hyprland() + if conn: + conn.send_command(f"dispatch focuswindow address:{address}") + + +def close_window(address: str): + conn = _ensure_hyprland() + if conn: + conn.send_command(f"dispatch closewindow address:{address}") + + +def launch_app(command_line: str): + if not command_line: + return + import re + + cleaned = re.sub(r"%\w+", "", command_line).strip() + from fabric.utils import exec_shell_command_async + + exec_shell_command_async( + f"hyprctl dispatch 'hl.dsp.exec_cmd([[uwsm app -- {cleaned}]])'" + ) + + +def open_trash(): + trash_path = os.path.expanduser("~/.local/share/Trash/files") + if not os.path.exists(trash_path): + return + for fm in ["nautilus", "dolphin", "thunar", "nemo", "caja", "pcmanfm"]: + try: + if find_binary(fm): + spawn_detached([fm, trash_path]) + return + except Exception: + continue + + +# --- Occlusion (uses Hyprland API above) --- + + +def check_occlusion(occlusion_region, workspace=None) -> bool: + if workspace is None: + workspace = get_active_workspace_id() + + if isinstance(occlusion_region, tuple) and len(occlusion_region) == 2: + side, size = occlusion_region + if isinstance(side, str): + screen_width, screen_height = get_screen_dimensions() + if side.lower() == "bottom": + occlusion_region = (0, screen_height - size, screen_width, size) + elif side.lower() == "top": + occlusion_region = (0, 0, screen_width, size) + elif side.lower() == "left": + occlusion_region = (0, 0, size, screen_height) + elif side.lower() == "right": + occlusion_region = (screen_width - size, 0, size, screen_height) + + if not isinstance(occlusion_region, tuple) or len(occlusion_region) != 4: + return False + + clients = get_clients() + occ_x, occ_y, occ_width, occ_height = occlusion_region + occ_x2 = occ_x + occ_width + occ_y2 = occ_y + occ_height + + for client in clients: + if not client.get("mapped", False): + continue + client_workspace = client.get("workspace", {}) + if client_workspace.get("id") != workspace: + continue + position = client.get("at") + size = client.get("size") + if not position or not size: + continue + x, y = position + width, height = size + win_x1, win_y1 = x, y + win_x2, win_y2 = x + width, y + height + if not ( + win_x2 <= occ_x or win_x1 >= occ_x2 or win_y2 <= occ_y or win_y1 >= occ_y2 + ): + return True + return False diff --git a/src/services/screencapture.py b/src/services/screencapture.py index f5e8dd4..6fa98e5 100644 --- a/src/services/screencapture.py +++ b/src/services/screencapture.py @@ -367,4 +367,9 @@ class ScreenCapture(Service): return is_app_running("wf-recorder") or is_app_running("gpu-screen-recorder") -screen_capture_service = ScreenCapture() +def __getattr__(name): + if name == "screen_capture_service": + import sys + + return sys.modules["services.modus"].screen_capture_service + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/src/shared/data.py b/src/shared/data.py index 0f4c506..4399734 100644 --- a/src/shared/data.py +++ b/src/shared/data.py @@ -16,9 +16,18 @@ HOME_DIR = GLib.get_home_dir() CONFIG_DIR = os.path.expanduser(f"~/.config/{APP_NAME}") -screen = Gdk.Screen.get_default() -CURRENT_WIDTH = screen.get_width() -CURRENT_HEIGHT = screen.get_height() + +def _get_screen_dimensions(): + try: + screen = Gdk.Screen.get_default() + if screen: + return screen.get_width(), screen.get_height() + except Exception: + pass + return 1920, 1080 + + +CURRENT_WIDTH, CURRENT_HEIGHT = _get_screen_dimensions() WALLPAPERS_DIR_DEFAULT = get_relative_path("../assets/wallpapers_example/") diff --git a/src/shared/dialogs/about.py b/src/shared/dialogs/about.py index 2ab9adf..aae1a39 100644 --- a/src/shared/dialogs/about.py +++ b/src/shared/dialogs/about.py @@ -2,7 +2,7 @@ import subprocess from fabric.utils import GdkPixbuf, Gtk, get_relative_path, logger, os, re -from utils.functions import escape_markup_text +from utils.functions import escape_markup_text, find_binary from utils.icon_resolver import IconResolver from utils.utils import setup_cursor_hover @@ -70,9 +70,9 @@ def get_executable_path(exec_string): # Search in PATH try: - result = subprocess.run(["which", executable], capture_output=True, text=True) - if result.returncode == 0: - return result.stdout.strip() + result = find_binary(executable) + if result: + return result except Exception as e: logger.error(f"An error occurred: {e}") @@ -207,11 +207,9 @@ def get_app_info(wmclass): # Fallback: try to find executable in PATH location = "" try: - result = subprocess.run( - ["which", wmclass.lower()], capture_output=True, text=True - ) - if result.returncode == 0: - location = result.stdout.strip() + result = find_binary(wmclass.lower()) + if result: + location = result except Exception as e: logger.error(f"An error occurred: {e}") diff --git a/src/shared/widgets/clipping_box.py b/src/shared/widgets/clipping_box.py index dc14306..da1d757 100644 --- a/src/shared/widgets/clipping_box.py +++ b/src/shared/widgets/clipping_box.py @@ -1,7 +1,7 @@ import math from typing import cast -import cairo +from fabric.utils import cairo from fabric.widgets.box import Box diff --git a/src/shared/widgets/custom_image.py b/src/shared/widgets/custom_image.py index a62612e..5cee888 100644 --- a/src/shared/widgets/custom_image.py +++ b/src/shared/widgets/custom_image.py @@ -1,8 +1,7 @@ import math from typing import cast -import cairo -from fabric.utils import Gtk +from fabric.utils import cairo, Gtk from fabric.widgets.image import Image diff --git a/src/shared/widgets/flat_scale.py b/src/shared/widgets/flat_scale.py index 3d12059..34e246a 100644 --- a/src/shared/widgets/flat_scale.py +++ b/src/shared/widgets/flat_scale.py @@ -1,8 +1,7 @@ import math from typing import Callable, Iterable, Literal, TypedDict -import cairo -from fabric.utils import Gdk, GObject, Gtk +from fabric.utils import Gdk, GObject, Gtk, cairo from fabric.utils.helpers import clamp from fabric.widgets.widget import Widget from gi.repository import Pango, PangoCairo # noqa: E402 diff --git a/src/shared/widgets/smooth_switch.py b/src/shared/widgets/smooth_switch.py index 30be208..77c3fe4 100644 --- a/src/shared/widgets/smooth_switch.py +++ b/src/shared/widgets/smooth_switch.py @@ -1,5 +1,4 @@ -import cairo -from fabric.utils import Gdk, Gtk +from fabric.utils import Gdk, Gtk, cairo from shared.widgets.animator import Animator, ease_in_out diff --git a/src/shared/window/dropdown.py b/src/shared/window/dropdown.py index 64c9b03..892686a 100644 --- a/src/shared/window/dropdown.py +++ b/src/shared/window/dropdown.py @@ -1,8 +1,7 @@ -from fabric.utils import logger +from fabric.utils import GLib, logger from fabric.widgets.box import Box from fabric.widgets.centerbox import CenterBox from fabric.widgets.eventbox import EventBox -from gi.repository import GLib from shared.window.applet_window import AppletWindow from utils.roam import modus_service diff --git a/src/utils/dbusmenu.py b/src/utils/dbusmenu.py index 1ec870e..c5896ea 100644 --- a/src/utils/dbusmenu.py +++ b/src/utils/dbusmenu.py @@ -4,10 +4,11 @@ Supports: - com.canonical.dbusmenu (KDE / Unity / XFCE) - GTK appmenu-module DBusMenu - GTK GMenuModel fallback (best-effort) +- Incremental updates via DBusMenu signals """ import threading -from typing import List, Optional +from typing import Callable, Dict, List, Optional, Tuple from fabric.utils import Any, Gio, GLib, logger @@ -25,7 +26,6 @@ def _get_bus() -> Gio.DBusConnection: return _SESSION_BUS -# DATA MODEL — __slots__ cuts ~200 bytes per instance class DBusMenuItem: __slots__ = ( "id", @@ -39,6 +39,7 @@ class DBusMenuItem: "children", "action_name", "action_target", + "parent_id", ) def __init__( @@ -54,6 +55,7 @@ class DBusMenuItem: children: Optional[List["DBusMenuItem"]] = None, action_name: str = "", action_target: Optional[Any] = None, + parent_id: int = -1, ): self.id = id self.label = label @@ -66,9 +68,9 @@ class DBusMenuItem: self.children: List["DBusMenuItem"] = children if children is not None else [] self.action_name = action_name self.action_target = action_target + self.parent_id = parent_id -# MAIN CLIENT class DBusMenuClient: __slots__ = ( "service_name", @@ -78,6 +80,11 @@ class DBusMenuClient: "_hash_cache", "_lock", "_fetching", + "_revision", + "_signal_ids", + "_on_layout_updated_cb", + "_on_items_updated_cb", + "_on_item_activated_cb", ) def __init__(self, service_name: str, object_path: str): @@ -87,11 +94,15 @@ class DBusMenuClient: self._cache: Optional[List[DBusMenuItem]] = None self._cache_valid = False self._hash_cache: Optional[str] = None + self._revision: int = 0 self._lock = threading.Lock() self._fetching = False + self._signal_ids: List[int] = [] + self._on_layout_updated_cb: Optional[Callable] = None + self._on_items_updated_cb: Optional[Callable] = None + self._on_item_activated_cb: Optional[Callable] = None - # DBUS CALL — uses shared session bus def _call(self, method, params, reply_type=None): try: return _get_bus().call_sync( @@ -108,7 +119,6 @@ class DBusMenuClient: except Exception: return None - # PUBLIC ENTRY def get_layout(self, force_refresh=False): if self._cache and self._cache_valid and not force_refresh: return self._cache @@ -124,7 +134,6 @@ class DBusMenuClient: with self._lock: self._fetching = False - # CORE FETCH LOGIC def _fetch(self): params = GLib.Variant("(iias)", (0, -1, [])) res = self._call("GetLayout", params) @@ -133,6 +142,9 @@ class DBusMenuClient: return [] try: + revision = res.get_child_value(0).get_uint32() + self._revision = revision + layout = res.get_child_value(1) while layout.is_of_type(GLib.VariantType("v")): layout = layout.get_variant() @@ -152,7 +164,6 @@ class DBusMenuClient: logger.error(f"[Menu] parse error: {e}") return [] - # HASH DIFF SYSTEM def _hash(self, items): flat: List[str] = [] @@ -164,8 +175,7 @@ class DBusMenuClient: walk(items) return str(hash("".join(flat))) - # DBUSMENU PARSER - def _parse_dbusmenu(self, node, _depth: int = 0): + def _parse_dbusmenu(self, node, _depth: int = 0, _parent_id: int = -1): if _depth > 20: return DBusMenuItem(id=0) item = DBusMenuItem(id=0) @@ -175,6 +185,7 @@ class DBusMenuClient: return item item.id = node.get_child_value(0).get_int32() + item.parent_id = _parent_id try: props = node.get_child_value(1).unpack() @@ -215,7 +226,7 @@ class DBusMenuClient: c = children.get_child_value(i) if c.is_of_type(GLib.VariantType("v")): c = c.get_variant() - item.children.append(self._parse_dbusmenu(c, _depth + 1)) + item.children.append(self._parse_dbusmenu(c, _depth + 1, item.id)) if item.has_submenu and not item.children: try: @@ -228,7 +239,6 @@ class DBusMenuClient: return item - # ACTIONS def about_to_show(self, item_id): try: res = self._call( @@ -250,7 +260,6 @@ class DBusMenuClient: except Exception: pass - # CACHE CONTROL def invalidate(self): self._cache_valid = False @@ -263,3 +272,130 @@ class DBusMenuClient: def on_focus(self): self.prefetch_async() + + def get_revision(self) -> int: + return self._revision + + def connect_signals( + self, + on_layout_updated: Optional[Callable] = None, + on_items_updated: Optional[Callable] = None, + on_item_activated: Optional[Callable] = None, + ): + self._on_layout_updated_cb = on_layout_updated + self._on_items_updated_cb = on_items_updated + self._on_item_activated_cb = on_item_activated + + bus = _get_bus() + + def handle_layout_updated( + _connection, _sender, _path, _iface, _signal, params, _user_data=None + ): + try: + parent_id, revision, properties = params.unpack() + self._revision = revision + if self._on_layout_updated_cb: + self._on_layout_updated_cb(parent_id, revision, properties) + except Exception as e: + logger.debug(f"[DBusMenuClient] LayoutUpdated handler error: {e}") + + def handle_items_properties_updated( + _connection, _sender, _path, _iface, _signal, params, _user_data=None + ): + try: + updated_props, removed_ids = params.unpack() + if self._on_items_updated_cb: + self._on_items_updated_cb(updated_props, removed_ids) + self._cache_valid = False + except Exception as e: + logger.debug(f"[DBusMenuClient] ItemsPropertiesUpdated error: {e}") + + def handle_item_activation_requested( + _connection, _sender, _path, _iface, _signal, params, _user_data=None + ): + try: + item_id, timestamp = params.unpack() + if self._on_item_activated_cb: + self._on_item_activated_cb(item_id, timestamp) + except Exception as e: + logger.debug(f"[DBusMenuClient] ItemActivationRequested error: {e}") + + for sig_name, handler in [ + ("LayoutUpdated", handle_layout_updated), + ("ItemsPropertiesUpdated", handle_items_properties_updated), + ("ItemActivationRequested", handle_item_activation_requested), + ]: + sid = bus.signal_subscribe( + self.service_name, + "com.canonical.dbusmenu", + sig_name, + self.object_path, + None, + Gio.DBusCallFlags.NONE, + handler, + None, + ) + if sid: + self._signal_ids.append(sid) + + def disconnect_signals(self): + if not self._signal_ids: + return + bus = _get_bus() + for sid in self._signal_ids: + try: + bus.signal_unsubscribe(sid) + except Exception: + pass + self._signal_ids.clear() + + def update_layout( + self, parent_id: int, revision: int + ) -> Optional[List[DBusMenuItem]]: + self._revision = revision + params = GLib.Variant("(iias)", (parent_id, revision, [])) + res = self._call("GetLayout", params) + if not res: + return None + try: + layout = res.get_child_value(1) + while layout.is_of_type(GLib.VariantType("v")): + layout = layout.get_variant() + parsed = self._parse_dbusmenu(layout).children + if parent_id == 0: + self._hash_cache = self._hash(parsed) + self._cache = parsed + self._cache_valid = True + return parsed + except Exception as e: + logger.error(f"[Menu] update_layout error: {e}") + return None + + def apply_items_update( + self, updated_props: List[Tuple[int, Dict]], removed_ids: List[int] + ): + if not self._cache: + return + removed_set = set(removed_ids) + + def walk(items): + result = [] + for item in items: + if item.id in removed_set: + continue + for uid, props in updated_props: + if item.id == uid: + if "label" in props: + item.label = str(props["label"]).replace("_", "") + if "enabled" in props: + item.enabled = bool(props["enabled"]) + if "visible" in props: + item.visible = bool(props["visible"]) + if "type" in props: + item.type = str(props["type"]) + item.children = walk(item.children) + result.append(item) + return result + + self._cache = walk(self._cache) + self._hash_cache = None diff --git a/src/utils/functions.py b/src/utils/functions.py index b90eb44..0ad0521 100644 --- a/src/utils/functions.py +++ b/src/utils/functions.py @@ -7,7 +7,6 @@ import threading from typing import Dict, NamedTuple, Optional, TypeVar from fabric.utils import ( - exec_shell_command, exec_shell_command_async, logger, ) @@ -187,28 +186,64 @@ 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 - ) +# --- Process management --- + + +def spawn_detached(args: list[str]) -> subprocess.Popen: + return subprocess.Popen( + args, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) -# 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_process_running(process_name: str) -> bool: + try: + result = subprocess.run(["pidof", process_name], capture_output=True, text=True) + return bool(result.stdout.strip()) + except Exception: + return False + + def is_app_running(app_name: str) -> bool: - return len(exec_shell_command(f"pidof {app_name}")) != 0 + return is_process_running(app_name) + + +def find_process_pid(process_name: str, timeout: float | None = None) -> list[str]: + try: + result = subprocess.run( + ["pgrep", "-f", process_name], + capture_output=True, + text=True, + timeout=timeout, + ) + return [pid for pid in result.stdout.strip().split() if pid] + except Exception: + return [] + + +def toggle_command(command: str, full_command: str): + if is_process_running(command): + kill_process(command) + else: + spawn_detached(full_command.split(" ")) + + +# --- Binary lookup --- + + +def find_binary(name: str) -> str | None: + try: + result = subprocess.run(["which", name], capture_output=True, text=True) + return result.stdout.strip() or None + except Exception: + return None # General utilities diff --git a/src/utils/gtkmenu.py b/src/utils/gtkmenu.py index 1fbb96a..d1d2e88 100644 --- a/src/utils/gtkmenu.py +++ b/src/utils/gtkmenu.py @@ -1,24 +1,43 @@ -import re -from typing import Dict, List, Optional +from typing import Any, Dict, List, Optional from fabric.utils import Gio, GLib, logger from utils.dbusmenu import DBusMenuItem, _get_bus -_SUBMENU_RE = re.compile(r"submenu") + +class _ActionInfo: + __slots__ = ("enabled", "parameter_type", "state", "state_type") + + def __init__( + self, + enabled: bool = True, + parameter_type: str = "", + state: Optional[Any] = None, + state_type: str = "", + ): + self.enabled = enabled + self.parameter_type = parameter_type + self.state = state + self.state_type = state_type class GtkMenuClient: - __slots__ = ("service_name", "object_path", "_action_path", "_cached_items") + __slots__ = ( + "service_name", + "object_path", + "_action_path", + "_cached_items", + "_action_cache", + ) def __init__(self, service_name: str, object_path: str): self.service_name = service_name self.object_path = object_path - # Actions live at the parent of /menus/... self._action_path = ( object_path.split("/menus/")[0] if "/menus/" in object_path else object_path ) self._cached_items: List[DBusMenuItem] = [] + self._action_cache: Dict[str, _ActionInfo] = {} def _call(self, interface, method, params, reply_type=None, path=None): try: @@ -37,10 +56,74 @@ class GtkMenuClient: logger.debug(f"[GtkMenuClient] {method} failed: {e}") return None - def get_layout(self, parent_id: int = 0) -> List[DBusMenuItem]: - items = self._fetch_menu(parent_id) - self._cached_items = items - return items + def get_layout( + self, parent_id: int = 0, force_refresh: bool = False + ) -> List[DBusMenuItem]: + self._refresh_action_cache() + if force_refresh or not self._cached_items: + items = self._fetch_menu(parent_id) + self._cached_items = items + return self._cached_items + + def _refresh_action_cache(self): + try: + res = self._call( + "org.gtk.Actions", + "DescribeAll", + None, + GLib.VariantType("(a{s(bgav)})"), + path=self._action_path, + ) + if not res: + return + data = res.get_child_value(0).unpack() + self._action_cache.clear() + for name, (enabled, param_type_v, state_v) in data.items(): + param_type = str(param_type_v) if param_type_v else "" + state = ( + state_v.unpack() if state_v and state_v.n_children() > 0 else None + ) + state_type = "" + if state_v and state_v.n_children() > 0: + st = state_v.get_child_value(0) + state_type = st.get_type_string() if st else "" + self._action_cache[name] = _ActionInfo( + enabled=enabled, + parameter_type=param_type, + state=state, + state_type=state_type, + ) + except Exception as e: + logger.debug(f"[GtkMenuClient] DescribeAll cache error: {e}") + + def _describe_action(self, action_name: str) -> Optional[_ActionInfo]: + cached = self._action_cache.get(action_name) + if cached is not None: + return cached + try: + res = self._call( + "org.gtk.Actions", + "Describe", + GLib.Variant("(s)", (action_name,)), + GLib.VariantType("(bgav)"), + path=self._action_path, + ) + if res: + enabled, param_type_v, state_v = res.unpack() + param_type = str(param_type_v) if param_type_v else "" + state = ( + state_v.unpack() if state_v and state_v.n_children() > 0 else None + ) + state_type = "" + if state_v and state_v.n_children() > 0: + st = state_v.get_child_value(0) + state_type = st.get_type_string() if st else "" + info = _ActionInfo(enabled, param_type, state, state_type) + self._action_cache[action_name] = info + return info + except Exception as e: + logger.debug(f"[GtkMenuClient] Describe error for {action_name}: {e}") + return None def _find_item( self, item_id: int, items: Optional[List[DBusMenuItem]] = None @@ -153,20 +236,52 @@ class GtkMenuClient: action_target = item.action_target if not action_name: return - # Strip the unity. prefix added by appmenu-gtk-module if action_name.startswith("unity."): action_name = action_name[6:] + + action_info = self._describe_action(action_name) + use_change_state = False + if action_info and action_info.state is not None: + if action_info.state_type == "b": + use_change_state = True + elif action_info.parameter_type: + use_change_state = True + try: - params_arr = [action_target] if action_target is not None else [] - # Try menu path first (Thunar), fall back to action path (LibreOffice) for act_path in (self.object_path, self._action_path): try: - self._call( - "org.gtk.Actions", - "Activate", - GLib.Variant("(sava{sv})", (action_name, params_arr, {})), - path=act_path, - ) + if ( + use_change_state + and action_info + and action_info.state is not None + ): + new_state = ( + not action_info.state + if action_info.state_type == "b" + else action_info.state + ) + self._call( + "org.gtk.Actions", + "ChangeState", + GLib.Variant( + "(sv)", + ( + action_name, + GLib.Variant(action_info.state_type, new_state), + ), + ), + path=act_path, + ) + else: + params_arr = ( + [action_target] if action_target is not None else [] + ) + self._call( + "org.gtk.Actions", + "Activate", + GLib.Variant("(sava{sv})", (action_name, params_arr, {})), + path=act_path, + ) return except Exception: continue @@ -175,21 +290,28 @@ class GtkMenuClient: class ActionMenuClient: - """Flat menu from org.gtk.Actions when org.gtk.Menus is unavailable.""" - - __slots__ = ("service_name", "object_path", "_cached_items", "_action_map") + __slots__ = ( + "service_name", + "object_path", + "_action_path", + "_cached_items", + "_action_map", + "_action_cache", + ) def __init__(self, service_name: str, object_path: str): self.service_name = service_name self.object_path = object_path + self._action_path = object_path self._cached_items: List[DBusMenuItem] = [] self._action_map: Dict[int, str] = {} + self._action_cache: Dict[str, _ActionInfo] = {} - def _call(self, interface, method, params, reply_type=None): + def _call(self, interface, method, params, reply_type=None, path=None): try: return _get_bus().call_sync( self.service_name, - self.object_path, + path or self.object_path, interface, method, params, @@ -201,9 +323,160 @@ class ActionMenuClient: except Exception: return None - def get_layout(self, parent_id: int = 0) -> List[DBusMenuItem]: + def _refresh_action_cache(self): + try: + res = self._call( + "org.gtk.Actions", + "DescribeAll", + None, + GLib.VariantType("(a{s(bgav)})"), + ) + if not res: + return + data = res.get_child_value(0).unpack() + self._action_cache.clear() + for name, (enabled, param_type_v, state_v) in data.items(): + param_type = str(param_type_v) if param_type_v else "" + state = ( + state_v.unpack() if state_v and state_v.n_children() > 0 else None + ) + state_type = "" + if state_v and state_v.n_children() > 0: + st = state_v.get_child_value(0) + state_type = st.get_type_string() if st else "" + self._action_cache[name] = _ActionInfo( + enabled=enabled, + parameter_type=param_type, + state=state, + state_type=state_type, + ) + except Exception as e: + logger.debug(f"[ActionMenuClient] DescribeAll cache error: {e}") + + def _describe_action(self, action_name: str) -> Optional[_ActionInfo]: + cached = self._action_cache.get(action_name) + if cached is not None: + return cached + try: + res = self._call( + "org.gtk.Actions", + "Describe", + GLib.Variant("(s)", (action_name,)), + GLib.VariantType("(bgav)"), + ) + if res: + enabled, param_type_v, state_v = res.unpack() + param_type = str(param_type_v) if param_type_v else "" + state = ( + state_v.unpack() if state_v and state_v.n_children() > 0 else None + ) + state_type = "" + if state_v and state_v.n_children() > 0: + st = state_v.get_child_value(0) + state_type = st.get_type_string() if st else "" + info = _ActionInfo(enabled, param_type, state, state_type) + self._action_cache[action_name] = info + return info + except Exception as e: + logger.debug(f"[ActionMenuClient] Describe error for {action_name}: {e}") + return None + + def _fetch_menu( + self, menu_id: int, _depth: int = 0, _visited: Optional[set] = None + ) -> List[DBusMenuItem]: + if _depth > 10: + return [] + if _visited is None: + _visited = set() + if menu_id in _visited: + return [] + _visited.add(menu_id) + res = self._call( + "org.gtk.Menus", + "Start", + GLib.Variant("(au)", ([menu_id],)), + GLib.VariantType("(a(uuaa{sv}))"), + ) + if not res: + return [] + + try: + data = res.get_child_value(0).unpack() + if not data: + return [] + + entry_map = {} + for entry in data: + gid, mid, sections = entry + entry_map[(gid, mid)] = sections + + items: List[DBusMenuItem] = [] + + def resolve(sections, visited=None): + if visited is None: + visited = set() + for props in sections: + if not isinstance(props, dict): + continue + if ":section" in props: + ref = props[":section"] + if isinstance(ref, (list, tuple)) and len(ref) == 2: + key = (int(ref[0]), int(ref[1])) + if key not in visited: + visited.add(key) + child = entry_map.get(key) + if child is not None: + resolve(child, visited) + continue + if not props.get("label") and not any( + k in props for k in ("action", ":submenu", "submenu") + ): + continue + item = DBusMenuItem(id=hash((menu_id, len(items)))) + item.label = props.get("label", "").replace("_", "") + item.enabled = props.get("enabled", True) + item.visible = props.get("visible", True) + sub_ref = props.get(":submenu") or props.get("submenu") + if ( + sub_ref + and isinstance(sub_ref, (list, tuple)) + and len(sub_ref) == 2 + ): + item.has_submenu = True + item.children = self._fetch_menu( + int(sub_ref[0]), _depth + 1, _visited + ) + if "action" in props: + item.action_name = props["action"] + if "target" in props: + try: + item.action_target = GLib.Variant.new_variant( + GLib.Variant("v", props["target"]) + ) + except Exception: + pass + items.append(item) + + if data: + resolve(data[0][2]) + + return items + except Exception as e: + logger.error(f"[ActionMenuClient] Parse error: {e}") + return [] + + def get_layout( + self, parent_id: int = 0, force_refresh: bool = False + ) -> List[DBusMenuItem]: if parent_id != 0: return [] + self._refresh_action_cache() + + menu_items = self._fetch_menu(0) + if menu_items: + self._cached_items = menu_items + return menu_items + res = self._call( "org.gtk.Actions", "DescribeAll", @@ -217,14 +490,30 @@ class ActionMenuClient: data = res.get_child_value(0).unpack() items: List[DBusMenuItem] = [] self._action_map = {} - for i, (name, (enabled, param_type, state)) in enumerate(data.items()): + for i, (name, (enabled, param_type_v, state_v)) in enumerate(data.items()): label = name.replace("-", " ").replace("_", " ").strip().title() + param_type = str(param_type_v) if param_type_v else "" + state = None + state_type = "" + if state_v and state_v.n_children() > 0: + st = state_v.get_child_value(0) + state_type = st.get_type_string() if st else "" + try: + state = st.get_value() + except Exception: + state = state_v.unpack() if state_v else None + self._action_cache[name] = _ActionInfo( + enabled, param_type, state, state_type + ) item = DBusMenuItem(id=hash((0, i))) item.action_name = name - # Hide actions that need parameters — they can't work in a flat menu - item.visible = not bool(param_type) - item.enabled = enabled if not param_type else False + item.visible = not bool(param_type) or param_type == "b" + can_activate = not bool(param_type) + item.enabled = enabled if can_activate else False item.label = label + if param_type == "b": + item.type = "toggle" + item.enabled = enabled self._action_map[item.id] = name items.append(item) self._cached_items = items @@ -233,25 +522,68 @@ class ActionMenuClient: logger.error(f"[ActionMenuClient] parse error: {e}") return self._cached_items + def _find_item( + self, item_id: int, items: Optional[List[DBusMenuItem]] = None + ) -> Optional[DBusMenuItem]: + if items is None: + items = self._cached_items + for item in items: + if item.id == item_id: + return item + if item.children: + found = self._find_item(item_id, item.children) + if found: + return found + return None + def about_to_show(self, item_id: int) -> bool: return False def invalidate(self): self._cached_items = [] self._action_map = {} + self._action_cache.clear() def click_item( self, item_id: int, action_name: str = "", action_target=None ) -> None: if not action_name: - action_name = self._action_map.get(item_id, "") + item = self._find_item(item_id) + if item and item.action_name: + action_name = item.action_name + action_target = item.action_target if not action_name: return + + action_info = self._describe_action(action_name) + use_change_state = False + if action_info and action_info.state is not None: + if action_info.state_type == "b": + use_change_state = True + elif action_info.parameter_type: + use_change_state = True + try: - self._call( - "org.gtk.Actions", - "Activate", - GLib.Variant("(sava{sv})", (action_name, [], {})), - ) + if use_change_state and action_info and action_info.state is not None: + new_state = ( + not action_info.state + if action_info.state_type == "b" + else action_info.state + ) + self._call( + "org.gtk.Actions", + "ChangeState", + GLib.Variant( + "(sv)", + (action_name, GLib.Variant(action_info.state_type, new_state)), + ), + ) + else: + params_arr = [action_target] if action_target is not None else [] + self._call( + "org.gtk.Actions", + "Activate", + GLib.Variant("(sava{sv})", (action_name, params_arr, {})), + ) except Exception as e: logger.error(f"[ActionMenuClient] click failed: {e}") diff --git a/src/utils/icon_resolver.py b/src/utils/icon_resolver.py index ae3a0c8..cdcf345 100644 --- a/src/utils/icon_resolver.py +++ b/src/utils/icon_resolver.py @@ -5,6 +5,7 @@ from fabric.utils import GLib, Gtk, logger, os import shared.data as data ICON_CACHE_FILE = data.CACHE_DIR + "/icons.json" +ICON_DICT_MAX = 500 if not os.path.exists(data.CACHE_DIR): os.makedirs(data.CACHE_DIR) @@ -23,6 +24,7 @@ class IconResolver: else: self._icon_dict = {} + self._icon_dict_order: list[str] = [] self.default_applicaiton_icon = default_applicaiton_icon def get_icon_name(self, app_id: str): @@ -58,6 +60,10 @@ class IconResolver: def _store_new_icon(self, app_id: str, icon: str): self._icon_dict[app_id] = icon + self._icon_dict_order.append(app_id) + while len(self._icon_dict) > ICON_DICT_MAX: + oldest = self._icon_dict_order.pop(0) + self._icon_dict.pop(oldest, None) with open(ICON_CACHE_FILE, "w") as f: json.dump(self._icon_dict, f) diff --git a/src/utils/memory_tracker.py b/src/utils/memory_tracker.py new file mode 100644 index 0000000..2589d90 --- /dev/null +++ b/src/utils/memory_tracker.py @@ -0,0 +1,280 @@ +import os +import gc +import tracemalloc + +PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +_snapshot_counter = 0 +_initial_snapshot = None + + +def _is_project_module(filename: str) -> bool: + return filename.startswith(PROJECT_ROOT) and ".venv" not in filename + + +def start_tracing() -> None: + """Start tracemalloc + file-triggered dumps. + + Only activates if MODUS_MEMTRACE env var is set. + Every 2s checks for /tmp/modus_dump. Touch that file to trigger a dump. + + MODUS_MEMTRACE=1 uv run start + touch /tmp/modus_dump # triggers dump + """ + if not os.environ.get("MODUS_MEMTRACE"): + return + try: + tracemalloc.start(10) + global _initial_snapshot + _initial_snapshot = tracemalloc.take_snapshot() + + from fabric.utils import GLib + + GLib.timeout_add(2000, _poll_dump_file) + + print("[memory_tracker] active — touch /tmp/modus_dump to dump", flush=True) + except Exception as e: + _log_error(f"start_tracing failed: {e}") + print( + f"[memory_tracker] FAILED: {e}", + file=__import__("sys").stderr, + flush=True, + ) + + +def _log_error(msg: str) -> None: + try: + with open("/tmp/modus_memtrace_err.log", "a") as f: + f.write(f"{msg}\n") + except Exception: + pass + + +def _poll_dump_file() -> bool: + if os.path.exists("/tmp/modus_dump"): + try: + os.unlink("/tmp/modus_dump") + dump_memory_usage() + except Exception as e: + _log_error(f"dump failed: {e}") + return True + + +def _get_process_rss() -> int: + """Return RSS in bytes from /proc/self/status.""" + try: + with open("/proc/self/status") as f: + for line in f: + if line.startswith("VmRSS:"): + return int(line.split()[1]) * 1024 + except Exception: + pass + return 0 + + +def _estimate_obj_bytes(obj) -> int: + """Best-effort byte estimate for a live object. + Uses special handling for known types, falls back to sys.getsizeof. + """ + import sys + + try: + tname = type(obj).__name__ + tmod = getattr(type(obj), "__module__", "") + + # -- known C types with accessible pixel data -- + if "Pixbuf" in tname: + try: + return obj.get_width() * obj.get_height() * obj.get_n_channels() + except Exception: + pass + if "Surface" in tname and "cairo" in tmod: + try: + return obj.get_width() * obj.get_height() * 4 + except Exception: + pass + + # -- strings (include internal buffer) -- + if isinstance(obj, str): + return sys.getsizeof(obj) + + # -- bytes / bytearray -- + if isinstance(obj, (bytes, bytearray)): + return sys.getsizeof(obj) + + # -- dicts / lists / tuples (shallow — contents counted separately) -- + if isinstance(obj, (dict, list, tuple, set, frozenset)): + return sys.getsizeof(obj) + + # -- GObject wrappers: return Python wrapper size (C struct not included) -- + return sys.getsizeof(obj) + + except Exception: + return 0 + + +def _scan_all_objects() -> dict: + """Scan ALL live Python objects, group by type, show counts + size estimates.""" + + by_type: dict[str, dict] = {} + total_count = 0 + total_bytes = 0 + + for obj in gc.get_objects(): + try: + tname = type(obj).__name__ + tmod = getattr(type(obj), "__module__", "") + key = f"{tname}" + if tmod and tmod != "builtins": + key = f"{tmod}.{tname}" + if key not in by_type: + by_type[key] = {"count": 0, "bytes": 0} + by_type[key]["count"] += 1 + b = _estimate_obj_bytes(obj) + by_type[key]["bytes"] += b + total_count += 1 + total_bytes += b + except Exception: + pass + + sorted_types = sorted(by_type.items(), key=lambda x: -x[1]["bytes"])[:40] + + lines = [] + lines.append("--- ALL LIVE OBJECTS BY TYPE (top 40) ---") + lines.append(f"{'Type':<55} {'Count':>8} {'Est. MB':>8} {'%':>6}") + lines.append("-" * 79) + + for key, data in sorted_types: + mb = data["bytes"] / 1024 / 1024 + pct = (data["bytes"] / total_bytes * 100) if total_bytes else 0 + lines.append(f"{key:<55} {data['count']:>8} {mb:>8.2f} {pct:>5.1f}%") + + # summarise 3rd-party C / external memory gap + rss = _get_process_rss() + gap_mb = (rss - total_bytes) / 1024 / 1024 + if gap_mb > 0: + lines.append("") + lines.append( + f"--- C / non-Python memory (RSS − Python estimate): {gap_mb:.0f} MB ---" + ) + lines.append( + " (GtkWidget C structs, GLib internals, stacks, mmap'd data, etc.)" + ) + + return { + "lines": lines, + "by_type": dict(sorted_types), + "total_count": total_count, + "total_bytes": total_bytes, + "gap_mb": gap_mb if gap_mb > 0 else 0, + } + + +def _get_module_label(filename: str) -> str: + if not _is_project_module(filename): + return "" + rel = os.path.relpath(filename, PROJECT_ROOT) + return rel.replace(".py", "").replace("/", ".") + + +def dump_memory_usage() -> dict: + global _snapshot_counter + _snapshot_counter += 1 + gc.collect() + + rss = _get_process_rss() + snapshot = tracemalloc.take_snapshot() + growth = ( + snapshot.compare_to(_initial_snapshot, "lineno") if _initial_snapshot else None + ) + + stats = snapshot.statistics("lineno") + + module_totals: dict[str, dict] = {} + external_stats = {"count": 0, "size": 0} + + for stat in stats: + filename = stat.traceback[0].filename + label = _get_module_label(filename) + if label != "": + if label not in module_totals: + module_totals[label] = {"size": 0, "count": 0, "growth": 0} + module_totals[label]["size"] += stat.size + module_totals[label]["count"] += stat.count + else: + external_stats["size"] += stat.size + external_stats["count"] += stat.count + + if growth: + for g in growth: + filename = g.traceback[0].filename + label = _get_module_label(filename) + if label in module_totals: + module_totals[label]["growth"] += g.size_diff + + sorted_modules = sorted( + module_totals.items(), key=lambda x: x[1]["size"], reverse=True + ) + + total_size = sum(m["size"] for m in module_totals.values()) + external_stats["size"] + total_count = ( + sum(m["count"] for m in module_totals.values()) + external_stats["count"] + ) + + lines = [] + + def w(s=""): + lines.append(s) + + # === Process memory === + w(f"=== Process RSS: {rss / 1024 / 1024:.1f} MB ===") + w("") + + # === ALL live objects by type === + scan = _scan_all_objects() + for line in scan["lines"]: + w(line) + w("") + + # === Python allocations === + w("--- Python Allocations (tracemalloc) ---") + w(f"{'Module':<50} {'Count':>8} {'Size (MB)':>10} {'Growth (KB)':>12} {'%':>6}") + w("-" * 88) + + for label, data in sorted_modules: + size_mb = data["size"] / 1024 / 1024 + growth_kb = data.get("growth", 0) / 1024 + pct = (data["size"] / total_size * 100) if total_size else 0 + w( + f"{label:<50} {data['count']:>8} {size_mb:>10.2f} {growth_kb:>+12.1f} {pct:>5.1f}%" + ) + + if external_stats["count"]: + w( + f"{'':<50} {external_stats['count']:>8} {external_stats['size'] / 1024 / 1024:>10.2f}" + ) + + w("-" * 88) + w(f"{'TOTAL Python':<50} {total_count:>8} {total_size / 1024 / 1024:>10.2f}") + + out = "\n".join(lines) + + dump_path = f"/tmp/modus_memdump_{_snapshot_counter}.txt" + with open(dump_path, "w") as f: + f.write(out + "\n") + + print(out, flush=True) + + return { + "rss": rss, + "modules": dict(module_totals), + "external_count": external_stats["count"], + "external_size": external_stats["size"], + "total_size": total_size, + "total_count": total_count, + } + + +if __name__ == "__main__": + print("Usage: MODUS_MEMTRACE=1 uv run start") + print("Then: touch /tmp/modus_dump") diff --git a/src/utils/occlusion.py b/src/utils/occlusion.py index afa0f1d..e3d20f2 100644 --- a/src/utils/occlusion.py +++ b/src/utils/occlusion.py @@ -1,141 +1,17 @@ -from fabric.utils import logger -import json -import subprocess - -import shared.data as data +from services.modus import ( + check_occlusion as _check_occlusion, + get_screen_dimensions as _get_screen_dimensions, + get_active_workspace_id, +) def get_current_workspace(): - """ - Get the current workspace ID using hyprctl. - """ - try: - result = subprocess.run( - ["hyprctl", "activeworkspace"], capture_output=True, text=True - ) - # Assume the output similar to: "ID " - # Extracting the number from the output - parts = result.stdout.split() - for i, part in enumerate(parts): - if part == "ID" and i + 1 < len(parts): - return int(parts[i + 1]) - except Exception as e: - logger.error(f"Error getting current workspace: {e}") - return -1 + return get_active_workspace_id() def get_screen_dimensions(): - """ - Get screen dimensions from hyprctl. - - Returns: - tuple: (width, height) of the monitor containing the current workspace - """ - try: - # Get current workspace - workspace_id = get_current_workspace() - - # Get monitor information - result = subprocess.run( - ["hyprctl", "-j", "monitors"], capture_output=True, text=True - ) - monitors = json.loads(result.stdout) - - # Find the monitor containing our workspace - for monitor in monitors: - if monitor.get("activeWorkspace", {}).get("id") == workspace_id: - return monitor.get("width", data.CURRENT_WIDTH), monitor.get( - "height", data.CURRENT_HEIGHT - ) - - # Fallback to first monitor - if monitors: - return monitors[0].get("width", data.CURRENT_WIDTH), monitors[0].get( - "height", data.CURRENT_HEIGHT - ) - except Exception as e: - logger.error(f"Error getting screen dimensions: {e}") - - # Default fallback values - return data.CURRENT_WIDTH, data.CURRENT_HEIGHT + return _get_screen_dimensions() def check_occlusion(occlusion_region, workspace=None): - """ - Check if a region is occupied by any window on a given workspace. - - Parameters: - occlusion_region: Can be one of: - - tuple (side, size): where side is "top", "bottom", "left", or "right" - and size is the pixel width of the region - - tuple (x, y, width, height): The full region coordinates (legacy format) - workspace (int, optional): The workspace ID to check. If None, the current workspace is used. - - Returns: - bool: True if any window overlaps with the occlusion region, False otherwise. - """ - if workspace is None: - workspace = get_current_workspace() - - # Handle simplified side-based format - if isinstance(occlusion_region, tuple) and len(occlusion_region) == 2: - side, size = occlusion_region - if isinstance(side, str): - # Convert side-based format to coordinates - screen_width, screen_height = get_screen_dimensions() - - if side.lower() == "bottom": - occlusion_region = (0, screen_height - size, screen_width, size) - elif side.lower() == "top": - occlusion_region = (0, 0, screen_width, size) - elif side.lower() == "left": - occlusion_region = (0, 0, size, screen_height) - elif side.lower() == "right": - occlusion_region = (screen_width - size, 0, size, screen_height) - - # Ensure occlusion_region is in the correct format (x, y, width, height) - if not isinstance(occlusion_region, tuple) or len(occlusion_region) != 4: - print(f"Invalid occlusion region format: {occlusion_region}") - return False - - try: - result = subprocess.run( - ["hyprctl", "-j", "clients"], capture_output=True, text=True - ) - clients = json.loads(result.stdout) - except Exception as e: - logger.error(f"Error retrieving client windows: {e}") - return False - - occ_x, occ_y, occ_width, occ_height = occlusion_region - occ_x2 = occ_x + occ_width - occ_y2 = occ_y + occ_height - - for client in clients: - # Check if client is mapped - if not client.get("mapped", False): - continue - - # Ensure client has proper workspace information and matches the workspace - client_workspace = client.get("workspace", {}) - if client_workspace.get("id") != workspace: - continue - - # Ensure client has position and size info - position = client.get("at") - size = client.get("size") - if not position or not size: - continue - - x, y = position - width, height = size - win_x1, win_y1 = x, y - win_x2, win_y2 = x + width, y + height - - # Check for intersection between the window and occlusion region - if not ( - win_x2 <= occ_x or win_x1 >= occ_x2 or win_y2 <= occ_y or win_y1 >= occ_y2 - ): - return True # Occlusion region is occupied - - return False # No window overlaps the occlusion region + return _check_occlusion(occlusion_region, workspace) diff --git a/src/utils/roam.py b/src/utils/roam.py index 20a791e..1d1c7b5 100644 --- a/src/utils/roam.py +++ b/src/utils/roam.py @@ -1,44 +1,19 @@ -from fabric.audio import Audio -from fabric.utils import logger +_services_mod = None -# ruff: noqa: I001 -from services.modus import ( - ModusService, - notification_service as notification_service_instance, -) -global modus_service -try: - modus_service = ModusService() -except Exception as e: - logger.error(f"[Main] Failed to create ModusService: {e}") - modus_service = None +def _mod(): + global _services_mod + if _services_mod is None: + import services.modus as _services_mod # noqa: F811 + return _services_mod -if modus_service is None: - logger.warning( - "[Main] ModusService was not initialized. Functionality may be limited." - ) -global notification_service -try: - notification_service = notification_service_instance -except Exception as e: - logger.error(f"[Main] Failed to create NotificationService: {e}") - notification_service = None - -if notification_service is None: - logger.warning( - "[Main] NotificationService was not initialized. Notifications may not work." - ) - -global audio_service -try: - audio_service = Audio() -except Exception as e: - logger.error(f"[Main] Failed to create AudioService: {e}") - audio_service = None - -if audio_service is None: - logger.warning( - "[Main] AudioService was not initialized. Audio features may not work." - ) +def __getattr__(name): + if name in ( + "modus_service", + "notification_service", + "audio_service", + "screen_capture_service", + ): + return getattr(_mod(), name) + raise AttributeError(f"module {__name__!r} has no attribute {name!r}")