media-player-server/tests/test_audio_analyzer.py

"""Tests for AudioAnalyzer.

Covers the pure-Python pieces that don't need real audio hardware:
- Logarithmic FFT bin edge layout
- Slow-AGC envelope follower (attack vs release behaviour)
- Lifecycle reset of the AGC reference on start()

Tests are skipped when numpy isn't installed in the host environment
so they don't block CI on a minimal interpreter.
"""

from __future__ import annotations

import pytest

from media_server.services.audio_analyzer import AudioAnalyzer, _load_numpy

np = _load_numpy()
needs_numpy = pytest.mark.skipif(np is None, reason="numpy not available")


@pytest.fixture
def analyzer() -> AudioAnalyzer:
    return AudioAnalyzer(num_bins=16, sample_rate=44100, chunk_size=1024)


# ── _compute_bin_edges ────────────────────────────────────────────


@needs_numpy
def test_bin_edges_count_matches_num_bins_plus_one(analyzer: AudioAnalyzer) -> None:
    edges = analyzer._compute_bin_edges()
    assert len(edges) == analyzer.num_bins + 1


@needs_numpy
def test_bin_edges_are_monotonic_non_decreasing(analyzer: AudioAnalyzer) -> None:
    edges = analyzer._compute_bin_edges()
    assert all(edges[i] <= edges[i + 1] for i in range(len(edges) - 1))


@needs_numpy
def test_bin_edges_stay_within_fft_size(analyzer: AudioAnalyzer) -> None:
    edges = analyzer._compute_bin_edges()
    fft_size = analyzer.chunk_size // 2 + 1
    assert max(edges) <= fft_size - 1
    assert min(edges) >= 0


# ── AGC envelope follower (the new behaviour) ─────────────────────


def _step_envelope(analyzer: AudioAnalyzer, peak: float) -> float:
    """Run one frame of the AGC update with a known peak value.

    Mirrors the math inside _capture_loop without spinning up a real
    capture thread or requiring numpy: pure Python on a single float.
    """
    if peak > analyzer._spectrum_ref:
        analyzer._spectrum_ref += (peak - analyzer._spectrum_ref) * 0.05
    else:
        analyzer._spectrum_ref += (peak - analyzer._spectrum_ref) * 0.005
    return analyzer._spectrum_ref


def test_agc_initial_reference_is_quiet(analyzer: AudioAnalyzer) -> None:
    assert analyzer._spectrum_ref == pytest.approx(0.01)


def test_agc_attacks_quickly_toward_loud_signal(analyzer: AudioAnalyzer) -> None:
    # Drive 30 frames of a loud signal; reference should climb sharply.
    for _ in range(30):
        _step_envelope(analyzer, peak=1.0)
    # 30 frames of attack=0.05 brings (1 - 0.99^30) ≈ 0.78 of the way to 1.0.
    assert analyzer._spectrum_ref > 0.5
    assert analyzer._spectrum_ref < 1.0


def test_agc_releases_slowly_toward_quiet_signal(analyzer: AudioAnalyzer) -> None:
    analyzer._spectrum_ref = 1.0
    for _ in range(30):
        _step_envelope(analyzer, peak=0.0)
    # Release coefficient is 0.005 — after 30 frames we should have shed
    # only ~14% of the headroom, not snap back to silent.
    assert analyzer._spectrum_ref > 0.7
    assert analyzer._spectrum_ref < 1.0


def test_agc_is_asymmetric_attack_faster_than_release(analyzer: AudioAnalyzer) -> None:
    a = AudioAnalyzer()
    b = AudioAnalyzer()
    a._spectrum_ref = 0.5
    b._spectrum_ref = 0.5
    # One attack frame toward 1.0
    _step_envelope(a, peak=1.0)
    # One release frame toward 0.0 (same magnitude of error: 0.5)
    _step_envelope(b, peak=0.0)
    attack_delta = a._spectrum_ref - 0.5
    release_delta = 0.5 - b._spectrum_ref
    # Attack coefficient (0.05) is 10× the release coefficient (0.005).
    assert attack_delta == pytest.approx(release_delta * 10, rel=1e-6)


# ── start() lifecycle reset ──────────────────────────────────────


def test_start_resets_spectrum_ref_when_unavailable(
    monkeypatch: pytest.MonkeyPatch, analyzer: AudioAnalyzer
) -> None:
    """Even when start() returns False (no hardware), the AGC state
    should remain at the documented quiet baseline."""
    # Force unavailable so start() short-circuits without spawning a thread.
    monkeypatch.setattr(
        AudioAnalyzer, "available", property(lambda self: False)
    )
    analyzer._spectrum_ref = 0.95  # leftover from prior session
    started = analyzer.start()
    assert started is False
    # start() returned early before the reset — by design (no capture
    # means no need to renormalize). Document the contract.
    assert analyzer._spectrum_ref == 0.95


def test_start_resets_spectrum_ref_when_available(
    monkeypatch: pytest.MonkeyPatch, analyzer: AudioAnalyzer
) -> None:
    """When capture actually starts, leftover AGC state from a prior
    session must be cleared so the first transients don't clip."""
    monkeypatch.setattr(
        AudioAnalyzer, "available", property(lambda self: True)
    )
    # Stub out the thread so we don't actually spin up a capture loop.
    monkeypatch.setattr(
        "media_server.services.audio_analyzer.threading.Thread",
        lambda *a, **kw: type("T", (), {"start": lambda self: None})(),
    )
    analyzer._spectrum_ref = 0.95  # leftover from prior session
    try:
        started = analyzer.start()
        assert started is True
        assert analyzer._spectrum_ref == pytest.approx(0.01)
    finally:
        analyzer._running = False


# ── get_frequency_data thread-safe contract ───────────────────────


def test_get_frequency_data_returns_none_before_capture(
    analyzer: AudioAnalyzer,
) -> None:
    assert analyzer.get_frequency_data() is None