Trixy7/trixy_core/hid/hidraw.py

# -*- coding: utf-8 -*-
"""
HID Raw Device Reader fuer Telephony-Tasten und LED-Steuerung.

Manche USB-Geraete (z.B. ANKER PowerConf S330) senden
Telephony-Tasten (Hook, Phone Mute) ueber die HID Telephony
Usage Page (0x0B), die von Linux evdev nicht erkannt wird.

Dieses Modul liest /dev/hidraw* direkt und dekodiert die Reports.

Report-Struktur (ANKER PowerConf S330):
  Report ID 1 (Consumer): Play, Next, Prev, Vol+, Vol-, Mute
                           → wird von evdev erkannt
  Report ID 2 (Telephony): Hook Switch, Flash, Drop, Phone Mute
                           → NUR ueber hidraw lesbar
  Report ID 3 (LED Out): Ring LED
  Report ID 4 (LED Out): Mute LED
  Report ID 5 (LED Out): Hold LED
"""

from __future__ import annotations

import asyncio
import os
import struct
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Callable, Awaitable

from trixy_core.hid.events import HIDEvent, HIDKeyType, HIDLedType
from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn

if TYPE_CHECKING:
    pass


@dataclass
class HidrawDeviceInfo:
    """Informationen ueber ein hidraw-Geraet."""
    path: str                  # z.B. /dev/hidraw0
    name: str = ""
    vendor_id: int = 0
    product_id: int = 0
    has_telephony: bool = False
    has_leds: bool = False


# Telephony Report ID 2 Bit-Mapping
_TELEPHONY_BITS: dict[int, HIDKeyType] = {
    0: HIDKeyType.HOOK,         # Bit 0: Hook Switch (Call/Answer)
    # Bit 1: Flash
    # Bit 2: Drop
    3: HIDKeyType.MIC_MUTE,     # Bit 3: Phone Mute
}

# LED Report IDs
_LED_REPORTS: dict[int, str] = {
    3: "ring",       # Ring-LED
    4: "mute",       # Mute-LED
    5: "hold",       # Hold-LED
}


