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

some simplifications of the Injector class api

Browse Source
pull/45/head
sezanzeb 4 years ago
parent ec35abf7df
commit a6b69be088
5 changed files with 72 additions and 67 deletions
Show Stats Download Patch File Download Diff File

2
keymapper/daemon.py
Unescape Escape View File

@ -443,7 +443,7 @@ class Daemon:
try:
injector = Injector(device, mapping)
injector.start_injecting()
injector.start()
self.injectors[device] = injector
except OSError:
# I think this will never happen, probably leftover from

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

@ -129,7 +129,7 @@ def is_in_capabilities(key, capabilities):
return False
class Injector:
class Injector(multiprocessing.Process):
"""Keeps injecting events in the background based on mapping and config.
Is a process to make it non-blocking for the rest of the code and to
@ -139,7 +139,7 @@ class Injector:
regrab_timeout = 0.5
def __init__(self, device, mapping):
"""Start injecting keycodes based on custom_mapping.
"""Setup a process to start injecting keycodes based on custom_mapping.
Parameters
----------
@ -148,14 +148,57 @@ class Injector:
mapping : Mapping
"""
self.device = device
self.mapping = mapping
self._process = None
self._msg_pipe = multiprocessing.Pipe()
self._key_to_code = self._map_keys_to_codes()
self._state = UNKNOWN
self._event_producer = None
self._state = UNKNOWN
self._msg_pipe = multiprocessing.Pipe()
super().__init__()
# Functions to interact with the running process:
def get_state(self):
"""Get the state of the injection.
Can be safely called from the main process.
"""
# slowly figure out what is going on
alive = self.is_alive()
if self._state == UNKNOWN and not alive:
# didn't start yet
return self._state
# if it is alive, it is definitely at least starting up
if self._state == UNKNOWN and alive:
self._state = STARTING
# if there is a message available, it might have finished starting up
if self._state == STARTING and self._msg_pipe[1].poll():
msg = self._msg_pipe[1].recv()
if msg == OK:
self._state = RUNNING
if msg == NO_GRAB:
self._state = NO_GRAB
if self._state in [STARTING, RUNNING] and not alive:
self._state = FAILED
logger.error('Injector was unexpectedly found stopped')
return self._state
@ensure_numlock
def stop_injecting(self):
"""Stop injecting keycodes.
Can be safely called from the main procss.
"""
logger.info('Stopping injecting keycodes for device "%s"', self.device)
self._msg_pipe[1].send(CLOSE)
self._state = STOPPED
# Process internal stuff:
def _forwards_joystick(self):
"""If at least one of the joysticks remains a regular joystick."""
@ -205,40 +248,6 @@ class Injector:
return key_to_code
def start_injecting(self):
"""Start injecting keycodes."""
if self._process is not None:
# So that there is less concern about integrity when putting
# stuff into self. Each injector object can only be
# started once.
raise Exception('Please construct a new injector instead')
if self.device not in get_devices():
logger.error('Cannot inject for unknown device "%s"', self.device)
return
self._state = STARTING
self._process = multiprocessing.Process(target=self._start_injecting)
self._process.start()
def get_state(self):
"""Get the state of the injection."""
# only at this point the actual state is figured out
if self._state == STARTING and self._msg_pipe[1].poll():
msg = self._msg_pipe[1].recv()
if msg == OK:
self._state = RUNNING
if msg == NO_GRAB:
self._state = NO_GRAB
alive = self._process is not None and self._process.is_alive()
if self._state in [STARTING, RUNNING] and not alive:
self._state = FAILED
logger.error('Injector was unexpectedly found stopped')
return self._state
def _grab_device(self, path):
"""Try to grab the device, return None if not needed/possible."""
try:
@ -386,7 +395,7 @@ class Injector:
loop.stop()
return
def _start_injecting(self):
def run(self):
"""The injection worker that keeps injecting until terminated.
Stuff is non-blocking by using asyncio in order to do multiple things
@ -395,6 +404,10 @@ class Injector:
Use this function as starting point in a process. It creates
the loops needed to read and map events and keeps running them.
"""
if self.device not in get_devices():
logger.error('Cannot inject for unknown device "%s"', self.device)
return
# create a new event loop, because somehow running an infinite loop
# that sleeps on iterations (event_producer) in one process causes
# another injection process to screw up reading from the grabbed
@ -565,10 +578,3 @@ class Injector:
'The consumer for "%s" stopped early',
source.path
)
@ensure_numlock
def stop_injecting(self):
"""Stop injecting keycodes."""
logger.info('Stopping injecting keycodes for device "%s"', self.device)
self._msg_pipe[1].send(CLOSE)
self._state = STOPPED

1
keymapper/injection/macros.py
Unescape Escape View File

