WIP: alt1 trap

README.md

@@ -92,7 +92,7 @@ alt1_pressed                  | ❌    | ✅
 game_activity                 | ❌    | ✅
 positionChanged               | ❌    | ✅
 scalingChanged                | ❌    | ❌
-focusChanged                  | ❌    | ✅
+focusChanged                  | ✅    | ✅
 set_taskbar_progress          | ❌    | ❌
 Window capture                | ✅    | ✅
 

runekit/game/x11/ge.py

@@ -0,0 +1,66 @@
+import xcffib
+
+opcode_cache = {}
+
+
+def get_extension_by_opcode(connection, opcode):
+    if opcode in opcode_cache:
+        return opcode_cache[opcode]
+
+    for name in connection.core.ListExtensions().reply().names:
+        ext_opcode = (
+            connection.core.QueryExtension(name.name_len, name.name)
+            .reply()
+            .major_opcode
+        )
+        opcode_cache[ext_opcode] = xcffib.ExtensionKey(name.name.to_string())
+
+    try:
+        return opcode_cache[opcode]
+    except KeyError:
+        raise ExtensionNotFound(opcode)
+
+
+class ExtensionNotFound(Exception):
+    def __init__(self, opcode):
+        self.opcode = opcode
+        super().__init__(f"Extension with opcode {opcode} was not found")
+
+
+class GeGenericEvent(xcffib.Event):
+    def __init__(self, unpacker):
+        if isinstance(unpacker, xcffib.Protobj):
+            unpacker = xcffib.MemoryUnpacker(unpacker.pack())
+
+        super().__init__(unpacker)
+        try:
+            base = unpacker.offset
+            (
+                self.extension,
+                self.sequence_number,
+                self.length,
+                self.evtype,
+            ) = unpacker.unpack("xBHIH")
+            self.payload = xcffib.List(unpacker, "I", 8 + self.length)
+            self.bufsize = unpacker.offset - base
+        except:
+            import traceback
+
+            traceback.print_exc()
+            raise
+
+    def to_event(self, connection):
+        extension = get_extension_by_opcode(connection, self.extension)
+
+        constructor = xcffib.extensions[extension][1][self.evtype]
+        unpacker = xcffib.MemoryUnpacker(self.payload.buf())
+        try:
+            return constructor(unpacker)
+        except:
+            import traceback
+
+            traceback.print_exc()
+            raise
+
+
+xcffib.core_events[35] = GeGenericEvent

runekit/game/x11/instance.py

@@ -2,10 +2,11 @@ import logging
 import time
 from typing import TYPE_CHECKING
 
-from PySide2.QtCore import QRect
+from PySide2.QtCore import QRect, Slot, Signal
 from PySide2.QtGui import QWindow, QGuiApplication
 import xcffib.xproto
 import xcffib.composite
+import xcffib.xinput
 from PIL import Image, ImageOps
 
 from runekit.game.instance import GameInstance
@@ -15,6 +16,8 @@ from .ximage import zpixmap_shm_to_image
 if TYPE_CHECKING:
     from .manager import X11GameManager
 
+KEY_1 = 10
+
 
 class X11GameInstance(QtGrabMixin, GameInstance):
     wid: int
@@ -24,6 +27,8 @@ class X11GameInstance(QtGrabMixin, GameInstance):
     game_last_grab = 0.0
     game_last_image = None
 
+    key_press = Signal(xcffib.xinput.KeyPressEvent)
+
     def __init__(self, manager: "X11GameManager", wid: int, **kwargs):
         super().__init__(**kwargs)
         self.manager = manager
@@ -31,22 +36,70 @@ class X11GameInstance(QtGrabMixin, GameInstance):
         self.qwindow = QWindow.fromWinId(wid)
         self.logger = logging.getLogger(__name__ + "." + self.__class__.__name__)
         self._setup()
+        self.key_press.connect(self.on_key_press)
         self._update_is_focused()
 
     def _setup(self):
