HTTP beat detection for stock WLED (works on ESP8266)

wled_beat.py

#!/usr/bin/env python3
"""
wled_beat.py - Audio reactive WLED controller
Reads from ALSA loopback, detects beats, controls WLED via HTTP API.

Requirements:
    pip3 install sounddevice numpy requests

Usage:
    python3 wled_beat.py --wled 192.168.1.x --device hw:3,1
"""

import argparse
import time
import threading
import queue
import logging
import sys

import numpy as np
import sounddevice as sd
import requests

# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("wled_beat")


# ---------------------------------------------------------------------------
# Configuration defaults (override with CLI args)
# ---------------------------------------------------------------------------
SAMPLE_RATE     = 44100
BLOCK_SIZE      = 2048       # samples per audio block (~46ms at 44100)
CHANNELS        = 2
WLED_PORT       = 80
WLED_TIMEOUT    = 0.2        # seconds - keep short so we don't block

# Beat detection tuning
BEAT_THRESHOLD  = 1.4        # energy must be this x the rolling average to count as a beat
BEAT_HISTORY    = 22         # number of blocks in the rolling energy window (~1 second)
BEAT_COOLDOWN   = 0.3        # minimum seconds between beats (prevents double-triggers)

# WLED brightness on beat vs decay
BRIGHTNESS_BEAT  = 255
BRIGHTNESS_IDLE  = 80
DECAY_STEP       = 8         # brightness drop per decay tick when no beat
DECAY_INTERVAL   = 0.05      # seconds between decay ticks

# Silence detection - stop sending to WLED after this many seconds of no beats
SILENCE_TIMEOUT  = 10.0


# ---------------------------------------------------------------------------
# WLED HTTP helper
# ---------------------------------------------------------------------------
class WLEDController:
    def __init__(self, host: str, port: int = WLED_PORT):
        self.base = f"http://{host}:{port}/json/state"
        self.info = f"http://{host}:{port}/json/info"
        self.session = requests.Session()

    def set_brightness(self, brightness: int):
        brightness = max(0, min(255, brightness))
        try:
            self.session.post(
                self.base,
                json={"bri": brightness, "on": brightness > 0},
                timeout=WLED_TIMEOUT,
            )
        except requests.RequestException as e:
            log.debug("WLED send failed: %s", e)

    def turn_off(self):
        try:
            self.session.post(
                self.base,
                json={"on": False},
                timeout=WLED_TIMEOUT,
            )
        except requests.RequestException as e:
            log.debug("WLED turn_off failed: %s", e)

    def set_state(self, payload: dict):
        try:
            self.session.post(self.base, json=payload, timeout=WLED_TIMEOUT)
        except requests.RequestException as e:
            log.debug("WLED send failed: %s", e)

    def ping(self) -> bool:
        try:
            r = self.session.get(self.info, timeout=2)
            return r.status_code == 200
        except requests.RequestException:
            return False


# ---------------------------------------------------------------------------
# Beat detector
# ---------------------------------------------------------------------------
class BeatDetector:
    def __init__(self, threshold: float = BEAT_THRESHOLD):
        self.threshold = threshold
        self.energy_history = np.zeros(BEAT_HISTORY)
        self.history_idx = 0
        self.last_beat_time = 0.0

    def process(self, indata: np.ndarray) -> bool:
        """
        Returns True if a beat is detected in this audio block.
        Uses energy-based detection: current block RMS vs rolling average.
        NOTE: must return quickly - runs inside the audio callback thread.
        No I/O, no locks, no sleeps.
        """
        mono = indata.mean(axis=1) if indata.ndim > 1 else indata
        energy = float(np.mean(mono ** 2))
        avg_energy = float(np.mean(self.energy_history)) or 1e-10

        self.energy_history[self.history_idx] = energy
        self.history_idx = (self.history_idx + 1) % BEAT_HISTORY

        now = time.monotonic()
        if (
            energy > avg_energy * self.threshold
            and (now - self.last_beat_time) > BEAT_COOLDOWN
        ):
            self.last_beat_time = now
            log.debug(
                "Beat! energy=%.6f avg=%.6f ratio=%.2f",
                energy, avg_energy, energy / avg_energy,
            )
            return True

        return False


