#!/usr/bin/env python3 # -*- coding: utf-8 -*- # input-remapper - GUI for device specific keyboard mappings # Copyright (C) 2025 sezanzeb # # This file is part of input-remapper. # # input-remapper is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # input-remapper is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with input-remapper. If not, see . import asyncio import unittest from typing import Iterable import evdev from evdev.ecodes import ( EV_KEY, EV_ABS, EV_REL, ABS_X, ABS_Y, REL_X, REL_Y, BTN_A, REL_HWHEEL, REL_WHEEL, REL_WHEEL_HI_RES, REL_HWHEEL_HI_RES, ABS_HAT0X, BTN_LEFT, BTN_B, KEY_A, ABS_HAT0Y, KEY_B, KEY_C, KEY_D, BTN_TL, KEY_1, ) from inputremapper.configs.mapping import ( Mapping, REL_XY_SCALING, WHEEL_SCALING, WHEEL_HI_RES_SCALING, DEFAULT_REL_RATE, ) from inputremapper.configs.preset import Preset from inputremapper.configs.keyboard_layout import keyboard_layout from inputremapper.configs.input_config import InputCombination, InputConfig from inputremapper.injection.context import Context from inputremapper.injection.event_reader import EventReader from inputremapper.injection.global_uinputs import GlobalUInputs, UInput from inputremapper.injection.mapping_handlers.mapping_parser import MappingParser from inputremapper.input_event import InputEvent from tests.lib.cleanup import cleanup from tests.lib.logger import logger from tests.lib.constants import MAX_ABS, MIN_ABS from tests.lib.fixtures import Fixture, fixtures from tests.lib.pipes import uinput_write_history from tests.lib.test_setup import test_setup class EventPipelineTestBase(unittest.IsolatedAsyncioTestCase): """Test the event pipeline form event_reader to UInput.""" def setUp(self): self.global_uinputs = GlobalUInputs(UInput) self.global_uinputs.prepare_all() self.mapping_parser = MappingParser(self.global_uinputs) self.global_uinputs.is_service = True self.global_uinputs.prepare_all() self.forward_uinput = evdev.UInput(name="test-forward-uinput") self.stop_event = asyncio.Event() def tearDown(self) -> None: cleanup() async def asyncTearDown(self) -> None: logger.info("setting stop_event for the reader") self.stop_event.set() await asyncio.sleep(0.5) @staticmethod async def send_events(events: Iterable[InputEvent], event_reader: EventReader): for event in events: logger.info("sending into event_pipeline: %s", event) await event_reader.handle(event) def create_event_reader( self, preset: Preset, source: Fixture, ) -> EventReader: """Create and start an EventReader.""" context = Context( preset, source_devices={}, forward_devices={source.get_device_hash(): self.forward_uinput}, mapping_parser=self.mapping_parser, ) reader = EventReader( context, evdev.InputDevice(source.path), self.stop_event, ) asyncio.ensure_future(reader.run()) return reader @test_setup class TestCombination(EventPipelineTestBase): async def test_any_event_as_button(self): """As long as there is an event handler and a mapping we should be able to map anything to a button""" # value needs to be higher than 10% below center of axis (absinfo) w_down = (EV_ABS, ABS_Y, -12345) w_up = (EV_ABS, ABS_Y, 0) s_down = (EV_ABS, ABS_Y, 12345) s_up = (EV_ABS, ABS_Y, 0) d_down = (EV_REL, REL_X, 100) d_up = (EV_REL, REL_X, 0) a_down = (EV_REL, REL_X, -100) a_up = (EV_REL, REL_X, 0) b_down = (EV_ABS, ABS_HAT0X, 1) b_up = (EV_ABS, ABS_HAT0X, 0) c_down = (EV_ABS, ABS_HAT0X, -1) c_up = (EV_ABS, ABS_HAT0X, 0) # first change the system mapping because Mapping will validate against it keyboard_layout.clear() code_w = 71 code_b = 72 code_c = 73 code_d = 74 code_a = 75 code_s = 76 keyboard_layout._set("w", code_w) keyboard_layout._set("d", code_d) keyboard_layout._set("a", code_a) keyboard_layout._set("s", code_s) keyboard_layout._set("b", code_b) keyboard_layout._set("c", code_c) preset = Preset() preset.add( Mapping.from_combination( InputCombination(InputCombination.from_tuples(b_down)), "keyboard", "b" ) ) preset.add( Mapping.from_combination( InputCombination(InputCombination.from_tuples(c_down)), "keyboard", "c" ) ) preset.add( Mapping.from_combination( InputCombination(InputCombination.from_tuples((*w_down[:2], -10))), "keyboard", "w", ) ) preset.add( Mapping.from_combination( InputCombination(InputCombination.from_tuples((*d_down[:2], 10))), "keyboard", "k(d)", ) ) preset.add( Mapping.from_combination( InputCombination(InputCombination.from_tuples((*s_down[:2], 10))), "keyboard", "s", ) ) preset.add( Mapping.from_combination( InputCombination(InputCombination.from_tuples((*a_down[:2], -10))), "keyboard", "a", ) ) # gamepad fixture event_reader = self.create_event_reader(preset, fixtures.gamepad) await self.send_events( [ InputEvent.from_tuple(b_down), InputEvent.from_tuple(c_down), InputEvent.from_tuple(w_down), InputEvent.from_tuple(d_down), InputEvent.from_tuple(s_down), InputEvent.from_tuple(a_down), InputEvent.from_tuple(b_up), InputEvent.from_tuple(c_up), InputEvent.from_tuple(w_up), InputEvent.from_tuple(d_up), InputEvent.from_tuple(s_up), InputEvent.from_tuple(a_up), ], event_reader, ) # wait a bit for the rel_to_btn handler to send the key up await asyncio.sleep(0.1) history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(history.count((EV_KEY, code_b, 1)), 1) self.assertEqual(history.count((EV_KEY, code_c, 1)), 1) self.assertEqual(history.count((EV_KEY, code_w, 1)), 1) self.assertEqual(history.count((EV_KEY, code_d, 1)), 1) self.assertEqual(history.count((EV_KEY, code_a, 1)), 1) self.assertEqual(history.count((EV_KEY, code_s, 1)), 1) self.assertEqual(history.count((EV_KEY, code_b, 0)), 1) self.assertEqual(history.count((EV_KEY, code_c, 0)), 1) self.assertEqual(history.count((EV_KEY, code_w, 0)), 1) self.assertEqual(history.count((EV_KEY, code_d, 0)), 1) self.assertEqual(history.count((EV_KEY, code_a, 0)), 1) self.assertEqual(history.count((EV_KEY, code_s, 0)), 1) async def test_reset_releases_keys(self): """Make sure that macros and keys are releases when the stop event is set.""" preset = Preset() input_cfg = InputCombination([InputConfig(type=EV_KEY, code=1)]).to_config() preset.add( Mapping.from_combination( input_combination=input_cfg, output_symbol="hold(a)" ) ) input_cfg = InputCombination([InputConfig(type=EV_KEY, code=2)]).to_config() preset.add( Mapping.from_combination(input_combination=input_cfg, output_symbol="b") ) input_cfg = InputCombination([InputConfig(type=EV_KEY, code=3)]).to_config() preset.add( Mapping.from_combination( input_combination=input_cfg, output_symbol="modify(c,hold(d))" ), ) event_reader = self.create_event_reader(preset, fixtures.foo_device_2_keyboard) a = keyboard_layout.get("a") b = keyboard_layout.get("b") c = keyboard_layout.get("c") d = keyboard_layout.get("d") await self.send_events( [ InputEvent.key(1, 1), InputEvent.key(2, 1), InputEvent.key(3, 1), ], event_reader, ) await asyncio.sleep(0.1) forwarded_history = self.forward_uinput.write_history keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(len(forwarded_history), 0) # a down, b down, c down, d down self.assertEqual(len(keyboard_history), 4) event_reader.context.reset() await asyncio.sleep(0.1) forwarded_history = self.forward_uinput.write_history keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(len(forwarded_history), 0) # all a, b, c, d down+up self.assertEqual(len(keyboard_history), 8) keyboard_history = keyboard_history[-4:] self.assertIn((EV_KEY, a, 0), keyboard_history) self.assertIn((EV_KEY, b, 0), keyboard_history) self.assertIn((EV_KEY, c, 0), keyboard_history) self.assertIn((EV_KEY, d, 0), keyboard_history) async def test_forward_abs(self): """Test if EV_ABS events are forwarded when other events of the same input are not.""" preset = Preset() # BTN_A -> 77 keyboard_layout._set("b", 77) preset.add( Mapping.from_combination( InputCombination([InputConfig(type=EV_KEY, code=BTN_A)]), "keyboard", "b", ) ) event_reader = self.create_event_reader(preset, fixtures.gamepad) # should forward them unmodified await self.send_events( [ InputEvent.abs(ABS_X, 10), InputEvent.abs(ABS_Y, 20), InputEvent.abs(ABS_X, -30), InputEvent.abs(ABS_Y, -40), # send them to keyboard 77 InputEvent.key(BTN_A, 1), InputEvent.key(BTN_A, 0), ], event_reader, ) history = self.forward_uinput.write_history keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(history.count((EV_ABS, ABS_X, 10)), 1) self.assertEqual(history.count((EV_ABS, ABS_Y, 20)), 1) self.assertEqual(history.count((EV_ABS, ABS_X, -30)), 1) self.assertEqual(history.count((EV_ABS, ABS_Y, -40)), 1) self.assertEqual(keyboard_history.count((EV_KEY, 77, 1)), 1) self.assertEqual(keyboard_history.count((EV_KEY, 77, 0)), 1) async def test_forward_rel(self): """Test if EV_REL events are forwarded when other events of the same input are not.""" preset = Preset() # BTN_A -> 77 keyboard_layout._set("b", 77) preset.add( Mapping.from_combination( InputCombination([InputConfig(type=EV_KEY, code=BTN_LEFT)]), "keyboard", "b", ) ) event_reader = self.create_event_reader(preset, fixtures.gamepad) # should forward them unmodified await self.send_events( [ InputEvent.rel(REL_X, 10), InputEvent.rel(REL_Y, 20), InputEvent.rel(REL_X, -30), InputEvent.rel(REL_Y, -40), # send them to keyboard 77 InputEvent.key(BTN_LEFT, 1), InputEvent.key(BTN_LEFT, 0), ], event_reader, ) await asyncio.sleep(0.1) history = self.forward_uinput.write_history keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(history.count((EV_REL, REL_X, 10)), 1) self.assertEqual(history.count((EV_REL, REL_Y, 20)), 1) self.assertEqual(history.count((EV_REL, REL_X, -30)), 1) self.assertEqual(history.count((EV_REL, REL_Y, -40)), 1) self.assertEqual(keyboard_history.count((EV_KEY, 77, 1)), 1) self.assertEqual(keyboard_history.count((EV_KEY, 77, 0)), 1) async def test_combination(self): """Test if combinations map to keys properly.""" a = keyboard_layout.get("a") b = keyboard_layout.get("b") c = keyboard_layout.get("c") origin = fixtures.gamepad origin_hash = origin.get_device_hash() mapping_1 = Mapping.from_combination( InputCombination( [ InputConfig( type=EV_ABS, code=ABS_X, analog_threshold=1, origin_hash=origin_hash, ) ] ), output_symbol="a", ) mapping_2 = Mapping.from_combination( InputCombination( [ InputConfig( type=EV_ABS, code=ABS_X, analog_threshold=1, origin_hash=origin_hash, ), InputConfig(type=EV_KEY, code=BTN_A, origin_hash=origin_hash), ] ), output_symbol="b", ) mapping_3 = Mapping.from_combination( InputCombination( [ InputConfig( type=EV_ABS, code=ABS_X, analog_threshold=1, origin_hash=origin_hash, ), InputConfig(type=EV_KEY, code=BTN_A, origin_hash=origin_hash), InputConfig(type=EV_KEY, code=BTN_B, origin_hash=origin_hash), ] ), output_symbol="c", ) assert mapping_1.release_combination_keys assert mapping_2.release_combination_keys assert mapping_3.release_combination_keys preset = Preset() preset.add(mapping_1) preset.add(mapping_2) preset.add(mapping_3) event_reader = self.create_event_reader(preset, origin) # send_events awaits the event_reader to do its thing await self.send_events( [ # forwarded InputEvent.key(BTN_A, 1, origin_hash), # triggers b, releases BTN_A, ABS_X InputEvent.abs(ABS_X, 1234, origin_hash), # triggers c, releases BTN_A, ABS_X, BTN_B InputEvent.key(BTN_B, 1, origin_hash), ], event_reader, ) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history # I don't remember the specifics. I guess if there is a combination ongoing, # it shouldn't trigger ABS_X -> a? self.assertNotIn((EV_KEY, a, 1), keyboard_history) # c and b should have been written, because the input from send_events # should trigger the combination self.assertEqual(keyboard_history.count((EV_KEY, c, 1)), 1) self.assertEqual(keyboard_history.count((EV_KEY, b, 1)), 1) self.assertEqual(forwarded_history.count((EV_KEY, BTN_A, 1)), 1) self.assertIn((EV_KEY, BTN_A, 0), forwarded_history) self.assertNotIn((EV_ABS, ABS_X, 1234), forwarded_history) self.assertNotIn((EV_KEY, BTN_B, 1), forwarded_history) # release b and c) await self.send_events( [InputEvent.abs(ABS_X, 0, origin_hash)], event_reader, ) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertNotIn((EV_KEY, a, 1), keyboard_history) self.assertNotIn((EV_KEY, a, 0), keyboard_history) self.assertEqual(keyboard_history.count((EV_KEY, c, 0)), 1) self.assertEqual(keyboard_history.count((EV_KEY, b, 0)), 1) async def test_combination_manual_release_in_press_order(self): """Test if combinations map to keys properly.""" # release_combination_keys is off # press 5, then 6, then release 5, then release 6 in_1 = keyboard_layout.get("7") in_2 = keyboard_layout.get("8") out = keyboard_layout.get("a") origin = fixtures.foo_device_2_keyboard origin_hash = origin.get_device_hash() input_combination = InputCombination( [ InputConfig( type=EV_KEY, code=in_1, origin_hash=origin_hash, ), InputConfig( type=EV_KEY, code=in_2, origin_hash=origin_hash, ), ] ) mapping = Mapping( input_combination=input_combination.to_config(), target_uinput="keyboard", output_symbol="a", release_combination_keys=False, ) assert not mapping.release_combination_keys preset = Preset() preset.add(mapping) event_reader = self.create_event_reader(preset, origin) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history # press the first key of the combination await self.send_events([InputEvent.key(in_1, 1, origin_hash)], event_reader) self.assertListEqual(forwarded_history, [(EV_KEY, in_1, 1)]) # then the second, it should trigger the combination await self.send_events([InputEvent.key(in_2, 1, origin_hash)], event_reader) self.assertListEqual(forwarded_history, [(EV_KEY, in_1, 1)]) self.assertListEqual(keyboard_history, [(EV_KEY, out, 1)]) # release the first key. A key-down event was injected for it previously, so # now we find a key-up event here as well. await self.send_events([InputEvent.key(in_1, 0, origin_hash)], event_reader) self.assertListEqual(forwarded_history, [(EV_KEY, in_1, 1), (EV_KEY, in_1, 0)]) self.assertListEqual(keyboard_history, [(EV_KEY, out, 1), (EV_KEY, out, 0)]) # release the second key. No key-down event was injected, so we don't have a # key-up event here either. await self.send_events([InputEvent.key(in_2, 0, origin_hash)], event_reader) self.assertListEqual(forwarded_history, [(EV_KEY, in_1, 1), (EV_KEY, in_1, 0)]) self.assertListEqual(keyboard_history, [(EV_KEY, out, 1), (EV_KEY, out, 0)]) async def test_releases_before_triggering(self): origin = fixtures.foo_device_2_keyboard origin_hash = origin.get_device_hash() input_combination = InputCombination( [ InputConfig( type=EV_KEY, code=KEY_A, origin_hash=origin_hash, ), InputConfig( type=EV_KEY, code=KEY_B, origin_hash=origin_hash, ), ] ) mapping = Mapping( input_combination=input_combination.to_config(), target_uinput="keyboard", output_symbol="1", release_combination_keys=True, ) preset = Preset() preset.add(mapping) event_reader = self.create_event_reader(preset, origin) await self.send_events( [ InputEvent.key(KEY_A, 1, origin_hash), InputEvent.key(KEY_B, 1, origin_hash), ], event_reader, ) # Other tests check forwarded_history and keyboard_history individually, # so here is one that looks at the order in uinput_write_history self.assertListEqual( uinput_write_history, [ (EV_KEY, KEY_A, 1), (EV_KEY, KEY_A, 0), (EV_KEY, KEY_1, 1), ], ) async def test_suppressed_doesnt_forward_releases(self): origin = fixtures.foo_device_2_keyboard origin_hash = origin.get_device_hash() input_combination_1 = InputCombination( [ InputConfig( type=EV_KEY, code=KEY_A, origin_hash=origin_hash, ), InputConfig( type=EV_KEY, code=KEY_B, origin_hash=origin_hash, ), InputConfig( type=EV_KEY, code=KEY_C, origin_hash=origin_hash, ), ] ) mapping_1 = Mapping( input_combination=input_combination_1.to_config(), target_uinput="keyboard", output_symbol="1", release_combination_keys=False, ) input_combination_2 = InputCombination( [ InputConfig( type=EV_KEY, code=KEY_A, origin_hash=origin_hash, ), InputConfig( type=EV_KEY, code=KEY_C, origin_hash=origin_hash, ), ] ) mapping_2 = Mapping( input_combination=input_combination_2.to_config(), target_uinput="keyboard", output_symbol="2", release_combination_keys=True, ) preset = Preset() preset.add(mapping_1) preset.add(mapping_2) event_reader = self.create_event_reader(preset, origin) # Trigger mapping_1, mapping_2 should be suppressed await self.send_events( [ InputEvent.key(KEY_A, 1, origin_hash), InputEvent.key(KEY_B, 1, origin_hash), InputEvent.key(KEY_C, 1, origin_hash), ], event_reader, ) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history # There used to be an incorrect KEY_A release, caused by # `release_combination_keys` on the suppressed mapping. self.assertListEqual( forwarded_history, [ (EV_KEY, KEY_A, 1), (EV_KEY, KEY_B, 1), ], ) self.assertListEqual( keyboard_history, [ (EV_KEY, KEY_1, 1), ], ) async def test_no_stuck_key(self): origin = fixtures.foo_device_2_keyboard origin_hash = origin.get_device_hash() input_combination_1 = InputCombination( [ InputConfig( type=EV_KEY, code=KEY_A, origin_hash=origin_hash, ), InputConfig( type=EV_KEY, code=KEY_B, origin_hash=origin_hash, ), InputConfig( type=EV_KEY, code=KEY_D, origin_hash=origin_hash, ), ] ) mapping_1 = Mapping( input_combination=input_combination_1.to_config(), target_uinput="keyboard", output_symbol="1", release_combination_keys=False, ) input_combination_2 = InputCombination( [ InputConfig( type=EV_KEY, code=KEY_C, origin_hash=origin_hash, ), InputConfig( type=EV_KEY, code=KEY_D, origin_hash=origin_hash, ), ] ) mapping_2 = Mapping( input_combination=input_combination_2.to_config(), target_uinput="keyboard", output_symbol="2", release_combination_keys=False, ) preset = Preset() preset.add(mapping_1) preset.add(mapping_2) event_reader = self.create_event_reader(preset, origin) # Trigger mapping_1, mapping_2 should be suppressed await self.send_events( [ InputEvent.key(KEY_A, 1, origin_hash), InputEvent.key(KEY_B, 1, origin_hash), InputEvent.key(KEY_C, 1, origin_hash), InputEvent.key(KEY_D, 1, origin_hash), # Release c -> mapping_2 transitions from suppressed to passive. # The release of c should be forwarded. InputEvent.key(KEY_C, 0, origin_hash), ], event_reader, ) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history self.assertListEqual( forwarded_history, [ (EV_KEY, KEY_A, 1), (EV_KEY, KEY_B, 1), (EV_KEY, KEY_C, 1), (EV_KEY, KEY_C, 0), ], ) self.assertListEqual( keyboard_history, [ (EV_KEY, KEY_1, 1), ], ) async def test_ignore_hold(self): # hold as in event-value 2, not in macro-hold. # linux will generate events with value 2 after input-remapper injected # the key-press, so input-remapper doesn't need to forward them. That # would cause duplicate events of those values otherwise. ev_1 = InputEvent.key(KEY_A, 1) ev_2 = InputEvent.key(KEY_A, 2) ev_3 = InputEvent.key(KEY_A, 0) preset = Preset() preset.add( Mapping.from_combination( input_combination=InputCombination( [InputConfig(type=EV_KEY, code=KEY_A)] ), output_symbol="a", ) ) a = keyboard_layout.get("a") event_reader = self.create_event_reader(preset, fixtures.gamepad) await self.send_events( [ev_1, ev_2, ev_3], event_reader, ) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history self.assertEqual(len(keyboard_history), 2) self.assertEqual(len(forwarded_history), 0) self.assertNotIn((EV_KEY, a, 2), keyboard_history) async def test_ignore_disabled(self): origin = fixtures.gamepad origin_hash = origin.get_device_hash() ev_1 = InputEvent.abs(ABS_HAT0Y, 1, origin_hash) ev_2 = InputEvent.abs(ABS_HAT0Y, 0, origin_hash) ev_3 = InputEvent.abs(ABS_HAT0X, 1, origin_hash) # disabled ev_4 = InputEvent.abs(ABS_HAT0X, 0, origin_hash) ev_5 = InputEvent.key(KEY_A, 1, origin_hash) ev_6 = InputEvent.key(KEY_A, 0, origin_hash) combi_1 = (ev_5, ev_3) combi_2 = (ev_3, ev_5) preset = Preset() preset.add( Mapping.from_combination( input_combination=InputCombination( [ InputConfig.from_input_event(ev_1), ] ), output_symbol="a", ) ) preset.add( Mapping.from_combination( input_combination=InputCombination( [ InputConfig.from_input_event(ev_3), ] ), output_symbol="disable", ) ) preset.add( Mapping.from_combination( input_combination=InputCombination( ( InputConfig.from_input_event(combi_1[0]), InputConfig.from_input_event(combi_1[1]), ) ), output_symbol="b", ) ) preset.add( Mapping.from_combination( input_combination=InputCombination( ( InputConfig.from_input_event(combi_2[0]), InputConfig.from_input_event(combi_2[1]), ) ), output_symbol="c", ) ) a = keyboard_layout.get("a") b = keyboard_layout.get("b") c = keyboard_layout.get("c") event_reader = self.create_event_reader(preset, origin) """Single keys""" await self.send_events( [ ev_1, # press a ev_3, # disabled ev_2, # release a ev_4, # disabled ], event_reader, ) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history self.assertIn((EV_KEY, a, 1), keyboard_history) self.assertIn((EV_KEY, a, 0), keyboard_history) self.assertEqual(len(keyboard_history), 2) self.assertEqual(len(forwarded_history), 0) """A combination that ends in a disabled key""" # ev_5 should be forwarded and the combination triggered await self.send_events(combi_1, event_reader) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history self.assertIn((EV_KEY, b, 1), keyboard_history) self.assertEqual(len(keyboard_history), 3) self.assertEqual(forwarded_history.count(ev_3), 0) self.assertEqual(forwarded_history.count(ev_5), 1) self.assertTrue(forwarded_history.count(ev_6) >= 1) # release what the combination maps to await self.send_events([ev_4, ev_6], event_reader) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history self.assertIn((EV_KEY, b, 0), keyboard_history) self.assertEqual(len(keyboard_history), 4) self.assertEqual(forwarded_history.count(ev_3), 0) self.assertTrue(forwarded_history.count(ev_6) >= 1) """A combination that starts with a disabled key""" # only the combination should get triggered await self.send_events(combi_2, event_reader) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history self.assertIn((EV_KEY, c, 1), keyboard_history) self.assertEqual(len(keyboard_history), 5) self.assertEqual(forwarded_history.count(ev_3), 0) self.assertEqual(forwarded_history.count(ev_5), 1) self.assertTrue(forwarded_history.count(ev_6) >= 1) # release what the combination maps to await self.send_events([ev_4, ev_6], event_reader) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history for event in keyboard_history: print(event.event_tuple) self.assertIn((EV_KEY, c, 0), keyboard_history) self.assertEqual(len(keyboard_history), 6) self.assertEqual(forwarded_history.count(ev_3), 0) self.assertTrue(forwarded_history.count(ev_6) >= 1) async def test_combination_keycode_macro_mix(self): """Ev_1 triggers macro, ev_1 + ev_2 triggers key while the macro is still running""" down_1 = (EV_ABS, ABS_HAT0X, 1) down_2 = (EV_ABS, ABS_HAT0Y, -1) up_1 = (EV_ABS, ABS_HAT0X, 0) up_2 = (EV_ABS, ABS_HAT0Y, 0) a = keyboard_layout.get("a") b = keyboard_layout.get("b") preset = Preset() preset.add( Mapping.from_combination( InputCombination(InputCombination.from_tuples(down_1)), output_symbol="h(k(a))", ) ) preset.add( Mapping.from_combination( InputCombination(InputCombination.from_tuples(down_1, down_2)), output_symbol="b", ) ) event_reader = self.create_event_reader(preset, fixtures.gamepad) # macro starts await self.send_events([InputEvent.from_tuple(down_1)], event_reader) await asyncio.sleep(0.05) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history self.assertEqual(len(forwarded_history), 0) self.assertGreater(len(keyboard_history), 1) self.assertNotIn((EV_KEY, b, 1), keyboard_history) self.assertIn((EV_KEY, a, 1), keyboard_history) self.assertIn((EV_KEY, a, 0), keyboard_history) # combination triggered await self.send_events([InputEvent.from_tuple(down_2)], event_reader) await asyncio.sleep(0) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertIn((EV_KEY, b, 1), keyboard_history) len_a = len(self.global_uinputs.get_uinput("keyboard").write_history) await asyncio.sleep(0.05) len_b = len(self.global_uinputs.get_uinput("keyboard").write_history) # still running self.assertGreater(len_b, len_a) # release await self.send_events([InputEvent.from_tuple(up_1)], event_reader) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(keyboard_history[-1], (EV_KEY, b, 0)) await asyncio.sleep(0.05) len_c = len(self.global_uinputs.get_uinput("keyboard").write_history) await asyncio.sleep(0.05) len_d = len(self.global_uinputs.get_uinput("keyboard").write_history) # not running anymore self.assertEqual(len_c, len_d) await self.send_events([InputEvent.from_tuple(up_2)], event_reader) await asyncio.sleep(0.05) len_e = len(self.global_uinputs.get_uinput("keyboard").write_history) self.assertEqual(len_e, len_d) async def test_wheel_combination_release_failure(self): # test based on a bug that once occurred # 1 | 22.6698, ((1, 276, 1)) -------------- forwarding # 2 | 22.9904, ((1, 276, 1), (2, 8, -1)) -- maps to 30 # 3 | 23.0103, ((1, 276, 1), (2, 8, -1)) -- duplicate key down # 4 | ... 34 more duplicate key downs (scrolling) # 5 | 23.7104, ((1, 276, 1), (2, 8, -1)) -- duplicate key down # 6 | 23.7283, ((1, 276, 0)) -------------- forwarding release # 7 | 23.7303, ((2, 8, -1)) --------------- forwarding # 8 | 23.7865, ((2, 8, 0)) ---------------- not forwarding release # line 7 should have been "duplicate key down" as well # line 8 should have released 30, instead it was never released # # Note: the test was modified for the new Event pipeline: # line 6 now releases the combination # line 7 get forwarded # line 8 get forwarded scroll = InputEvent.from_tuple((2, 8, -1)) scroll_release = InputEvent.from_tuple((2, 8, 0)) btn_down = InputEvent.key(276, 1) btn_up = InputEvent.key(276, 0) combination = InputCombination( InputCombination.from_tuples((1, 276, 1), (2, 8, -1)) ) keyboard_layout.clear() keyboard_layout._set("a", 30) a = 30 m = Mapping.from_combination(combination, output_symbol="a") m.release_timeout = 0.1 # a higher release timeout to give time for assertions preset = Preset() preset.add(m) event_reader = self.create_event_reader(preset, fixtures.foo_device_2_mouse) await self.send_events([btn_down], event_reader) forwarded_history = self.forward_uinput.write_history self.assertEqual(forwarded_history[0], btn_down) await self.send_events([scroll], event_reader) # "maps to 30" keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(keyboard_history[0], (EV_KEY, a, 1)) await self.send_events([scroll] * 5, event_reader) # nothing new since all of them were duplicate key downs keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(len(keyboard_history), 1) await self.send_events([btn_up], event_reader) # releasing the combination keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(keyboard_history[1], (EV_KEY, a, 0)) # more scroll events # it should be ignored as duplicate key-down await self.send_events([scroll] * 5, event_reader) forwarded_history = self.forward_uinput.write_history self.assertEqual(forwarded_history.count(scroll), 5) await self.send_events([scroll_release], event_reader) forwarded_history = self.forward_uinput.write_history self.assertEqual(forwarded_history[-1], scroll_release) async def test_can_not_map(self): """Inject events to wrong or invalid uinput.""" ev_1 = InputEvent.key(KEY_A, 1) ev_2 = InputEvent.key(KEY_B, 1) ev_3 = InputEvent.key(KEY_C, 1) ev_4 = InputEvent.key(KEY_A, 0) ev_5 = InputEvent.key(KEY_B, 0) ev_6 = InputEvent.key(KEY_C, 0) mapping_1 = Mapping( input_combination=InputCombination([InputConfig.from_input_event(ev_2)]), target_uinput="keyboard", output_type=EV_KEY, output_code=BTN_TL, ) mapping_2 = Mapping( input_combination=InputCombination([InputConfig.from_input_event(ev_3)]), target_uinput="keyboard", output_type=EV_KEY, output_code=KEY_A, ) preset = Preset() preset.add(mapping_1) preset.add(mapping_2) event_reader = self.create_event_reader(preset, fixtures.foo_device_2_mouse) # send key-down and up await self.send_events( [ ev_1, ev_2, ev_3, ev_4, ev_5, ev_6, ], event_reader, ) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history self.assertEqual(len(forwarded_history), 4) self.assertEqual(len(keyboard_history), 2) self.assertIn(ev_1, forwarded_history) self.assertIn(ev_2, forwarded_history) self.assertIn(ev_4, forwarded_history) self.assertIn(ev_5, forwarded_history) self.assertNotIn(ev_3, forwarded_history) self.assertNotIn(ev_6, forwarded_history) self.assertIn((EV_KEY, KEY_A, 1), keyboard_history) self.assertIn((EV_KEY, KEY_A, 0), keyboard_history) async def test_axis_switch(self): """Test a mapping for an axis that can be switched on or off.""" rel_rate = 60 # rate [Hz] at which events are produced gain = 0.5 # halve the speed of the rel axis preset = Preset() mouse = self.global_uinputs.get_uinput("mouse") forward_history = self.forward_uinput.write_history mouse_history = mouse.write_history # ABS_X to REL_Y if ABS_Y is above 10% combination = InputCombination( InputCombination.from_tuples((EV_ABS, ABS_X, 0), (EV_ABS, ABS_Y, 10)) ) cfg = { "input_combination": combination.to_config(), "target_uinput": "mouse", "output_type": EV_REL, "output_code": REL_X, "rel_rate": rel_rate, "gain": gain, "deadzone": 0, } m1 = Mapping(**cfg) preset.add(m1) event_reader = self.create_event_reader(preset, fixtures.gamepad) # set ABS_X input to 100% await event_reader.handle(InputEvent.abs(ABS_X, MAX_ABS)) # wait a bit more for nothing to sum up, because ABS_Y is still 0 await asyncio.sleep(0.2) self.assertEqual(len(mouse_history), 0) self.assertEqual(len(forward_history), 1) self.assertEqual( InputEvent.from_event(forward_history[0]), (EV_ABS, ABS_X, MAX_ABS), ) # move ABS_Y above 10% await self.send_events( ( InputEvent.abs(ABS_Y, int(MAX_ABS * 0.05)), InputEvent.abs(ABS_Y, int(MAX_ABS * 0.11)), InputEvent.abs(ABS_Y, int(MAX_ABS * 0.5)), ), event_reader, ) # wait a bit more for it to sum up sleep = 0.5 await asyncio.sleep(sleep) self.assertAlmostEqual(len(mouse_history), rel_rate * sleep, delta=3) self.assertEqual(len(forward_history), 1) # send some more x events await self.send_events( ( InputEvent.abs(ABS_X, MAX_ABS), InputEvent.abs(ABS_X, int(MAX_ABS * 0.9)), ), event_reader, ) # stop it await event_reader.handle(InputEvent.abs(ABS_Y, int(MAX_ABS * 0.05))) await asyncio.sleep(0.2) # wait a bit more for nothing to sum up if mouse_history[0].type == EV_ABS: raise AssertionError( "The injector probably just forwarded them unchanged" # possibly in addition to writing mouse events ) self.assertAlmostEqual(len(mouse_history), rel_rate * sleep, delta=3) # does not contain anything else expected_rel_event = (EV_REL, REL_X, int(gain * REL_XY_SCALING)) count_x = mouse_history.count(expected_rel_event) self.assertEqual(len(mouse_history), count_x) async def test_key_axis_combination_to_disable(self): combination = InputCombination( [ InputConfig(type=EV_ABS, code=ABS_X), InputConfig(type=EV_ABS, code=ABS_Y, analog_threshold=5), ] ) preset = Preset() forward_history = self.forward_uinput.write_history mapping = Mapping( input_combination=combination, output_symbol="disable", target_uinput="keyboard", ) preset.add(mapping) event_reader = self.create_event_reader(preset, fixtures.gamepad) await self.send_events( [ InputEvent.from_tuple((EV_ABS, ABS_X, 10)), # forwarded InputEvent.from_tuple((EV_ABS, ABS_Y, int(0.1 * MAX_ABS))), InputEvent.from_tuple((EV_ABS, ABS_X, 20)), # disabled InputEvent.from_tuple((EV_ABS, ABS_Y, int(0.02 * MAX_ABS))), InputEvent.from_tuple((EV_ABS, ABS_X, 30)), # forwarded ], event_reader, ) self.assertEqual( forward_history, [ InputEvent.from_tuple((EV_ABS, ABS_X, 10)), InputEvent.from_tuple((EV_ABS, ABS_X, 30)), ], ) async def test_multiple_axis_in_hierarchy_handler(self): preset = Preset() # Add two mappings that map EV_REL to EV_ABS. We want to test that they don't # suppress each other when they are part of a hierarchy handler. So having at # least two of them is important for this test. cutoff = 2 for in_, out in [(REL_X, ABS_X), (REL_Y, ABS_Y)]: input_combination = InputCombination( [ InputConfig(type=EV_KEY, code=KEY_A), InputConfig(type=EV_REL, code=in_), ] ) mapping = Mapping( input_combination=input_combination.to_config(), target_uinput="gamepad", output_type=EV_ABS, output_code=out, gain=0.5, rel_to_abs_input_cutoff=cutoff, release_timeout=0.1, deadzone=0, ) preset.add(mapping) # Add a non-analog mapping, to make sure a HierarchyHandler exists key_input = InputCombination( ( InputConfig(type=EV_KEY, code=KEY_A), InputConfig(type=EV_KEY, code=KEY_B), ) ) key_mapping = Mapping( input_combination=key_input.to_config(), target_uinput="keyboard", output_type=EV_KEY, output_code=KEY_C, ) preset.add(key_mapping) event_reader = self.create_event_reader(preset, fixtures.gamepad) # Trigger all of them at the same time value = int(REL_XY_SCALING * cutoff) await asyncio.sleep(0.1) await self.send_events( [ InputEvent(0, 0, EV_KEY, KEY_A, 1), InputEvent(0, 0, EV_REL, REL_X, value), InputEvent(0, 0, EV_REL, REL_Y, value), InputEvent(0, 0, EV_KEY, KEY_B, 1), InputEvent(0, 0, EV_KEY, KEY_B, 0), ], event_reader, ) await asyncio.sleep(0.4) # We expect all of it to be present. No mapping was suppressed. # So we can trigger combinations that inject keys while a joystick is being # simulated in multiple directions. self.assertEqual( uinput_write_history, [ InputEvent.from_tuple((EV_ABS, ABS_X, MAX_ABS / 2)), InputEvent.from_tuple((EV_ABS, ABS_Y, MAX_ABS / 2)), InputEvent.from_tuple((EV_KEY, KEY_C, 1)), InputEvent.from_tuple((EV_KEY, KEY_C, 0)), InputEvent.from_tuple((EV_ABS, ABS_X, 0)), InputEvent.from_tuple((EV_ABS, ABS_Y, 0)), ], ) @test_setup class TestAbsToAbs(EventPipelineTestBase): async def test_abs_to_abs(self): gain = 0.5 # left x to mouse x input_config = InputConfig(type=EV_ABS, code=ABS_X) mapping_config = { "input_combination": InputCombination([input_config]).to_config(), "target_uinput": "gamepad", "output_type": EV_ABS, "output_code": ABS_X, "gain": gain, "deadzone": 0, } mapping_1 = Mapping(**mapping_config) preset = Preset() preset.add(mapping_1) input_config = InputConfig(type=EV_ABS, code=ABS_Y) mapping_config["input_combination"] = InputCombination( [input_config] ).to_config() mapping_config["output_code"] = ABS_Y mapping_2 = Mapping(**mapping_config) preset.add(mapping_2) x = MAX_ABS y = MAX_ABS event_reader = self.create_event_reader(preset, fixtures.gamepad) await self.send_events( [ InputEvent.abs(ABS_X, -x), InputEvent.abs(ABS_Y, y), ], event_reader, ) await asyncio.sleep(0.2) history = self.global_uinputs.get_uinput("gamepad").write_history self.assertEqual( history, [ InputEvent.from_tuple((3, 0, MIN_ABS / 2)), InputEvent.from_tuple((3, 1, MAX_ABS / 2)), ], ) async def test_abs_to_abs_with_input_switch(self): gain = 0.5 input_combination = InputCombination( ( InputConfig(type=EV_ABS, code=0), InputConfig(type=EV_ABS, code=1, analog_threshold=10), ) ) # left x to mouse x mapping_config = { "input_combination": input_combination.to_config(), "target_uinput": "gamepad", "output_type": EV_ABS, "output_code": ABS_X, "gain": gain, "deadzone": 0, } mapping_1 = Mapping(**mapping_config) preset = Preset() preset.add(mapping_1) x = MAX_ABS y = MAX_ABS event_reader = self.create_event_reader(preset, fixtures.gamepad) await self.send_events( [ InputEvent.abs(ABS_X, -x // 5), # will not map InputEvent.abs(ABS_X, -x), # will map later # switch axis on sends initial position (previous event) InputEvent.abs(ABS_Y, y), InputEvent.abs(ABS_X, x), # normally mapped InputEvent.abs(ABS_Y, y // 15), # off, re-centers axis InputEvent.abs(ABS_X, -x // 5), # will not map ], event_reader, ) await asyncio.sleep(0.2) history = self.global_uinputs.get_uinput("gamepad").write_history self.assertEqual( history, [ InputEvent.from_tuple((3, 0, MIN_ABS / 2)), InputEvent.from_tuple((3, 0, MAX_ABS / 2)), InputEvent.from_tuple((3, 0, 0)), ], ) @test_setup class TestRelToAbs(EventPipelineTestBase): def setUp(self): self.timestamp = 0 super().setUp() def next_usec_time(self): self.timestamp += 1000000 / DEFAULT_REL_RATE return self.timestamp async def test_rel_to_abs(self): # first mapping # left mouse x to abs x gain = 0.5 cutoff = 2 input_combination = InputCombination([InputConfig(type=EV_REL, code=REL_X)]) mapping_config = { "input_combination": input_combination.to_config(), "target_uinput": "gamepad", "output_type": EV_ABS, "output_code": ABS_X, "gain": gain, "rel_to_abs_input_cutoff": cutoff, "release_timeout": 0.5, "deadzone": 0, } mapping_1 = Mapping(**mapping_config) preset = Preset() preset.add(mapping_1) # second mapping input_combination = InputCombination([InputConfig(type=EV_REL, code=REL_Y)]) mapping_config["input_combination"] = input_combination.to_config() mapping_config["output_code"] = ABS_Y mapping_2 = Mapping(**mapping_config) preset.add(mapping_2) event_reader = self.create_event_reader(preset, fixtures.gamepad) next_time = self.next_usec_time() await self.send_events( [ InputEvent(0, next_time, EV_REL, REL_X, -int(REL_XY_SCALING * cutoff)), InputEvent(0, next_time, EV_REL, REL_Y, int(REL_XY_SCALING * cutoff)), ], event_reader, ) await asyncio.sleep(0.1) history = self.global_uinputs.get_uinput("gamepad").write_history self.assertEqual( history, [ InputEvent.from_tuple((3, 0, MIN_ABS / 2)), InputEvent.from_tuple((3, 1, MAX_ABS / 2)), ], ) # send more events, then wait until the release timeout next_time = self.next_usec_time() await self.send_events( [ InputEvent(0, next_time, EV_REL, REL_X, -int(REL_XY_SCALING)), InputEvent(0, next_time, EV_REL, REL_Y, int(REL_XY_SCALING)), ], event_reader, ) await asyncio.sleep(0.7) history = self.global_uinputs.get_uinput("gamepad").write_history self.assertEqual( history, [ InputEvent.from_tuple((3, 0, MIN_ABS / 2)), InputEvent.from_tuple((3, 1, MAX_ABS / 2)), InputEvent.from_tuple((3, 0, MIN_ABS / 4)), InputEvent.from_tuple((3, 1, MAX_ABS / 4)), InputEvent.from_tuple((3, 0, 0)), InputEvent.from_tuple((3, 1, 0)), ], ) async def test_rel_to_abs_reset_multiple(self): # Recenters correctly when triggering the mapping a second time. # Could only be reproduced if a key input is part of the combination, that is # released and pressed again. # left mouse x to abs x gain = 0.5 cutoff = 2 input_combination = InputCombination( [ InputConfig(type=EV_KEY, code=KEY_A), InputConfig(type=EV_REL, code=REL_X), ] ) mapping_config = { "input_combination": input_combination.to_config(), "target_uinput": "gamepad", "output_type": EV_ABS, "output_code": ABS_X, "gain": gain, "rel_to_abs_input_cutoff": cutoff, "release_timeout": 0.1, "deadzone": 0, } mapping_1 = Mapping(**mapping_config) preset = Preset() preset.add(mapping_1) event_reader = self.create_event_reader(preset, fixtures.gamepad) for _ in range(3): next_time = self.next_usec_time() value = int(REL_XY_SCALING * cutoff) await event_reader.handle(InputEvent(0, next_time, EV_KEY, KEY_A, 1)) await event_reader.handle(InputEvent(0, next_time, EV_REL, REL_X, value)) await asyncio.sleep(0.2) history = self.global_uinputs.get_uinput("gamepad").write_history self.assertIn( InputEvent.from_tuple((3, 0, 0)), history, ) await event_reader.handle(InputEvent(0, next_time, EV_KEY, KEY_A, 0)) await asyncio.sleep(0.05) self.global_uinputs.get_uinput("gamepad").write_history = [] async def test_rel_to_abs_with_input_switch(self): # use 0 everywhere, because that will cause the handler to not update the rate, # and we are able to test things without worrying about that at all timestamp = 0 gain = 0.5 cutoff = 1 input_combination = InputCombination( ( InputConfig(type=EV_REL, code=REL_X), InputConfig(type=EV_REL, code=REL_Y, analog_threshold=10), ) ) # left mouse x to x mapping_config = { "input_combination": input_combination.to_config(), "target_uinput": "gamepad", "output_type": EV_ABS, "output_code": ABS_X, "gain": gain, "rel_to_abs_input_cutoff": cutoff, "deadzone": 0, } mapping_1 = Mapping(**mapping_config) preset = Preset() preset.add(mapping_1) event_reader = self.create_event_reader(preset, fixtures.gamepad) # if the cutoff is higher, the test sends higher values to overcome the cutoff await self.send_events( [ # will not map InputEvent(0, timestamp, EV_REL, REL_X, -REL_XY_SCALING / 4 * cutoff), # switch axis on InputEvent(0, timestamp, EV_REL, REL_Y, REL_XY_SCALING / 5 * cutoff), # normally mapped InputEvent(0, timestamp, EV_REL, REL_X, REL_XY_SCALING * cutoff), # off, re-centers axis InputEvent(0, timestamp, EV_REL, REL_Y, REL_XY_SCALING / 20 * cutoff), # will not map InputEvent(0, timestamp, EV_REL, REL_X, REL_XY_SCALING / 2 * cutoff), ], event_reader, ) await asyncio.sleep(0.2) history = self.global_uinputs.get_uinput("gamepad").write_history self.assertEqual( history, [ InputEvent.from_tuple((3, 0, MAX_ABS / 2)), InputEvent.from_tuple((3, 0, 0)), ], ) @test_setup class TestAbsToRel(EventPipelineTestBase): async def test_abs_to_rel(self): """Map gamepad EV_ABS events to EV_REL events.""" rel_rate = 60 # rate [Hz] at which events are produced gain = 0.5 # halve the speed of the rel axis # left x to mouse x input_config = InputConfig(type=EV_ABS, code=ABS_X) mapping_config = { "input_combination": InputCombination([input_config]).to_config(), "target_uinput": "mouse", "output_type": EV_REL, "output_code": REL_X, "rel_rate": rel_rate, "gain": gain, "deadzone": 0, } mapping_1 = Mapping(**mapping_config) preset = Preset() preset.add(mapping_1) # left y to mouse y input_config = InputConfig(type=EV_ABS, code=ABS_Y) mapping_config["input_combination"] = InputCombination( [input_config] ).to_config() mapping_config["output_code"] = REL_Y mapping_2 = Mapping(**mapping_config) preset.add(mapping_2) # set input axis to 100% in order to move # (gain * REL_XY_SCALING) pixel per event x = MAX_ABS y = MAX_ABS event_reader = self.create_event_reader(preset, fixtures.gamepad) await self.send_events( [ InputEvent.abs(ABS_X, -x), InputEvent.abs(ABS_Y, -y), ], event_reader, ) # wait a bit more for it to sum up sleep = 0.5 await asyncio.sleep(sleep) # stop it await self.send_events( [ InputEvent.abs(ABS_X, 0), InputEvent.abs(ABS_Y, 0), ], event_reader, ) mouse_history = self.global_uinputs.get_uinput("mouse").write_history if mouse_history[0].type == EV_ABS: raise AssertionError( "The injector probably just forwarded them unchanged" # possibly in addition to writing mouse events ) self.assertAlmostEqual(len(mouse_history), rel_rate * sleep * 2, delta=5) # those may be in arbitrary order expected_value = -gain * REL_XY_SCALING * (rel_rate / DEFAULT_REL_RATE) count_x = mouse_history.count((EV_REL, REL_X, expected_value)) count_y = mouse_history.count((EV_REL, REL_Y, expected_value)) self.assertGreater(count_x, 1) self.assertGreater(count_y, 1) # only those two types of events were written self.assertEqual(len(mouse_history), count_x + count_y) async def test_abs_to_wheel_hi_res_quirk(self): """When mapping to wheel events we always expect to see both, REL_WHEEL and REL_WHEEL_HI_RES events with an accumulative value ratio of 1/120 """ rel_rate = 60 # rate [Hz] at which events are produced gain = 1 # left x to mouse x input_config = InputConfig(type=EV_ABS, code=ABS_X) mapping_config = { "input_combination": InputCombination([input_config]).to_config(), "target_uinput": "mouse", "output_type": EV_REL, "output_code": REL_WHEEL, "rel_rate": rel_rate, "gain": gain, "deadzone": 0, } mapping_1 = Mapping(**mapping_config) preset = Preset() preset.add(mapping_1) # left y to mouse y input_config = InputConfig(type=EV_ABS, code=ABS_Y) mapping_config["input_combination"] = InputCombination( [input_config] ).to_config() mapping_config["output_code"] = REL_HWHEEL_HI_RES mapping_2 = Mapping(**mapping_config) preset.add(mapping_2) # set input axis to 100% in order to move # speed*gain*rate=1*0.5*60 pixel per second x = MAX_ABS y = MAX_ABS event_reader = self.create_event_reader(preset, fixtures.gamepad) await self.send_events( [ InputEvent.abs(ABS_X, x), InputEvent.abs(ABS_Y, -y), ], event_reader, ) # wait a bit more for it to sum up sleep = 0.8 await asyncio.sleep(sleep) # stop it await self.send_events( [ InputEvent.abs(ABS_X, 0), InputEvent.abs(ABS_Y, 0), ], event_reader, ) m_history = self.global_uinputs.get_uinput("mouse").write_history rel_wheel = sum([event.value for event in m_history if event.code == REL_WHEEL]) rel_wheel_hi_res = sum( [event.value for event in m_history if event.code == REL_WHEEL_HI_RES] ) rel_hwheel = sum( [event.value for event in m_history if event.code == REL_HWHEEL] ) rel_hwheel_hi_res = sum( [event.value for event in m_history if event.code == REL_HWHEEL_HI_RES] ) self.assertAlmostEqual(rel_wheel, rel_wheel_hi_res / 120, places=0) self.assertAlmostEqual(rel_hwheel, rel_hwheel_hi_res / 120, places=0) @test_setup class TestRelToBtn(EventPipelineTestBase): async def test_rel_to_btn(self): """Rel axis mapped to buttons are automatically released if no new rel event arrives.""" # map those two to stuff w_up = (EV_REL, REL_WHEEL, -1) hw_right = (EV_REL, REL_HWHEEL, 1) # should be forwarded and present in the capabilities hw_left = (EV_REL, REL_HWHEEL, -1) keyboard_layout.clear() code_b = 91 code_c = 92 keyboard_layout._set("b", code_b) keyboard_layout._set("c", code_c) # set a high release timeout to make sure the tests pass release_timeout = 0.2 mapping_1 = Mapping.from_combination( InputCombination(InputCombination.from_tuples(hw_right)), "keyboard", "k(b)" ) mapping_2 = Mapping.from_combination( InputCombination(InputCombination.from_tuples(w_up)), "keyboard", "c" ) mapping_1.release_timeout = release_timeout mapping_2.release_timeout = release_timeout preset = Preset() preset.add(mapping_1) preset.add(mapping_2) event_reader = self.create_event_reader(preset, fixtures.foo_device_2_mouse) await self.send_events( [InputEvent.from_tuple(hw_right), InputEvent.from_tuple(w_up)] * 5, event_reader, ) # wait less than the release timeout and send more events await asyncio.sleep(release_timeout / 5) await self.send_events( [InputEvent.from_tuple(hw_right), InputEvent.from_tuple(w_up)] * 5 + [InputEvent.from_tuple(hw_left)] * 3, # one event will release hw_right, the others are forwarded event_reader, ) # wait more than the release_timeout to make sure all handlers finish await asyncio.sleep(release_timeout * 1.2) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history self.assertEqual(keyboard_history.count((EV_KEY, code_b, 1)), 1) self.assertEqual(keyboard_history.count((EV_KEY, code_c, 1)), 1) self.assertEqual(keyboard_history.count((EV_KEY, code_b, 0)), 1) self.assertEqual(keyboard_history.count((EV_KEY, code_c, 0)), 1) # the unmapped wheel direction self.assertEqual(forwarded_history.count(hw_left), 2) # the unmapped wheel won't get a debounced release command, it's # forwarded as is self.assertNotIn((EV_REL, REL_HWHEEL, 0), forwarded_history) async def test_rel_trigger_threshold(self): """Test that different activation points for rel_to_btn work correctly.""" # at 5 map to a mapping_1 = Mapping.from_combination( InputCombination( [InputConfig(type=EV_REL, code=REL_X, analog_threshold=5)] ), output_symbol="a", ) # at 15 map to b mapping_2 = Mapping.from_combination( InputCombination( [InputConfig(type=EV_REL, code=REL_X, analog_threshold=15)] ), output_symbol="b", ) release_timeout = 0.2 # give some time to do assertions before the release mapping_1.release_timeout = release_timeout mapping_2.release_timeout = release_timeout preset = Preset() preset.add(mapping_1) preset.add(mapping_2) a = keyboard_layout.get("a") b = keyboard_layout.get("b") event_reader = self.create_event_reader(preset, fixtures.foo_device_2_mouse) await self.send_events( [ InputEvent.rel(REL_X, -5), # forward InputEvent.rel(REL_X, 0), # forward InputEvent.rel(REL_X, 3), # forward InputEvent.rel(REL_X, 10), # trigger a ], event_reader, ) await asyncio.sleep(release_timeout * 1.5) # release a keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(keyboard_history, [(EV_KEY, a, 1), (EV_KEY, a, 0)]) self.assertEqual(keyboard_history.count((EV_KEY, a, 1)), 1) self.assertEqual(keyboard_history.count((EV_KEY, a, 0)), 1) self.assertNotIn((EV_KEY, b, 1), keyboard_history) await self.send_events( [ InputEvent.rel(REL_X, 10), # trigger a InputEvent.rel(REL_X, 20), # trigger b InputEvent.rel(REL_X, 10), # release b ], event_reader, ) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(keyboard_history.count((EV_KEY, a, 1)), 2) self.assertEqual(keyboard_history.count((EV_KEY, b, 1)), 1) self.assertEqual(keyboard_history.count((EV_KEY, b, 0)), 1) self.assertEqual(keyboard_history.count((EV_KEY, a, 0)), 1) await asyncio.sleep(release_timeout * 1.5) # release a keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history self.assertEqual(keyboard_history.count((EV_KEY, a, 0)), 2) self.assertEqual( forwarded_history, [(EV_REL, REL_X, -5), (EV_REL, REL_X, 0), (EV_REL, REL_X, 3)], ) @test_setup class TestAbsToBtn(EventPipelineTestBase): async def test_abs_trigger_threshold(self): """Test that different activation points for abs_to_btn work correctly.""" # at 30% map to a mapping_1 = Mapping.from_combination( InputCombination( [InputConfig(type=EV_ABS, code=ABS_X, analog_threshold=30)] ), output_symbol="a", ) # at 70% map to b mapping_2 = Mapping.from_combination( InputCombination( [InputConfig(type=EV_ABS, code=ABS_X, analog_threshold=70)] ), output_symbol="b", ) preset = Preset() preset.add(mapping_1) preset.add(mapping_2) a = keyboard_layout.get("a") b = keyboard_layout.get("b") event_reader = self.create_event_reader(preset, fixtures.gamepad) await self.send_events( [ # -10%, do nothing InputEvent.abs(ABS_X, MIN_ABS // 10), # 0%, do noting InputEvent.abs(ABS_X, 0), # 10%, do nothing InputEvent.abs(ABS_X, MAX_ABS // 10), # 50%, trigger a InputEvent.abs(ABS_X, MAX_ABS // 2), ], event_reader, ) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(keyboard_history.count((EV_KEY, a, 1)), 1) self.assertNotIn((EV_KEY, a, 0), keyboard_history) self.assertNotIn((EV_KEY, b, 1), keyboard_history) await self.send_events( [ # 80%, trigger b InputEvent.abs(ABS_X, int(MAX_ABS * 0.8)), InputEvent.abs(ABS_X, MAX_ABS // 2), # 50%, release b ], event_reader, ) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history self.assertEqual(keyboard_history.count((EV_KEY, a, 1)), 1) self.assertEqual(keyboard_history.count((EV_KEY, b, 1)), 1) self.assertEqual(keyboard_history.count((EV_KEY, b, 0)), 1) self.assertNotIn((EV_KEY, a, 0), keyboard_history) # 0% release a await event_reader.handle(InputEvent.abs(ABS_X, 0)) keyboard_history = self.global_uinputs.get_uinput("keyboard").write_history forwarded_history = self.forward_uinput.write_history self.assertEqual(keyboard_history.count((EV_KEY, a, 0)), 1) self.assertEqual(len(forwarded_history), 0) @test_setup class TestRelToRel(EventPipelineTestBase): async def _test(self, input_code, input_value, output_code, output_value, gain=1): preset = Preset() input_config = InputConfig(type=EV_REL, code=input_code) mapping = Mapping( input_combination=InputCombination([input_config]).to_config(), target_uinput="mouse", output_type=EV_REL, output_code=output_code, deadzone=0, gain=gain, ) preset.add(mapping) event_reader = self.create_event_reader(preset, fixtures.gamepad) await self.send_events( [InputEvent(0, 0, EV_REL, input_code, input_value)], event_reader, ) history = self.global_uinputs.get_uinput("mouse").write_history self.assertEqual(len(history), 1) self.assertEqual( history[0], InputEvent(0, 0, EV_REL, output_code, output_value), ) async def test_wheel_to_y(self): await self._test( input_code=REL_WHEEL, input_value=2 * WHEEL_SCALING, output_code=REL_Y, output_value=2 * REL_XY_SCALING, ) async def test_hi_res_wheel_to_y(self): await self._test( input_code=REL_WHEEL_HI_RES, input_value=3 * WHEEL_HI_RES_SCALING, output_code=REL_Y, output_value=3 * REL_XY_SCALING, ) async def test_x_to_hwheel(self): # injects both hi_res and regular wheel events at the same time input_code = REL_X input_value = 100 output_code = REL_HWHEEL gain = 2 output_value = int(input_value / REL_XY_SCALING * WHEEL_SCALING * gain) output_value_hi_res = int( input_value / REL_XY_SCALING * WHEEL_HI_RES_SCALING * gain ) preset = Preset() input_config = InputConfig(type=EV_REL, code=input_code) mapping = Mapping( input_combination=InputCombination([input_config]).to_config(), target_uinput="mouse", output_type=EV_REL, output_code=output_code, deadzone=0, gain=gain, ) preset.add(mapping) event_reader = self.create_event_reader(preset, fixtures.gamepad) await self.send_events( [InputEvent(0, 0, EV_REL, input_code, input_value)], event_reader, ) history = self.global_uinputs.get_uinput("mouse").write_history # injects both REL_WHEEL and REL_WHEEL_HI_RES events self.assertEqual(len(history), 2) self.assertEqual( history[0], InputEvent( 0, 0, EV_REL, REL_HWHEEL, output_value, ), ) self.assertEqual( history[1], InputEvent( 0, 0, EV_REL, REL_HWHEEL_HI_RES, output_value_hi_res, ), ) async def test_remainder(self): preset = Preset() history = self.global_uinputs.get_uinput("mouse").write_history # REL_WHEEL_HI_RES to REL_Y input_config = InputConfig(type=EV_REL, code=REL_WHEEL_HI_RES) gain = 0.01 mapping = Mapping( input_combination=InputCombination([input_config]).to_config(), target_uinput="mouse", output_type=EV_REL, output_code=REL_Y, deadzone=0, gain=gain, ) preset.add(mapping) event_reader = self.create_event_reader(preset, fixtures.gamepad) events_until_one_rel_y_written = int( WHEEL_HI_RES_SCALING / REL_XY_SCALING / gain ) # due to the low gain and low input value, it needs to be sent many times # until one REL_Y event is written await self.send_events( [InputEvent(0, 0, EV_REL, REL_WHEEL_HI_RES, 1)] * (events_until_one_rel_y_written - 1), event_reader, ) self.assertEqual(len(history), 0) # write the final event that causes the input to accumulate to 1 # plus one extra event because of floating-point math await self.send_events( [InputEvent(0, 0, EV_REL, REL_WHEEL_HI_RES, 1)], event_reader, ) self.assertEqual(len(history), 1) self.assertEqual( history[0], InputEvent(0, 0, EV_REL, REL_Y, 1), ) # repeat it one more time to see if the remainder is reset correctly await self.send_events( [InputEvent(0, 0, EV_REL, REL_WHEEL_HI_RES, 1)] * (events_until_one_rel_y_written - 1), event_reader, ) self.assertEqual(len(history), 1) # the event that causes the second REL_Y to be written # this should never need the one extra if the remainder is reset correctly await self.send_events( [InputEvent(0, 0, EV_REL, REL_WHEEL_HI_RES, 1)], event_reader, ) self.assertEqual(len(history), 2) self.assertEqual( history[1], InputEvent(0, 0, EV_REL, REL_Y, 1), ) if __name__ == "__main__": unittest.main()