KeyBinder.handle_keypress() - Code Metrics - Inspection of "keybinder.KeyBinder: Refactor XGrabKey callback" - ssokolow/quicktile - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 858ae1...f1c0c2 )

by Stephan

created 2017-08-22 08:59 UTC

KeyBinder.handle_keypress() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
dl	0
loc	17
rs	9.4285
c	1
b	0
f	0

"""Xlib-based global hotkey-binding code"""

__author__ = "Stephan Sokolow (deitarion/SSokolow)"
__license__ = "GNU GPL 2.0 or later"

import logging

import gobject, gtk
from Xlib import X
from Xlib.display import Display
from Xlib.error import BadAccess, DisplayConnectionError
from Xlib.protocol.event import KeyPress as XKeyPress

from .util import powerset, XInitError

# Allow MyPy to work without depending on the `typing` package
# (And silence complaints from only using the imported types in comments)
try:
    # pylint: disable=unused-import
    from typing import (Any, Callable, Dict, Iterable, Iterator, List,  # NOQA
                        Optional, Sequence, Sized, Tuple)

    from Xlib.error import XError          # NOQA
    from .commands import CommandRegistry  # NOQA
    from .wm import WindowManager          # NOQA
    from .util import CommandCB            # NOQA
except:  # pylint: disable=bare-except
    pass

