hms-mediaengine/_entwicklung/tests/unit/test_audio.py

"""Unit-Tests Audio-Analyse und Mapping (PLAN.md §20, §29.1)."""

from __future__ import annotations

import math

import pytest
from hms_audio import (
    AudioAnalyzer,
    AudioFeatures,
    BeatDetector,
    RingBuffer,
    compute_band_energy,
    compute_fft_magnitude,
    compute_peak,
    compute_rms,
    compute_spectral_flux,
)
from hms_audio.mapping import (
    LFO,
    AudioBinding,
    CurveType,
    ModulatorEngine,
    RandomModulator,
    StepSequencer,
    apply_curve,
)

# ---------- RMS/Peak (§20.2) ----------


def test_rms_of_silence_is_zero() -> None:
    assert compute_rms([]) == 0.0
    assert compute_rms([0.0] * 100) == 0.0


def test_rms_of_constant_signal() -> None:
    assert compute_rms([0.5] * 100) == pytest.approx(0.5)
    assert compute_rms([1.0, -1.0] * 50) == pytest.approx(1.0)


def test_peak_finds_absolute_maximum() -> None:
    assert compute_peak([0.3, -0.8, 0.5]) == 0.8
    assert compute_peak([]) == 0.0


# ---------- RingBuffer (§20.3) ----------


def test_ringbuffer_capacity_bounded() -> None:
    buf = RingBuffer(8)
    for i in range(20):
        buf.push(float(i))
    assert len(buf) == 8
    assert buf.capacity == 8


def test_ringbuffer_latest_returns_chronological() -> None:
    buf = RingBuffer(4)
    buf.extend([1.0, 2.0, 3.0, 4.0, 5.0])  # überschreibt die ältesten
    latest = buf.latest(3)
    assert latest == [3.0, 4.0, 5.0]  # chronologisch, nicht reversed


def test_ringbuffer_rejects_zero_capacity() -> None:
    with pytest.raises(ValueError):
        RingBuffer(0)


# ---------- FFT und Bänder (§20.2) ----------


def test_fft_of_sine_finds_dominant_frequency() -> None:
    """Ein 100-Hz-Sinus muss seinen Peak bei ~100 Hz haben."""
    sample_rate = 1000.0
    freq = 100.0
    n = 256
    samples = [math.sin(2.0 * math.pi * freq * t / sample_rate) for t in range(n)]
    magnitudes = compute_fft_magnitude(samples, sample_rate)
    assert len(magnitudes) == n // 2
    peak_bin = magnitudes.index(max(magnitudes))
    peak_freq = peak_bin * sample_rate / n
    assert 80.0 < peak_freq < 120.0  # innerhalb der FFT-Auflösung


def test_band_energy_isolated() -> None:
    """Bassband-Energie mit reinem Bass-Signal > Trebleband-Energie."""
    sample_rate = 44100.0
    bass_freq = 100.0
    n = 512
    samples = [math.sin(2.0 * math.pi * bass_freq * t / sample_rate) for t in range(n)]
    magnitudes = compute_fft_magnitude(samples, sample_rate)
    bass = compute_band_energy(magnitudes, sample_rate, 0, 250)
    treble = compute_band_energy(magnitudes, sample_rate, 8000, 20000)
    assert bass > treble  # Energie steckt im Bass, nicht im Höhenband


def test_band_energy_empty_magnitudes() -> None:
    assert compute_band_energy([], 44100, 0, 20000) == 0.0


# ---------- Spectral Flux (§20.2) ----------


def test_spectral_flux_positive_changes_only() -> None:
    current = [0.5, 0.3, 0.7]
    previous = [0.2, 0.4, 0.5]
    flux = compute_spectral_flux(current, previous)
    # positive: (0.5-0.2)=0.3, (0.7-0.5)=0.2; negative: (0.3-0.4) verworfen
    assert flux == pytest.approx(0.5)


def test_spectral_flux_empty() -> None:
    assert compute_spectral_flux([], []) == 0.0
    assert compute_spectral_flux([1.0], []) == 0.0


# ---------- BeatDetector (§20.2) ----------


def test_beat_detector_recovers_bpm() -> None:
    """Regelmäßige Flux-Spitzen bei 120 BPM = 0.5 s Peak-zu-Peak-Intervall.

    Peaks alle 2 Perioden à 0.25 s = 0.5 s zwischen Beats = 120 BPM.
    """
    det = BeatDetector(min_interval_s=0.3)
    ns_per_period = int(0.25 * 1e9)  # 250 ms pro Periode
    beat_count = 0
    for period in range(40):
        t = period * ns_per_period
        flux = 10.0 if period % 2 == 0 else 0.1  # Beat alle 0.5 s
        if det.feed(flux, t):
            beat_count += 1
    assert beat_count >= 5  # die meisten Beats erkannt
    assert 100.0 < det.bpm < 140.0  # um 120 BPM
    assert det.confidence > 0.3


