# -*- coding: utf-8 -*- """ HID Media Key Service. Lauscht auf USB-HID-Eingabegeraete (Konferenzmikrofone, Headsets) und wandelt Media-Tasten in Trixy-Events um. Unterstuetzt u.a.: - ANKER PowerConf S330: MUTE, VOL+/-, MICMUTE, PLAY, NEXT/PREV + LED - Jabra Speak: MUTE, VOL+/-, HOOK + LED - Poly/Plantronics: MUTE, VOL+/-, HOOK - Generische USB-HID Media Keys Benoetigt: evdev (pip install evdev) Benoetigt: Leserechte auf /dev/input/event* (Gruppe 'input' oder root) """ from __future__ import annotations import asyncio import time from dataclasses import dataclass, field from typing import TYPE_CHECKING, Callable, Awaitable from trixy_core.service import IService, ServicePriority, ServiceGroup from trixy_core.hid.events import HIDEvent, HIDKeyType, HIDLedType from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn if TYPE_CHECKING: from trixy_core.application import BaseApplication # evdev Key-Codes → HIDKeyType Mapping _KEY_MAP: dict[int, HIDKeyType] = {} # evdev LED-Codes → HIDLedType Mapping _LED_MAP: dict[int, HIDLedType] = {} try: from evdev import ecodes _KEY_MAP = { ecodes.KEY_VOLUMEUP: HIDKeyType.VOLUME_UP, ecodes.KEY_VOLUMEDOWN: HIDKeyType.VOLUME_DOWN, ecodes.KEY_MUTE: HIDKeyType.MUTE, ecodes.KEY_PLAY: HIDKeyType.PLAY, ecodes.KEY_PLAYPAUSE: HIDKeyType.PLAY_PAUSE, ecodes.KEY_PAUSECD: HIDKeyType.PLAY_PAUSE, ecodes.KEY_PLAYCD: HIDKeyType.PLAY, ecodes.KEY_NEXTSONG: HIDKeyType.NEXT_TRACK, ecodes.KEY_PREVIOUSSONG: HIDKeyType.PREV_TRACK, ecodes.KEY_PHONE: HIDKeyType.HOOK, ecodes.KEY_MEDIA: HIDKeyType.HOOK, } # Geraete-spezifische Tasten if hasattr(ecodes, "KEY_MICMUTE"): _KEY_MAP[ecodes.KEY_MICMUTE] = HIDKeyType.MIC_MUTE if hasattr(ecodes, "KEY_HOOKSWITCH"): _KEY_MAP[ecodes.KEY_HOOKSWITCH] = HIDKeyType.HOOK _LED_MAP = { ecodes.LED_MUTE: HIDLedType.MUTE, } if hasattr(ecodes, "LED_MAIL"): _LED_MAP[ecodes.LED_MAIL] = HIDLedType.MAIL except ImportError: pass @dataclass class HIDServiceConfig: """Konfiguration fuer den HID-Service.""" # Automatisch alle HID-Geraete mit Media-Tasten erkennen auto_detect: bool = True # Manuelle Geraete-Pfade (zusaetzlich zur Auto-Erkennung) device_paths: list[str] = field(default_factory=list) # Lautstaerke-Schrittweite (0.0-1.0) volume_step: float = 0.05 # Hook-Taste als Wakeword-Trigger hook_triggers_wakeword: bool = True # MicMute-Taste: kurzer Druck = Mute, langer Druck = Wakeword-Trigger micmute_long_press_as_wakeword: bool = True # PLAY-Taste als Wakeword-Trigger (fuer Geraete ohne Hook) play_triggers_wakeword: bool = True # Langer Druck Schwelle (Sekunden) long_press_seconds: float = 2.0 @dataclass class _LedDevice: """Ein Geraet mit LED-Steuerung.""" device: object # evdev.InputDevice led_codes: dict[HIDLedType, int] = field(default_factory=dict) class HIDService(IService): """ Service fuer USB-HID Media Keys und LED-Steuerung. Erkennt automatisch HID-Geraete, lauscht auf Tasten, misst Drueck-Dauer und steuert LEDs. """ PRIORITY = ServicePriority.OPTIONAL GROUP = ServiceGroup.AUDIO DEPENDENCIES = [] def __init__( self, application: "BaseApplication", config: HIDServiceConfig | None = None, ) -> None: super().__init__(application) self._config = config or HIDServiceConfig() self._tasks: list[asyncio.Task] = [] self._key_devices: list = [] # evdev Geraete mit Tasten self._led_devices: list[_LedDevice] = [] # evdev Geraete mit LEDs self._hidraw_readers: list = [] # HidrawReader fuer Telephony self._callbacks: list[Callable[[HIDEvent], Awaitable[None]]] = [] self._available = False self._mic_muted = False # Zeitstempel fuer Hold-Duration self._press_times: dict[tuple[str, int], float] = {} # === Properties === @property def available(self) -> bool: """True wenn evdev verfuegbar und Geraete gefunden.""" return self._available @property def device_count(self) -> int: """Anzahl der verbundenen HID-Geraete (Tasten).""" return len(self._key_devices) @property def device_names(self) -> list[str]: """Namen der verbundenen HID-Geraete.""" return [d.name for d in self._key_devices] @property def has_leds(self) -> bool: """True wenn LED-steuerbare Geraete vorhanden.""" return len(self._led_devices) > 0 @property def mic_muted(self) -> bool: """Mikrofon-Stummschaltungsstatus.""" return self._mic_muted # === Service Lifecycle === async def start(self) -> None: """Startet den HID-Service.""" try: import evdev self._available = True except ImportError: pdebug("evdev nicht installiert — HID Media Keys deaktiviert") return key_devices, led_devices = self._find_devices() if not key_devices and not led_devices: pdebug("Keine HID-Geraete gefunden") return self._key_devices = key_devices self._led_devices = led_devices for dev in key_devices: task = asyncio.create_task(self._listen_device(dev)) self._tasks.append(task) pinfo(f"HID-Tasten: {dev.name} ({dev.path})") for ld in led_devices: leds = [lt.value for lt in ld.led_codes] pinfo(f"HID-LEDs: {ld.device.name} ({ld.device.path}) — {', '.join(leds)}") # Hidraw-Reader fuer Telephony-Geraete (Hook, Phone Mute) try: from trixy_core.hid.hidraw import find_hidraw_devices, HidrawReader hidraw_devs = find_hidraw_devices() for hdev in hidraw_devs: reader = HidrawReader( hdev.path, hdev.name, on_key=self._on_hidraw_key, ) await reader.start() if reader.is_running: self._hidraw_readers.append(reader) except Exception as e: pdebug(f"HID-Raw nicht verfuegbar: {e}") total = len(key_devices) + len(self._hidraw_readers) if total == 0 and not led_devices: pdebug("Keine HID-Geraete gefunden") return self._available = True pinfo( f"HID-Service gestartet: {len(key_devices)} evdev, " f"{len(self._hidraw_readers)} hidraw, " f"{len(led_devices)} LED-Geraet(e)" ) async def stop(self) -> None: """Stoppt den HID-Service.""" for task in self._tasks: task.cancel() self._tasks.clear() for dev in self._key_devices: try: dev.close() except Exception: pass self._key_devices.clear() # LED-Devices nicht schliessen wenn sie auch Key-Device sind self._led_devices.clear() # Hidraw-Reader stoppen for reader in self._hidraw_readers: await reader.stop() self._hidraw_readers.clear() if self._available: pdebug("HID-Service gestoppt") def on_event(self, callback: Callable[[HIDEvent], Awaitable[None]]) -> None: """Registriert einen Callback fuer HID-Events.""" self._callbacks.append(callback) # === LED-Steuerung === async def _on_hidraw_key(self, event: HIDEvent) -> None: """Callback fuer hidraw-Events (Telephony-Tasten).""" # Gleiche Verarbeitung wie evdev-Events if event.pressed: await self._handle_press(event) else: await self._handle_release(event) # Callbacks for callback in self._callbacks: try: await callback(event) except Exception as e: perror(f"HID-Callback Fehler: {e}") # Allgemeines Event events = getattr(self._application, "events", None) if events: await events.emit("hid_key", { "key": event.key_type.value, "pressed": event.pressed, "hold_seconds": event.hold_seconds, "device": event.device_name, "source": "hidraw", }) async def set_led(self, led_type: HIDLedType, on: bool) -> bool: """ Schaltet eine LED auf einem HID-Geraet. Args: led_type: Welche LED (MUTE, MAIL) on: True = an, False = aus Returns: True wenn erfolgreich """ import evdev for ld in self._led_devices: led_code = ld.led_codes.get(led_type) if led_code is not None: try: ld.device.write(evdev.ecodes.EV_LED, led_code, 1 if on else 0) ld.device.syn() pdebug(f"HID LED {led_type.value}: {'AN' if on else 'AUS'} ({ld.device.name})") return True except Exception as e: perror(f"HID LED-Fehler: {e}") return False async def set_mute_led(self, on: bool) -> bool: """Schaltet die Mute-LED (evdev oder hidraw).""" # Zuerst evdev versuchen if await self.set_led(HIDLedType.MUTE, on): return True # Fallback: hidraw for reader in self._hidraw_readers: if reader.set_led("mute", on): return True return False async def set_ring_led(self, on: bool) -> bool: """Schaltet die Ring-LED (nur hidraw).""" for reader in self._hidraw_readers: if reader.set_led("ring", on): return True return False async def set_hold_led(self, on: bool) -> bool: """Schaltet die Hold-LED (nur hidraw).""" for reader in self._hidraw_readers: if reader.set_led("hold", on): return True return False async def set_status_led(self, on: bool) -> bool: """Schaltet die Status-LED (LED_MAIL via evdev).""" return await self.set_led(HIDLedType.MAIL, on) # === Geraete-Erkennung === def _find_devices(self) -> tuple[list, list[_LedDevice]]: """Findet HID-Geraete mit Tasten und/oder LEDs.""" import evdev key_devices = [] led_devices: list[_LedDevice] = [] seen_paths: set[str] = set() # Manuelle Pfade for path in self._config.device_paths: try: dev = evdev.InputDevice(path) self._classify_device(dev, key_devices, led_devices) seen_paths.add(path) except (FileNotFoundError, PermissionError) as e: pwarn(f"HID-Geraet nicht verfuegbar: {path} ({e})") # Auto-Detect if self._config.auto_detect: for path in sorted(evdev.list_devices()): if path in seen_paths: continue try: dev = evdev.InputDevice(path) added = self._classify_device(dev, key_devices, led_devices) if not added: dev.close() except (PermissionError, OSError): pass return key_devices, led_devices def _classify_device( self, dev, key_devices: list, led_devices: list[_LedDevice], ) -> bool: """Klassifiziert ein Geraet (Tasten, LEDs oder beides). Gibt True zurueck wenn relevant.""" import evdev caps = dev.capabilities(verbose=False) added = False # Tasten pruefen key_caps = caps.get(evdev.ecodes.EV_KEY, []) has_keys = any(kc in _KEY_MAP for kc in key_caps) if has_keys: key_devices.append(dev) keys = [_KEY_MAP[kc].value for kc in key_caps if kc in _KEY_MAP] pdebug(f"HID Tasten: {dev.name} ({dev.path}) — {', '.join(keys)}") added = True # LEDs pruefen led_caps = caps.get(evdev.ecodes.EV_LED, []) if led_caps: led_codes = {} for lc in led_caps: led_type = _LED_MAP.get(lc) if led_type: led_codes[led_type] = lc if led_codes: led_devices.append(_LedDevice(device=dev, led_codes=led_codes)) pdebug(f"HID LEDs: {dev.name} ({dev.path}) — {[lt.value for lt in led_codes]}") added = True return added # === Event-Verarbeitung === async def _listen_device(self, device) -> None: """Lauscht auf Events eines einzelnen Geraets.""" import evdev try: async for event in device.async_read_loop(): if event.type != evdev.ecodes.EV_KEY: continue key_type = _KEY_MAP.get(event.code) if key_type is None: continue # Nur key-down (1) und key-up (0), kein auto-repeat (2) if event.value not in (0, 1): continue pressed = event.value == 1 device_key = (device.path, event.code) hold_seconds = 0.0 if pressed: self._press_times[device_key] = time.monotonic() else: press_time = self._press_times.pop(device_key, None) if press_time: hold_seconds = time.monotonic() - press_time hid_event = HIDEvent( key_type=key_type, device_name=device.name, pressed=pressed, hold_seconds=hold_seconds, ) # Debug-Ausgabe if pressed: pinfo(f"HID [{device.name}] {key_type.value} GEDRUECKT") else: pinfo(f"HID [{device.name}] {key_type.value} LOSGELASSEN ({hold_seconds:.1f}s)") # Aktionen bei key-down if pressed: await self._handle_press(hid_event) # Aktionen bei key-up (langer Druck) if not pressed: await self._handle_release(hid_event) # Callbacks for callback in self._callbacks: try: await callback(hid_event) except Exception as e: perror(f"HID-Callback Fehler: {e}") # Allgemeines Event events = getattr(self._application, "events", None) if events: await events.emit("hid_key", { "key": key_type.value, "pressed": pressed, "hold_seconds": hold_seconds, "device": device.name, }) except asyncio.CancelledError: pass except OSError: pwarn(f"HID-Geraet getrennt: {device.name}") async def _handle_press(self, event: HIDEvent) -> None: """Verarbeitet Tastendruck (key-down).""" events = getattr(self._application, "events", None) if not events: return if event.key_type == HIDKeyType.VOLUME_UP: step_pct = int(self._config.volume_step * 100) pinfo(f"HID: Lautstaerke +{step_pct}%") await events.emit("volume_change", { "direction": "up", "step": self._config.volume_step, "device": event.device_name, }) elif event.key_type == HIDKeyType.VOLUME_DOWN: step_pct = int(self._config.volume_step * 100) pinfo(f"HID: Lautstaerke -{step_pct}%") await events.emit("volume_change", { "direction": "down", "step": self._config.volume_step, "device": event.device_name, }) elif event.key_type == HIDKeyType.MUTE: pinfo("HID: Lautsprecher-Mute umschalten") await events.emit("mute_toggle", { "device": event.device_name, "type": "speaker", }) elif event.key_type == HIDKeyType.MIC_MUTE: # Mikrofon-Mute: Toggle und LED aktualisieren self._mic_muted = not self._mic_muted pinfo(f"HID: Mikrofon {'STUMM' if self._mic_muted else 'AKTIV'}") await self.set_mute_led(self._mic_muted) await events.emit("mute_toggle", { "device": event.device_name, "type": "microphone", "muted": self._mic_muted, }) elif event.key_type == HIDKeyType.HOOK: if self._config.hook_triggers_wakeword: pinfo("HID: Hook-Taste → Wakeword-Trigger") await events.emit("wakeword_manual_trigger", { "device": event.device_name, "forced": True, }) elif event.key_type == HIDKeyType.PLAY: if self._config.play_triggers_wakeword: pinfo("HID: Play-Taste → Wakeword-Trigger") await events.emit("wakeword_manual_trigger", { "device": event.device_name, "forced": True, }) else: await events.emit("playback_toggle", {"device": event.device_name}) elif event.key_type == HIDKeyType.PLAY_PAUSE: await events.emit("playback_toggle", {"device": event.device_name}) elif event.key_type == HIDKeyType.NEXT_TRACK: pinfo("HID: Naechster Track") await events.emit("music_next", {"device": event.device_name}) elif event.key_type == HIDKeyType.PREV_TRACK: pinfo("HID: Vorheriger Track") await events.emit("music_previous", {"device": event.device_name}) async def _handle_release(self, event: HIDEvent) -> None: """Verarbeitet Tastenloslassen (key-up) — prueft langen Druck.""" if event.hold_seconds < self._config.long_press_seconds: return events = getattr(self._application, "events", None) if not events: return pinfo( f"HID: Langer Druck {event.key_type.value} " f"({event.hold_seconds:.1f}s) von {event.device_name}" ) await events.emit("hid_long_press", { "key": event.key_type.value, "hold_seconds": event.hold_seconds, "device": event.device_name, }) # MicMute lang gedrueckt → Wakeword-Trigger if (event.key_type == HIDKeyType.MIC_MUTE and self._config.micmute_long_press_as_wakeword): pinfo("HID: MicMute langer Druck → Wakeword-Trigger") # Mute aufheben fuer Aufnahme if self._mic_muted: self._mic_muted = False await self.set_mute_led(False) await events.emit("wakeword_manual_trigger", { "device": event.device_name, "forced": True, })