Blomios
af10ec8bd1
Le suivi du focus laissait les injections app-controlled actives lors d'une perte de focus réelle (fermeture de la fenêtre courante, retour à un bureau vide). On corrige la détection sur toute la chaîne : - FocusService: un app_id vide est désormais traité comme un unfocus réel (et non un focus transitoire ignoré), ce qui stoppe les injections liées à l'application. - Sway: on ignore les events 'new' (pas un changement de focus) et on relit le focus courant sur un event 'title' non focused. - WlrForeignToplevelBackend: suit _current_handle, le clear sur stop() et émet un FocusEvent vide quand le toplevel actuellement focus se ferme. Tests: test_focus_service + test_focus_watcher (67) OK; lot feature historique (295) OK. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
279 lines
10 KiB
Python
279 lines
10 KiB
Python
# -*- 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/>.
|
|
|
|
"""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"""
|
|
<node>
|
|
<interface name='{FOCUS.interface_name}'>
|
|
<method name='get_focused_app'>
|
|
<arg type='s' name='response' direction='out'/>
|
|
</method>
|
|
<method name='reload'>
|
|
</method>
|
|
<method name='set_enabled'>
|
|
<arg type='b' name='enabled' direction='in'/>
|
|
</method>
|
|
<method name='get_status'>
|
|
<arg type='s' name='response' direction='out'/>
|
|
</method>
|
|
<signal name='focus_changed'>
|
|
<arg type='s' name='app'/>
|
|
</signal>
|
|
</interface>
|
|
</node>
|
|
"""
|
|
|
|
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)."""
|
|
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]
|