# -*- coding: utf-8 -*- # input-remapper - GUI for device specific keyboard mappings # Copyright (C) 2025 sezanzeb # # 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 . 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, )