use event origin information

inputremapper/configs/input_config.py

@@ -20,7 +20,7 @@
 from __future__ import annotations
 
 import itertools
-from typing import Tuple, Iterable, Union, List, Dict, Optional
+from typing import Tuple, Iterable, Union, List, Dict, Optional, Hashable
 
 from evdev import ecodes
 from pydantic import BaseModel, root_validator, validator
@@ -51,6 +51,16 @@ class InputConfig(BaseModel):
     origin: Optional[int] = None
     analog_threshold: Optional[int] = None
 
+    @property
+    def input_match_hash(self) -> Hashable:
+        """a Hashable object which is intended to match the InputConfig with a
+        InputEvent.
+
+        InputConfig itself is hashable, but can not be used to match InputEvent's
+        because its hash includes the analog_threshold
+        """
+        return self.type, self.code, self.origin
+
     @property
     def defines_analog_input(self) -> bool:
         """Whether this defines an analog input"""

inputremapper/configs/preset.py

@@ -76,7 +76,7 @@ def __init__(
 
     def __iter__(self) -> Iterator[MappingModel]:
         """Iterate over Mapping objects."""
-        return iter(self._mappings.values())
+        return iter(self._mappings.copy().values())
 
     def __len__(self) -> int:
         return len(self._mappings)

inputremapper/injection/context.py

@@ -20,15 +20,18 @@
 
 """Stores injection-process wide information."""
 from collections import defaultdict
-from typing import List, Dict, Tuple, Set
+from typing import List, Dict, Tuple, Set, Hashable
 
 from inputremapper.configs.preset import Preset
 from inputremapper.injection.mapping_handlers.mapping_handler import (
     InputEventHandler,
     EventListener,
     NotifyCallback,
 )
-from inputremapper.injection.mapping_handlers.mapping_parser import parse_mappings
+from inputremapper.injection.mapping_handlers.mapping_parser import (
+    parse_mappings,
+    EventPipelines,
+)
 from inputremapper.input_event import InputEvent
 
 
@@ -61,8 +64,8 @@ class Context:
     """
 
     listeners: Set[EventListener]
-    notify_callbacks: Dict[Tuple[int, int], List[NotifyCallback]]
-    _handlers: Dict[InputEvent, Set[InputEventHandler]]
+    notify_callbacks: Dict[Hashable, List[NotifyCallback]]
+    _handlers: EventPipelines
 
     def __init__(self, preset: Preset):
         self.listeners = set()
@@ -79,7 +82,7 @@ def reset(self) -> None:
 
     def _create_callbacks(self) -> None:
         """Add the notify method from all _handlers to self.callbacks."""
-        for event, handler_list in self._handlers.items():
-            self.notify_callbacks[event.type_and_code].extend(
+        for input_config, handler_list in self._handlers.items():
+            self.notify_callbacks[input_config.input_match_hash].extend(
                 handler.notify for handler in handler_list
             )

inputremapper/injection/event_reader.py

@@ -22,10 +22,11 @@
 
 import asyncio
 import os
-from typing import AsyncIterator, Protocol, Set, Dict, Tuple, List
+from typing import AsyncIterator, Protocol, Set, Dict, List, Hashable
 
 import evdev
 
+from inputremapper.utils import get_device_hash
 from inputremapper.injection.mapping_handlers.mapping_handler import (
     EventListener,
     NotifyCallback,
@@ -36,7 +37,7 @@
 
 class Context(Protocol):
     listeners: Set[EventListener]
-    notify_callbacks: Dict[Tuple[int, int], List[NotifyCallback]]
+    notify_callbacks: Dict[Hashable, List[NotifyCallback]]
 
     def reset(self):
         ...
@@ -71,6 +72,7 @@ def __init__(
             Should be an UInput with capabilities that work for all forwarded
             events, so ideally they should be copied from source.
         """
+        self._device_hash = get_device_hash(source)
         self._source = source
         self._forward_to = forward_to
         self.context = context
@@ -119,7 +121,7 @@ def send_to_handlers(self, event: InputEvent) -> bool:
             return False
 
         results = set()
-        notify_callbacks = self.context.notify_callbacks.get(event.type_and_code)
+        notify_callbacks = self.context.notify_callbacks.get(event.input_match_hash)
         if notify_callbacks:
             for notify_callback in notify_callbacks:
                 results.add(
@@ -191,7 +193,7 @@ async def run(self):
             self._source.fd,
         )
         async for event in self.read_loop():
-            await self.handle(InputEvent.from_event(event))
+            await self.handle(InputEvent.from_event(event, origin=self._device_hash))
 
         self.context.reset()
         logger.info("read loop for %s stopped", self._source.path)

inputremapper/injection/injector.py

@@ -32,7 +32,7 @@
 
 import evdev
 
-from inputremapper.configs.input_config import InputCombination
+from inputremapper.configs.input_config import InputCombination, InputConfig
 from inputremapper.configs.preset import Preset
 from inputremapper.groups import (
     _Group,
@@ -44,6 +44,7 @@
 from inputremapper.injection.event_reader import EventReader
 from inputremapper.injection.numlock import set_numlock, is_numlock_on, ensure_numlock
 from inputremapper.logger import logger
+from inputremapper.utils import get_device_hash
 
 CapabilitiesDict = Dict[int, List[int]]
 GroupSources = List[evdev.InputDevice]
@@ -188,7 +189,23 @@ def stop_injecting(self) -> None:
 
     """Process internal stuff."""
 
-    def _grab_devices(self) -> GroupSources:
+    @staticmethod
+    def _find_input_device(
+        devices: List[evdev.InputDevice], input_config: InputConfig
+    ) -> Optional[evdev.InputDevice]:
+        devices_by_hash = {get_device_hash: device for device in devices}
+
+        def has_type_and_code(device, type, code) -> bool:
+            return code in device.capabilities(absinfo=False).get(type, [])
+
+        if device := devices_by_hash.get(input_config.origin):
+            if has_type_and_code(device, *input_config.type_and_code):
+                return device
+
+        """Fallback logic"""
+        logger.warning(
+            f"Can not find correct device for {input_config.origin} " f"trying fallback"
+        )
         ranking = [
             DeviceType.KEYBOARD,
             DeviceType.GAMEPAD,
@@ -198,6 +215,23 @@ def _grab_devices(self) -> GroupSources:
             DeviceType.CAMERA,
             DeviceType.UNKNOWN,
         ]
+        candidates: List[evdev.InputDevice] = [
+            device
+            for device in devices
+            if input_config.code
+            in device.capabilities(absinfo=False).get(input_config.type, [])
+        ]
+        if len(candidates) > 1:
+            # there is more than on input device which can be used for this
+            # event we choose only one determined by the ranking
+            return sorted(candidates, key=lambda d: ranking.index(classify(d)))[0]
+        elif len(candidates) == 1:
+            return candidates.pop()
+
+        logger.error(f"Could not find input for {input_config}")
+        return None
+
+    def _grab_devices(self) -> GroupSources:
 
         # all devices in this group
         devices: List[evdev.InputDevice] = []
@@ -213,25 +247,20 @@ def _grab_devices(self) -> GroupSources:
         needed_devices = {}
 
         for mapping in self.preset:
-            for event in mapping.input_combination:
-                candidates: List[evdev.InputDevice] = [
-                    device
-                    for device in devices
-                    if event.code
-                    in device.capabilities(absinfo=False).get(event.type, [])
-                ]
-                if len(candidates) > 1:
-                    # there is more than on input device which can be used for this
-                    # event we choose only one determined by the ranking
-                    device = sorted(
-                        candidates, key=lambda d: ranking.index(classify(d))
-                    )[0]
-                elif len(candidates) == 1:
-                    device = candidates.pop()
-                else:
-                    logger.error("Could not find input for %s in %s", event, mapping)
-                    continue
-                needed_devices[device.path] = device
+            update_configs = {}
+            for input_config in mapping.input_combination:
+                if device := self._find_input_device(devices, input_config):
+                    needed_devices[device.path] = device
+                    if input_config.origin is None:
+                        update_configs[input_config] = input_config.modify(
+                            origin=get_device_hash(device)
+                        )
+
+            if update_configs:
+                combination = list(mapping.input_combination)
+                for old, new in update_configs.items():
+                    combination[combination.index(old)] = new
+                mapping.input_combination = combination
 
         grabbed_devices = []
         for device in needed_devices.values():
@@ -324,16 +353,16 @@ def run(self) -> None:
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
 
+        # 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
+        # This also updates the mappings with origin information for each input_config
+        sources = self._grab_devices()
+
         # create this within the process after the event loop creation,
         # so that the macros use the correct loop
         self.context = Context(self.preset)
         self._stop_event = asyncio.Event()
 
-        # 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()
-
         if len(sources) == 0:
             # maybe the preset was empty or something
             logger.error("Did not grab any device")

inputremapper/injection/mapping_handlers/mapping_parser.py

@@ -24,7 +24,7 @@
 
 from evdev.ecodes import EV_KEY, EV_ABS, EV_REL
 
-from inputremapper.configs.input_config import InputCombination
+from inputremapper.configs.input_config import InputCombination, InputConfig
 from inputremapper.configs.mapping import Mapping
 from inputremapper.configs.preset import Preset
 from inputremapper.configs.system_mapping import DISABLE_CODE, DISABLE_NAME
@@ -56,7 +56,7 @@
 from inputremapper.logger import logger
 from inputremapper.utils import get_evdev_constant_name
 
-EventPipelines = Dict[InputEvent, Set[InputEventHandler]]
+EventPipelines = Dict[InputConfig, Set[InputEventHandler]]
 
 mapping_handler_classes: Dict[HandlerEnums, Optional[Type[MappingHandler]]] = {
     # all available mapping_handlers
@@ -130,14 +130,14 @@ def parse_mappings(preset: Preset, context: ContextProtocol) -> EventPipelines:
     event_pipelines: EventPipelines = defaultdict(set)
     for handler in handlers:
         assert handler.input_configs
-        for event in handler.input_configs:
+        for input_config in handler.input_configs:
             logger.debug(
                 "event-pipeline with entry point: %s %s",
-                get_evdev_constant_name(*event.type_and_code),
-                event.type_and_code,
+                get_evdev_constant_name(*input_config.type_and_code),
+                input_config.input_match_hash,
             )
             logger.debug_mapping_handler(handler)
-            event_pipelines[event].add(handler)
+            event_pipelines[input_config].add(handler)
 
     return event_pipelines
 

inputremapper/input_event.py

@@ -21,7 +21,7 @@
 
 import enum
 from dataclasses import dataclass
-from typing import Tuple, Optional
+from typing import Tuple, Optional, Hashable
 
 import evdev
 from evdev import ecodes
@@ -63,11 +63,27 @@ def __eq__(self, other: InputEvent | evdev.InputEvent | Tuple[int, int, int]):
             return self.event_tuple == other
         raise TypeError(f"cannot compare {type(other)} with InputEvent")
 
+    @property
+    def input_match_hash(self) -> Hashable:
+        """a Hashable object which is intended to match the InputEvent with a
+        InputConfig.
+        """
+        return self.type, self.code, self.origin
+
     @classmethod
-    def from_event(cls, event: evdev.InputEvent) -> InputEvent:
+    def from_event(
+        cls, event: evdev.InputEvent, origin: Optional[int] = None
+    ) -> InputEvent:
         """Create a InputEvent from another InputEvent or evdev.InputEvent."""
         try:
-            return cls(event.sec, event.usec, event.type, event.code, event.value)
+            return cls(
+                event.sec,
+                event.usec,
+                event.type,
+                event.code,
+                event.value,
+                origin=origin,
+            )
         except AttributeError as exception:
             raise TypeError(
                 f"Failed to create InputEvent from {event = }"

inputremapper/utils.py

@@ -21,7 +21,7 @@
 """Utility functions."""
 
 import sys
-from typing import Optional
+from typing import Optional, Hashable
 
 import evdev
 
@@ -30,6 +30,10 @@ def is_service() -> bool:
     return sys.argv[0].endswith("input-remapper-service")
 
 
+def get_device_hash(device: evdev.InputDevice) -> int:
+    return hash(str(device.capabilities(absinfo=False)) + device.name)
+
+
 def get_evdev_constant_name(type_: int, code: int, *_) -> Optional[str]:
     """Handy function to get the evdev constant name."""
     # this is more readable than