80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
"""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)
|