# ---------------------------------------------------------------------------
# Brightness decay thread
# ---------------------------------------------------------------------------
class BrightnessManager:
    """
    Manages WLED brightness with instant-on on beat and smooth decay.
    All WLED HTTP calls happen here, never in the audio callback thread.
    Stops sending to WLED after SILENCE_TIMEOUT seconds of no beats,
    so HA can turn WLED off without this script fighting it.
    """
    def __init__(self, wled: WLEDController):
        self.wled = wled
        self._brightness = BRIGHTNESS_IDLE
        self._lock = threading.Lock()
        self._running = False
        self._active = False          # True when audio is playing
        self._last_beat = time.monotonic()

    def beat(self):
        with self._lock:
            self._brightness = BRIGHTNESS_BEAT
            self._last_beat = time.monotonic()
            if not self._active:
                self._active = True
                log.info("Audio detected - WLED activated")
        self.wled.set_brightness(BRIGHTNESS_BEAT)

    def start(self):
        self._running = True
        t = threading.Thread(target=self._decay_loop, daemon=True)
        t.start()

    def stop(self):
        self._running = False

    def _decay_loop(self):
        while self._running:
            time.sleep(DECAY_INTERVAL)
            with self._lock:
                # Check for silence timeout
                silent_for = time.monotonic() - self._last_beat
                if self._active and silent_for >= SILENCE_TIMEOUT:
                    self._active = False
                    self._brightness = BRIGHTNESS_IDLE
                    log.info("No beats for %.0fs - releasing WLED to HA control", silent_for)
                    self.wled.turn_off()
                    continue

                # Only send decay updates while active
                if self._active and self._brightness > BRIGHTNESS_IDLE:
                    self._brightness = max(BRIGHTNESS_IDLE, self._brightness - DECAY_STEP)
                    self.wled.set_brightness(self._brightness)


# ---------------------------------------------------------------------------
# Main audio capture loop
# ---------------------------------------------------------------------------
def run(wled_host: str, alsa_device: str, threshold: float = BEAT_THRESHOLD):
    log.info("Connecting to WLED at %s ...", wled_host)
    wled = WLEDController(wled_host)
    if not wled.ping():
        log.error("Cannot reach WLED at %s - check IP and that WLED is on.", wled_host)
        sys.exit(1)
    log.info("WLED reachable. Starting audio capture on %s", alsa_device)

    detector = BeatDetector(threshold)
    brightness = BrightnessManager(wled)
    brightness.start()

    beat_count = 0
    start_time = time.monotonic()

    # Queue used to signal beats from the audio callback to the main thread.
    # This keeps all network I/O off the time-critical audio callback thread.
    beat_queue = queue.Queue()

    def audio_callback(indata, frames, time_info, status):
        nonlocal beat_count
        if status:
            log.warning("Audio status: %s", status)
        if detector.process(indata):
            beat_count += 1
            beat_queue.put_nowait(True)  # signal only - no I/O here

    try:
        with sd.InputStream(
            device=alsa_device,
            samplerate=SAMPLE_RATE,
            blocksize=BLOCK_SIZE,
            channels=CHANNELS,
            dtype="float32",
            callback=audio_callback,
        ):
            log.info("Listening for beats. Press Ctrl+C to stop.")
            last_log = start_time
            while True:
                # Drain beat signals and handle them here in the main thread
                # so WLED HTTP calls never block the audio callback
                try:
                    while True:
                        beat_queue.get_nowait()
                        brightness.beat()
                except queue.Empty:
                    pass

                time.sleep(0.02)  # 20ms poll - responsive but lightweight

                # Log stats every 5 seconds
                now = time.monotonic()
                if now - last_log >= 5:
                    elapsed = now - start_time
                    bpm = (beat_count / elapsed) * 60 if elapsed > 0 else 0
                    log.info(
                        "Running %.0fs | beats: %d | est. BPM: %.0f",
                        elapsed, beat_count, bpm,
                    )
                    last_log = now

    except KeyboardInterrupt:
        log.info("Stopped by user.")
    except Exception as e:
        log.error("Audio capture error: %s", e)
        raise
    finally:
        brightness.stop()
        wled.turn_off()


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
    parser = argparse.ArgumentParser(
        description="WLED beat-reactive controller via ALSA loopback"
    )
    parser.add_argument(
        "--wled",
        required=True,
        metavar="IP",
        help="IP address of your WLED device (e.g. 192.168.1.50)",
    )
    parser.add_argument(
        "--device",
        default="hw:3,1",
        metavar="ALSA_DEVICE",
        help="ALSA capture device - loopback read side (default: hw:3,1)",
    )
    parser.add_argument(
        "--threshold",
        type=float,
        default=1.4,
        metavar="FLOAT",
        help="Beat sensitivity multiplier (default: 1.4). "
             "Lower = more sensitive, higher = less sensitive.",
    )
    parser.add_argument(
        "--silence-timeout",
        type=float,
        default=SILENCE_TIMEOUT,
        metavar="SECONDS",
        help=f"Seconds of silence before releasing WLED to HA control (default: {SILENCE_TIMEOUT})",
    )
    parser.add_argument(
        "--debug",
        action="store_true",
        help="Enable debug logging (shows every beat detection)",
    )
    args = parser.parse_args()

    if args.debug:
        logging.getLogger().setLevel(logging.DEBUG)

    global SILENCE_TIMEOUT
    SILENCE_TIMEOUT = args.silence_timeout

    run(args.wled, args.device, args.threshold)


if __name__ == "__main__":
    main()