Trixy7/trixy_core/hid/service.py

# -*- 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,
            })