feat: add pynput hotkey backend for Windows
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
+54
-2
@@ -4,8 +4,6 @@ from unittest.mock import AsyncMock, MagicMock, patch
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
pytestmark = pytest.mark.skipif(sys.platform != "linux", reason="evdev only available on Linux")
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def listener():
|
def listener():
|
||||||
@@ -13,6 +11,7 @@ def listener():
|
|||||||
return EvdevHotkeyListener(key_name="KEY_F12")
|
return EvdevHotkeyListener(key_name="KEY_F12")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(sys.platform != "linux", reason="evdev only available on Linux")
|
||||||
class TestHotkeyListener:
|
class TestHotkeyListener:
|
||||||
def test_init_stores_key_name(self, listener):
|
def test_init_stores_key_name(self, listener):
|
||||||
assert listener.key_name == "KEY_F12"
|
assert listener.key_name == "KEY_F12"
|
||||||
@@ -37,3 +36,56 @@ class TestHotkeyListener:
|
|||||||
async def test_no_callback_no_error(self, listener):
|
async def test_no_callback_no_error(self, listener):
|
||||||
await listener._handle_key_event(key_down=True)
|
await listener._handle_key_event(key_down=True)
|
||||||
await listener._handle_key_event(key_down=False)
|
await listener._handle_key_event(key_down=False)
|
||||||
|
|
||||||
|
|
||||||
|
class TestPynputKeyMapping:
|
||||||
|
"""Tests für die Key-Name-Übersetzung von evdev zu pynput."""
|
||||||
|
|
||||||
|
def test_map_function_key(self):
|
||||||
|
from whisper_local.hotkey._pynput import _evdev_to_pynput_key
|
||||||
|
from pynput.keyboard import Key
|
||||||
|
assert _evdev_to_pynput_key("KEY_F12") == Key.f12
|
||||||
|
|
||||||
|
def test_map_f1(self):
|
||||||
|
from whisper_local.hotkey._pynput import _evdev_to_pynput_key
|
||||||
|
from pynput.keyboard import Key
|
||||||
|
assert _evdev_to_pynput_key("KEY_F1") == Key.f1
|
||||||
|
|
||||||
|
def test_map_unknown_key_raises(self):
|
||||||
|
from whisper_local.hotkey._pynput import _evdev_to_pynput_key
|
||||||
|
with pytest.raises(ValueError, match="Unbekannter Key-Name"):
|
||||||
|
_evdev_to_pynput_key("KEY_NONEXISTENT_999")
|
||||||
|
|
||||||
|
|
||||||
|
class TestPynputHotkeyListener:
|
||||||
|
def test_init_stores_key_name(self):
|
||||||
|
from whisper_local.hotkey._pynput import PynputHotkeyListener
|
||||||
|
listener = PynputHotkeyListener("KEY_F12")
|
||||||
|
assert listener.key_name == "KEY_F12"
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_handle_key_event_press(self):
|
||||||
|
from whisper_local.hotkey._pynput import PynputHotkeyListener
|
||||||
|
listener = PynputHotkeyListener("KEY_F12")
|
||||||
|
on_press = AsyncMock()
|
||||||
|
listener.on_press = on_press
|
||||||
|
|
||||||
|
await listener._handle_key_event(key_down=True)
|
||||||
|
on_press.assert_awaited_once()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_handle_key_event_release(self):
|
||||||
|
from whisper_local.hotkey._pynput import PynputHotkeyListener
|
||||||
|
listener = PynputHotkeyListener("KEY_F12")
|
||||||
|
on_release = AsyncMock()
|
||||||
|
listener.on_release = on_release
|
||||||
|
|
||||||
|
await listener._handle_key_event(key_down=False)
|
||||||
|
on_release.assert_awaited_once()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_no_callback_no_error(self):
|
||||||
|
from whisper_local.hotkey._pynput import PynputHotkeyListener
|
||||||
|
listener = PynputHotkeyListener("KEY_F12")
|
||||||
|
await listener._handle_key_event(key_down=True)
|
||||||
|
await listener._handle_key_event(key_down=False)
|
||||||
|
|||||||
@@ -0,0 +1,78 @@
|
|||||||
|
"""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
|
||||||
|
|
||||||
|
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:
|
||||||
|
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:
|
||||||
|
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()
|
||||||
|
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 stop_event.wait() # Blockiert bis zum Programmende
|
||||||
|
finally:
|
||||||
|
listener.stop()
|
||||||
Reference in New Issue
Block a user