From 4b039c1ff39b292b7df70c43b11949a6d40daaff Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Sun, 12 Jul 2026 02:44:35 +0545 Subject: [PATCH] =?UTF-8?q?chore(src/utils):=20remove=20deprecated=20DBus?= =?UTF-8?q?=20menu=20client=20and=20dev=E2=80=91members.json,=20add=20uinp?= =?UTF-8?q?ut=20module=20for=20mouse=20click=20simulation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drop unused dbusmenu.py and related files, update pyproject.toml and requirements.txt to reflect new dependencies, and add pure‑Python uinput click helper. --- src/utils/dbusmenu.py | 395 ------------------------ src/utils/functions.py | 5 +- src/utils/gtkmenu.py | 589 ------------------------------------ src/utils/memory_tracker.py | 280 ----------------- src/utils/uinput.py | 65 ++++ 5 files changed, 69 insertions(+), 1265 deletions(-) delete mode 100644 src/utils/dbusmenu.py delete mode 100644 src/utils/gtkmenu.py delete mode 100644 src/utils/memory_tracker.py create mode 100644 src/utils/uinput.py diff --git a/src/utils/dbusmenu.py b/src/utils/dbusmenu.py deleted file mode 100644 index f40748e..0000000 --- a/src/utils/dbusmenu.py +++ /dev/null @@ -1,395 +0,0 @@ -""" -Universal DBus + GTK Global Menu Client -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 Callable, Dict, List, Optional, Tuple - -from fabric.utils import Any, Gio, GLib, logger - -# SHARED BUS — one connection for the entire process -_SESSION_BUS: Optional[Gio.DBusConnection] = None -_BUS_LOCK = threading.Lock() - - -def _get_bus() -> Gio.DBusConnection: - global _SESSION_BUS - if _SESSION_BUS is None: - with _BUS_LOCK: - if _SESSION_BUS is None: - _SESSION_BUS = Gio.bus_get_sync(Gio.BusType.SESSION, None) - return _SESSION_BUS - - -class DBusMenuItem: - __slots__ = ( - "id", - "label", - "enabled", - "visible", - "type", - "shortcut", - "icon_name", - "has_submenu", - "children", - "action_name", - "action_target", - "parent_id", - ) - - def __init__( - self, - id: int, - label: str = "", - enabled: bool = True, - visible: bool = True, - type: str = "standard", - shortcut: str = "", - icon_name: str = "", - has_submenu: bool = False, - children: Optional[List["DBusMenuItem"]] = None, - action_name: str = "", - action_target: Optional[Any] = None, - parent_id: int = -1, - ): - self.id = id - self.label = label - self.enabled = enabled - self.visible = visible - self.type = type - self.shortcut = shortcut - self.icon_name = icon_name - self.has_submenu = has_submenu - 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 - - -class DBusMenuClient: - __slots__ = ( - "service_name", - "object_path", - "_cache", - "_cache_valid", - "_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): - self.service_name = service_name - self.object_path = object_path - - 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 - - def _call(self, method, params, reply_type=None): - try: - return _get_bus().call_sync( - self.service_name, - self.object_path, - "com.canonical.dbusmenu", - method, - params, - reply_type, - Gio.DBusCallFlags.NONE, - 3000, - None, - ) - except Exception: - return None - - def get_layout(self, force_refresh=False): - if self._cache and self._cache_valid and not force_refresh: - return self._cache - - with self._lock: - if self._fetching: - return self._cache or [] - self._fetching = True - - try: - return self._fetch() - finally: - with self._lock: - self._fetching = False - - def _fetch(self): - params = GLib.Variant("(iias)", (0, -1, [])) - res = self._call("GetLayout", params) - - if not res: - 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() - - parsed = self._parse_dbusmenu(layout).children - - h = self._hash(parsed) - if h == self._hash_cache: - return self._cache or parsed - - self._hash_cache = h - self._cache = parsed - self._cache_valid = True - return parsed - - except Exception as e: - logger.error(f"[Menu] parse error: {e}") - return [] - - def _hash(self, items): - flat: List[str] = [] - - def walk(nodes): - for n in nodes: - flat.append(f"{n.id}:{n.label}:{n.enabled}") - walk(n.children) - - walk(items) - return str(hash("".join(flat))) - - def _parse_dbusmenu(self, node, _depth: int = 0, _parent_id: int = -1): - if _depth > 20: - return DBusMenuItem(id=0) - item = DBusMenuItem(id=0) - - try: - if node.n_children() != 3: - return item - - item.id = node.get_child_value(0).get_int32() - item.parent_id = _parent_id - - try: - props = node.get_child_value(1).unpack() - except Exception: - props = {} - - item.label = props.get("label", "").replace("_", "") - item.enabled = props.get("enabled", True) - item.visible = props.get("visible", True) - item.type = props.get("type", "standard") - item.icon_name = props.get("icon-name", "") - - if props.get("children-display") == "submenu": - item.has_submenu = True - - sc = props.get("shortcut") - if sc: - try: - parts = [] - if isinstance(sc, list): - flat = sc[0] if isinstance(sc[0], list) else sc - mapping = { - "Control": "Ctrl", - "Shift": "Shift", - "Alt": "Alt", - "Meta": "Super", - } - for s in flat: - parts.append(mapping.get(s, str(s).upper())) - item.shortcut = "+".join(parts) - except Exception: - item.shortcut = "" - - children = node.get_child_value(2) - for i in range(children.n_children()): - if _depth >= 20: - break - 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.id)) - - if item.has_submenu and not item.children: - try: - self.about_to_show(item.id) - except Exception: - pass - - except Exception: - pass - - return item - - def about_to_show(self, item_id): - try: - res = self._call( - "AboutToShow", - GLib.Variant("(i)", (item_id,)), - GLib.VariantType("(b)"), - ) - return res.get_child_value(0).get_boolean() if res else False - except Exception: - return False - - def click_item(self, item_id): - try: - self._call( - "Event", - GLib.Variant("(isvu)", (item_id, "clicked", GLib.Variant("s", ""), 0)), - ) - self.invalidate() - except Exception: - pass - - def invalidate(self): - self._cache_valid = False - - def prefetch_async(self): - if self._fetching: - return - threading.Thread( - target=self.get_layout, kwargs={"force_refresh": True}, daemon=True - ).start() - - 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 13677a8..5176331 100644 --- a/src/utils/functions.py +++ b/src/utils/functions.py @@ -141,7 +141,10 @@ def is_special_workspace(client: dict) -> bool: def escape_markup_text(text: str) -> str: - return html.escape(text.replace("\n", " ")) + import re + + clean = re.sub(r"<[^>]+>", "", text) + return html.escape(clean.replace("\n", " ")) # --- Process management --- diff --git a/src/utils/gtkmenu.py b/src/utils/gtkmenu.py deleted file mode 100644 index d1d2e88..0000000 --- a/src/utils/gtkmenu.py +++ /dev/null @@ -1,589 +0,0 @@ -from typing import Any, Dict, List, Optional - -from fabric.utils import Gio, GLib, logger - -from utils.dbusmenu import DBusMenuItem, _get_bus - - -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", - "_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.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: - return _get_bus().call_sync( - self.service_name, - path or self.object_path, - interface, - method, - params, - reply_type, - Gio.DBusCallFlags.NONE, - 3000, - None, - ) - except Exception as e: - logger.debug(f"[GtkMenuClient] {method} failed: {e}") - return None - - 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 - ) -> 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 _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"[GtkMenuClient] Parse error: {e}") - return [] - - def about_to_show(self, item_id: int) -> bool: - return False - - def click_item( - self, item_id: int, action_name: str = "", action_target=None - ) -> None: - if not action_name: - 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 - 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: - for act_path in (self.object_path, self._action_path): - try: - 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 - except Exception as e: - logger.error(f"[GtkMenuClient] click failed: {e}") - - -class ActionMenuClient: - __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, path=None): - try: - return _get_bus().call_sync( - self.service_name, - path or self.object_path, - interface, - method, - params, - reply_type, - Gio.DBusCallFlags.NONE, - 3000, - None, - ) - except Exception: - return None - - 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", - None, - GLib.VariantType("(a{s(bgav)})"), - ) - if not res: - return self._cached_items - - try: - data = res.get_child_value(0).unpack() - items: List[DBusMenuItem] = [] - self._action_map = {} - 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 - 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 - return items - except Exception as e: - 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: - 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: - 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/memory_tracker.py b/src/utils/memory_tracker.py deleted file mode 100644 index 2589d90..0000000 --- a/src/utils/memory_tracker.py +++ /dev/null @@ -1,280 +0,0 @@ -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/uinput.py b/src/utils/uinput.py new file mode 100644 index 0000000..85e9965 --- /dev/null +++ b/src/utils/uinput.py @@ -0,0 +1,65 @@ +"""Pure-Python mouse click simulation via /dev/uinput.""" + +import os +import struct +import time as _time + +_UI_DEV_CREATE = 0x5501 +_UI_SET_EVBIT = 0x40045564 +_UI_SET_KEYBIT = 0x40045565 +_UI_SET_RELBIT = 0x40045567 + +_EV_KEY = 0x01 +_EV_REL = 0x02 +_REL_X = 0x00 +_REL_Y = 0x01 +_BTN_LEFT = 0x110 +_SYN_REPORT = 0 + +_ABS_CNT = 64 +_INPUT_EVENT_FMT = " bool: + """Move cursor to (x, y) and left-click using a virtual uinput mouse. + + Returns True on success. + """ + try: + fd = os.open("/dev/uinput", os.O_WRONLY | os.O_NONBLOCK) + + import fcntl + + for cap in (_EV_KEY, _EV_REL): + fcntl.ioctl(fd, _UI_SET_EVBIT, cap) + fcntl.ioctl(fd, _UI_SET_KEYBIT, _BTN_LEFT) + for rel in (_REL_X, _REL_Y): + fcntl.ioctl(fd, _UI_SET_RELBIT, rel) + + name = b"OpenCodeMouse\x00" + b"\x00" * 66 + dev = name + dev += struct.pack("