Trixy7/trixy_core/audio/microphone.py

# -*- coding: utf-8 -*-
"""
Mikrofon-Aufnahme via sounddevice.

Erfasst Audio vom Mikrofon in einem PortAudio-Thread
und leitet es per Callback an den Consumer weiter.
16kHz, 16-bit, mono, 80ms Frames (1280 Samples).
"""

from typing import Callable

import numpy as np

from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn

try:
    import sounddevice as sd
    HAS_SOUNDDEVICE = True
except ImportError:
    sd = None
    HAS_SOUNDDEVICE = False


class MicrophoneCapture:
    """
    Mikrofon-Aufnahme via sounddevice.InputStream.

    Callback läuft in PortAudio-Thread (eigener Thread).
    16kHz, 16-bit, mono, 80ms Frames (1280 Samples).
    """

    def __init__(
        self,
        sample_rate: int = 16000,
        channels: int = 1,
        frame_ms: int = 80,
        device: str | int | None = None,
    ) -> None:
        """
        Initialisiert MicrophoneCapture.

        Args:
            sample_rate: Abtastrate in Hz
            channels: Anzahl Kanäle (1=Mono)
            frame_ms: Frame-Größe in Millisekunden
            device: Audio-Eingabegerät (None=Standard)
        """
        if not HAS_SOUNDDEVICE:
            raise ImportError("sounddevice ist nicht installiert: pip install sounddevice")

        self._sample_rate = sample_rate
        self._channels = channels
        self._frame_ms = frame_ms
        self._frame_size = int(sample_rate * frame_ms / 1000)  # 1280 bei 16kHz/80ms
        self._device = device

        self._stream: sd.InputStream | None = None
        self._callback: Callable[[bytes], None] | None = None

    def set_callback(self, callback: Callable[[bytes], None]) -> None:
        """
        Setzt den Audio-Callback.

        Args:
            callback: Funktion die Audio-Bytes empfängt (16-bit PCM)
        """
        self._callback = callback

    def start(self) -> None:
        """Erstellt und startet InputStream mit _audio_callback."""
        if self._stream is not None:
            pwarn("Mikrofon läuft bereits")
            return

        if not self._callback:
            raise RuntimeError("Kein Callback gesetzt - set_callback() aufrufen")

        # Gerät auflösen (Name → Index)
        device = self._resolve_device(self._device)

        self._stream = sd.InputStream(
            samplerate=self._sample_rate,
            channels=self._channels,
            dtype="int16",
            blocksize=self._frame_size,
            device=device,
            callback=self._audio_callback,
        )
        self._stream.start()
        pinfo(f"Mikrofonaufnahme gestartet ({self._sample_rate}Hz, {self._channels}ch, {self._frame_ms}ms Frames)")

    def stop(self) -> None:
        """Stoppt und schließt Stream."""
        if self._stream is not None:
            try:
                self._stream.stop()
                self._stream.close()
            except Exception as e:
                pdebug(f"Mikrofon-Stop-Fehler: {e}")
            finally:
                self._stream = None
            pdebug("Mikrofonaufnahme gestoppt")

    def _audio_callback(self, indata: np.ndarray, frames: int, time_info, status) -> None:
        """
        sounddevice-Callback: numpy → bytes → self._callback.

        Läuft in PortAudio-Thread, darf nicht blockieren.
        """
        if status:
            pdebug(f"Mikrofon-Status: {status}")

        if self._callback:
            # numpy int16 Array → bytes (Little-Endian 16-bit PCM)
            audio_bytes = indata.tobytes()
            try:
                self._callback(audio_bytes)
            except Exception:
                # Callback-Fehler nicht in PortAudio-Thread propagieren
                pass

    @property
    def is_running(self) -> bool:
        """Prüft ob Aufnahme läuft."""
        return self._stream is not None and self._stream.active

    @property
    def sample_rate(self) -> int:
        """Abtastrate."""
        return self._sample_rate

    @property
    def channels(self) -> int:
        """Anzahl Kanäle."""
        return self._channels

    @property
    def frame_size(self) -> int:
        """Frame-Größe in Samples."""
        return self._frame_size

    @staticmethod
    def _resolve_device(device: str | int | None) -> int | None:
        """
        Löst Gerätenamen zu Index auf.

        Args:
            device: Gerätename, Index oder None

        Returns:
            Geräte-Index oder None (Standard)
        """
        if device is None or device == "":
            return None

        if isinstance(device, int):
            return device

        # Name → Index suchen
        try:
            devices = sd.query_devices()
            for i, dev in enumerate(devices):
                if device.lower() in dev["name"].lower() and dev["max_input_channels"] > 0:
                    pdebug(f"Mikrofon-Gerät gefunden: [{i}] {dev['name']}")
                    return i
            pwarn(f"Mikrofon-Gerät '{device}' nicht gefunden, verwende Standard")
        except Exception as e:
            pwarn(f"Gerätesuche fehlgeschlagen: {e}")

        return None

    @staticmethod
    def list_devices() -> list[dict]:
        """
        Listet verfügbare Eingabegeräte auf.

        Returns:
            Liste von Geräte-Informationen
        """
        if not HAS_SOUNDDEVICE:
            return []

        result = []
        try:
            devices = sd.query_devices()
            for i, dev in enumerate(devices):
                if dev["max_input_channels"] > 0:
                    result.append({
                        "index": i,
                        "name": dev["name"],
                        "channels": dev["max_input_channels"],
                        "sample_rate": dev["default_samplerate"],
                    })
        except Exception:
            pass

        return result