-        self.manager.composite.RedirectWindow(
+        self.manager.xcomposite.RedirectWindow(
             self.wid, xcffib.composite.Redirect.Automatic
         )
         self.pixmap_id = self.manager.connection.generate_id()
-        self.manager.composite.NameWindowPixmap(
+        self.manager.xcomposite.NameWindowPixmap(
             self.wid, self.pixmap_id, is_checked=True
         )
 
+        self._grab_alt1()
+        # self.manager.xinput.XIGrabDevice(
+        #     self.wid,
+        #     xcffib.xproto.Time.CurrentTime,
+        #     xcffib.xproto.Cursor._None,
+        #     xcffib.xinput.Device.All,
+        #     xcffib.xinput.GrabMode22.Async,
+        #     xcffib.xinput.GrabMode22.Async,
+        #     True,
+        #     1,
+        #     (
+        #         xcffib.xinput.XIEventMask.KeyPress
+        #         | xcffib.xinput.XIEventMask.ButtonPress,
+        #     ),
+        # )
+
+    def _grab_alt1(self):
+        self.manager.xinput.XIPassiveGrabDevice(
+            xcffib.xproto.Time.CurrentTime,
+            self.wid,
+            xcffib.xproto.Cursor._None,
+            KEY_1,
+            xcffib.xinput.Device.All,
+            1,
+            1,
+            xcffib.xinput.GrabType.Keycode,
+            xcffib.xinput.GrabMode22.Async,
+            xcffib.xproto.GrabMode.Async,
+            True,
+            (
+                xcffib.xinput.XIEventMask.KeyPress
+                | xcffib.xinput.XIEventMask.KeyRelease,
+            ),
+            (xcffib.xproto.KeyButMask.Mod1,),
+        )
+
+    def _ungrab_alt1(self):
+        self.manager.xinput.XIPassiveUngrabDevice(
+            self.wid,
+            KEY_1,
+            xcffib.xinput.Device.All,
+            1,
+            xcffib.xinput.GrabType.Keycode,
+            (xcffib.xproto.KeyButMask.Mod1,),
+        )
+
     def __del__(self):
         self.manager.connection.core.FreePixmap(self.pixmap_id)
-        self.manager.composite.UnredirectWindow(
+        self.manager.xcomposite.UnredirectWindow(
             self.wid, xcffib.composite.Redirect.Automatic
         )
+        self._ungrab_alt1()
 
     def get_position(self) -> QRect:
         geom = self.manager.connection.core.GetGeometry(self.wid).reply()
@@ -68,6 +121,22 @@ class X11GameInstance(QtGrabMixin, GameInstance):
         # TODO: If progress = 1 set urgency flag
         pass
 
+    @Slot(xcffib.xinput.KeyPressEvent)
+    def on_key_press(self, evt: xcffib.xinput.KeyPressEvent):
+        # Check for alt1
+        if evt.flags & xcffib.xinput.KeyEventFlags.KeyRepeat:
+            return
+
+        if evt.mods.effective & xcffib.xproto.KeyButMask.Mod1 and evt.detail == KEY_1:
+            if isinstance(evt, xcffib.xinput.KeyPressEvent):
+                self.alt1_pressed.emit()
+                self.logger.debug("alt1 pressed %s", repr(evt))
+            else:
+                self._ungrab_alt1()
+                self._grab_alt1()
+
+            return
+
     def grab_game(self):
         if (time.monotonic() - self.game_last_grab) * 1000 < self.refresh_rate:
             return self.game_last_image
@@ -77,7 +146,7 @@ class X11GameInstance(QtGrabMixin, GameInstance):
             geom = self.manager.connection.core.GetGeometry(self.pixmap_id).reply()
             xid, shm = self.manager.get_shm(geom.width * geom.height * 4)
             size = (
-                self.manager.shm.GetImage(
+                self.manager.xshm.GetImage(
                     self.pixmap_id,
                     0,
                     0,

runekit/game/x11/manager.py

@@ -12,82 +12,13 @@ import sysv_ipc
 
 from runekit.game import GameManager
 from .instance import GameInstance, X11GameInstance
+from . import ge
+from .utils import ClassDict
 
 MAX_SHM = 10
 NET_ACTIVE_WINDOW = "_NET_ACTIVE_WINDOW"
 
 
-class X11EventWorker(QObject):
-    on_active_window_changed = Signal(int)
-    on_alt1 = Signal(int)
-    stop = False
-
-    def __init__(self, manager: "X11GameManager", **kwargs):
-        super().__init__(**kwargs)
-        self.manager = manager
-        self.logger = logging.getLogger(__name__ + "." + self.__class__.__name__)
-
-        self.handlers = {
-            # ge.GenericEventCode: self.dispatch_ge,
-            xcffib.xproto.PropertyNotifyEvent: self.on_property_change,
-        }
-        # self.ge_handlers = {xinput.KeyPress: self.on_key_press}
-        self.active_win_id = self.manager.get_active_window()
-
-    @Slot()
-    def run(self):
-        self.manager.connection.core.ChangeWindowAttributes(
-            self.manager.screen.root,
-            xcffib.xproto.CW.EventMask,
-            [xcffib.xproto.EventMask.PropertyChange],
-            is_checked=True,
-        )
-
-        while True:
-            if self.stop:
-                return
-
-            evt = self.manager.connection.wait_for_event()
-
-            for wanted_type, handler in self.handlers.items():
-                if isinstance(evt, wanted_type):
-                    try:
-                        handler(evt)
-                    except:
-                        self.logger.error(
-                            "Error handling event %s", repr(evt), exc_info=True
-                        )
-
-    #
-    #     def dispatch_ge(self, evt: ge.GenericEvent):
-    #         handler = self.ge_handlers.get(evt.evtype)
-    #         if handler:
-    #             handler(evt)
-    #
-    #     def on_key_press(self, evt):
-    #         window = evt.data.event
-    #         # alt1
-    #         if evt.data.mods.effective_mods & X.Mod1Mask and evt.data.detail == 10:
-    #             self.on_alt1.emit(window.id)
-    #             self.logger.debug("alt1 pressed %s", repr(evt))
-    #
-    #         window.xinput_ungrab_keycode(evt.data.deviceid, evt.data.detail, (X.Mod1Mask,))
-    #         self.manager._setup_grab(window)
-    #
-    def on_property_change(self, evt: xcffib.xproto.PropertyNotifyEvent):
-        if evt.atom == self.manager.get_atom(NET_ACTIVE_WINDOW):
-            active_win_id = self.manager.get_active_window()
-
-            if self.active_win_id == active_win_id:
-                return
-
-            self.active_win_id = active_win_id
-            self.logger.debug(
-                "Active window changed to %d - %s", active_win_id, repr(evt)
-            )
-            self.on_active_window_changed.emit(active_win_id)
-
-
 class X11GameManager(GameManager):
     connection: xcffib.Connection
 
@@ -117,7 +48,6 @@ class X11GameManager(GameManager):
         self.event_worker.on_active_window_changed.connect(
             self.on_active_window_changed
         )
-        self.event_worker.on_alt1.connect(self.on_alt1)
 
         self.event_thread.start()
 
@@ -132,6 +62,8 @@ class X11GameManager(GameManager):
                         instance = X11GameInstance(self, wid, parent=self)
                         self._instance[wid] = instance
 
+                    return
+
             query = self.connection.core.QueryTree(wid).reply()
             for child in query.children:
                 visit(child)
@@ -140,19 +72,6 @@ class X11GameManager(GameManager):
 
         return list(self._instance.values())
 
-    # def _setup_grab(self, window: Window):
-    #     # alt1
-    #     window.xinput_grab_keycode(
-    #         xinput.AllDevices,
-    #         X.CurrentTime,
-    #         10,
-    #         xinput.GrabModeAsync,
-    #         xinput.GrabModeAsync,
-    #         True,
-    #         xinput.KeyPressMask,
-    #         (X.Mod1Mask,),
-    #     )
-
     def get_active_window(self) -> int:
         return self.get_property(self.screen.root, "_NET_ACTIVE_WINDOW")
 
@@ -165,11 +84,6 @@ class X11GameManager(GameManager):
                 instance.is_focused = active
                 instance.focusChanged.emit(active)
 
-    @Slot(int)
-    def on_alt1(self, winid):
-        if winid in self._instance:
-            self._instance[winid].alt1_pressed.emit()
-
     def _setup_composite(self):
         self.xcomposite.QueryVersion(0, 4, is_checked=True)
 
@@ -235,3 +149,102 @@ class X11GameManager(GameManager):
             self.xshm.Detach(xid)
             shm.detach()
             shm.remove()
+
+
+class X11EventWorker(QObject):
+    on_active_window_changed = Signal(int)
+    stop = False
+
+    def __init__(self, manager: "X11GameManager", **kwargs):
+        super().__init__(**kwargs)
+        self.manager = manager
+        self.logger = logging.getLogger(__name__ + "." + self.__class__.__name__)
+
+        self.handlers = ClassDict(
+            {
+                xcffib.xproto.PropertyNotifyEvent: self.on_property_change,
+                xcffib.xinput.KeyPressEvent: self.on_key_press,
+                xcffib.xinput.KeyReleaseEvent: self.on_key_press,
+            }
+        )
+        self.active_win_id = self.manager.get_active_window()
+
+    @Slot()
+    def run(self):
+        self.manager.connection.core.ChangeWindowAttributes(
+            self.manager.screen.root,
+            xcffib.xproto.CW.EventMask,
+            [xcffib.xproto.EventMask.PropertyChange],
+            is_checked=True,
+        )
+
+        while True:
+            if self.stop:
+                self.logger.debug("Exiting")
+                return
+
+            evt = self.manager.connection.wait_for_event()
+
+            if isinstance(evt, ge.GeGenericEvent):
+                evt = evt.to_event(self.manager.connection)
+
+            try:
+                handler = self.handlers[evt]
+            except KeyError:
+                self.logger.debug("Unhandled event %s", repr(evt))
+                continue
+
+            try:
+                handler(evt)
+            except:
+                self.logger.error("Error handling event %s", repr(evt), exc_info=True)
+
+    def on_key_press(self, evt: xcffib.xinput.KeyPressEvent):
+        print(
+            {
+                "deviceid": evt.deviceid,
+                "time": evt.time,
+                "detail": evt.detail,
+                "root": evt.root,
+                "event": evt.event,
+                "child": evt.child,
+                "root_x": evt.root_x,
+                "root_y": evt.root_y,
+                "event_x": evt.event_x,
+                "event_y": evt.event_y,
+                "buttons_len": evt.buttons_len,
+                "valuators_len": evt.valuators_len,
+                "sourceid": evt.sourceid,
+                "flags": evt.flags,
+            }
+        )
+        try:
+            instance = self.manager._instance[evt.root]
+        except KeyError:
+            self.logger.debug(
+                "Unknown KeyPressEvent to instance %s: %s", evt.root, repr(evt)
+            )
+            self.manager.xinput.XIPassiveUngrabDevice(
+                evt.root,
+                evt.detail,
+                evt.deviceid,
+                1,
+                xcffib.xinput.GrabType.Keycode,
+                evt.mods.effective,
+            )
+            return
+
+        instance.key_press.emit(evt)
+
+    def on_property_change(self, evt: xcffib.xproto.PropertyNotifyEvent):
+        if evt.atom == self.manager.get_atom(NET_ACTIVE_WINDOW):
+            active_win_id = self.manager.get_active_window()
+
+            if self.active_win_id == active_win_id:
+                return
+
+            self.active_win_id = active_win_id
+            self.logger.debug(
+                "Active window changed to %d - %s", active_win_id, repr(evt)
+            )
+            self.on_active_window_changed.emit(active_win_id)

runekit/game/x11/utils.py

@@ -0,0 +1,7 @@
+class ClassDict(dict[type]):
+    def __getitem__(self, item):
+        for key, value in self.items():
+            if isinstance(item, key):
+                return value
+
+        raise KeyError