@ -112,7 +112,6 @@ class _Macro:
handler : function
Will receive int code and value for an EV_KEY event to write
"""
# TODO test handler
self.running = True
for _, task in self.tasks:
coroutine = task(handler)

4
readme/pylint.svg
Unescape Escape View File

@ -17,7 +17,7 @@
<text x="22.0" y="14">pylint</text>
</g>
<g fill="#fff" text-anchor="middle" font-family="DejaVu Sans,Verdana,Geneva,sans-serif" font-size="11">
<text x="63.0" y="15" fill="#010101" fill-opacity=".3">9.74</text>
<text x="62.0" y="14">9.74</text>
<text x="63.0" y="15" fill="#010101" fill-opacity=".3">9.75</text>
<text x="62.0" y="14">9.75</text>
</g>
</svg>
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 1.0 KiB

After

Width:  |  Height:  |  Size: 1.0 KiB

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

@ -103,12 +103,12 @@ class TestInjector(unittest.TestCase):
self.assertGreaterEqual(self.failed, 1)
self.assertEqual(self.injector.get_state(), UNKNOWN)
self.injector.start_injecting()
self.injector.start()
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)
self.assertFalse(self.injector._process.is_alive())
self.assertFalse(self.injector.is_alive())
self.assertEqual(self.injector.get_state(), NO_GRAB)
def test_grab_device_1(self):
@ -333,7 +333,7 @@ class TestInjector(unittest.TestCase):
]
self.injector = Injector('gamepad', custom_mapping)
self.injector.start_injecting()
self.injector.start()
# wait for the injector to start sending, at most 1s
uinput_write_history_pipe[0].poll(1)
@ -381,7 +381,7 @@ class TestInjector(unittest.TestCase):
custom_mapping.change(Key((1, BTN_A, 1)), 'b')
system_mapping._set('b', 77)
self.injector = Injector('gamepad', custom_mapping)
self.injector.start_injecting()
self.injector.start()
# wait for the injector to start sending, at most 1s
uinput_write_history_pipe[0].poll(1)
@ -411,7 +411,7 @@ class TestInjector(unittest.TestCase):
custom_mapping.change(Key((EV_ABS, ABS_Z, 1)), 'b')
system_mapping._set('b', 77)
self.injector = Injector('gamepad', custom_mapping)
self.injector.start_injecting()
self.injector.start()
# wait for the injector to start sending, at most 1s
uinput_write_history_pipe[0].poll(1)
@ -428,10 +428,10 @@ class TestInjector(unittest.TestCase):
custom_mapping.set('gamepad.joystick.right_purpose', NONE)
self.injector = Injector('gamepad', custom_mapping)
# the stop message will be available in the pipe right away,
# so _start_injecting won't block and just stop. all the stuff
# so run won't block and just stop. all the stuff
# will be initialized though, so that stuff can be tested
self.injector.stop_injecting()
self.injector._start_injecting()
self.injector.run()
# not in a process, so the event_producer state can be checked
self.assertEqual(self.injector._event_producer.max_abs, MAX_ABS)
self.assertIsNotNone(self.injector._event_producer.mouse_uinput)
@ -441,7 +441,7 @@ class TestInjector(unittest.TestCase):
custom_mapping.set('gamepad.joystick.right_purpose', BUTTONS)
self.injector = Injector('gamepad', custom_mapping)
self.injector.stop_injecting()
self.injector._start_injecting()
self.injector.run()
self.assertIsNone(self.injector._event_producer.max_abs, MAX_ABS)
self.assertIsNone(self.injector._event_producer.mouse_uinput)
@ -450,7 +450,7 @@ class TestInjector(unittest.TestCase):
custom_mapping.set('gamepad.joystick.right_purpose', WHEEL)
self.injector = Injector('device 1', custom_mapping)
self.injector.stop_injecting()
self.injector._start_injecting()
self.injector.run()
# not a gamepad, so _event_producer is not initialized for that.
# it can still debounce stuff though
self.assertIsNone(self.injector._event_producer.max_abs)
@ -494,7 +494,7 @@ class TestInjector(unittest.TestCase):
self.injector = Injector('device 2', custom_mapping)
self.assertEqual(self.injector.get_state(), UNKNOWN)
self.injector.start_injecting()
self.injector.start()
self.assertEqual(self.injector.get_state(), STARTING)
uinput_write_history_pipe[0].poll(timeout=1)
@ -553,7 +553,7 @@ class TestInjector(unittest.TestCase):
self.assertEqual(history[6], (3124, 3564, 6542))
time.sleep(0.1)
self.assertTrue(self.injector._process.is_alive())
self.assertTrue(self.injector.is_alive())
numlock_after = is_numlock_on()
self.assertEqual(numlock_before, numlock_after)
@ -603,7 +603,7 @@ class TestInjector(unittest.TestCase):
input = InputDevice('/dev/input/event30')
self.injector._grab_device = lambda *args: input
self.injector.start_injecting()
self.injector.start()
uinput_write_history_pipe[0].poll(timeout=1)
time.sleep(EVENT_READ_TIMEOUT * 10)
return read_write_history_pipe()
@ -668,7 +668,7 @@ class TestInjector(unittest.TestCase):
self.assertIn(REL_HWHEEL, device.capabilities()[EV_REL])
self.assertIn(device.path, get_devices()[device_name]['paths'])
self.injector.start_injecting()
self.injector.start()
# wait for the first injected key down event
uinput_write_history_pipe[0].poll(timeout=1)
@ -724,7 +724,7 @@ class TestInjector(unittest.TestCase):
self.injector._modify_capabilities = _modify_capabilities
try:
self.injector._start_injecting()
self.injector.run()
except Stop:
pass

Write Preview
Loading…
Cancel
Save