beefcarver/beefy-player
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity
Files
dc6ed00abed34431d97a0fc509da512cf97023c9
beefy-player/tests/test_spectrum.py
T

448 lines
15 KiB
Python
Raw Blame History

from __future__ import annotations
import sys
import time
from pathlib import Path
import numpy as np
from beefy_player import spectrum as spectrum_module
from beefy_player.spectrum import Frame, SpectrumEngine
PCM_SCRIPT = f"#!{sys.executable}\n" + """import math, struct, sys, time
sr = 44100
samples = [int(0.6 * 32767 * math.sin(2 * math.pi * 1000 * i / sr)) for i in range(4096)]
chunk = struct.pack("<%dh" % len(samples), *samples)
out = sys.stdout.buffer
try:
while True:
out.write(chunk)
out.flush()
time.sleep(0.08)
except (BrokenPipeError, KeyboardInterrupt):
pass
"""
def write_pcm_script(path: Path) -> None:
path.write_text(PCM_SCRIPT)
path.chmod(0o755)
def tap_pcm(
freq: float = 1000,
frames: int = 4096,
sample_rate: int = 48000,
channels: int = 2,
) -> bytes:
"""Interleaved float32 stereo PCM, as the macOS tap delivers it."""
samples = np.arange(frames) / sample_rate
tone = (0.5 * np.sin(2 * np.pi * freq * samples)).astype("<f4")
return np.repeat(tone, channels).tobytes()
class FakeTap:
"""Stand-in for ``beefy_player.macos_tap.SystemAudioTap``."""
def __init__(
self,
*,
silent: bool = False,
idle: bool = False,
error: Exception | None = None,
sample_rate: int = 48000,
channels: int = 2,
) -> None:
self.sample_rate = sample_rate
self.channels = channels
self.bytes_per_frame = 4 * channels
self.stopped = False
self._idle = idle
self._error = error
self._chunk = (
b"\x00" * (4096 * self.bytes_per_frame)
if silent
else tap_pcm(1000, 4096, sample_rate, channels)
)
def start(self) -> None:
if self._error is not None:
raise self._error
def read(self, timeout: float | None = 1.0) -> bytes | None:
if self.stopped or self._idle:
return None
return self._chunk
def stop(self) -> None:
self.stopped = True
def wait_for_spectrum(engine: SpectrumEngine, status: str, timeout: float = 6.0):
deadline = time.monotonic() + timeout
snap = engine.snapshot()
while time.monotonic() < deadline:
snap = engine.snapshot()
if snap.status == status and sum(snap.bands) > 1.0:
break
time.sleep(0.05)
return snap
def test_fft_detects_a_1khz_tone() -> None:
engine = SpectrumEngine(ffmpeg_path="")
sample_rate = 44100
t = np.arange(8192) / sample_rate
tone = np.sin(2 * np.pi * 1000 * t) * 0.5
engine.feed_pcm((tone * 32767).astype("<i2").tobytes())
snap = engine.snapshot()
assert snap.status == "live"
assert snap.label == "FFT"
assert len(snap.bands) == 72
top = max(range(len(snap.bands)), key=lambda index: snap.bands[index])
expected = 72 * np.log(1000.0 / 35.0) / np.log(16000.0 / 35.0)
assert abs(top - expected) <= 2
assert snap.bands[top] > 0.8
assert snap.peaks[top] >= snap.bands[top] - 1e-6
assert any(abs(value) > 0.3 for value in snap.wave)
assert 0.0 < snap.level <= 1.0
def test_frame_labels() -> None:
assert Frame((), (), (), 0.0, "monitor").label == "SYNC"
assert Frame((), (), (), 0.0, "live").label == "FFT"
assert Frame((), (), (), 0.0, "sim").label == "SIM"
assert Frame((), (), (), 0.0, "idle").label == "IDLE"
assert Frame((), (), (), 0.0, "off").label == "OFF"
assert Frame((), (), (), 0.0, "monitor").is_live is True
def test_simulation_fallback() -> None:
engine = SpectrumEngine(ffmpeg_path="", source_mode="sim")
engine.start()
try:
engine.set_source("https://example.invalid/demo")
deadline = time.monotonic() + 5
snap = engine.snapshot()
while time.monotonic() < deadline:
snap = engine.snapshot()
if snap.status == "sim" and max(snap.bands) > 0.2:
break
time.sleep(0.05)
assert snap.status == "sim"
assert snap.label == "SIM"
assert max(snap.bands) > 0.2
assert 0.0 <= snap.level <= 1.0
finally:
engine.stop()
def test_monitor_capture_prefers_real_output(tmp_path, monkeypatch) -> None:
script = tmp_path / "parec"
write_pcm_script(script)
def fake_which(name: str) -> str | None:
return str(script) if name == "parec" else None
monkeypatch.setattr(spectrum_module.shutil, "which", fake_which)
engine = SpectrumEngine(source_mode="auto")
assert engine.monitor_available is True
engine.start()
try:
engine.set_source("https://example.invalid/live")
snap = wait_for_spectrum(engine, "monitor")
assert snap.status == "monitor", snap
assert snap.label == "SYNC"
top = max(range(len(snap.bands)), key=lambda index: snap.bands[index])
assert 36 <= top <= 42
assert snap.bands[top] > 0.5
engine.set_source(None)
time.sleep(0.3)
assert engine.snapshot().status == "idle"
finally:
engine.stop()
def test_stream_tap_pipeline(tmp_path) -> None:
script = tmp_path / "fake-ffmpeg"
write_pcm_script(script)
engine = SpectrumEngine(ffmpeg_path=str(script), source_mode="stream")
engine.start()
try:
engine.set_source("https://example.invalid/stream")
snap = wait_for_spectrum(engine, "live")
assert snap.status == "live", snap
assert snap.label == "FFT"
top = max(range(len(snap.bands)), key=lambda index: snap.bands[index])
assert 36 <= top <= 42
assert snap.bands[top] > 0.5
engine.set_source(None)
time.sleep(0.3)
assert engine.snapshot().status == "idle"
finally:
engine.stop()
def test_macos_tap_is_the_sync_monitor(monkeypatch) -> None:
engine = SpectrumEngine(ffmpeg_path="")
engine._parec = None
engine._darwin = True
monkeypatch.setattr(spectrum_module.macos_tap, "available", lambda: True)
tap = FakeTap()
monkeypatch.setattr(spectrum_module.macos_tap, "SystemAudioTap", lambda: tap)
assert engine.monitor_available is True
assert engine._monitor_kind() == "tap"
engine.start()
try:
engine.set_source("https://example.invalid/live")
snap = wait_for_spectrum(engine, "monitor")
assert snap.status == "monitor", snap
assert snap.label == "SYNC"
top = max(range(len(snap.bands)), key=lambda index: snap.bands[index])
assert 36 <= top <= 42
assert snap.bands[top] > 0.5
engine.set_source(None)
time.sleep(0.3)
assert engine.snapshot().status == "idle"
assert tap.stopped is True
finally:
engine.stop()
def test_silent_tap_falls_back_instead_of_flat_bars(monkeypatch) -> None:
# A tap that was never granted System Audio Recording permission is
# created happily but only ever hands out digital silence.
engine = SpectrumEngine(ffmpeg_path="")
engine._parec = None
engine._darwin = True
monkeypatch.setattr(spectrum_module.macos_tap, "available", lambda: True)
tap = FakeTap(silent=True)
monkeypatch.setattr(spectrum_module.macos_tap, "SystemAudioTap", lambda: tap)
engine.start()
try:
engine.set_source("https://example.invalid/live")
snap = wait_for_spectrum(engine, "sim")
assert snap.status == "sim", snap
assert engine.monitor_available is False
assert tap.stopped is True
notice = engine.take_notice()
assert notice is not None and "System Audio Recording" in notice
assert engine.take_notice() is None
finally:
engine.stop()
def test_tap_start_failure_falls_back_and_explains(monkeypatch) -> None:
engine = SpectrumEngine(ffmpeg_path="")
engine._parec = None
engine._darwin = True
monkeypatch.setattr(spectrum_module.macos_tap, "available", lambda: True)
monkeypatch.setattr(
spectrum_module.macos_tap,
"SystemAudioTap",
lambda: FakeTap(error=spectrum_module.macos_tap.TapError("boom")),
)
engine.start()
try:
engine.set_source("https://example.invalid/live")
snap = wait_for_spectrum(engine, "sim")
assert snap.status == "sim", snap
assert engine.monitor_available is False
notice = engine.take_notice()
assert notice is not None and "macOS audio capture failed" in notice
finally:
engine.stop()
def test_idle_tap_retries_instead_of_latching(monkeypatch) -> None:
# If the output device renders nothing yet (slow stream, or mpv pointed at
# another device) the stream tap covers us and the tap is retried later.
monkeypatch.setattr(spectrum_module, "TAP_IDLE_SECONDS", 0.05)
engine = SpectrumEngine(ffmpeg_path="")
engine._parec = None
engine._darwin = True
monkeypatch.setattr(spectrum_module.macos_tap, "available", lambda: True)
tap = FakeTap(idle=True)
monkeypatch.setattr(spectrum_module.macos_tap, "SystemAudioTap", lambda: tap)
engine.start()
try:
engine.set_source("/tmp/tone.wav")
snap = wait_for_spectrum(engine, "sim")
assert snap.status == "sim", snap
assert engine.monitor_available is True # not a permanent failure
assert engine._tap_retry_at > time.monotonic()
notice = engine.take_notice()
assert notice is not None and "default output device" in notice
finally:
engine.stop()
def test_tap_audio_is_downmixed_and_retuned() -> None:
engine = SpectrumEngine(ffmpeg_path="")
engine._retune(48000)
engine.feed_audio(tap_pcm(1000, sample_rate=48000), channels=2)
snap = engine.snapshot()
assert snap.status == "monitor"
top = max(range(len(snap.bands)), key=lambda index: snap.bands[index])
assert 36 <= top <= 42
assert snap.bands[top] > 0.8
def test_dead_monitor_falls_back_instead_of_flat_bars(tmp_path) -> None:
script = tmp_path / "silent-parec"
script.write_text(
f"#!{sys.executable}\n"
"import sys\n"
"block = b'\\0' * 4096\n"
"for _ in range(1000):\n"
" sys.stdout.buffer.write(block)\n"
"sys.stdout.buffer.flush()\n"
)
script.chmod(0o755)
engine = SpectrumEngine(ffmpeg_path="")
engine.set_source("https://example.invalid/live")
proc = engine._spawn([str(script)], "monitor")
assert proc is not None
assert engine.snapshot().status == "monitor"
try:
assert engine._capture(
proc, "https://example.invalid/live", stop_on_url_change=False, status="monitor"
) is False
finally:
engine._stop_proc(proc)
engine.stop()
def test_monitor_survives_silence_after_audio(tmp_path, monkeypatch) -> None:
# Tuning a station leaves mpv's sink silent while the new stream buffers.
# That gap must not be mistaken for a dead monitor: only a capture that has
# never delivered audio is written off, otherwise SYNC would be lost (and
# replaced by the drifting stream tap) for the rest of the session.
script = tmp_path / "parec"
script.write_text(
f"#!{sys.executable}\n"
"import math, struct, sys, time\n"
"sr = 44100\n"
"n = 2048\n"
"tone = struct.pack('<%dh' % n, *["
"int(0.6 * 32767 * math.sin(2 * math.pi * 1000 * i / sr)) for i in range(n)])\n"
"silence = b'\\0' * (2 * n)\n"
"out = sys.stdout.buffer\n"
"for _ in range(40):\n"
" out.write(tone)\n"
"out.flush()\n"
"for _ in range(400):\n"
" out.write(silence)\n"
"out.flush()\n"
"time.sleep(30)\n"
)
script.chmod(0o755)
def fake_which(name: str) -> str | None:
return str(script) if name == "parec" else None
monkeypatch.setattr(spectrum_module.shutil, "which", fake_which)
engine = SpectrumEngine(ffmpeg_path="")
engine.start()
try:
engine.set_source("https://example.invalid/live")
assert wait_for_spectrum(engine, "monitor").status == "monitor"
# The script now feeds far more than SILENT_MONITOR_BLOCKS of digital
# silence. Give the analyser time to consume it, then make sure the
# capture is still alive and still SYNC.
time.sleep(1.0)
snap = engine.snapshot()
assert snap.status == "monitor", snap
assert snap.label == "SYNC"
assert engine.monitor_available is True
assert engine._monitor_failed is False
finally:
engine.stop()
class SequencedTap(FakeTap):
"""A tap that hands out a fixed list of buffers, then goes quiet."""
def __init__(self, chunks: list[bytes], **kwargs) -> None:
super().__init__(**kwargs)
self._chunks = list(chunks)
def read(self, timeout: float | None = 1.0) -> bytes | None:
if self.stopped:
return None
return self._chunks.pop(0) if self._chunks else None
def test_tap_survives_silence_after_audio(monkeypatch) -> None:
# A real tap that has already captured audio is allowed to go quiet while
# a new station buffers; only a tap that never produced audio means the
# System Audio Recording permission is missing.
engine = SpectrumEngine(ffmpeg_path="")
engine._parec = None
engine._darwin = True
monkeypatch.setattr(spectrum_module.macos_tap, "available", lambda: True)
chunks = [tap_pcm(1000, 4096)] + [b"\x00" * (4096 * 8)] * 200
tap = SequencedTap(chunks)
monkeypatch.setattr(spectrum_module.macos_tap, "SystemAudioTap", lambda: tap)
engine.start()
try:
engine.set_source("https://example.invalid/live")
assert wait_for_spectrum(engine, "monitor").status == "monitor"
time.sleep(0.5)
snap = engine.snapshot()
assert snap.status == "monitor", snap
assert engine.monitor_available is True
assert engine._monitor_failed is False
assert tap.stopped is False
finally:
engine.stop()
def test_tuning_clears_a_monitor_failure_and_retries_sync(tmp_path, monkeypatch) -> None:
# Once an output capture has failed (or been given up on after a tuning
# gap), a station change must clear that state and try SYNC again instead
# of staying on the drifting FFT stream for the rest of the session.
script = tmp_path / "parec"
write_pcm_script(script)
def fake_which(name: str) -> str | None:
return str(script) if name == "parec" else None
monkeypatch.setattr(spectrum_module.shutil, "which", fake_which)
engine = SpectrumEngine(ffmpeg_path="")
engine.start()
try:
engine.set_source("https://example.invalid/a")
assert wait_for_spectrum(engine, "monitor").status == "monitor"
# Simulate the latch that a silent retry used to leave behind.
engine._monitor_failed = True
engine._tap_retry_at = time.monotonic() + 60.0
for index in range(10):
engine.set_source(f"https://example.invalid/b{index}")
time.sleep(0.02)
snap = wait_for_spectrum(engine, "monitor")
assert snap.status == "monitor", snap
assert engine._monitor_failed is False
assert engine.monitor_available is True
assert engine._monitor_worked is True
finally:
engine.stop()
Reference in new issue View Git Blame Copy Permalink