File: utils.py

package info (click to toggle)
input-remapper 2.2.0%2Bgit20251231.d605b42-1
  • links: PTS, VCS
  • area: main
  • in suites: experimental
  • size: 2,896 kB
  • sloc: python: 27,435; sh: 45; xml: 33; makefile: 9
file content (272 lines) | stat: -rw-r--r-- 8,519 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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
# -*- 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/>.

from __future__ import annotations

import time
from dataclasses import dataclass
from typing import List, Callable, Dict, Optional

from gi.repository import Gtk, GLib, Gdk

from inputremapper.logging.logger import logger

# status ctx ids

CTX_SAVE = 0
CTX_APPLY = 1
CTX_KEYCODE = 2
CTX_ERROR = 3
CTX_WARNING = 4
CTX_MAPPING = 5


@dataclass()
class DebounceInfo:
    # constant after register:
    function: Optional[Callable]
    other: object
    key: int

    # can change when called again:
    args: list
    kwargs: dict
    glib_timeout: Optional[int]


class DebounceManager:
    """Stops all debounced functions if needed."""

    debounce_infos: Dict[int, DebounceInfo] = {}

    def _register(self, other, function):
        debounce_info = DebounceInfo(
            function=function,
            glib_timeout=None,
            other=other,
            args=[],
            kwargs={},
            key=self._get_key(other, function),
        )
        key = self._get_key(other, function)
        self.debounce_infos[key] = debounce_info
        return debounce_info

    def get(self, other: object, function: Callable) -> Optional[DebounceInfo]:
        """Find the debounce_info that matches the given callable."""
        key = self._get_key(other, function)
        return self.debounce_infos.get(key)

    def _get_key(self, other, function):
        return f"{id(other)},{function.__name__}"

    def debounce(self, other, function, timeout_ms, *args, **kwargs):
        """Call this function with the given args later."""
        debounce_info = self.get(other, function)
        if debounce_info is None:
            debounce_info = self._register(other, function)

        debounce_info.args = args
        debounce_info.kwargs = kwargs

        glib_timeout = debounce_info.glib_timeout
        if glib_timeout is not None:
            GLib.source_remove(glib_timeout)

        def run():
            self.stop(other, function)
            return function(other, *args, **kwargs)

        debounce_info.glib_timeout = GLib.timeout_add(
            timeout_ms,
            lambda: run(),
        )

    def stop(self, other: object, function: Callable):
        """Stop the current debounce timeout of this function and don't call it.

        New calls to that function will be debounced again.
        """
        debounce_info = self.get(other, function)
        if debounce_info is None:
            logger.debug("Tried to stop function that is not currently scheduled")
            return

        if debounce_info.glib_timeout is not None:
            GLib.source_remove(debounce_info.glib_timeout)
            debounce_info.glib_timeout = None

    def stop_all(self):
        """No debounced function should be called anymore after this.

        New calls to that function will be debounced again.
        """
        for debounce_info in self.debounce_infos.values():
            self.stop(debounce_info.other, debounce_info.function)

    def run_all_now(self):
        """Don't wait any longer."""
        for debounce_info in self.debounce_infos.values():
            if debounce_info.glib_timeout is None:
                # nothing is currently waiting for this function to be called
                continue

            self.stop(debounce_info.other, debounce_info.function)
            try:
                logger.warning(
                    'Running "%s" now without waiting',
                    debounce_info.function.__name__,
                )
                debounce_info.function(
                    debounce_info.other,
                    *debounce_info.args,
                    **debounce_info.kwargs,
                )
            except Exception as exception:
                # if individual functions fails, continue calling the others.
                # also, don't raise this because there is nowhere this exception
                # could be caught in a useful way
                logger.error(exception)


def debounce(timeout):
    """Debounce a method call to improve performance.

    Calling this with a millisecond value creates the decorator, so use something like

    @debounce(50)
    def function(self):
        ...

    In tests, run_all_now can be used to avoid waiting to speed them up.
    """
    # the outside `debounce` function is needed to obtain the millisecond value

    def decorator(function):
        # the regular decorator.
        # @decorator
        # def foo():
        #   ...
        def wrapped(self, *args, **kwargs):
            # this is the function that will actually be called
            debounce_manager.debounce(self, function, timeout, *args, **kwargs)

        wrapped.__name__ = function.__name__

        return wrapped

    return decorator


debounce_manager = DebounceManager()


class HandlerDisabled:
    """Safely modify a widget without causing handlers to be called.

    Use in a `with` statement.
    """

    def __init__(self, widget: Gtk.Widget, handler: Callable):
        self.widget = widget
        self.handler = handler

    def __enter__(self):
        try:
            self.widget.handler_block_by_func(self.handler)
        except TypeError as error:
            # if nothing is connected to the given signal, it is not critical
            # at all
            logger.warning('HandlerDisabled entry failed: "%s"', error)

    def __exit__(self, *_):
        try:
            self.widget.handler_unblock_by_func(self.handler)
        except TypeError as error:
            logger.warning('HandlerDisabled exit failed: "%s"', error)


def gtk_iteration(iterations=0):
    """Iterate while events are pending."""
    while Gtk.events_pending():
        Gtk.main_iteration()
    for _ in range(iterations):
        time.sleep(0.002)
        while Gtk.events_pending():
            Gtk.main_iteration()


class Colors:
    """Looks up colors from the GTK theme.

    Defaults to libadwaita-light theme colors if the lookup fails.
    """

    fallback_accent = Gdk.RGBA(0.21, 0.52, 0.89, 1)
    fallback_background = Gdk.RGBA(0.98, 0.98, 0.98, 1)
    fallback_base = Gdk.RGBA(1, 1, 1, 1)
    fallback_border = Gdk.RGBA(0.87, 0.87, 0.87, 1)
    fallback_font = Gdk.RGBA(0.20, 0.20, 0.20, 1)

    @staticmethod
    def get_color(names: List[str], fallback: Gdk.RGBA) -> Gdk.RGBA:
        """Get theme colors. Provide multiple names for fallback purposes."""
        for name in names:
            found, color = Gtk.StyleContext().lookup_color(name)
            if found:
                return color

        return fallback

    @staticmethod
    def get_accent_color() -> Gdk.RGBA:
        """Look up the accent color from the current theme."""
        return Colors.get_color(
            ["accent_bg_color", "theme_selected_bg_color"],
            Colors.fallback_accent,
        )

    @staticmethod
    def get_background_color() -> Gdk.RGBA:
        """Look up the background-color from the current theme."""
        return Colors.get_color(
            ["theme_bg_color"],
            Colors.fallback_background,
        )

    @staticmethod
    def get_base_color() -> Gdk.RGBA:
        """Look up the base-color from the current theme."""
        return Colors.get_color(
            ["theme_base_color"],
            Colors.fallback_base,
        )

    @staticmethod
    def get_border_color() -> Gdk.RGBA:
        """Look up the border from the current theme."""
        return Colors.get_color(["borders"], Colors.fallback_border)

    @staticmethod
    def get_font_color() -> Gdk.RGBA:
        """Look up the border from the current theme."""
        return Colors.get_color(
            ["theme_fg_color"],
            Colors.fallback_font,
        )