def test_beat_detector_respects_min_interval() -> None:
    """Beats näher als min_interval werden ignoriert (§20.2)."""
    det = BeatDetector(min_interval_s=0.5)
    det._flux_history = [1.0] * 10  # genug Basisdaten
    t0 = 1_000_000_000  # > 0: vermeidet Sentinel-Verwirrung
    t1 = t0 + int(0.1 * 1e9)  # nur 100 ms später
    assert det.feed(10.0, t0) is True  # erster Beat
    assert det.feed(10.0, t1) is False  # zu nah: ignoriert


def test_beat_detector_needs_warmup() -> None:
    """Vor 4 Werten gibt es keine Beats (Ausreißerschutz)."""
    det = BeatDetector()
    assert det.feed(100.0, 0) is False  # erst 1 Wert: kein Beat
    assert det.feed(100.0, 1) is False
    assert det.feed(100.0, 2) is False


# ---------- AudioAnalyzer (§20.2, §20.3) ----------


def test_analyzer_silence_detection() -> None:
    an = AudioAnalyzer()
    an.feed([0.0] * 512)
    features = an.analyze(now_ns=1_000_000_000)
    assert features.silence is True
    assert features.rms < 0.001


def test_analyzer_detects_tone() -> None:
    """Ein 440-Hz-Ton: RMS deutlich über 0, Bassband hat Energie."""
    an = AudioAnalyzer()
    sample_rate = AudioAnalyzer.SAMPLE_RATE
    samples = [
        0.5 * math.sin(2.0 * math.pi * 440.0 * t / sample_rate)
        for t in range(512)
    ]
    an.feed(samples)
    features = an.analyze(now_ns=1_000_000_000)
    assert features.silence is False
    assert features.rms > 0.1
    assert features.bass > 0.0  # 440 Hz fällt ins Low-Mid, aber Bass hat Anteil
    assert features.monotonic_ns == 1_000_000_000  # timestamped (§20.3)


def test_analyzer_insufficient_data_returns_last() -> None:
    """Weniger als halbes Fenster: letzter Snapshot wird zurückgegeben."""
    an = AudioAnalyzer()
    an.feed([0.1] * 10)  # viel zu wenig
    features = an.analyze()
    assert features == AudioFeatures()  # Initial-Snapshot (alles 0)


# ---------- Kurven (§20.4) ----------


def test_apply_curve_types() -> None:
    assert apply_curve(0.5, CurveType.LINEAR) == pytest.approx(0.5)
    assert apply_curve(0.5, CurveType.QUADRATIC) == pytest.approx(0.25)
    assert apply_curve(0.5, CurveType.CUBIC) == pytest.approx(0.125)
    assert apply_curve(2.0, CurveType.LINEAR) == 1.0  # clamp
    assert apply_curve(-1.0, CurveType.LINEAR) == 0.0  # clamp


# ---------- AudioBinding (§20.4) ----------


def test_binding_full_pipeline() -> None:
    """Feature → Gate → Kurve → Attack → Min/Max."""
    binding = AudioBinding(
        id="b1",
        feature="bass",
        parameter_path="composition/x/layer/y/opacity",
        threshold=0.1,
        gain=2.0,
        curve=CurveType.LINEAR,
        attack_s=0.01,
        release_s=0.1,
        min_value=0.2,
        max_value=0.9,
    )
    features = AudioFeatures(bass=0.5, monotonic_ns=1_000_000_000)
    value = binding.process(features, 1_000_000_000)
    # Gate: (0.5-0.1)/(1-0.1)=0.444, Gain: 0.889, Clamp: 0.889
    # Min/Max: 0.2 + 0.889*0.7 = 0.822
    assert 0.5 < value < 0.9


def test_binding_gate_below_threshold() -> None:
    binding = AudioBinding(
        id="b2",
        feature="rms",
        parameter_path="master/intensity",
        threshold=0.5,
    )
    features = AudioFeatures(rms=0.3, monotonic_ns=1_000_000)
    value = binding.process(features, 1_000_000)
    assert value == pytest.approx(0.0)  # unter Schwelle → 0