class HidrawReader:
    """
    Liest HID-Reports direkt von /dev/hidraw*.

    Erkennt Telephony-Tasten die evdev nicht sieht
    und steuert LEDs ueber Output-Reports.

    Beispiel:
        ```python
        reader = HidrawReader("/dev/hidraw0", "ANKER PowerConf S330")
        reader.on_key = my_callback
        await reader.start()

        # LED steuern
        reader.set_led("mute", True)
        ```
    """

    def __init__(
        self,
        device_path: str,
        device_name: str = "",
        on_key: Callable[[HIDEvent], Awaitable[None]] | None = None,
    ) -> None:
        self._path = device_path
        self._name = device_name or device_path
        self._on_key = on_key
        self._fd: int | None = None
        self._task: asyncio.Task | None = None
        self._running = False

        # State: welche Tasten sind gerade gedrueckt (fuer press/release Erkennung)
        self._prev_report2: int = 0
        # Zeitstempel fuer Hold-Duration
        self._press_times: dict[int, float] = {}

    @property
    def is_running(self) -> bool:
        return self._running

    async def start(self) -> None:
        """Startet das Lesen von hidraw."""
        try:
            # Non-blocking oeffnen
            self._fd = os.open(self._path, os.O_RDWR | os.O_NONBLOCK)
            self._running = True
            self._task = asyncio.create_task(self._read_loop())
            pinfo(f"HID-Raw gestartet: {self._name} ({self._path})")
        except PermissionError:
            pwarn(f"HID-Raw: Keine Berechtigung fuer {self._path} (root oder Gruppe 'input' noetig)")
        except OSError as e:
            pwarn(f"HID-Raw: Kann {self._path} nicht oeffnen: {e}")

    async def stop(self) -> None:
        """Stoppt das Lesen."""
        self._running = False
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass
            self._task = None

        if self._fd is not None:
            try:
                os.close(self._fd)
            except OSError:
                pass
            self._fd = None

    def set_led(self, led_name: str, on: bool) -> bool:
        """
        Setzt eine LED ueber einen Output-Report.

        Args:
            led_name: "mute", "ring" oder "hold"
            on: True = an, False = aus

        Returns:
            True bei Erfolg
        """
        if self._fd is None:
            return False

        # Report-ID fuer die LED finden
        report_id = None
        for rid, name in _LED_REPORTS.items():
            if name == led_name:
                report_id = rid
                break

        if report_id is None:
            pdebug(f"HID-Raw: Unbekannte LED '{led_name}'")
            return False

        try:
            # Output-Report: [Report-ID, Value (1 Bit + 7 Padding)]
            value = 0x01 if on else 0x00
            data = struct.pack("BB", report_id, value)
            os.write(self._fd, data)
            pdebug(f"HID-Raw LED '{led_name}': {'AN' if on else 'AUS'}")
            return True
        except OSError as e:
            perror(f"HID-Raw LED-Fehler: {e}")
            return False

    async def _read_loop(self) -> None:
        """Haupt-Leseschleife fuer hidraw."""
        import time

        loop = asyncio.get_running_loop()

        try:
            while self._running:
                try:
                    # Non-blocking read via Event-Loop
                    data = await loop.run_in_executor(None, self._read_report)
                    if data:
                        await self._process_report(data)
                except asyncio.CancelledError:
                    break
                except OSError:
                    if self._running:
                        pwarn(f"HID-Raw: Lesefehler auf {self._path}")
                    break
        except asyncio.CancelledError:
            pass

    def _read_report(self) -> bytes | None:
        """Liest einen Report (blockierend, wird in Executor ausgefuehrt)."""
        import select
        import time

        if self._fd is None:
            return None

        # select mit Timeout damit wir nicht ewig blockieren
        try:
            ready, _, _ = select.select([self._fd], [], [], 0.5)
            if ready:
                return os.read(self._fd, 64)
        except (OSError, ValueError):
            pass
        return None

    async def _process_report(self, data: bytes) -> None:
        """Verarbeitet einen HID-Report."""
        import time

        if len(data) < 2:
            return

        report_id = data[0]
        report_data = data[1]

        # Nur Report ID 2 (Telephony) verarbeiten
        # Report ID 1 (Consumer) wird von evdev behandelt
        if report_id != 2:
            return

        # Bit-Aenderungen erkennen (press/release)
        changed = report_data ^ self._prev_report2
        self._prev_report2 = report_data

        if changed == 0:
            return

        now = time.monotonic()

        for bit, key_type in _TELEPHONY_BITS.items():
            mask = 1 << bit
            if not (changed & mask):
                continue

            pressed = bool(report_data & mask)
            hold_seconds = 0.0

            if pressed:
                self._press_times[bit] = now
            else:
                press_time = self._press_times.pop(bit, None)
                if press_time:
                    hold_seconds = now - press_time

            event = HIDEvent(
                key_type=key_type,
                device_name=self._name,
                pressed=pressed,
                hold_seconds=hold_seconds,
            )

            if pressed:
                pinfo(f"HID-Raw [{self._name}] {key_type.value} GEDRUECKT")
            else:
                pinfo(f"HID-Raw [{self._name}] {key_type.value} LOSGELASSEN ({hold_seconds:.1f}s)")

            if self._on_key:
                await self._on_key(event)


def find_hidraw_devices() -> list[HidrawDeviceInfo]:
    """
    Findet hidraw-Geraete mit Telephony-Support.

    Scannt /sys/class/hidraw/*/device/report_descriptor
    nach der Telephony Usage Page (0x050B).
    """
    devices: list[HidrawDeviceInfo] = []

    hidraw_dir = Path("/sys/class/hidraw")
    if not hidraw_dir.exists():
        return devices

    for entry in sorted(hidraw_dir.iterdir()):
        dev_path = f"/dev/{entry.name}"
        if not os.path.exists(dev_path):
            continue

        info = HidrawDeviceInfo(path=dev_path)

        # Report-Descriptor lesen
        desc_path = entry / "device" / "report_descriptor"
        if desc_path.exists():
            try:
                desc = desc_path.read_bytes()
                # Telephony Usage Page: 05 0B
                info.has_telephony = b"\x05\x0b" in desc
                # LED Usage Page: 05 08
                info.has_leds = b"\x05\x08" in desc
            except OSError:
                continue

        # Geraetename
        name_path = entry / "device" / "name"
        if not name_path.exists():
            # Alternativ: uevent parsen
            uevent_path = entry / "device" / "uevent"
            if uevent_path.exists():
                try:
                    for line in uevent_path.read_text().splitlines():
                        if line.startswith("HID_NAME="):
                            info.name = line.split("=", 1)[1]
                            break
                except OSError:
                    pass

        if info.has_telephony:
            pdebug(f"HID-Raw Telephony: {info.name} ({dev_path})")
            devices.append(info)

    return devices