led-sync-studio/src/led_sync/audio/beat_detector.py

"""
BeatDetector: real-time beat/onset detection via sounddevice + aubio.
Implements AUD-03 (system audio capture) and AUD-04 (beat detection).

Threading (D-04/D-05):
  sounddevice _audio_callback runs on OS audio thread.
  Beat events bridged to asyncio via loop.call_soon_threadsafe().
  NEVER call asyncio APIs directly from _audio_callback.

PipeWire (D-06):
  device=None -> PipeWire default input (usually mic).
  For system audio loopback, pass the monitor source name:
    e.g. device="alsa_output.pci-0000_00_1f.3.analog-stereo.monitor"
  List devices: python -c "import sounddevice; print(sounddevice.query_devices())"
"""
import asyncio
import logging
import threading
import time
from typing import Callable, Coroutine

logger = logging.getLogger(__name__)

SAMPLE_RATE = 44100
HOP_SIZE = 512          # latency = 512/44100 ≈ 11ms
BUF_SIZE = 1024         # aubio FFT window
BEAT_HOLD_MS = 200      # suppress double-triggers within 200ms


class BeatDetector:
    """
    Real-time beat detector. Runs sounddevice InputStream on OS audio thread.
    Posts beat events into asyncio event loop via call_soon_threadsafe (D-05).

    Usage:
      detector = BeatDetector(loop=asyncio.get_running_loop(), on_beat=my_async_fn)
      detector.start()          # starts audio capture
      ...
      detector.stop()           # stops capture, closes stream
    """

    def __init__(
        self,
        loop: asyncio.AbstractEventLoop,
        on_beat: Callable[[float], Coroutine],
    ):
        self._loop = loop
        self._on_beat = on_beat
        self._stream = None
        self._lock = threading.Lock()
        self._beat_count: int = 0
        self._bpm: float = 0.0
        self._last_beat_time: float = 0.0  # for hold suppression
        self._tempo = None  # aubio tempo object; created in start()

    @property
    def beat_count(self) -> int:
        with self._lock:
            return self._beat_count

    @property
    def bpm(self) -> float:
        with self._lock:
            return self._bpm

    def start(self, device=None) -> None:
        """
        Open sounddevice InputStream and begin beat detection.
        device=None uses PipeWire default. Pass monitor source name for loopback.
        Raises RuntimeError if sounddevice or aubio is unavailable.
        """
        try:
            import sounddevice as sd
            import aubio
            import numpy as np
        except ImportError as e:
            raise RuntimeError(
                f"Beat detection requires sounddevice, aubio, and numpy: {e}"
            ) from e

        # Create aubio tempo detector
        self._tempo = aubio.tempo(
            method="default",
            buf_size=BUF_SIZE,
            hop_size=HOP_SIZE,
            samplerate=SAMPLE_RATE,
        )

        # Capture hold time in seconds for double-trigger suppression
        self._hold_seconds = BEAT_HOLD_MS / 1000.0
        self._last_beat_time = 0.0

        def _audio_callback(indata, frames, time_info, status):
            """
            Called on OS audio thread by sounddevice.
            MUST NOT call asyncio APIs directly (D-05).
            """
            if status:
                logger.debug("sounddevice status: %s", status)

            # Mono float32 samples for aubio
            samples = indata[:, 0].astype("float32")
            is_beat = self._tempo(samples)

            if is_beat[0]:
                beat_time = time_info.inputBufferAdcTime
                # Double-trigger suppression
                now = time.monotonic()
                with self._lock:
                    if now - self._last_beat_time < self._hold_seconds:
                        return
                    self._last_beat_time = now
                    self._beat_count += 1
                    self._bpm = float(self._tempo.get_bpm())

                logger.debug("Beat detected at %.3fs, BPM=%.1f", beat_time, self._bpm)

                # Bridge to asyncio: non-blocking, thread-safe (D-05)
                self._loop.call_soon_threadsafe(
                    self._loop.create_task,
                    self._on_beat(beat_time),
                )

        try:
            self._stream = sd.InputStream(
                samplerate=SAMPLE_RATE,
                blocksize=HOP_SIZE,
                channels=1,
                dtype="float32",
                callback=_audio_callback,
                device=device,
            )
            self._stream.start()
            logger.info(
                "BeatDetector started: device=%s, sample_rate=%d, hop=%d (%.1fms latency)",
                device, SAMPLE_RATE, HOP_SIZE, HOP_SIZE / SAMPLE_RATE * 1000,
            )
        except Exception as e:
            raise RuntimeError(f"Failed to open audio input stream: {e}") from e

    def stop(self) -> None:
        """Stop beat detection and close the sounddevice stream."""
        if self._stream is not None:
            try:
                self._stream.stop()
                self._stream.close()
            except Exception:
                pass
            self._stream = None
            logger.info("BeatDetector stopped (total beats: %d)", self.beat_count)