Refactored injection

.gitignore

@@ -58,6 +58,9 @@ coverage.xml
 .hypothesis/
 .pytest_cache/
 
+# pyreverse graphs
+*.dot
+
 # Translations
 *.mo
 

inputremapper/groups.py

@@ -485,7 +485,12 @@ def loads(self, dump):
         """Load a serialized representation created via dumps."""
         self._groups = [_Group.loads(group) for group in json.loads(dump)]
 
-    def find(self, name=None, key=None, path=None, include_inputremapper=False):
+    def find(self,
+             name: str = None,
+             key: str = None,
+             path: str = None,
+             include_inputremapper: bool = False
+             ) -> _Group:
         """Find a group that matches the provided parameters.
 
         Parameters

inputremapper/injection/consumer_control.py

@@ -23,19 +23,27 @@
 
 
 import asyncio
-
 import evdev
 
 from inputremapper.injection.consumers.joystick_to_mouse import JoystickToMouse
 from inputremapper.injection.consumers.keycode_mapper import KeycodeMapper
 from inputremapper.logger import logger
-
+from inputremapper.injection.context import Context
 
 consumer_classes = [
     KeycodeMapper,
     JoystickToMouse,
 ]
 
+def copy_event(event: evdev.InputEvent) -> evdev.InputEvent:
+    return evdev.InputEvent(
+        sec=event.sec,
+        usec=event.usec,
+        type=event.type,
+        code=event.code,
+        value=event.value,
+    )
+
 
 class ConsumerControl:
     """Reads input events from a single device and distributes them.
@@ -48,7 +56,11 @@ class ConsumerControl:
     needs to be created multiple times.
     """
 
-    def __init__(self, context, source, forward_to):
+    def __init__(self,
+                 context: Context,
+                 source: evdev.InputDevice,
+                 forward_to: evdev.UInput,
+                 ) -> None:
         """Initialize all consumers
 
         Parameters
