storm/cthulhu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Lots of work on the web side of things. More coming.

Browse Source
This commit is contained in:
Storm Dragon
2026-07-19 15:00:25 -04:00
parent 3e97c1ce33
commit 3d44f6d50b
8 changed files with 1738 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cthulhu/input_event_manager.py
+152 -4
View File
@@ -46,6 +46,7 @@ import gi
gi.require_version("Atspi", "2.0")
from gi.repository import Atspi
from gi.repository import GLib
import pyatspi
from . import debug
from . import focus_manager
@@ -61,6 +62,44 @@ from .ax_utilities_application import AXUtilitiesApplication
if TYPE_CHECKING:
from . import keybindings
_X11_SHIFT_MASK = 1 << int(Atspi.ModifierType.SHIFT)
_X11_SHIFT_LOCK_MASK = 1 << int(Atspi.ModifierType.SHIFTLOCK)
_X11_NUM_LOCK_MASK = 1 << int(Atspi.ModifierType.NUMLOCK)
_X11_CONSUMABLE_MODIFIER_MASKS = (
0,
_X11_SHIFT_MASK,
_X11_SHIFT_LOCK_MASK,
_X11_SHIFT_MASK | _X11_SHIFT_LOCK_MASK,
_X11_NUM_LOCK_MASK,
_X11_SHIFT_MASK | _X11_NUM_LOCK_MASK,
_X11_SHIFT_LOCK_MASK | _X11_NUM_LOCK_MASK,
_X11_SHIFT_MASK | _X11_SHIFT_LOCK_MASK | _X11_NUM_LOCK_MASK,
)
_X11_CONSUMABLE_KEYSYMS = (
"a", "b", "c", "d", "e", "f", "g", "h", "i", "k", "l", "m",
"o", "p", "q", "r", "s", "t", "u", "v", "x", "y",
"1", "2", "3", "4", "5", "6", "comma",
)
def _get_x11_consumable_key_set() -> List[Atspi.KeyDefinition]:
"""Returns AT-SPI definitions for unmodified structural-navigation keys."""
# Importing here avoids a module-level cycle: keybindings obtains this
# manager lazily when bindings install their AT-SPI grabs.
from . import keybindings # pylint: disable=import-outside-toplevel
keycodes = sorted({
keycode
for keysym in _X11_CONSUMABLE_KEYSYMS
if (keycode := keybindings.get_keycodes(keysym)[1])
})
keySet = []
for keycode in keycodes:
definition = Atspi.KeyDefinition()
definition.keycode = keycode
keySet.append(definition)
return keySet
class InputEventManager:
"""Provides utilities for managing input events."""
@@ -68,6 +107,11 @@ class InputEventManager:
self._last_input_event: Optional[input_event.InputEvent] = None
self._last_non_modifier_key_event: Optional[input_event.KeyboardEvent] = None
self._device: Optional[Atspi.Device] = None
self._key_pressed_id: int = 0
self._key_released_id: int = 0
self._selective_legacy_listener_active: bool = False
self._selective_legacy_key_set: List[Atspi.KeyDefinition] = []
self._selective_legacy_keycodes: set[int] = set()
self._pointer_moved_id: int = 0
self._mapped_keycodes: List[int] = []
self._mapped_keysyms: List[int] = []
@@ -101,13 +145,56 @@ class InputEventManager:
msg = "INPUT EVENT MANAGER: Starting key watcher."
debug.print_message(debug.LEVEL_INFO, msg, True)
self.activate_device().add_key_watcher(self.process_keyboard_event)
device = self.activate_device()
atspiVersion = Atspi.get_version()
if atspiVersion[0] > 2 or atspiVersion[1] >= 60:
msg = "INPUT EVENT MANAGER: Using AT-SPI key signals."
debug.print_message(debug.LEVEL_INFO, msg, True)
self._key_pressed_id = device.connect("key-pressed", self._on_key_pressed)
self._key_released_id = device.connect("key-released", self._on_key_released)
if os.environ.get("XDG_SESSION_TYPE", "").lower() == "x11":
self._selective_legacy_key_set = _get_x11_consumable_key_set()
if self._selective_legacy_key_set:
self._selective_legacy_keycodes = {
definition.keycode for definition in self._selective_legacy_key_set
}
msg = "INPUT EVENT MANAGER: Adding selective consumable X11 key listener."
debug.print_message(debug.LEVEL_INFO, msg, True)
pyatspi.Registry.registerKeystrokeListener(
self._process_selective_legacy_keyboard_event,
key_set=self._selective_legacy_key_set,
mask=list(_X11_CONSUMABLE_MODIFIER_MASKS),
kind=(Atspi.EventType.KEY_PRESSED_EVENT, Atspi.EventType.KEY_RELEASED_EVENT),
synchronous=True,
preemptive=True,
)
self._selective_legacy_listener_active = True
else:
msg = "INPUT EVENT MANAGER: Using legacy AT-SPI key watcher."
debug.print_message(debug.LEVEL_INFO, msg, True)
device.add_key_watcher(self.process_keyboard_event)
def stop_key_watcher(self) -> None:
"""Starts the watcher for keyboard input events."""
"""Stops the watcher for keyboard input events."""
msg = "INPUT EVENT MANAGER: Stopping key watcher."
debug.print_message(debug.LEVEL_INFO, msg, True)
if self._selective_legacy_listener_active:
pyatspi.Registry.deregisterKeystrokeListener(
self._process_selective_legacy_keyboard_event,
key_set=self._selective_legacy_key_set,
mask=list(_X11_CONSUMABLE_MODIFIER_MASKS),
kind=(Atspi.EventType.KEY_PRESSED_EVENT, Atspi.EventType.KEY_RELEASED_EVENT),
)
self._selective_legacy_listener_active = False
self._selective_legacy_key_set = []
self._selective_legacy_keycodes.clear()
if self._device is not None:
for handlerId in (self._key_pressed_id, self._key_released_id):
if handlerId:
self._device.disconnect(handlerId)
self._key_pressed_id = 0
self._key_released_id = 0
self._device = None
def get_debug_snapshot(self) -> Dict[str, object]:
@@ -229,6 +316,11 @@ class InputEventManager:
grab_ids = []
for kd in binding.key_definitions():
definitionKeycode = kd.keycode or binding.keycode
if self._signal_event_uses_selective_legacy_listener(
definitionKeycode, kd.modifiers
):
continue
grab_id = self._device.add_key_grab(kd, None)
if grab_id == 0:
continue
@@ -756,6 +848,62 @@ class InputEventManager:
# pylint: disable=too-many-arguments
# pylint: disable=too-many-positional-arguments
def _process_selective_legacy_keyboard_event(self, event: Atspi.DeviceEvent) -> bool:
"""Returns an X11 consume decision for unmodified browse-navigation keys."""
pressed = event.type == Atspi.EventType.KEY_PRESSED_EVENT
return self.process_keyboard_event(
self._device,
pressed,
event.hw_code,
event.id,
event.modifiers,
event.event_string or "",
)
def _signal_event_uses_selective_legacy_listener(self, keycode: int, modifiers: int) -> bool:
"""Returns True if the X11 compatibility listener owns this event."""
return (
self._selective_legacy_listener_active
and keycode in self._selective_legacy_keycodes
and modifiers in _X11_CONSUMABLE_MODIFIER_MASKS
)
def _on_key_pressed(
self,
device: Atspi.Device,
keycode: int,
keysym: int,
modifiers: int,
text: str,
) -> None:
"""Processes a key-pressed signal from AT-SPI 2.60 and later."""
if self._signal_event_uses_selective_legacy_listener(keycode, modifiers):
msg = "INPUT EVENT MANAGER: Deferring key-pressed signal to X11 consumable listener."
debug.print_message(debug.LEVEL_INFO, msg, True)
return
self.process_keyboard_event(device, True, keycode, keysym, modifiers, text)
def _on_key_released(
self,
device: Atspi.Device,
keycode: int,
keysym: int,
modifiers: int,
text: str,
) -> None:
"""Processes a key-released signal from AT-SPI 2.60 and later."""
if self._signal_event_uses_selective_legacy_listener(keycode, modifiers):
msg = "INPUT EVENT MANAGER: Deferring key-released signal to X11 consumable listener."
debug.print_message(debug.LEVEL_INFO, msg, True)
return
self.process_keyboard_event(device, False, keycode, keysym, modifiers, text)
def process_keyboard_event(
self,
_device: Atspi.Device,
@@ -849,7 +997,7 @@ class InputEventManager:
event._finalize_initialization()
event.set_click_count(self._determine_keyboard_event_click_count(event))
event.process()
consumed = event.process()
if event.is_modifier_key():
if self.is_release_for(event, self._last_input_event):
@@ -859,7 +1007,7 @@ class InputEventManager:
else:
self._last_non_modifier_key_event = event
self._last_input_event = event
return True
return consumed
# pylint: enable=too-many-arguments
# pylint: enable=too-many-positional-arguments