feat(globalmenu): got a prototype of working globalmenu for some apps

This commit is contained in:
sankalp committed 2026-06-22 15:58:10 +05:45
1 parent dd9cd58106
commit 2a15d71c33
10 files changed
+930 -123

No files matched your search

+3
View File
@@ -1,5 +1,8 @@
local modus = "$HOME/.config/Modus"
-- Global Menu: auto-set env vars so Qt/GTK apps export their menus
hl.env("GTK_MODULES", "appmenu-gtk-module")
hl.on("hyprland.start", function()
local cmds = {
"uwsm app -- awww-daemon",
+17
View File
@@ -48,6 +48,10 @@ PACKAGES=(
uwsm
cinnamon-desktop
ddcutil
at-spi2-core
gcc
make
pkgconf
)
# Colors and formatting
@@ -286,6 +290,19 @@ else
success "All packages are up-to-date"
fi
progress "Building Global Menu C library"
step "Compiling libglobalmenu.so..."
if [ -f "$INSTALL_DIR/native/Makefile" ]; then
if make -C "$INSTALL_DIR/native" clean >/dev/null 2>&1 && make -C "$INSTALL_DIR/native" >/dev/null 2>&1; then
success "libglobalmenu.so built successfully"
else
warn "Failed to build libglobalmenu.so - global menu may not work"
fi
else
warn "native/Makefile not found - skipping C library build"
fi
progress "Configuring Hyprland"
HYPR_CONFIG="$HOME/.config/hypr/hyprland.lua"
+341
View File
@@ -0,0 +1,341 @@
"""Global Menu Service - extracts and manages application menus.
Uses the pure Python DBusMenu extractor via Gio.DBus.
"""
import threading
import json
import subprocess
from typing import Optional, List
from fabric.core.service import Property, Service, Signal
from fabric.utils import logger
from gi.repository import Gio, GLib
from utils.dbusmenu import DBusMenuClient, DBusMenuItem
REGISTRAR_XML = """
<node>
<interface name="com.canonical.AppMenu.Registrar">
<method name="RegisterWindow">
<arg type="u" name="windowId" direction="in"/>
<arg type="o" name="menuObjectPath" direction="in"/>
</method>
<method name="UnregisterWindow">
<arg type="u" name="windowId" direction="in"/>
</method>
<method name="GetMenuForWindow">
<arg type="u" name="windowId" direction="in"/>
<arg type="s" name="service" direction="out"/>
<arg type="o" name="menuObjectPath" direction="out"/>
</method>
<method name="GetMenus">
<arg type="a(uso)" name="menus" direction="out"/>
</method>
</interface>
</node>
"""
class GlobalMenuService(Service):
"""Service that extracts and provides application menus for the global menu bar."""
@Signal
def menu_changed(self, menu_items: object) -> None: ...
@Signal
def active_app_changed(self, app_name: str, wm_class: str) -> None: ...
@Property(object, flags="read-write")
def current_menu(self) -> Optional[List[DBusMenuItem]]:
return self._current_menu
@Property(str, flags="read-write")
def current_app(self) -> str:
return self._current_app
@Property(str, flags="read-write")
def current_wm_class(self) -> str:
return self._current_wm_class
def __init__(self):
super().__init__()
self._current_menu: Optional[List[DBusMenuItem]] = None
self._current_app: str = ""
self._current_wm_class: str = ""
# Cache holds DBusMenuClient instances
self._client_cache: dict[str, Optional[DBusMenuClient]] = {}
# Cache holds the extracted top-level menus
self._menu_cache: dict[str, Optional[List[DBusMenuItem]]] = {}
# Registrar mapping: sender -> object_path
self._registered_menus: dict[str, str] = {}
self._setup_registrar()
self._lock = threading.Lock()
self._extraction_seq = 0 # Monotonic counter to discard stale results
logger.info(
"[GlobalMenuService] Initialized native Python DBusMenu parser successfully"
)
def _setup_registrar(self):
"""Host the AppMenu Registrar on DBus to allow GTK/KDE apps to register menus."""
try:
self._bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
self._node = Gio.DBusNodeInfo.new_for_xml(REGISTRAR_XML)
self._reg_id = self._bus.register_object(
"/com/canonical/AppMenu/Registrar",
self._node.interfaces[0],
self._handle_registrar_method,
None,
None,
)
Gio.bus_own_name_on_connection(
self._bus,
"com.canonical.AppMenu.Registrar",
Gio.BusNameOwnerFlags.NONE,
None,
None,
)
logger.info("[GlobalMenuService] AppMenu Registrar DBus service started")
except Exception as e:
logger.error(f"[GlobalMenuService] Failed to start Registrar: {e}")
def _handle_registrar_method(
self,
connection,
sender,
object_path,
interface_name,
method_name,
parameters,
invocation,
):
"""Handles incoming DBus requests from apps registering their menus."""
try:
if method_name == "RegisterWindow":
window_id, menu_path = parameters.unpack()
logger.debug(
f"[GlobalMenuService] App {sender} registered menu {menu_path} for window {window_id}"
)
self._registered_menus[sender] = menu_path
invocation.return_value(None)
elif method_name == "UnregisterWindow":
invocation.return_value(None)
elif method_name == "GetMenuForWindow":
invocation.return_value(
GLib.Variant("(so)", (sender, "/com/canonical/menu/0"))
)
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}"
)
invocation.return_error_literal(Gio.DBusError.FAILED, str(e))
def update_active_window(self, app_name: str, wm_class: str):
"""Called when the active window changes. Triggers menu extraction."""
if wm_class == self._current_wm_class:
return
logger.info(
f"[GlobalMenuService] Active window changed: {app_name} ({wm_class})"
)
self._current_app = app_name
self._current_wm_class = wm_class
self._extraction_seq += 1
seq = self._extraction_seq
self.active_app_changed(app_name, wm_class)
# Check cache first
if wm_class in self._menu_cache and wm_class in self._client_cache:
cached_menu = self._menu_cache[wm_class]
self._current_menu = cached_menu
logger.info(f"[GlobalMenuService] Cache hit for '{wm_class}'")
self.menu_changed(cached_menu if cached_menu else [])
return
# Clear menu immediately for new app while extraction runs
self._current_menu = None
self.menu_changed([])
# Extract menu in background thread
logger.info(f"[GlobalMenuService] Starting extraction for '{wm_class}'")
thread = threading.Thread(
target=self._extract_menu,
args=(wm_class, seq),
daemon=True,
)
thread.start()
def _discover_dbus_client(self) -> Optional[DBusMenuClient]:
"""Discovers the DBusMenu object path for the active Wayland window using Gio.DBus."""
try:
out = subprocess.check_output(["hyprctl", "activewindow", "-j"], timeout=2)
data = json.loads(out)
target_pid = data.get("pid", 0)
if target_pid <= 0:
return None
bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
res = bus.call_sync(
"org.freedesktop.DBus",
"/org/freedesktop/DBus",
"org.freedesktop.DBus",
"ListNames",
None,
GLib.VariantType("(as)"),
Gio.DBusCallFlags.NONE,
-1,
None,
)
names = res.get_child_value(0).unpack()
for svc in names:
if (
svc.startswith("org.freedesktop.systemd")
or svc.startswith("org.freedesktop.login1")
or svc == "org.freedesktop.DBus"
):
continue
try:
pid_res = bus.call_sync(
"org.freedesktop.DBus",
"/org/freedesktop/DBus",
"org.freedesktop.DBus",
"GetConnectionUnixProcessID",
GLib.Variant("(s)", (svc,)),
GLib.VariantType("(u)"),
Gio.DBusCallFlags.NONE,
1000,
None,
)
pid = pid_res.get_child_value(0).get_uint32()
if pid == target_pid:
# 1. Check if this app explicitly registered its menu via Registrar
if svc in self._registered_menus:
path = self._registered_menus[svc]
logger.info(
f"[GlobalMenuService] Found Registrar path {path} for {svc}"
)
return DBusMenuClient(svc, path)
# 2. Fallback to common paths
for path in [
"/com/canonical/menu/0",
"/com/canonical/menu/1",
"/MenuBar",
"/MenuBar/1",
"/org/xfce/Thunar/menus/menubar/0",
"/com/canonical/AppMenu/Registrar/0",
]:
try:
introspect_res = bus.call_sync(
svc,
path,
"org.freedesktop.DBus.Introspectable",
"Introspect",
None,
GLib.VariantType("(s)"),
Gio.DBusCallFlags.NONE,
500,
None,
)
xml = introspect_res.get_child_value(0).get_string()
if "com.canonical.dbusmenu" in xml:
logger.info(
f"[GlobalMenuService] Discovered DBusMenu natively at {svc} {path}"
)
return DBusMenuClient(svc, path)
except Exception:
pass
except Exception:
pass
except Exception as e:
logger.debug(f"[GlobalMenuService] DBus discovery failed: {e}")
return None
def _extract_menu(self, wm_class: str, seq: int):
"""Extract menu for a given WM class in a background thread."""
try:
client = self._discover_dbus_client()
items = []
if client:
items = client.get_layout()
with self._lock:
self._client_cache[wm_class] = client
self._menu_cache[wm_class] = items
if items:
logger.info(
f"[GlobalMenuService] Found menu for '{wm_class}': {len(items)} items"
)
else:
logger.info(
f"[GlobalMenuService] No DBus menu found for '{wm_class}'"
)
# Only update UI if this extraction is still current
if seq != self._extraction_seq:
return
# Update UI on main thread via GLib idle
GLib.idle_add(self._emit_menu_changed, items)
except Exception as e:
logger.error(
f"[GlobalMenuService] Error extracting menu for '{wm_class}': {e}"
)
def _emit_menu_changed(self, items: List[DBusMenuItem]):
"""Emit menu_changed signal on the main thread."""
self._current_menu = items
self.menu_changed(items)
return False # Remove from idle queue
def click_item(self, item_id: int) -> bool:
"""Click a menu item by ID."""
client = self._client_cache.get(self._current_wm_class)
if client:
client.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."""
client = self._client_cache.get(self._current_wm_class)
if client:
return client.about_to_show(item_id)
return False
def refresh_menu_sync(self) -> List[DBusMenuItem]:
"""Synchronously refetches the menu for the current app."""
client = self._client_cache.get(self._current_wm_class)
if client:
items = client.get_layout()
with self._lock:
self._menu_cache[self._current_wm_class] = items
return items
return []
_global_menu_svc_instance = None
def get_global_menu_service() -> GlobalMenuService:
global _global_menu_svc_instance
if _global_menu_svc_instance is None:
_global_menu_svc_instance = GlobalMenuService()
return _global_menu_svc_instance
+2 -1
View File
@@ -324,8 +324,9 @@ class ModusService(Service):
else:
name = "Finder"
self.current_active_app_name = name
# 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}")
+56 -43
View File
@@ -67,12 +67,9 @@ class PlayerService(Service):
@Property(str, "readable", default_value="")
def arturl(self) -> str:
try:
meta = self._player.props.metadata
if meta and "mpris:artUrl" in meta.keys():
return meta["mpris:artUrl"]
except Exception as e:
logger.error(f"An error occurred: {e}")
return ""
return self._player.props.metadata["mpris:artUrl"] or ""
except Exception:
return ""
@Property(str, "readable", default_value="")
def title(self) -> str:
@@ -84,22 +81,16 @@ class PlayerService(Service):
@Property(object, "readable")
def artist(self) -> list:
try:
meta = self._player.props.metadata
if meta and "xesam:artist" in meta.keys():
return list(meta["xesam:artist"]) or []
except Exception as e:
logger.error(f"An error occurred: {e}")
return []
return list(self._player.props.metadata["xesam:artist"]) or []
except Exception:
return []
@Property(str, "readable", default_value="")
def album(self) -> str:
try:
meta = self._player.props.metadata
if meta and "xesam:album" in meta.keys():
return meta["xesam:album"] or ""
except Exception as e:
logger.error(f"An error occurred: {e}")
return ""
return self._player.props.metadata["xesam:album"] or ""
except Exception:
return ""
@Property(str, "readable", default_value="Stopped")
def playback_status(self) -> str:
@@ -117,24 +108,17 @@ class PlayerService(Service):
@Property(int, "readable", default_value=0)
def length(self) -> int:
try:
meta = self._player.props.metadata
if meta and "mpris:length" in meta.keys():
return int(meta["mpris:length"])
except Exception as e:
logger.error(f"An error occurred: {e}")
return 0
return int(self._player.props.metadata["mpris:length"])
except Exception:
return 0
@Property(int, "read-write", default_value=0)
@Property(int, "readable", default_value=0)
def position(self) -> int:
try:
return int(self._player.get_position())
except Exception:
return 0
@position.setter
def position(self, value: int):
self.set_position(value / 1_000_000)
def play_pause(self, *_):
try:
self._player.play_pause()
@@ -187,7 +171,7 @@ class PlayerService(Service):
metadata = self._player.props.metadata
if metadata:
self.meta_change(metadata, self._player)
self._handle_artwork(metadata, metadata.keys())
self._handle_artwork(metadata)
except Exception as e:
logger.warning(f"Failed to initialize metadata: {e}")
@@ -207,10 +191,30 @@ class PlayerService(Service):
if self._is_cleaning_up:
return
self.pos_fabricator.stop()
current = self.get_position() / 1_000_000
try:
self._player.set_position(int(pos * 1_000_000))
except GLib.Error:
try:
offset = pos - current
self._player.seek(int(offset * 1_000_000))
except GLib.Error as e:
logger.error(f"Failed to seek: {e}")
finally:
if self.playback_status.lower() == "playing":
self.pos_fabricator.start()
def seek(self, offset: float):
if self._is_cleaning_up:
return
self.pos_fabricator.stop()
try:
self._player.seek(int(offset * 1_000_000))
except GLib.Error as e:
logger.error(f"Failed to seek: {e}")
finally:
if self.playback_status.lower() == "playing":
self.pos_fabricator.start()
def poll_progress(self):
if self._is_cleaning_up:
@@ -220,20 +224,26 @@ class PlayerService(Service):
else:
self.pos_fabricator.stop()
def fabricating(self):
def fabricating(self, metadata=None):
if self._is_cleaning_up:
return
try:
pos = self._player.get_position() / 1_000_000
keys = self._player.props.metadata.keys()
dur = (
self._player.props.metadata["mpris:length"] / 1_000_000
if "mpris:length" in keys
else 0
)
self.track_position(pos, dur)
except GLib.Error as e:
logger.warning(f"Failed to get position: {e}")
return
dur = 0
if metadata is None:
try:
metadata = self._player.props.metadata
except Exception:
metadata = None
if metadata is not None:
try:
dur = metadata["mpris:length"] / 1_000_000
except Exception:
dur = 0
self.track_position(pos, dur)
def on_seeked(self, player, position):
if self._is_cleaning_up:
@@ -266,13 +276,16 @@ class PlayerService(Service):
if self._is_cleaning_up:
return
self.meta_change(metadata, player)
self._handle_artwork(metadata, metadata.keys())
self._handle_artwork(metadata)
self.fabricating(metadata)
def _handle_artwork(self, metadata, keys):
if self._is_cleaning_up or "mpris:artUrl" not in keys:
def _handle_artwork(self, metadata):
if self._is_cleaning_up:
return
try:
art_url = metadata["mpris:artUrl"]
except Exception:
return
art_url = metadata["mpris:artUrl"]
artwork_hash = hashlib.md5(art_url.encode()).hexdigest()
if artwork_hash == self._current_artwork_hash:
+1 -1
View File
@@ -73,8 +73,8 @@ class AppName:
name = name.split(".")[-1]
if update:
modus_service.current_active_app_name = name
modus_service.current_active_wm_class = wmclass
modus_service.current_active_app_name = name
return name
+148
View File
@@ -0,0 +1,148 @@
"""Pure Python DBusMenu extractor using Gio.DBus."""
from dataclasses import dataclass, field
from typing import List, Optional
from gi.repository import GLib, Gio
from fabric.utils import logger
@dataclass
class DBusMenuItem:
id: int
label: str = ""
enabled: bool = True
visible: bool = True
type: str = "standard" # standard, separator
shortcut: str = ""
icon_name: str = ""
has_submenu: bool = False
children: List["DBusMenuItem"] = field(default_factory=list)
class DBusMenuClient:
def __init__(self, service_name: str, object_path: str):
self.service_name = service_name
self.object_path = object_path
self.bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
def _call_method(
self,
method: str,
params: Optional[GLib.Variant],
reply_type: Optional[GLib.VariantType],
) -> Optional[GLib.Variant]:
try:
return self.bus.call_sync(
self.service_name,
self.object_path,
"com.canonical.dbusmenu",
method,
params,
reply_type,
Gio.DBusCallFlags.NONE,
1000,
None,
)
except Exception as e:
logger.debug(f"[DBusMenuClient] Call to {method} failed: {e}")
return None
def get_layout(
self, parent_id: int = 0, recursion_depth: int = -1
) -> List[DBusMenuItem]:
"""Fetches and parses the full menu layout."""
# Signature: iias -> u(ia{sv}av)
params = GLib.Variant("(iias)", (parent_id, recursion_depth, []))
res = self._call_method("GetLayout", params, None)
if not res:
return []
try:
# Result is usually a tuple: (u, (i, a{sv}, av)) or (u, v) depending on the DBus implementation
# We bypass the strict type by unpacking the child
if res.n_children() < 2:
return []
layout_var = res.get_child_value(1)
if layout_var.is_of_type(GLib.VariantType("v")):
layout_var = layout_var.get_variant()
return self._parse_node(layout_var).children
except Exception as e:
logger.error(f"[DBusMenuClient] Error parsing GetLayout: {e}")
return []
def _parse_node(self, node: GLib.Variant) -> DBusMenuItem:
"""Parses a layout node (ia{sv}av) into a DBusMenuItem."""
item = DBusMenuItem(id=0)
try:
# node is a struct with 3 elements: id (i), properties (a{sv}), children (av)
if node.n_children() != 3:
return item
item.id = node.get_child_value(0).get_int32()
props_dict = node.get_child_value(1).unpack()
# Map properties
item.label = props_dict.get("label", "").replace("_", "")
item.enabled = props_dict.get("enabled", True)
item.visible = props_dict.get("visible", True)
item.type = props_dict.get("type", "standard")
item.icon_name = props_dict.get("icon-name", "")
# Submenus
if props_dict.get("children-display") == "submenu":
item.has_submenu = True
# Shortcut parsing (usually a nested list of strings)
shortcut_data = props_dict.get("shortcut")
if (
shortcut_data
and isinstance(shortcut_data, list)
and len(shortcut_data) > 0
):
parts = []
for s in shortcut_data[0]:
if s == "Control":
parts.append("Ctrl")
elif s == "Shift":
parts.append("Shift")
elif s == "Alt":
parts.append("Alt")
else:
parts.append(s.upper())
item.shortcut = "+".join(parts)
# Parse children
children_array = node.get_child_value(2)
for i in range(children_array.n_children()):
child_var = children_array.get_child_value(i)
if child_var.is_of_type(GLib.VariantType("v")):
child_var = child_var.get_variant()
child_item = self._parse_node(child_var)
if child_item.visible:
item.children.append(child_item)
except Exception as e:
logger.debug(f"[DBusMenuClient] Error parsing node: {e}")
return item
def about_to_show(self, item_id: int) -> bool:
"""Tells the app to populate a dynamic submenu."""
# Signature: i -> b
res = self._call_method(
"AboutToShow", GLib.Variant("(i)", (item_id,)), GLib.VariantType("(b)")
)
if res:
return res.get_child_value(0).get_boolean()
return False
def click_item(self, item_id: int) -> None:
"""Triggers the activation event on a menu item."""
# Signature: isvu -> nothing
# Event(id, eventId, data, timestamp)
params = GLib.Variant("(isvu)", (item_id, "clicked", GLib.Variant("s", ""), 0))
self._call_method("Event", params, None)
+132 -34
View File
@@ -1,6 +1,6 @@
import weakref
from fabric.utils import bulk_connect, logger
from fabric.utils import GLib, bulk_connect, logger
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.image import Image
@@ -279,6 +279,8 @@ class PlayerBox(Box):
self._signal_connections = []
self._property_bindings = []
self._seekbar_signal_ids = []
self._init_metadata_polled = False
self._cached_duration = 0
self.album_cover = Box(style_classes="album-image-c")
self.album_cover.set_style(
@@ -492,11 +494,49 @@ class PlayerBox(Box):
self._signal_connections.append((self.player, handler_id))
self.seek_bar.connect("realize", self._on_seek_bar_realized)
if self.seek_bar.get_realized():
self._on_seek_bar_realized(self.seek_bar)
result = apply_player_art(self.album_cover, self.player)
if result:
self.cover_path = result
self._load_initial_metadata()
GLib.timeout_add(500, self._poll_initial_metadata)
def _load_initial_metadata(self):
try:
metadata = self.player._player.props.metadata
if metadata:
keys = metadata.keys()
self.track_title.set_label(
metadata["xesam:title"] if "xesam:title" in keys else "No Title"
)
self.track_artist.set_label(
", ".join(metadata["xesam:artist"])
if "xesam:artist" in keys and metadata["xesam:artist"]
else "No Artist"
)
self.track_album.set_label(
metadata["xesam:album"] if "xesam:album" in keys else "No Album"
)
self._update_duration_from_metadata()
self.position_label.set_label(
self.length_str(self.player.position or 0)
)
except Exception:
pass
def _poll_initial_metadata(self):
if self.exit:
return False
if self._init_metadata_polled:
return False
self._init_metadata_polled = True
self._load_initial_metadata()
self.set_image()
return False
def update_buttons(self, player_buttons, show_buttons):
self.stack_buttons_box.children = []
if show_buttons and len(player_buttons) > 1:
@@ -532,24 +572,35 @@ class PlayerBox(Box):
else f"{minutes}:{seconds:02d}"
)
def _on_metadata(self, *_):
def _on_metadata(self, *args):
if self.exit or self.player is None:
return
self.track_title.set_label(self.player.title or "No Title")
self.track_artist.set_label(
", ".join(self.player.artist) if self.player.artist else "No Artist"
)
self.track_album.set_label(self.player.album or "No Album")
self.set_image()
duration = self.player.length or 0
if duration > 0:
self.length_label.set_label(self.length_str(duration))
if self.seek_bar.get_realized():
metadata = args[1] if len(args) >= 2 else None
if metadata is not None:
keys = metadata.keys()
self.track_title.set_label(
metadata["xesam:title"] if "xesam:title" in keys else "No Title"
)
self.track_artist.set_label(
", ".join(metadata["xesam:artist"])
if "xesam:artist" in keys and metadata["xesam:artist"]
else "No Artist"
)
self.track_album.set_label(
metadata["xesam:album"] if "xesam:album" in keys else "No Album"
)
if "mpris:length" in keys:
duration = int(metadata["mpris:length"])
self.length_label.set_label(self.length_str(duration))
self.seek_bar.set_range(0, min(2147483647, duration))
self.position_label.set_label(self.length_str(self.player.position or 0))
else:
self.length_label.set_label("0:00")
if self.seek_bar.get_realized():
self.seek_bar.set_range(0, 100)
self.track_title.set_label(self.player.title or "No Title")
self.track_artist.set_label(
", ".join(self.player.artist) if self.player.artist else "No Artist"
)
self.track_album.set_label(self.player.album or "No Album")
self.set_image()
def suspend(self):
pass
@@ -612,24 +663,25 @@ class PlayerBox(Box):
self.cover_path = result
def _on_track_position(self, service, pos: float, dur: float):
if self.exit or self._user_seeking:
if self.exit:
return
try:
position_micros = int(pos * 1_000_000)
duration_micros = int(dur * 1_000_000)
if duration_micros <= 0:
duration_micros = self.player.length or 0
if duration_micros > 0:
self._cached_duration = duration_micros
self.length_label.set_label(self.length_str(duration_micros))
self.position_label.set_label(self.length_str(position_micros))
if duration_micros > 0:
self.length_label.set_label(self.length_str(duration_micros))
if self.seek_bar.get_realized() and self.seek_bar.get_adjustment():
if (
duration_micros > 0
and self.seek_bar.get_adjustment().get_upper() <= 100
):
adj = self.seek_bar.get_adjustment()
if duration_micros > 0:
self.seek_bar.set_range(0, min(2147483647, duration_micros))
if self.seek_bar.get_adjustment().get_upper() > 0:
if not self._user_seeking and adj.get_upper() > 0:
self.seek_bar.set_value(min(2147483647, position_micros))
except Exception as e:
logger.error(f"[_on_track_position] Error: {e}")
@@ -640,26 +692,72 @@ class PlayerBox(Box):
def _on_seek_end(self, widget, event):
self._user_seeking = False
if not self.player or self.exit:
return False
try:
adj = widget.get_adjustment()
upper = adj.get_upper()
value = max(0, int(widget.get_value()))
duration = self._cached_duration or self.player.length or 0
if duration <= 0 and upper > 100:
duration = int(upper)
if upper > 100:
new_position = value
elif duration > 0 and upper > 0:
new_position = int(value / upper * duration)
else:
new_position = value
self.player.set_position(new_position / 1_000_000)
except Exception as e:
logger.error(f"An error occurred: {e}")
return False
def _update_duration_from_metadata(self):
try:
duration = self._cached_duration or self.player.length or 0
if duration <= 0 and self.seek_bar.get_realized():
adj = self.seek_bar.get_adjustment()
if adj.get_upper() > 100:
duration = int(adj.get_upper())
if duration > 0:
self.seek_bar.set_range(0, min(2147483647, duration))
self.length_label.set_label(self.length_str(duration))
return True
except Exception:
pass
return False
def _on_seek_bar_realized(self, widget):
try:
if not self.exit and self.player:
duration = self.player.length or 0
self.seek_bar.set_range(
0, min(2147483647, duration) if duration > 0 else 100
self._update_duration_from_metadata()
self.position_label.set_label(
self.length_str(self.player.position or 0)
)
except Exception as e:
logger.error(f"An error occurred: {e}")
def _on_scale_value_changed(self, scale: Scale):
if self.player and not self.exit and self._user_seeking:
try:
new_position = max(-2147483648, min(2147483647, int(scale.get_value())))
self.player.position = new_position
self.position_label.set_label(self.length_str(new_position))
except Exception as e:
logger.error(f"An error occurred: {e}")
if not self.player or self.exit:
return
try:
adj = scale.get_adjustment()
upper = adj.get_upper()
value = max(0, int(scale.get_value()))
duration = self._cached_duration or self.player.length or 0
if duration <= 0 and upper > 100:
duration = int(upper)
if upper > 100:
new_position = value
elif duration > 0 and upper > 0:
new_position = int(value / upper * duration)
else:
new_position = value
self.position_label.set_label(self.length_str(new_position))
except Exception as e:
logger.error(f"An error occurred: {e}")
class ExpandedPlayer(Window):
+32 -5
View File
@@ -429,17 +429,44 @@ class PlayerBox(Box):
if result:
self.cover_path = result
try:
metadata = self.player._player.props.metadata
if metadata:
keys = metadata.keys()
self.track_title.set_label(
metadata["xesam:title"] if "xesam:title" in keys else "No Title"
)
self.track_artist.set_label(
", ".join(metadata["xesam:artist"])
if "xesam:artist" in keys and metadata["xesam:artist"]
else "No Artist"
)
except Exception:
pass
def _on_outer_box_clicked(self, *_):
if self.control_center and hasattr(self.control_center, "open_expanded_player"):
self.control_center.open_expanded_player()
def _on_metadata(self, *_):
def _on_metadata(self, *args):
if self.exit or self.player is None:
return
self.track_title.set_label(self.player.title or "No Title")
self.track_artist.set_label(
", ".join(self.player.artist) if self.player.artist else "No Artist"
)
metadata = args[1] if len(args) >= 2 else None
if metadata is not None:
keys = metadata.keys()
self.track_title.set_label(
metadata["xesam:title"] if "xesam:title" in keys else "No Title"
)
self.track_artist.set_label(
", ".join(metadata["xesam:artist"])
if "xesam:artist" in keys and metadata["xesam:artist"]
else "No Artist"
)
else:
self.track_title.set_label(self.player.title or "No Title")
self.track_artist.set_label(
", ".join(self.player.artist) if self.player.artist else "No Artist"
)
self.set_image()
def refresh_icon(self):
+198 -39
View File
@@ -7,6 +7,7 @@ from fabric.widgets.label import Label
from shared.dialogs.about import AboutApp, get_about_window
from shared.window.dropdown import ModusDropdown, dropdown_divider, dropdowns
from services.globalmenu import get_global_menu_service
from utils.app_name_resolver import format_window
from utils.roam import modus_service
from utils.utils import setup_cursor_hover
@@ -45,7 +46,7 @@ def create_dropdown_with_capture(dropdown_id, parent, dropdown_children, layer="
def manage_button_style_classes(buttons, active_button=None, style_class="active"):
for button in buttons:
if button: # Check if button exists (some might be None)
if button:
button.remove_style_class(style_class)
if active_button:
@@ -103,6 +104,32 @@ def dropdown_option(
return btn
def _make_menu_item_click(item_id):
"""Create a click handler for a menu item."""
def handler(_, __=None):
global_menu_svc = get_global_menu_service()
if global_menu_svc:
global_menu_svc.click_item(item_id)
return handler
def _menu_item_to_dropdown(item):
"""Convert a DBusMenuItem to a dropdown_option widget."""
if item.type == "separator":
return dropdown_divider("---------------------")
keybind = item.shortcut if item.shortcut else ""
return dropdown_option(
item.label if item.label else "",
keybind,
on_click=None,
on_clicked=_make_menu_item_click(item.id),
)
class SystemDropdown(ModusDropdown):
def __init__(self, parent, **kwargs):
super().__init__(
@@ -131,17 +158,15 @@ class SystemDropdown(ModusDropdown):
class GlobalMenuDropdowns:
def __init__(self, parent):
def __init__(self, parent, menu_box=None):
self.parent = parent
self._menu_box = menu_box # The GlobalMenu Box widget to update
# Get the global menu service
self._global_menu_svc = get_global_menu_service()
self._has_extracted_menu = False
self.system_dropdown = SystemDropdown(parent=parent)
self.menu_button_dropdown = self.system_dropdown
self.menu_button = Button(
label="Modus",
name="global-menu",
on_clicked=lambda _: self.menu_button_dropdown.toggle(),
)
self.menu_button_dropdown.set_pointing_to(self.menu_button)
self.global_title_menu_about = dropdown_option(
f"About {modus_service.current_active_app_name}",
@@ -153,6 +178,10 @@ class GlobalMenuDropdowns:
[self.global_title_menu_about],
)
# Dynamic menu dropdowns (populated from extracted app menus)
self._dynamic_dropdowns: list[ModusDropdown] = []
# Fallback static Hyprland menu
self.global_menu_view = create_dropdown_with_capture(
"global-menu-view",
parent,
@@ -237,6 +266,8 @@ class GlobalMenuDropdowns:
"current-active-app-name-changed", self._on_active_app_changed
)
# Connect to global menu service signals (Moved to the bottom of __init__)
self.global_menu_button_title = Button(
child=ActiveWindow(
formatter=FormattedString(
@@ -249,11 +280,8 @@ class GlobalMenuDropdowns:
)
self.global_menu_title.set_pointing_to(self.global_menu_button_title)
# File, Edit and Go buttons are placeholders - no dropdowns implemented yet
self.global_menu_button_file = create_menu_button("File")
self.global_menu_button_edit = create_menu_button("Edit")
self.global_menu_button_go = create_menu_button("Go")
# Fallback buttons (shown when no app menu is extracted)
self.global_menu_button_view = create_menu_button(
"View", lambda _: self.global_menu_view.toggle()
)
@@ -267,24 +295,14 @@ class GlobalMenuDropdowns:
)
self.global_menu_help.set_pointing_to(self.global_menu_button_help)
# Start with only title
self.all_menu_buttons = [
self.menu_button,
self.global_menu_button_title,
self.global_menu_button_file,
self.global_menu_button_edit,
self.global_menu_button_view,
self.global_menu_button_go,
self.global_menu_button_window,
self.global_menu_button_help,
]
self.dropdown_button_map = {
"os-menu": self.menu_button,
"global-menu-title": self.global_menu_button_title,
"global-menu-file": self.global_menu_button_file,
"global-menu-edit": self.global_menu_button_edit,
"global-menu-view": self.global_menu_button_view,
"global-menu-go": self.global_menu_button_go,
"global-menu-window": self.global_menu_button_window,
"global-menu-help": self.global_menu_button_help,
}
@@ -292,13 +310,148 @@ class GlobalMenuDropdowns:
modus_service.connect("current-dropdown-changed", self.changed_dropdown)
modus_service.connect("dropdowns-hide-changed", self.hide_dropdowns)
# Connect to global menu service signals
if self._global_menu_svc:
self._global_menu_svc.connect("menu-changed", self._on_menu_changed)
def _on_title_button_clicked(self, _):
if has_active_window():
self.global_menu_title.toggle()
def _on_active_app_changed(self, _, value):
logger.info(f"[GlobalMenu] Active app changed: {value}")
self.global_title_menu_about.set_property("label", f"About {value}")
# Notify the global menu service about the active window change
if self._global_menu_svc:
wm_class = getattr(modus_service, "current_active_wm_class", "")
logger.info(
f"[GlobalMenu] Calling update_active_window('{value}', '{wm_class}')"
)
self._global_menu_svc.update_active_window(value, wm_class)
def _on_menu_changed(self, _, menu_items: list):
"""Handle menu changes from the GlobalMenuService."""
logger.info(
f"[GlobalMenu] _on_menu_changed called with {len(menu_items)} items"
)
self._rebuild_dynamic_menus(menu_items)
def _rebuild_dynamic_menus(self, menu_items: list):
"""Rebuild the menu bar buttons based on extracted menu items.
If the app has a menu bar → show extracted menus (File, Edit, etc.)
If not → show fallback Hyprland window controls (View, Window, Help)
"""
logger.info(f"[GlobalMenu] _rebuild_dynamic_menus: {len(menu_items)} items")
# Clean up any previous dynamic dropdowns
for dd in self._dynamic_dropdowns:
try:
dd.destroy()
except Exception:
pass
self._dynamic_dropdowns.clear()
# Filter to top-level menu items only
top_level = [
item for item in menu_items if hasattr(item, "label") and item.label
]
logger.info(f"[GlobalMenu] top_level after filter: {len(top_level)} items")
if not top_level:
# No extracted menu → show only title button
self._has_extracted_menu = False
self.all_menu_buttons = [
self.global_menu_button_title,
]
else:
# Has extracted menu → build dynamic buttons
self._has_extracted_menu = True
new_buttons = [
self.global_menu_button_title,
]
for item in top_level:
is_action = not item.has_submenu and not item.children
if is_action:
# Top-level item without children (e.g., VLC Play/Stop)
btn = create_menu_button(
item.label,
_make_menu_item_click(item.id),
)
new_buttons.append(btn)
else:
# Build initial children for this dropdown
dropdown_children = []
for child in item.children if item.children else []:
dropdown_children.append(_menu_item_to_dropdown(child))
if not dropdown_children:
dropdown_children = [
dropdown_option(f"No items in {item.label}")
]
dropdown_id = f"global-menu-{item.label.lower()}"
dropdown = create_dropdown_with_capture(
dropdown_id, self.parent, dropdown_children
)
btn = create_menu_button(
item.label,
lambda _, dd=dropdown: dd.toggle(),
)
dropdown.set_pointing_to(btn)
# Dynamically fetch submenu on hover
def make_on_enter(menu_item, dd):
def on_enter(widget, event):
if not menu_item.has_submenu:
return False
svc = get_global_menu_service()
if svc:
svc.about_to_show(menu_item.id)
new_menu = svc.refresh_menu_sync()
if new_menu:
# find the updated item
new_item = next(
(i for i in new_menu if i.id == menu_item.id),
None,
)
if new_item:
children = []
for child in (
new_item.children
if new_item.children
else []
):
children.append(
_menu_item_to_dropdown(child)
)
if not children:
children = [
dropdown_option(
f"No items in {new_item.label}"
)
]
dd.dropdown.children = children
return False
return on_enter
btn.connect("enter-notify-event", make_on_enter(item, dropdown))
self._dynamic_dropdowns.append(dropdown)
self.dropdown_button_map[dropdown_id] = btn
new_buttons.append(btn)
self.all_menu_buttons = new_buttons
# Update the panel (the actual GTK Box widget)
if self._menu_box is not None:
self._menu_box.children = self.all_menu_buttons
def hide_dropdowns(self, *_):
manage_button_style_classes(self.all_menu_buttons)
@@ -315,9 +468,24 @@ class GlobalMenuDropdowns:
except Exception as e:
logger.error(f"An error occurred: {e}")
# Destroy all dropdown captures
# Disconnect global menu service
if self._global_menu_svc:
try:
self._global_menu_svc.disconnect_by_func(self._on_menu_changed)
except Exception:
pass
# Destroy dynamic dropdowns
for dd in self._dynamic_dropdowns:
try:
dd.destroy()
except Exception as e:
logger.error(f"An error occurred: {e}")
self._dynamic_dropdowns.clear()
# Destroy static dropdown captures
dropdown_captures = [
getattr(self, "menu_button_dropdown", None),
getattr(self, "system_dropdown", None),
getattr(self, "global_menu_title", None),
getattr(self, "global_menu_view", None),
getattr(self, "global_menu_window", None),
@@ -342,22 +510,13 @@ class GlobalMenu(Box):
name="globalmenu", orientation="horizontal", spacing=0, **kwargs
)
self.dropdown_system = GlobalMenuDropdowns(parent=parent_window)
self.dropdown_system = GlobalMenuDropdowns(parent=parent_window, menu_box=self)
self.children = [
self.dropdown_system.global_menu_button_title,
self.dropdown_system.global_menu_button_file,
self.dropdown_system.global_menu_button_edit,
self.dropdown_system.global_menu_button_view,
self.dropdown_system.global_menu_button_go,
self.dropdown_system.global_menu_button_window,
self.dropdown_system.global_menu_button_help,
]
self.children = self.dropdown_system.all_menu_buttons
def show_system_dropdown(self, imac_button):
self.dropdown_system.menu_button_dropdown.set_pointing_to(imac_button)
mouse_capture = self.dropdown_system.menu_button_dropdown
mouse_capture.toggle()
self.dropdown_system.system_dropdown.set_pointing_to(imac_button)
self.dropdown_system.system_dropdown.toggle()
def destroy(self):
"""Clean up the global menu and its dropdowns"""