Fixing release events when keys are released in order of pressing

inputremapper/injection/mapping_handlers/combination_handler.py

@@ -18,7 +18,7 @@
 # along with input-remapper. If not, see <https://www.gnu.org/licenses/>.
 
 from __future__ import annotations # needed for the TYPE_CHECKING import
-from typing import TYPE_CHECKING, Dict, Hashable
+from typing import TYPE_CHECKING, Dict, Hashable, Tuple
 
 import evdev
 from evdev.ecodes import EV_ABS, EV_REL
@@ -42,9 +42,12 @@ class CombinationHandler(MappingHandler):
 
  # map of InputEvent.input_match_hash -> bool , keep track of the combination state
  _pressed_keys: Dict[Hashable, bool]
- _output_state: bool # the last update we sent to a sub-handler
+ # the last update we sent to a sub-handler. If this is true, the output key is
+ # still being held down.
+ _output_active: bool
  _sub_handler: InputEventHandler
  _handled_input_hashes: list[Hashable]
+ _requires_a_release: Dict[Tuple[int, int], bool]
 
  def __init__(
  self,
@@ -56,8 +59,9 @@ def __init__(
  logger.debug(str(mapping))
  super().__init__(combination, mapping)
  self._pressed_keys = {}
- self._output_state = False
+ self._output_active = False
  self._context = context
+ self._requires_a_release = {}
 
  # prepare a key map for all events with non-zero value
  for input_config in combination:
@@ -98,8 +102,6 @@ def notify(
  # we are not responsible for the event
  return False
 
- was_activated = self.is_activated()
-
  # update the state
  # The value of non-key input should have been changed to either 0 or 1 at this
  # point by other handlers.
@@ -108,22 +110,28 @@ def notify(
  # maybe this changes the activation status (triggered/not-triggered)
  is_activated = self.is_activated()
 
- if is_activated == was_activated or is_activated == self._output_state:
+ if is_activated == self._output_active:
  # nothing changed
- if self._output_state:
- # combination is active, consume the event
- return True
+ if is_pressed:
+ # If the output is active, a key-down event triggered it, which then
+ # did not get forwarded, therefore it doesn't require a release.
+ self.require_release_later(not self._output_active, event)
+ # output is active: consume the event
+ # output inactive: forward the event
+ return self._output_active
  else:
- # combination inactive, forward the event
- return False
+ # Else if it is released: Returning `False` means that the event-reader
+ # will forward the release.
+ return not self.should_release_event(event)
 
  if is_activated:
- # send key up events to the forwarded uinput
+ # Send key up events to the forwarded uinput if configured to do so.
  self.forward_release()
  event = event.modify(value=1)
  else:
- if self._output_state or self.mapping.is_axis_mapping():
- # we ignore the suppress argument for release events
+ # If not activated. All required keys are not yet pressed.
+ if self._output_active or self.mapping.is_axis_mapping():
+ # we ignore the `suppress` argument for release events
  # otherwise we might end up with stuck keys
  # (test_event_pipeline.test_combination)
 
@@ -136,14 +144,47 @@ def notify(
  return False
 
  logger.debug("Sending %s to sub-handler", self.mapping.input_combination)
- self._output_state = bool(event.value)
- return self._sub_handler.notify(event, source, suppress)
+ self._output_active = bool(event.value)
+ sub_handler_result = self._sub_handler.notify(event, source, suppress)
+
+ if is_pressed:
+ # If the sub-handler return True, it handled the event, so the user never
+ # sees this key-down event. In that case, we don't require a release event.
+ self.require_release_later(not sub_handler_result, event)
+ return sub_handler_result
+ else:
+ # Else if it is released: Returning `False` means that the event-reader
+ # will forward the release.
+ return not self.should_release_event(event)
+
+ def should_release_event(self, event: InputEvent) -> bool:
+ """Check if the key-up event should be forwarded by the event-reader.
+
+ After this, the release event needs to be injected by someone, otherwise the
+ dictionary was modified erroneously. If there is no entry, we assume that there
+ was no key-down event to release. Maybe a duplicate event arrived.
+ """
+ # Ensure that all injected key-down events will get their release event
+ # injected eventually.
+ # If a key-up event arrives that will inactivate the combination, but
+ # for which previously a key-down event was injected (because it was
+ # an earlier key in the combination chain), then we need to ensure that its
+ # release is injected as well. So we get two release events in that case:
+ # one for the key, and one for the output.
+ assert event.value == 0
+ return self._requires_a_release.pop(event.type_and_code, False)
+
+ def require_release_later(self, require: bool, event: InputEvent) -> None:
+ """Remember if this key-down event will need a release event later on."""
+ assert event.value == 1
+ self._requires_a_release[event.type_and_code] = require
 
  def reset(self) -> None:
  self._sub_handler.reset()
  for key in self._pressed_keys:
  self._pressed_keys[key] = False
- self._output_state = False
+ self._requires_a_release = {}
+ self._output_active = False
 
  def is_activated(self) -> bool:
  """Return if all keys in the keymap are set to True."""
@@ -165,6 +206,9 @@ def forward_release(self) -> None:
  logger.debug("Forwarding release for %s", self.mapping.input_combination)
 
  for input_config in keys_to_release:
+ if not self._requires_a_release.get(input_config.type_and_code):
+ continue
+
  origin_hash = input_config.origin_hash
  if origin_hash is None:
  logger.error(
@@ -177,6 +221,9 @@ def forward_release(self) -> None:
  forward_to.write(*input_config.type_and_code, 0)
  forward_to.syn()
 
+ # We are done with this key, forget about it
+ del self._requires_a_release[input_config.type_and_code]
+
  def needs_ranking(self) -> bool:
  return bool(self.input_configs)
 

inputremapper/injection/mapping_handlers/key_handler.py

@@ -68,7 +68,6 @@ def child(self): # used for logging
 
  def notify(self, event: InputEvent, *_, **__) -> bool:
  """Inject event.value to the target key."""
-
  event_tuple = (*self._maps_to, event.value)
  try:
  global_uinputs.write(event_tuple, self.mapping.target_uinput)

tests/unit/test_event_pipeline/test_event_pipeline.py

@@ -17,6 +17,9 @@
 #
 # You should have received a copy of the GNU General Public License
 # along with input-remapper. If not, see <https://www.gnu.org/licenses/>.
+
+from tests import test
+
 import asyncio
 import unittest
 from typing import Iterable
@@ -108,7 +111,7 @@ def create_event_reader(
  return reader
 
 
-class TestIdk(EventPipelineTestBase):
+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"""
@@ -416,6 +419,10 @@ async def test_combination(self):
  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)
@@ -440,6 +447,8 @@ async def test_combination(self):
 
  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
@@ -465,6 +474,70 @@ async def test_combination(self):
  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 = system_mapping.get("7")
+ in_2 = system_mapping.get("8")
+ out = system_mapping.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 = 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_ignore_hold(self):
  # hold as in event-value 2, not in macro-hold.
  # linux will generate events with value 2 after input-remapper injected

tests/unit/test_event_pipeline/test_mapping_handlers.py

@@ -21,6 +21,7 @@
 
 """See TestEventPipeline for more tests."""
 
+from tests import test
 
 import asyncio
 import unittest
@@ -59,13 +60,16 @@
 from inputremapper.injection.mapping_handlers.hierarchy_handler import HierarchyHandler
 from inputremapper.injection.mapping_handlers.key_handler import KeyHandler
 from inputremapper.injection.mapping_handlers.macro_handler import MacroHandler
-from inputremapper.injection.mapping_handlers.mapping_handler import MappingHandler
+from inputremapper.injection.mapping_handlers.mapping_handler import (
+ MappingHandler,
+ InputEventHandler,
+)
 from inputremapper.injection.mapping_handlers.rel_to_abs_handler import RelToAbsHandler
 from inputremapper.input_event import InputEvent, EventActions
 
 from tests.lib.cleanup import cleanup
 from tests.lib.logger import logger
-from tests.lib.patches import InputDevice
+from tests.lib.patches import InputDevice, UInput
 from tests.lib.constants import MAX_ABS
 from tests.lib.fixtures import fixtures
 
@@ -286,26 +290,29 @@ def setUp(self):
  input_combination=input_combination.to_config(),
  target_uinput="mouse",
  output_symbol="BTN_LEFT",
+ release_combination_keys=True,
  ),
  self.context_mock,
  )
 
+ sub_handler_mock = MagicMock(InputEventHandler)
+ self.handler.set_sub_handler(sub_handler_mock)
+
+ # insert our own test-uinput to see what is being written to it
+ self.uinputs = {
+ self.mouse_hash: UInput(),
+ self.keyboard_hash: UInput(),
+ self.gamepad_hash: UInput(),
+ }
+ self.context_mock.get_forward_uinput = lambda origin_hash: self.uinputs[
+ origin_hash
+ ]
+
  def test_forward_correctly(self):
  # In the past, if a mapping has inputs from two different sub devices, it
  # always failed to send the release events to the correct one.
  # Nowadays, self._context.get_forward_uinput(origin_hash) is used to
  # release them correctly.
- mock = MagicMock()
- self.handler.set_sub_handler(mock)
-
- # insert our own test-uinput to see what is being written to it
- uinputs = {
- self.mouse_hash: evdev.UInput(),
- self.keyboard_hash: evdev.UInput(),
- self.gamepad_hash: evdev.UInput(),
- }
- self.context_mock.get_forward_uinput = lambda origin_hash: uinputs[origin_hash]
-
  # 1. trigger the combination
  self.handler.notify(
  InputEvent.rel(
@@ -335,30 +342,23 @@ def test_forward_correctly(self):
  # 2. expect release events to be written to the correct devices, as indicated
  # by the origin_hash of the InputConfigs
  self.assertListEqual(
- uinputs[self.mouse_hash].write_history,
+ self.uinputs[self.mouse_hash].write_history,
  [InputEvent.rel(self.input_combination[0].code, 0)],
  )
  self.assertListEqual(
- uinputs[self.keyboard_hash].write_history,
+ self.uinputs[self.keyboard_hash].write_history,
  [InputEvent.key(self.input_combination[1].code, 0)],
  )
- self.assertListEqual(
- uinputs[self.gamepad_hash].write_history,
- [InputEvent.key(self.input_combination[2].code, 0)],
- )
+
+ # We do not expect a release event for this, because there was no key-down
+ # event when the combination triggered.
+ # self.assertListEqual(
+ # self.uinputs[self.gamepad_hash].write_history,
+ # [InputEvent.key(self.input_combination[2].code, 0)],
+ # )
 
  def test_no_forwards(self):
  # if a combination is not triggered, nothing is released
- mock = MagicMock()
- self.handler.set_sub_handler(mock)
-
- # insert our own test-uinput to see what is being written to it
- uinputs = {
- self.mouse_hash: evdev.UInput(),
- self.keyboard_hash: evdev.UInput(),
- }
- self.context_mock.get_forward_uinput = lambda origin_hash: uinputs[origin_hash]
-
  # 1. inject any two events
  self.handler.notify(
  InputEvent.rel(
@@ -378,8 +378,8 @@ def test_no_forwards(self):
  )
 
  # 2. expect no release events to be written
- self.assertListEqual(uinputs[self.mouse_hash].write_history, [])
- self.assertListEqual(uinputs[self.keyboard_hash].write_history, [])
+ self.assertListEqual(self.uinputs[self.mouse_hash].write_history, [])
+ self.assertListEqual(self.uinputs[self.keyboard_hash].write_history, [])
 
 
 class TestHierarchyHandler(BaseTests, unittest.IsolatedAsyncioTestCase):