File: context.py

package info (click to toggle)
input-remapper 2.1.1-2
  • links: PTS, VCS
  • area: main
  • in suites: forky
  • size: 2,856 kB
  • sloc: python: 27,277; sh: 191; xml: 33; makefile: 3
file content (132 lines) | stat: -rw-r--r-- 4,867 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
# -*- coding: utf-8 -*-
# input-remapper - GUI for device specific keyboard mappings
# Copyright (C) 2025 sezanzeb <b8x45ygc9@mozmail.com>
#
# 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/>.


"""Stores injection-process wide information."""

from __future__ import annotations

from collections import defaultdict
from typing import List, Dict, Set, Hashable

import evdev

from inputremapper.configs.preset import Preset
from inputremapper.injection.mapping_handlers.mapping_handler import (
    EventListener,
    NotifyCallback,
)
from inputremapper.injection.mapping_handlers.mapping_parser import (
    MappingParser,
    EventPipelines,
)
from inputremapper.input_event import InputEvent
from inputremapper.logging.logger import logger
from inputremapper.utils import DeviceHash


class Context:
    """Stores injection-process wide information.

    In some ways this is a wrapper for the preset that derives some
    information that is specifically important to the injection.

    The information in the context does not change during the injection.

    One Context exists for each injection process, which is shared
    with all coroutines and used objects.

    Benefits of the context:
    - less redundant passing around of parameters
    - easier to add new process wide information without having to adjust
      all function calls in unittests
    - makes the injection class shorter and more specific to a certain task,
      which is actually spinning up the injection.

    Note, that for the reader_service a ContextDummy is used.

    Members
    -------
    preset : Preset
        The preset holds all Mappings for the injection process
    listeners : Set[EventListener]
        A set of callbacks which receive all events
    callbacks : Dict[Tuple[int, int], List[NotifyCallback]]
        All entry points to the event pipeline sorted by InputEvent.type_and_code
    """

    listeners: Set[EventListener]
    _notify_callbacks: Dict[Hashable, List[NotifyCallback]]
    _handlers: EventPipelines
    _forward_devices: Dict[DeviceHash, evdev.UInput]
    _source_devices: Dict[DeviceHash, evdev.InputDevice]

    def __init__(
        self,
        preset: Preset,
        source_devices: Dict[DeviceHash, evdev.InputDevice],
        forward_devices: Dict[DeviceHash, evdev.UInput],
        mapping_parser: MappingParser,
    ):
        if len(forward_devices) == 0:
            logger.warning("forward_devices not set")

        if len(source_devices) == 0:
            logger.warning("source_devices not set")

        self.listeners = set()
        self._source_devices = source_devices
        self._forward_devices = forward_devices
        self._notify_callbacks = defaultdict(list)
        self._handlers = mapping_parser.parse_mappings(preset, self)

        self._create_callbacks()

    def reset(self) -> None:
        """Call the reset method for each handler in the context."""
        for handlers in self._handlers.values():
            for handler in handlers:
                handler.reset()

    def _create_callbacks(self) -> None:
        """Add the notify method from all _handlers to self.callbacks."""
        for input_config, handler_list in self._handlers.items():
            input_match_hash = input_config.input_match_hash
            logger.debug("Adding NotifyCallback for %s", input_match_hash)
            self._notify_callbacks[input_match_hash].extend(
                handler.notify for handler in handler_list
            )

    def get_notify_callbacks(self, input_event: InputEvent) -> List[NotifyCallback]:
        input_match_hash = input_event.input_match_hash
        return self._notify_callbacks[input_match_hash]

    def get_forward_uinput(self, origin_hash: DeviceHash) -> evdev.UInput:
        """Get the "forward" uinput events from the given origin should go into."""
        return self._forward_devices[origin_hash]

    def get_source(self, key: DeviceHash) -> evdev.InputDevice:
        return self._source_devices[key]

    def get_leds(self) -> Set[int]:
        """Get a set of LED_* ecodes that are currently on."""
        leds = set()
        for device in self._source_devices.values():
            leds.update(device.leds())
        return leds