whisper-local/whisper_local/hotkey/_pynput.py

"""Hotkey-Listener via pynput für Push-to-Talk (Windows / Fallback)."""

import asyncio
import logging
import re

from pynput.keyboard import Key, Listener

from whisper_local.hotkey import AsyncCallback

logger = logging.getLogger(__name__)


def _evdev_to_pynput_key(evdev_name: str) -> Key:
    """Übersetzt evdev Key-Namen (z.B. KEY_F12) ins pynput-Format."""
    match = re.match(r"^KEY_(F\d+)$", evdev_name)
    if match:
        attr = match.group(1).lower()
        key = getattr(Key, attr, None)
        if key is not None:
            return key

    # Versuche direkte Zuordnung (KEY_SPACE -> space, etc.)
    suffix = evdev_name.removeprefix("KEY_").lower()
    key = getattr(Key, suffix, None)
    if key is not None:
        return key

    raise ValueError(f"Unbekannter Key-Name: {evdev_name}")


class PynputHotkeyListener:
    def __init__(self, key_name: str = "KEY_F12"):
        self.key_name = key_name
        self._target_key = _evdev_to_pynput_key(key_name)
        self.on_press: AsyncCallback | None = None
        self.on_release: AsyncCallback | None = None
        self._loop: asyncio.AbstractEventLoop | None = None
        self._pressed = False
        self._stop_event: asyncio.Event | None = None

    def stop(self) -> None:
        """Signalisiert dem listen()-Loop zu beenden."""
        if self._loop is not None and self._stop_event is not None:
            self._loop.call_soon_threadsafe(self._stop_event.set)

    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()

    def _schedule_callback(self, key_down: bool) -> None:
        """Bridged den Thread-Callback zu asyncio."""
        if self._loop is None:
            return
        self._loop.call_soon_threadsafe(
            asyncio.ensure_future,
            self._handle_key_event(key_down),
        )

    def _on_press(self, key) -> None:
        if key == self._target_key and not self._pressed:
            self._pressed = True
            logger.debug("%s gedrückt", self.key_name)
            self._schedule_callback(key_down=True)

    def _on_release(self, key) -> None:
        if key == self._target_key and self._pressed:
            self._pressed = False
            logger.debug("%s losgelassen", self.key_name)
            self._schedule_callback(key_down=False)

    async def listen(self) -> None:
        """Startet den pynput Keyboard-Listener und blockiert async."""
        self._loop = asyncio.get_running_loop()
        self._stop_event = asyncio.Event()

        listener = Listener(on_press=self._on_press, on_release=self._on_release)
        listener.start()
        logger.info("Lausche auf %s via pynput", self.key_name)

        try:
            await self._stop_event.wait()
        finally:
            listener.stop()