chore(src/utils): remove deprecated DBus menu client and dev‑members.json, add uinput module for mouse click simulation
Drop unused dbusmenu.py and related files, update pyproject.toml and requirements.txt to reflect new dependencies, and add pure‑Python uinput click helper.
This commit is contained in:
5 files changed
+69
-1265
No files matched your search
@@ -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
|
||||
@@ -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 ---
|
||||
|
||||
@@ -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}")
|
||||
@@ -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 "<external>"
|
||||
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 != "<external>":
|
||||
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"{'<external>':<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")
|
||||
@@ -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 = "<qqHHi"
|
||||
|
||||
|
||||
def uinput_click(x: int, y: int, move_first: bool = True) -> 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("<HHHHI", 0x03, 0x1234, 0x5678, 1, 0)
|
||||
dev += b"\x00" * (_ABS_CNT * 4 * 4)
|
||||
os.write(fd, dev)
|
||||
fcntl.ioctl(fd, _UI_DEV_CREATE)
|
||||
_time.sleep(0.15)
|
||||
|
||||
def _emit(etype: int, code: int, value: int):
|
||||
os.write(fd, struct.pack(_INPUT_EVENT_FMT, 0, 0, etype, code, value))
|
||||
|
||||
if move_first:
|
||||
_emit(_EV_REL, _REL_X, x)
|
||||
_emit(_EV_REL, _REL_Y, y)
|
||||
_emit(_SYN_REPORT, 0, 0)
|
||||
_time.sleep(0.03)
|
||||
|
||||
_emit(_EV_KEY, _BTN_LEFT, 1)
|
||||
_emit(_SYN_REPORT, 0, 0)
|
||||
_time.sleep(0.03)
|
||||
_emit(_EV_KEY, _BTN_LEFT, 0)
|
||||
_emit(_SYN_REPORT, 0, 0)
|
||||
|
||||
os.close(fd)
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
Reference in new issue
Block a user