def test_binding_disabled_returns_current() -> None:
    binding = AudioBinding(
        id="b3",
        feature="rms",
        parameter_path="x",
        enabled=False,
    )
    features = AudioFeatures(rms=0.8)
    assert binding.process(features, 1_000_000) == 0.0  # bleibt bei 0


def test_binding_attack_smoothing() -> None:
    """Attack glättet: bei schneller Zeitänderung nähert sich der Wert."""
    binding = AudioBinding(
        id="b4",
        feature="rms",
        parameter_path="x",
        attack_s=1.0,  # langsam
    )
    t0 = 1_000_000_000
    t1 = t0 + 100_000_000  # 100 ms später
    binding.process(AudioFeatures(rms=1.0), t0)
    v1 = binding.process(AudioFeatures(rms=1.0), t1)
    # Erster Schritt setzt _current=1.0; zweiter bleibt bei 1.0
    assert v1 == pytest.approx(1.0)


# ---------- LFO / Random / Sequencer (§20.5) ----------


def test_lfo_sine_periodicity() -> None:
    lfo = LFO(id="l1", waveform="sine", rate_hz=1.0)
    t0 = 0
    t_half = int(0.5 * 1e9)  # halbe Periode
    v0 = lfo.process(t0)
    v_half = lfo.process(t_half)
    lfo.process(int(1.0 * 1e9))  # volle Periode: nur Nebenprodukt
    assert v0 != v_half  # unterschiedliche Phasen
    assert 0.0 <= v0 <= 1.0
    assert 0.0 <= v_half <= 1.0


def test_lfo_square_waveform() -> None:
    lfo = LFO(id="l2", waveform="square", rate_hz=1.0)
    v_low = lfo.process(int(0.25 * 1e9))  # erste Hälfte
    v_high = lfo.process(int(0.75 * 1e9))  # zweite Hälfte
    assert v_low == 1.0
    assert v_high == 0.0


def test_random_modulator_deterministic_with_seed() -> None:
    """Gleicher Seed → gleiche Sequenz (§20.5: Random mit Seed)."""
    r1 = RandomModulator(id="r1", seed=42, rate_hz=100)
    r2 = RandomModulator(id="r2", seed=42, rate_hz=100)
    t = int(0.01 * 1e9)
    v1 = [r1.process(t + i * 10_000_000) for i in range(10)]
    v2 = [r2.process(t + i * 10_000_000) for i in range(10)]
    assert v1 == v2  # deterministisch


def test_step_sequencer_cycles_through_steps() -> None:
    seq = StepSequencer(
        id="s1",
        steps=[0.0, 1.0, 0.5, 0.0],
        bpm=240.0,  # 4 Steps pro Sekunde
    )
    t0 = 0
    t1 = int(0.25 * 1e9)  # Step 1
    t2 = int(0.50 * 1e9)  # Step 2
    v0 = seq.process(t0)
    v1 = seq.process(t1)
    v2 = seq.process(t2)
    assert v0 == pytest.approx(0.0)
    assert v1 == pytest.approx(1.0)
    assert v2 == pytest.approx(0.5)


# ---------- ModulatorEngine (§20.4, §20.5) ----------


def test_engine_routes_audio_bindings() -> None:
    engine = ModulatorEngine()
    engine.add_audio_binding(
        AudioBinding(id="a1", feature="bass", parameter_path="layer/x/opacity")
    )
    engine.add_audio_binding(
        AudioBinding(id="a2", feature="rms", parameter_path="master/intensity")
    )
    features = AudioFeatures(bass=0.8, rms=0.3, monotonic_ns=1_000_000_000)
    results = engine.process_audio(features, 1_000_000_000)
    assert "layer/x/opacity" in results
    assert "master/intensity" in results
    assert results["layer/x/opacity"] > results["master/intensity"]  # bass > rms


def test_engine_disabled_binding_skipped() -> None:
    engine = ModulatorEngine()
    engine.add_audio_binding(
        AudioBinding(id="a1", feature="bass", parameter_path="x", enabled=False)
    )
    results = engine.process_audio(AudioFeatures(bass=0.5), 1_000_000)
    assert results == {}  # nichts aktiv


def test_engine_processes_all_modulator_types() -> None:
    engine = ModulatorEngine()
    engine.add_lfo(LFO(id="l1", rate_hz=2.0))
    engine.add_random(RandomModulator(id="r1", seed=1))
    engine.add_sequencer(StepSequencer(id="s1", steps=[1.0, 0.0]))
    results = engine.process_modulators(int(0.1 * 1e9))
    assert "lfo" in results and "l1" in results["lfo"]
    assert "random" in results and "r1" in results["random"]
    assert "sequencer" in results and "s1" in results["sequencer"]