feat: add stop() method to HotkeyListener protocol and PynputHotkeyListener
This commit is contained in:
@@ -112,3 +112,18 @@ class TestPynputHotkeyListener:
|
||||
# Erneutes Release ohne Press — wird ignoriert
|
||||
listener._on_release(Key.f12)
|
||||
assert listener._loop.call_soon_threadsafe.call_count == 2
|
||||
|
||||
|
||||
class TestPynputHotkeyListenerStop:
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_ends_listen(self):
|
||||
from whisper_local.hotkey._pynput import PynputHotkeyListener
|
||||
from unittest.mock import patch, MagicMock
|
||||
listener = PynputHotkeyListener("KEY_F12")
|
||||
|
||||
with patch("whisper_local.hotkey._pynput.Listener") as mock_listener_cls:
|
||||
mock_listener_cls.return_value = MagicMock()
|
||||
listen_task = asyncio.create_task(listener.listen())
|
||||
await asyncio.sleep(0) # Loop einen Schritt weiter
|
||||
listener.stop()
|
||||
await asyncio.wait_for(listen_task, timeout=1.0)
|
||||
|
||||
@@ -12,6 +12,7 @@ class HotkeyListener(Protocol):
|
||||
on_release: AsyncCallback | None
|
||||
|
||||
async def listen(self) -> None: ...
|
||||
def stop(self) -> None: ...
|
||||
|
||||
|
||||
def create_listener(key_name: str = "KEY_F12") -> HotkeyListener:
|
||||
|
||||
@@ -56,6 +56,10 @@ class EvdevHotkeyListener:
|
||||
logger.debug("%s losgelassen (via %s)", self.key_name, device.path)
|
||||
await self._handle_key_event(key_down=False)
|
||||
|
||||
def stop(self) -> None:
|
||||
"""Stub — evdev-Listener läuft bis zum Prozessende."""
|
||||
pass
|
||||
|
||||
async def listen(self) -> None:
|
||||
"""Lauscht auf evdev-Events der konfigurierten Taste auf allen passenden Devices."""
|
||||
devices = find_keyboard_devices(self.key_name)
|
||||
|
||||
@@ -37,6 +37,12 @@ class PynputHotkeyListener:
|
||||
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."""
|
||||
@@ -69,13 +75,13 @@ class PynputHotkeyListener:
|
||||
async def listen(self) -> None:
|
||||
"""Startet den pynput Keyboard-Listener und blockiert async."""
|
||||
self._loop = asyncio.get_running_loop()
|
||||
stop_event = asyncio.Event()
|
||||
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 stop_event.wait() # Blockiert bis zum Programmende
|
||||
await self._stop_event.wait()
|
||||
finally:
|
||||
listener.stop()
|
||||
|
||||
Reference in New Issue
Block a user