ledgrab/server/tests/test_reactive_palette.py

"""Tests for audio-reactive palette modulation on procedural effects.

Covers the model round-trip, the AudioEnergyTap (resolve/acquire/energy via
fakes), and the per-frame brightness/saturation modulation applied to a
rendered effect frame.
"""

from __future__ import annotations

from datetime import datetime, timezone

import numpy as np

from ledgrab.core.processing.audio_energy_tap import AudioEnergyTap
from ledgrab.core.processing.effect_stream import EffectColorStripStream
from ledgrab.storage.color_strip_source import EffectColorStripSource


def _make_source(**overrides) -> EffectColorStripSource:
    base = dict(
        id="fx1",
        name="fx",
        source_type="effect",
        created_at=datetime.now(timezone.utc),
        updated_at=datetime.now(timezone.utc),
        effect_type="plasma",
    )
    base.update(overrides)
    return EffectColorStripSource.create_from_kwargs(**base)


# ── Model ───────────────────────────────────────────────────────────────


class TestModel:
    def test_defaults(self):
        s = _make_source()
        assert s.audio_reactive is False
        assert s.reactive_mode == "brightness"
        assert s.reactive_audio_source_id == ""

    def test_round_trip_preserves_reactive_fields(self):
        s = _make_source(
            audio_reactive=True,
            reactive_audio_source_id="as_42",
            reactive_mode="both",
            reactive_intensity=0.5,
        )
        back = EffectColorStripSource.from_dict(s.to_dict())
        assert back.audio_reactive is True
        assert back.reactive_audio_source_id == "as_42"
        assert back.reactive_mode == "both"
        assert back.reactive_intensity.value == 0.5

    def test_apply_update_changes_reactive_fields(self):
        s = _make_source()
        s.apply_update(audio_reactive=True, reactive_mode="saturation", reactive_intensity=0.9)
        assert s.audio_reactive is True
        assert s.reactive_mode == "saturation"
        assert s.reactive_intensity.value == 0.9


# ── AudioEnergyTap ──────────────────────────────────────────────────────


class _Analysis:
    def __init__(self, rms):
        self.rms = rms


class _CaptureStream:
    def __init__(self, rms=0.25):
        self._rms = rms

    def get_latest_analysis(self):
        return _Analysis(self._rms)


class _Resolved:
    device_index = 3
    is_loopback = True
    audio_template_id = ""


class _SourceStore:
    def resolve_audio_source(self, sid):
        return _Resolved()


class _Manager:
    def __init__(self):
        self.acquired = []
        self.released = []
        self.stream = _CaptureStream()

    def acquire(self, device, loopback, engine_type=None, engine_config=None):
        self.acquired.append((device, loopback))
        return self.stream

    def release(self, device, loopback, engine_type=None):
        self.released.append((device, loopback))


class TestAudioEnergyTap:
    def test_unavailable_without_manager(self):
        tap = AudioEnergyTap(None)
        assert tap.available is False
        tap.start()
        assert tap.energy() == 0.0

    def test_configure_resolves_capture_params(self):
        mgr = _Manager()
        tap = AudioEnergyTap(mgr, _SourceStore())
        tap.configure("as_1")
        tap.start()
        assert tap.active is True
        assert mgr.acquired == [(3, True)]

    def test_energy_smooths_rms(self):
        mgr = _Manager()
        tap = AudioEnergyTap(mgr, _SourceStore())
        tap.configure("as_1")
        tap.start()
        # rms 0.25 * gain 4 = 1.0 (clamped); EMA rises toward 1.0
        e1 = tap.energy()
        e2 = tap.energy()
        assert 0.0 < e1 < e2 <= 1.0

    def test_stop_releases_and_resets(self):
        mgr = _Manager()
        tap = AudioEnergyTap(mgr, _SourceStore())
        tap.configure("as_1")
        tap.start()
        tap.stop()
        assert tap.active is False
        assert mgr.released == [(3, True)]
        assert tap.energy() == 0.0


# ── Modulation ──────────────────────────────────────────────────────────


class _FixedTap:
    def __init__(self, energy):
        self._e = energy

    def energy(self, smoothing=0.4):
        return self._e


class TestModulation:
    def _stream(self, mode, intensity, energy):
        src = _make_source(audio_reactive=True, reactive_mode=mode, reactive_intensity=intensity)
        st = EffectColorStripStream(src)
        st._audio_tap = _FixedTap(energy)
        return st

    def test_brightness_silence_dims_to_zero(self):
        st = self._stream("brightness", 1.0, 0.0)
        buf = np.array([[200, 100, 50]], dtype=np.uint8)
        st._apply_audio_modulation(buf)
        assert np.array_equal(buf, np.array([[0, 0, 0]], dtype=np.uint8))

    def test_brightness_full_energy_preserves(self):
        st = self._stream("brightness", 1.0, 1.0)
        buf = np.array([[200, 100, 50]], dtype=np.uint8)
        st._apply_audio_modulation(buf)
        assert np.array_equal(buf, np.array([[200, 100, 50]], dtype=np.uint8))

    def test_saturation_silence_desaturates_to_luminance(self):
        st = self._stream("saturation", 1.0, 0.0)
        buf = np.array([[200, 100, 50]], dtype=np.uint8)
        st._apply_audio_modulation(buf)
        # All channels collapse to the luminance value (greyscale).
        assert buf[0, 0] == buf[0, 1] == buf[0, 2]

    def test_zero_intensity_is_noop(self):
        st = self._stream("both", 0.0, 0.0)
        buf = np.array([[200, 100, 50]], dtype=np.uint8)
        st._apply_audio_modulation(buf)
        assert np.array_equal(buf, np.array([[200, 100, 50]], dtype=np.uint8))