class KeyBinder(object):
    """A convenience class for wrapping C{XGrabKey}."""

    #: @todo: Figure out how to set the modifier mask in X11 and use
    #:        C{gtk.accelerator_get_default_mod_mask()} to feed said code.
    ignored_modifiers = ['Mod2Mask', 'LockMask']

    #: Used to pass state from L{cb_xerror}
    keybind_failed = False

    def __init__(self, xdisplay=None):  # type: (Optional[Display]) -> None
        """Connect to X11 and the Glib event loop.

        @param xdisplay: A C{python-xlib} display handle.
        @type xdisplay: C{Xlib.display.Display}
        """
        try:
            self.xdisp = xdisplay or Display()
        except (UnicodeDecodeError, DisplayConnectionError), err:
            raise XInitError("python-xlib failed with %s when asked to open"
                             " a connection to the X server. Cannot bind keys."
                             "\n\tIt's unclear why this happens, but it is"
                             " usually fixed by deleting your ~/.Xauthority"
                             " file and rebooting."
                             % err.__class__.__name__)

        self.xroot = self.xdisp.screen().root
        self._keys = {}  # type: Dict[Tuple[int, int], Callable]

        # Resolve these at runtime to avoid NameErrors
        self._ignored_modifiers = [getattr(X, name) for name in
                                   self.ignored_modifiers]  # type: List[int]

        # We want to receive KeyPress events
        self.xroot.change_attributes(event_mask=X.KeyPressMask)

        # Set up a handler to catch XGrabKey() failures
        self.xdisp.set_error_handler(self.cb_xerror)

        # Merge python-xlib into the Glib event loop
        # Source: http://www.pygtk.org/pygtk2tutorial/sec-MonitoringIO.html
        gobject.io_add_watch(self.xroot.display,
                             gobject.IO_IN, self.cb_xevent)

    def bind(self, accel, callback):  # type: (str, Callable[[], None]) -> bool
        """Bind a global key combination to a callback.

        @param accel: An accelerator as either a string to be parsed by
            C{gtk.accelerator_parse()} or a tuple as returned by it.)
        @param callback: The function to call when the key is pressed.

        @type accel: C{str} or C{(int, gtk.gdk.ModifierType)} or C{(int, int)}
        @type callback: C{function}

        @returns: A boolean indicating whether the provided keybinding was
            parsed successfully. (But not whether it was registered
            successfully due to the asynchronous nature of the C{XGrabKey}
            request.)
        @rtype: C{bool}
        """
        keycode, modmask = self.parse_accel(accel)
        if keycode is None or modmask is None:
            return False

        # Ignore modifiers like Mod2 (NumLock) and Lock (CapsLock)
        self._keys[(keycode, 0)] = callback  # Null modifiers seem to be a risk
        for mmask in self._vary_modmask(modmask, self._ignored_modifiers):
            self._keys[(keycode, mmask)] = callback
            self.xroot.grab_key(keycode, mmask,
                                1, X.GrabModeAsync, X.GrabModeAsync)

        # If we don't do this, then nothing works.
        # I assume it flushes the XGrabKey calls to the server.
        self.xdisp.sync()

        # React to any cb_xerror that might have resulted from xdisp.sync()
        if self.keybind_failed:
            self.keybind_failed = False
            logging.warning("Failed to bind key. It may already be in use: %s",
                            accel)
            return False

        return True

    def cb_xerror(self, err, _):  # type: (XError, Any) -> None
        """Used to identify when attempts to bind keys fail.
        @note: If you can make python-xlib's C{CatchError} actually work or if
               you can retrieve more information to show, feel free.
        """
        if isinstance(err, BadAccess):
            self.keybind_failed = True
        else:
            self.xdisp.display.default_error_handler(err)

    def cb_xevent(self, src, cond, handle=None):  # pylint: disable=W0613
        # type: (Any, Any, Optional[Display]) -> bool
        """Callback to dispatch X events to more specific handlers.

        @rtype: C{True}

        @todo: Make sure uncaught exceptions are prevented from making
            quicktile unresponsive in the general case.
        """
        handle = handle or self.xroot.display

        for _ in range(0, handle.pending_events()):
            xevent = handle.next_event()
            if xevent.type == X.KeyPress:
                self.handle_keypress(xevent)

        # Necessary for proper function
        return True

    def handle_keypress(self, xevent):  # type: (XKeyPress) -> None
        """Dispatch C{XKeyPress} events to their callbacks."""
        keysig = (xevent.detail, xevent.state)
        if keysig not in self._keys:
            logging.error("Received an event for an unrecognized keybind: "
                          "%s, %s", xevent.detail, xevent.state)
            return

        # Display a meaningful debug message
        # FIXME: Only call this code if --debug
        # FIXME: Proper "index" arg for keycode_to_keysym
        keysym = self.xdisp.keycode_to_keysym(keysig[0], 0)
        kbstr = gtk.accelerator_name(keysym, keysig[1]) # pylint: disable=E1101
        logging.debug("Received keybind: %s", kbstr)

        # Call the associated callback
        self._keys[keysig]()

    def parse_accel(self, accel  # type: str
                    ):  # type: (...) -> Tuple[Optional[int], Optional[int]]
        """Convert an accelerator string into the form XGrabKey needs."""

        keysym, modmask = gtk.accelerator_parse(accel)
        if not gtk.accelerator_valid(keysym, modmask):  # pylint: disable=E1101
            logging.error("Invalid keybinding: %s", accel)
            return None, None

        if modmask > 2**16 - 1:
            logging.error("Modifier out of range for XGrabKey "
                          "(int(modmask) > 65535). "
                          "Did you use <Super> instead of <Mod4>?")
            return None, None

        # Convert to what XGrabKey expects
        keycode = self.xdisp.keysym_to_keycode(keysym)
        if isinstance(modmask, gtk.gdk.ModifierType):
            modmask = modmask.real

        return keycode, modmask

    @staticmethod
    def _vary_modmask(modmask, ignored):
        # type: (int, Sequence[int]) -> Iterator[int]
        """Generate all possible variations on C{modmask} that need to be
        taken into consideration if we can't properly ignore the modifiers in
        C{ignored}. (Typically NumLock and CapsLock)

        @param modmask: A bitfield to be combinatorically grown.
        @param ignored: Modifiers to be combined with C{modmask}.

        @type modmask: C{int} or C{gtk.gdk.ModifierType}
        @type ignored: C{list(int)}

        @rtype: generator of C{type(modmask)}
        """

        for ignored in powerset(ignored):
            imask = reduce(lambda x, y: x | y, ignored, 0)
            yield modmask | imask

def init(modmask,   # type: Optional[str]
         mappings,  # type: Dict[str, CommandCB]
         commands,  # type: CommandRegistry
         winman     # type: WindowManager
         ):         # type: (...) -> Optional[KeyBinder]
    """Initialize the keybinder and bind the requested mappings"""
    # Allow modmask to be empty for keybinds which don't share a common prefix
    if not modmask or modmask.lower() == 'none':
        modmask = ''

    try:
        keybinder = KeyBinder()
    except XInitError as err:
        logging.error("%s", err)
        return None
    else:
        # TODO: Take a mapping dict with pre-modmasked keys
        #       and pre-closured commands
        for key, func in mappings.items():
            def call(func=func):
                """Closure to resolve `func` and call it on a
                   `WindowManager` instance"""
                commands.call(func, winman)

            keybinder.bind(modmask + key, call)
    return keybinder


