Archives
/
input-remapper
mirror of https://github.com/sezanzeb/input-remapper
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

#72 ungrab and earlier grab

Browse Source
xkb
sezanzeb 3 years ago committed by sezanzeb
parent b581eca3c8
commit 3b28ab832b
3 changed files with 55 additions and 37 deletions
Show Stats Download Patch File Download Diff File

46
keymapper/injection/injector.py
Unescape Escape View File

@ -78,7 +78,7 @@ class Injector(multiprocessing.Process):
make running multiple injector easier. There is one process per
hardware-device that is being mapped.
"""
regrab_timeout = 0.5
regrab_timeout = 0.2
def __init__(self, group, mapping):
"""Setup a process to start injecting keycodes based on custom_mapping.
@ -145,8 +145,25 @@ class Injector(multiprocessing.Process):
"""Process internal stuff"""
def _grab_devices(self):
"""Grab all devices that are needed for the injection."""
sources = []
for path in self.group.paths:
source = self._grab_device(path)
if source is None:
# this path doesn't need to be grabbed for injection, because
# it doesn't provide the events needed to execute the mapping
continue
sources.append(source)
return sources
def _grab_device(self, path):
"""Try to grab the device, return None if not needed/possible."""
"""Try to grab the device, return None if not needed/possible.
Without grab, original events from it would reach the display server
even though they are mapped.
"""
try:
device = evdev.InputDevice(path)
except (FileNotFoundError, OSError):
@ -177,10 +194,6 @@ class Injector(multiprocessing.Process):
attempts = 0
while True:
try:
# grab to avoid e.g. the disabled keycode of 10 to confuse
# X, especially when one of the buttons of your mouse also
# uses 10. This also avoids having to load an empty xkb
# symbols file to prevent writing any unwanted keys.
device.grab()
logger.debug('Grab %s', path)
break
@ -191,7 +204,7 @@ class Injector(multiprocessing.Process):
# it was previously grabbed.
logger.debug('Failed attempts to grab %s: %d', path, attempts)
if attempts >= 4:
if attempts >= 10:
logger.error('Cannot grab %s, it is possibly in use', path)
logger.error(str(error))
return None
@ -328,6 +341,11 @@ class Injector(multiprocessing.Process):
# so that the macros use the correct loop
self.context = Context(self.mapping)
# grab devices as early as possible. If events appear that won't get
# released anymore before the grab they appear to be held down
# forever
sources = self._grab_devices()
self._event_producer = EventProducer(self.context)
numlock_state = is_numlock_on()
@ -341,14 +359,7 @@ class Injector(multiprocessing.Process):
events=self._construct_capabilities(GAMEPAD in self.group.types)
)
# Watch over each one of the potentially multiple devices per hardware
for path in self.group.paths:
source = self._grab_device(path)
if source is None:
# this path doesn't need to be grabbed for injection, because
# it doesn't provide the events needed to execute the mapping
continue
for source in sources:
# certain capabilities can have side effects apparently. with an
# EV_ABS capability, EV_REL won't move the mouse pointer anymore.
# so don't merge all InputDevices into one UInput device.
@ -398,6 +409,11 @@ class Injector(multiprocessing.Process):
# reached otherwise.
logger.debug('asyncio coroutines ended')
for source in sources:
# ungrab at the end to make the next injection process not fail
# its grabs
source.ungrab()
async def _event_consumer(self, source, forward_to):
"""Reads input events to inject keycodes or talk to the event_producer.

29
tests/test.py
Unescape Escape View File

@ -47,6 +47,10 @@ assert not os.getcwd().endswith('tests')
os.environ['UNITTEST'] = '1'
def grey_log(*msgs):
print(f'\033[90m{" ".join(msgs)}\033[0m')
def is_service_running():
"""Check if the daemon is running."""
try:
@ -67,11 +71,7 @@ def join_children():
for child in children:
if i > 10:
child.kill()
print(
f'\033[90m' # color
f'Killed pid {child.pid} because it didn\'t finish in time'
'\033[0m' # end style
)
grey_log(f'Killed pid {child.pid} because it didn\'t finish in time')
children = this.children(recursive=True)
time.sleep(EVENT_READ_TIMEOUT)
@ -326,17 +326,16 @@ class InputDevice:
return self.fd
def log(self, key, msg):
print(
f'\033[90m' # color
f'{msg} "{self.name}" "{self.path}" {key}'
'\033[0m' # end style
)
grey_log(f'{msg} "{self.name}" "{self.path}" {key}')
def absinfo(self, *args):
raise Exception('Ubuntus version of evdev doesn\'t support .absinfo')
def grab(self):
pass
grey_log('grab', self.name, self.path)
def ungrab(self):
grey_log('ungrab', self.name, self.path)
async def async_read_loop(self):
if pending_events.get(self.group_key) is None:
@ -435,11 +434,7 @@ class UInput:
uinput_write_history.append(event)
uinput_write_history_pipe[1].send(event)
self.write_history.append(event)
print(
f'\033[90m' # color
f'{(type, code, value)} written'
'\033[0m' # end style
)
grey_log(f'{(type, code, value)} written')
def syn(self):
pass
@ -521,7 +516,7 @@ from keymapper.injection.macros import macro_variables
from keymapper.injection.keycode_mapper import active_macros, unreleased
# no need for a high number in tests
Injector.regrab_timeout = 0.15
Injector.regrab_timeout = 0.05
_fixture_copy = copy.deepcopy(fixtures)

17
tests/testcases/test_injector.py
Unescape Escape View File

@ -77,10 +77,11 @@ class TestInjector(unittest.TestCase):
def test_grab(self):
# path is from the fixtures
path = '/dev/input/event10'
custom_mapping.change(Key(EV_KEY, 10, 1), 'a')
self.injector = Injector(groups.find(key='Foo Device 2'), custom_mapping)
path = '/dev/input/event10'
# this test needs to pass around all other constraints of
# _grab_device
self.injector.context = Context(custom_mapping)
@ -92,7 +93,7 @@ class TestInjector(unittest.TestCase):
self.assertEqual(device.name, fixtures[path]['name'])
def test_fail_grab(self):
self.make_it_fail = 10
self.make_it_fail = 999
custom_mapping.change(Key(EV_KEY, 10, 1), 'a')
self.injector = Injector(groups.find(key='Foo Device 2'), custom_mapping)
@ -107,7 +108,7 @@ class TestInjector(unittest.TestCase):
self.assertEqual(self.injector.get_state(), STARTING)
# since none can be grabbed, the process will terminate. But that
# actually takes quite some time.
time.sleep(1.5)
time.sleep(self.injector.regrab_timeout * 12)
self.assertFalse(self.injector.is_alive())
self.assertEqual(self.injector.get_state(), NO_GRAB)
@ -403,7 +404,8 @@ class TestInjector(unittest.TestCase):
self.assertEqual(history.count((EV_KEY, 77, 1)), 1)
self.assertEqual(history.count((EV_ABS, ABS_RZ, value)), 1)
def test_gamepad_to_mouse_event_producer(self):
@mock.patch('evdev.InputDevice.ungrab')
def test_gamepad_to_mouse_event_producer(self, ungrab_patch):
custom_mapping.set('gamepad.joystick.left_purpose', MOUSE)
custom_mapping.set('gamepad.joystick.right_purpose', NONE)
self.injector = Injector(groups.find(name='gamepad'), custom_mapping)
@ -424,6 +426,8 @@ class TestInjector(unittest.TestCase):
MOUSE
)
self.assertEqual(ungrab_patch.call_count, 1)
def test_gamepad_to_buttons_event_producer(self):
custom_mapping.set('gamepad.joystick.left_purpose', BUTTONS)
custom_mapping.set('gamepad.joystick.right_purpose', BUTTONS)
@ -459,7 +463,8 @@ class TestInjector(unittest.TestCase):
'key-mapper abcd forwarded'
)
def test_capabilities_and_uinput_presence(self):
@mock.patch('evdev.InputDevice.ungrab')
def test_capabilities_and_uinput_presence(self, ungrab_patch):
custom_mapping.change(Key(EV_KEY, KEY_A, 1), 'c')
custom_mapping.change(Key(EV_REL, REL_HWHEEL, 1), 'k(b)')
self.injector = Injector(groups.find(key='Foo Device 2'), custom_mapping)
@ -518,6 +523,8 @@ class TestInjector(unittest.TestCase):
keyboard_keys
)
self.assertEqual(ungrab_patch.call_count, 2)
def test_injector(self):
# the tests in test_keycode_mapper.py test this stuff in detail

Write Preview
Loading…
Cancel
Save