whisper-local/whisper_local/hotkey/_evdev.py

"""Hotkey-Listener via evdev für Push-to-Talk (Linux)."""

import asyncio
import logging

import evdev
from evdev import InputDevice, categorize, ecodes

from whisper_local.hotkey import AsyncCallback

logger = logging.getLogger(__name__)


def find_keyboard_devices(key_name: str) -> list[InputDevice]:
    """Findet alle Devices die den angegebenen Key unterstützen."""
    matches = []
    for path in evdev.list_devices():
        device = InputDevice(path)
        capabilities = device.capabilities(verbose=True)
        for (etype_name, _etype_code), events in capabilities.items():
            if etype_name == "EV_KEY":
                key_names = [name for name, _code in events]
                if key_name in key_names:
                    logger.info("Device mit %s gefunden: %s (%s)", key_name, device.name, device.path)
                    matches.append(device)
                    break
    if not matches:
        raise RuntimeError(f"Kein Device mit {key_name} gefunden in /dev/input/")
    return matches


class EvdevHotkeyListener:
    def __init__(self, key_name: str = "KEY_F12"):
        self.key_name = key_name
        self.key_code = ecodes.ecodes.get(key_name)
        if self.key_code is None:
            raise ValueError(f"Unbekannter Key-Name: {key_name}")
        self.on_press: AsyncCallback | None = None
        self.on_release: AsyncCallback | None = None
        self._tasks: list[asyncio.Task] = []
        self._devices: list[InputDevice] = []

    async def _handle_key_event(self, key_down: bool) -> None:
        """Ruft den passenden Callback auf."""
        if key_down and self.on_press:
            await self.on_press()
        elif not key_down and self.on_release:
            await self.on_release()

    async def _read_device(self, device: InputDevice) -> None:
        """Liest Events von einem einzelnen Device."""
        async for event in device.async_read_loop():
            if event.type == ecodes.EV_KEY and event.code == self.key_code:
                if event.value == 1:  # Key down
                    logger.debug("%s gedrückt (via %s)", self.key_name, device.path)
                    await self._handle_key_event(key_down=True)
                elif event.value == 0:  # Key up
                    logger.debug("%s losgelassen (via %s)", self.key_name, device.path)
                    await self._handle_key_event(key_down=False)

    def stop(self) -> None:
        """Cancelt laufende Read-Tasks und schließt Devices."""
        for task in self._tasks:
            task.cancel()
        for dev in self._devices:
            try:
                dev.close()
            except Exception:
                pass
        self._tasks = []
        self._devices = []

    async def listen(self) -> None:
        """Lauscht auf evdev-Events der konfigurierten Taste auf allen passenden Devices."""
        self._devices = find_keyboard_devices(self.key_name)
        for dev in self._devices:
            logger.info("Lausche auf %s auf %s (%s)", self.key_name, dev.name, dev.path)
        self._tasks = [asyncio.create_task(self._read_device(dev)) for dev in self._devices]
        await asyncio.gather(*self._tasks, return_exceptions=True)