@@ -62,6 +74,7 @@ def __init__(self, context, source, forward_to):
         """
         self._source = source
         self._forward_to = forward_to
+        self.context = context
 
         # add all consumers that are enabled for this particular configuration
         self._consumers = []
@@ -93,17 +106,21 @@ async def run(self):
                 # won't appear, no need to forward or map them.
                 continue
 
+            tasks = []
+            for callback in self.context.callbacks:
+                tasks.append(callback(copy_event(event)))
+            results = await asyncio.gather(*tasks)
+
+            if True in results:
+                continue
+
+            # try legacy injection if the new injection did not do anything
+            # TODO: Remove. only keep the call to forward
             handled = False
             for consumer in self._consumers:
                 # copy so that the consumer doesn't screw this up for
                 # all other future consumers
-                event_copy = evdev.InputEvent(
-                    sec=event.sec,
-                    usec=event.usec,
-                    type=event.type,
-                    code=event.code,
-                    value=event.value,
-                )
+                event_copy = copy_event(event)
                 if consumer.is_handled(event_copy):
                     await consumer.notify(event_copy)
                     handled = True

inputremapper/injection/consumers/mapping_handler.py

@@ -0,0 +1,178 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+# input-remapper - GUI for device specific keyboard mappings
+# Copyright (C) 2022 sezanzeb <proxima@sezanzeb.de>
+#
+# 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 <https://www.gnu.org/licenses/>.
+
+
+"""Consumer base class.
+
+Can be notified of new events so that inheriting classes can map them and
+inject new events based on them.
+"""
+import copy
+from typing import Dict, Tuple, Type, List
+
+import evdev
+
+from inputremapper import utils
+from inputremapper.logger import logger
+from inputremapper.system_mapping import system_mapping
+from inputremapper.key import Key
+from inputremapper.injection.global_uinputs import global_uinputs
+
+
+class MappingHandler:
+    """Can be notified of new events to inject them. Base class."""
+    _config: Dict[str, any]
+    _key: Key
+    _target: str
+
+    # hierarich mapping handlers e.g. KeysToKeyHandler will get sorted to make sure
+    # only one handler will receive the event. This allows mapping of combinations
+    # to override other combinations.
+    # Non hierarich handlers will always receive all events they care about
+    hierarich = False
+
+    def __init__(self, config_dict: Dict[str, any]) -> None:
+        """Initialize event consuming functionality."""
+        self._config = copy.deepcopy(config_dict)
+        self._key = Key(self._config['key'])
+        self._target = self._config["target"]
+
+    def might_handle(self, event_type: int, event_code: int) -> bool:
+        """Check if the handler cares about this at all"""
+        return self._key.contains_event(event_type, event_code)
+
+    async def notify(self, event: evdev.InputEvent) -> bool:
+        """A new event is ready.
+
+        return if the event was actually taken care of
+
+        Overwrite this function if the consumer should do something each time
+        a new event arrives. E.g. mapping a single button once clicked.
+        """
+        raise NotImplementedError
+
+    async def run(self) -> None:
+        """Start doing things.
+
+        Overwrite this function if the consumer should do something
+        continuously even if no new event arrives. e.g. continuously injecting
+        mouse movement events.
+        """
+        raise NotImplementedError
+
+    def inject(self, event_tuple: Tuple) -> None:
+        logger.debug("injecting form new mapping handler: %s", event_tuple)
+        global_uinputs.write(event_tuple, self._target)
+
+
+class KeysToKeyHandler(MappingHandler):
+
+    _key_map: Dict[Tuple[int, int], bool]
+    _active: bool  # keep track if the target key is pressed down
+    maps_to: Tuple[int, int]
+    hierarich = True
+
+    def __init__(self, config: Dict[str, any]) -> None:
+        super().__init__(config)
+
+        self._active = False
+        self._key_map = {}
+        for sub_key in self._key:  # prepare key_map
+            self._key_map[sub_key[:2]] = False
+
+        self.maps_to = (evdev.ecodes.EV_KEY, system_mapping.get(self._config["symbol"]))
+
+    async def notify(self, event: evdev.InputEvent) -> bool:
+        map_key = (event.type, event.code)
+        if map_key not in self._key_map.keys():
+            return False  # we are not responsible for the event
+
+        self._key_map[map_key] = event.value == 1
+        if self.get_active() == self._active:
+            return False  # nothing changed ignore this event
+
+        self._active = self.get_active() and not utils.is_key_up(event.value)
+        if self._active:
+            value = 1
+        else:
+            value = 0
+
+        print("warum kommt der folgende key_spam nicht immer an?")
+        logger.key_spam(self._key, "maps to (%s)", self.maps_to)
+        self.inject((*self.maps_to, value))
+        return True
+
+    async def run(self) -> None:  # no debouncer or anything (yet)
+        pass
+
+    def get_active(self) -> bool:
+        return False not in self._key_map.values()
+
+
+class HierarchyHandler(MappingHandler):
+    """handler consisting of an ordered list of KeysToKeyHandler
+
+    only the first handler which successfully handles a key_down event will execute the key_down.
+    All handlers receive key up events, but in reversed order.
+    """
+    hierarich = False
+
+    def __init__(self, handlers: List[MappingHandler]) -> None:
+        cfg = {
+            "key": None,
+            "target": None,
+        }
+        super().__init__(cfg)
+        self.handlers = handlers
+
+    async def run(self) -> None:
+        pass
+
+    async def notify(self, event: evdev.InputEvent) -> bool:
+        if event.value == 1:
+            return self.handle_key_down(event)
+        else:
+            return self.handle_key_up(event)
+
+    def handle_key_up(self, event: evdev.InputEvent) -> bool:
+        success = False
+        for handler in self.handlers[::-1]:
+            success = await handler.notify(event)
+        return success
+
+    def handle_key_down(self, event: evdev.InputEvent) -> bool:
+        success = False
+        for handler in self.handlers:
+            success = await handler.notify(event)
+            if success:
+                break
+
+        return success
+
+
+mapping_handler_classes: Dict[str, Type[MappingHandler]] = {
+    # all available mapping_handlers
+    "keys_to_key": KeysToKeyHandler,
+}
+
+
+def create_handler(config: Dict[str, any]) -> MappingHandler:
+    """return the MappingHandler"""
+    return mapping_handler_classes[config['type']](config)