# -*- 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 . """User-level service that applies presets based on the focused application. This process runs as the logged-in user (never root, no pkexec). It reads the user config, detects the focused window through a ``FocusBackend`` and drives the existing root daemon over the system D-Bus. It exposes a small interface on the session bus (``inputremapper.Focus``) for the GUI. """ from __future__ import annotations import json from typing import Dict, List, Optional import gi from dasbus.connection import SessionMessageBus from dasbus.identifier import DBusServiceIdentifier from dasbus.loop import EventLoop from dasbus.signal import Signal from inputremapper.configs.app_binding import AppBinding from inputremapper.configs.global_config import GlobalConfig from inputremapper.daemon import DaemonProxy from inputremapper.focus.focus_backend import FocusBackend, FocusEvent from inputremapper.logging.logger import logger gi.require_version("GLib", "2.0") from gi.repository import GLib # noqa: E402 SESSION_BUS = SessionMessageBus() FOCUS = DBusServiceIdentifier( namespace=("inputremapper", "Focus"), message_bus=SESSION_BUS, ) # How long to wait for the focus to settle before reconciling, in milliseconds. # This collapses rapid alt-tabbing into a single reconciliation. DEFAULT_DEBOUNCE_MS = 150 class FocusService: """Reconciles injected presets with the currently focused application. The service only ever stops injections that it started itself (tracked in ``_app_controlled``). Presets applied manually (e.g. from the GUI) are left untouched until another *bound* application takes focus and reclaims that device, which implements the "manual apply is a temporary override" behavior. """ __dbus_xml__ = f""" """ def __init__( self, global_config: GlobalConfig, backend: FocusBackend, daemon: DaemonProxy, debounce_ms: int = DEFAULT_DEBOUNCE_MS, ) -> None: self.global_config = global_config self.backend = backend self.daemon = daemon self._debounce_ms = debounce_ms # The signal attribute that dasbus connects to the bus. self.focus_changed = Signal() self._bindings: List[AppBinding] = [] self._enabled = False # group_key -> preset currently injected because of a binding self._app_controlled: Dict[str, str] = {} self._current_event: Optional[FocusEvent] = None self._pending_event: Optional[FocusEvent] = None self._debounce_id: Optional[int] = None self._load_config() # -- lifecycle --------------------------------------------------------- def _load_config(self) -> None: self.global_config.load_config() self._enabled = self.global_config.get_app_binding_enabled() self._bindings = self.global_config.get_app_bindings() logger.info( "Loaded %d app binding(s), enabled=%s", len(self._bindings), self._enabled, ) def publish(self) -> None: """Make the session-bus interface available to the GUI.""" try: SESSION_BUS.publish_object(FOCUS.object_path, self) SESSION_BUS.register_service(FOCUS.service_name) except ConnectionError as error: logger.error("Is the focus-service already running? (%s)", str(error)) raise RuntimeError("Failed to publish focus-service on D-Bus") from error def start(self) -> None: """Publish the interface and start watching for focus changes.""" self.publish() self.backend.start(self._on_focus) self._current_event = self.backend.get_current() if self._current_event is not None and self._enabled: self._reconcile(self._current_event) logger.info("Focus-service started with backend '%s'", self.backend.name) def run(self) -> None: """Start the service and block on the GLib main loop.""" self.start() loop = EventLoop() loop.run() def stop(self) -> None: """Stop the backend and any injections this service controls.""" if self._debounce_id is not None: GLib.source_remove(self._debounce_id) self._debounce_id = None self.backend.stop() for group_key in list(self._app_controlled): self.daemon.stop_injecting(group_key) self._app_controlled.clear() # -- focus handling ---------------------------------------------------- def _on_focus(self, event: FocusEvent) -> None: """Called by the backend on every focus change (debounced here).""" if not event.app_id: # Transient focus (menus, popups, no focused window). Ignore it so # we don't tear down injections during alt-tab. logger.debug("Ignoring transient focus event: %s", event) return self._pending_event = event if self._debounce_id is not None: GLib.source_remove(self._debounce_id) self._debounce_id = GLib.timeout_add(self._debounce_ms, self._flush) def _flush(self) -> bool: self._debounce_id = None event = self._pending_event if event is None: return False changed = event != self._current_event self._current_event = event if changed: self._emit_focus_changed(event) if self._enabled: self._reconcile(event) return False # GLib: do not repeat def _compute_desired(self, event: FocusEvent) -> Dict[str, str]: """Map of group_key -> preset that should be injected for this event.""" desired: Dict[str, str] = {} for binding in self._bindings: if binding.matches(event): for bound in binding.presets: desired[bound.group_key] = bound.preset return desired def _reconcile(self, event: FocusEvent) -> None: desired = self._compute_desired(event) # Stop app-controlled injections that are no longer wanted. Manually # applied presets are not in _app_controlled, so they are never stopped # here. for group_key in list(self._app_controlled): if group_key not in desired: logger.info("Focus reconcile: stopping '%s'", group_key) self.daemon.stop_injecting(group_key) del self._app_controlled[group_key] # Start or replace the wanted injections. for group_key, preset in desired.items(): if self._app_controlled.get(group_key) == preset: continue logger.info( "Focus reconcile: injecting '%s' on '%s' for app '%s'", preset, group_key, event.app_id, ) if self.daemon.start_injecting(group_key, preset): self._app_controlled[group_key] = preset else: logger.error( "Failed to start injecting '%s' on '%s'", preset, group_key ) def _emit_focus_changed(self, event: FocusEvent) -> None: self.focus_changed(self._event_to_json(event)) @staticmethod def _event_to_json(event: Optional[FocusEvent]) -> str: if event is None: return json.dumps({"app_id": "", "title": ""}) return json.dumps({"app_id": event.app_id, "title": event.title}) # -- D-Bus interface --------------------------------------------------- def get_focused_app(self) -> str: """Return json {app_id, title} for the currently focused window.""" return self._event_to_json(self._current_event) def reload(self) -> None: """Re-read the config from disk and reconcile the current focus.""" logger.info("Reloading focus-service config") self._load_config() if self._current_event is not None and self._enabled: self._reconcile(self._current_event) elif not self._enabled: self._stop_app_controlled() def set_enabled(self, enabled: bool) -> None: """Toggle focus-driven binding at runtime (does not write the config).""" logger.info("Setting focus-binding enabled=%s", enabled) self._enabled = bool(enabled) if self._enabled: if self._current_event is not None: self._reconcile(self._current_event) else: self._stop_app_controlled() def get_status(self) -> str: """Return json describing the service state for the GUI.""" return json.dumps( { "backend": self.backend.name, "enabled": self._enabled, "focused": { "app_id": self._current_event.app_id if self._current_event else "", "title": self._current_event.title if self._current_event else "", }, "app_controlled": dict(self._app_controlled), } ) def _stop_app_controlled(self) -> None: for group_key in list(self._app_controlled): self.daemon.stop_injecting(group_key) del self._app_controlled[group_key]