1			"""Xlib-based global hotkey-binding code"""
2
3			__author__ = "Stephan Sokolow (deitarion/SSokolow)"
4			__license__ = "GNU GPL 2.0 or later"
5
6			import logging
7
8			import gobject, gtk
9			from Xlib import X
10			from Xlib.display import Display
11			from Xlib.error import BadAccess, DisplayConnectionError
12			from Xlib.protocol.event import KeyPress as XKeyPress
13
14			from .util import powerset, XInitError
15
16			# Allow MyPy to work without depending on the `typing` package
17			# (And silence complaints from only using the imported types in comments)
18			try:
19			# pylint: disable=unused-import
20			from typing import (Any, Callable, Dict, Iterable, Iterator, List, # NOQA
21			Optional, Sequence, Sized, Tuple)
22
23			from Xlib.error import XError # NOQA
24			from .commands import CommandRegistry # NOQA
25			from .wm import WindowManager # NOQA
26			from .util import CommandCB # NOQA
27			except: # pylint: disable=bare-except
28			pass
29
30			class KeyBinder(object):
31			"""A convenience class for wrapping C{XGrabKey}."""
32
33			#: @todo: Figure out how to set the modifier mask in X11 and use
34			#: C{gtk.accelerator_get_default_mod_mask()} to feed said code.
35			ignored_modifiers = ['Mod2Mask', 'LockMask']
36
37			#: Used to pass state from L{cb_xerror}
38			keybind_failed = False
39
40			def __init__(self, xdisplay=None): # type: (Optional[Display]) -> None
41			"""Connect to X11 and the Glib event loop.
42
43			@param xdisplay: A C{python-xlib} display handle.
44			@type xdisplay: C{Xlib.display.Display}
45			"""
46			try:
47			self.xdisp = xdisplay or Display()
48			except (UnicodeDecodeError, DisplayConnectionError), err:
49			raise XInitError("python-xlib failed with %s when asked to open"
50			" a connection to the X server. Cannot bind keys."
51			"\n\tIt's unclear why this happens, but it is"
52			" usually fixed by deleting your ~/.Xauthority"
53			" file and rebooting."
54			% err.__class__.__name__)
55
56			self.xroot = self.xdisp.screen().root
57			self._keys = {} # type: Dict[Tuple[int, int], Callable]
58
59			# Resolve these at runtime to avoid NameErrors
60			self._ignored_modifiers = [getattr(X, name) for name in
61			self.ignored_modifiers] # type: List[int]
62
63			# We want to receive KeyPress events
64			self.xroot.change_attributes(event_mask=X.KeyPressMask)
65
66			# Set up a handler to catch XGrabKey() failures
67			self.xdisp.set_error_handler(self.cb_xerror)
68
69			# Merge python-xlib into the Glib event loop
70			# Source: http://www.pygtk.org/pygtk2tutorial/sec-MonitoringIO.html
71			gobject.io_add_watch(self.xroot.display,
72			gobject.IO_IN, self.cb_xevent)
73
74			def bind(self, accel, callback): # type: (str, Callable[[], None]) -> bool
75			"""Bind a global key combination to a callback.
76
77			@param accel: An accelerator as either a string to be parsed by
78			C{gtk.accelerator_parse()} or a tuple as returned by it.)
79			@param callback: The function to call when the key is pressed.
80
81			@type accel: C{str} or C{(int, gtk.gdk.ModifierType)} or C{(int, int)}
82			@type callback: C{function}
83
84			@returns: A boolean indicating whether the provided keybinding was
85			parsed successfully. (But not whether it was registered
86			successfully due to the asynchronous nature of the C{XGrabKey}
87			request.)
88			@rtype: C{bool}
89			"""
90			keycode, modmask = self.parse_accel(accel)
91			if keycode is None or modmask is None:
92			return False
93
94			# Ignore modifiers like Mod2 (NumLock) and Lock (CapsLock)
95			self._keys[(keycode, 0)] = callback # Null modifiers seem to be a risk
96			for mmask in self._vary_modmask(modmask, self._ignored_modifiers):
97			self._keys[(keycode, mmask)] = callback
98			self.xroot.grab_key(keycode, mmask,
99			1, X.GrabModeAsync, X.GrabModeAsync)
100
101			# If we don't do this, then nothing works.
102			# I assume it flushes the XGrabKey calls to the server.
103			self.xdisp.sync()
104
105			# React to any cb_xerror that might have resulted from xdisp.sync()
106			if self.keybind_failed:
107			self.keybind_failed = False
108			logging.warning("Failed to bind key. It may already be in use: %s",
109			accel)
110			return False
111
112			return True
113
114			def cb_xerror(self, err, _): # type: (XError, Any) -> None
115			"""Used to identify when attempts to bind keys fail.
116			@note: If you can make python-xlib's C{CatchError} actually work or if
117			you can retrieve more information to show, feel free.
118			"""
119			if isinstance(err, BadAccess):
120			self.keybind_failed = True
121			else:
122			self.xdisp.display.default_error_handler(err)
123
124			def cb_xevent(self, src, cond, handle=None): # pylint: disable=W0613
125			# type: (Any, Any, Optional[Display]) -> bool
126			"""Callback to dispatch X events to more specific handlers.
127
128			@rtype: C{True}
129
130			@todo: Make sure uncaught exceptions are prevented from making
131			quicktile unresponsive in the general case.
132			"""
133			handle = handle or self.xroot.display
134
135			for _ in range(0, handle.pending_events()):
136			xevent = handle.next_event()
137			if xevent.type == X.KeyPress:
138			self.handle_keypress(xevent)
139
140			# Necessary for proper function
141			return True
142
143			def handle_keypress(self, xevent): # type: (XKeyPress) -> None
144			"""Dispatch C{XKeyPress} events to their callbacks."""
145			keysig = (xevent.detail, xevent.state)
146			if keysig not in self._keys:
147			logging.error("Received an event for an unrecognized keybind: "
148			"%s, %s", xevent.detail, xevent.state)
149			return
150
151			# Display a meaningful debug message
152			# FIXME: Only call this code if --debug
153			# FIXME: Proper "index" arg for keycode_to_keysym
154			keysym = self.xdisp.keycode_to_keysym(keysig[0], 0)
155			kbstr = gtk.accelerator_name(keysym, keysig[1]) # pylint: disable=E1101
156			logging.debug("Received keybind: %s", kbstr)
157
158			# Call the associated callback
159			self._keys[keysig]()
160
161			def parse_accel(self, accel # type: str
162			): # type: (...) -> Tuple[Optional[int], Optional[int]]
163			"""Convert an accelerator string into the form XGrabKey needs."""
164
165			keysym, modmask = gtk.accelerator_parse(accel)
166			if not gtk.accelerator_valid(keysym, modmask): # pylint: disable=E1101
167			logging.error("Invalid keybinding: %s", accel)
168			return None, None
169
170			if modmask > 2**16 - 1:
171			logging.error("Modifier out of range for XGrabKey "
172			"(int(modmask) > 65535). "
173			"Did you use <Super> instead of <Mod4>?")
174			return None, None
175
176			# Convert to what XGrabKey expects
177			keycode = self.xdisp.keysym_to_keycode(keysym)
178			if isinstance(modmask, gtk.gdk.ModifierType):
179			modmask = modmask.real
180
181			return keycode, modmask
182
183			@staticmethod
184			def _vary_modmask(modmask, ignored):
185			# type: (int, Sequence[int]) -> Iterator[int]
186			"""Generate all possible variations on C{modmask} that need to be
187			taken into consideration if we can't properly ignore the modifiers in
188			C{ignored}. (Typically NumLock and CapsLock)
189
190			@param modmask: A bitfield to be combinatorically grown.
191			@param ignored: Modifiers to be combined with C{modmask}.
192
193			@type modmask: C{int} or C{gtk.gdk.ModifierType}
194			@type ignored: C{list(int)}
195
196			@rtype: generator of C{type(modmask)}
197			"""
198
199			for ignored in powerset(ignored):
200			imask = reduce(lambda x, y: x \| y, ignored, 0)
201			yield modmask \| imask
202
203			def init(modmask, # type: Optional[str]
204			mappings, # type: Dict[str, CommandCB]
205			commands, # type: CommandRegistry
206			winman # type: WindowManager
207			): # type: (...) -> Optional[KeyBinder]
208			"""Initialize the keybinder and bind the requested mappings"""
209			# Allow modmask to be empty for keybinds which don't share a common prefix
210			if not modmask or modmask.lower() == 'none':
211			modmask = ''
212
213			try:
214			keybinder = KeyBinder()
215			except XInitError as err:
216			logging.error("%s", err)
217			return None
218			else:
219			# TODO: Take a mapping dict with pre-modmasked keys
220			# and pre-closured commands
221			for key, func in mappings.items():
222			def call(func=func):
223			"""Closure to resolve `func` and call it on a
224			`WindowManager` instance"""
225			commands.call(func, winman)
226
227			keybinder.bind(modmask + key, call)
228			return keybinder
229

ssokolow / quicktile

Push — master ( 858ae1...f1c0c2 )

KeyBinder.handle_keypress() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like