from __future__ import annotations import time from pathlib import Path import numpy as np from somafm_tui import spectrum as spectrum_module from somafm_tui.spectrum import Frame, SpectrumEngine PCM_SCRIPT = """#!/usr/bin/env python3 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 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(" 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_avfoundation_loopback_is_preferred_on_macos(monkeypatch) -> None: engine = SpectrumEngine(ffmpeg_path="/opt/homebrew/bin/ffmpeg") engine._parec = None # pretend there is no PulseAudio on this host engine._darwin = True monkeypatch.delenv("SOMAFM_SPECTRUM_DEVICE", raising=False) monkeypatch.setattr( engine, "_avf_audio_devices", lambda: [(0, "MacBook Pro Microphone"), (1, "BlackHole 2ch")], ) engine._avf_checked = False assert engine._avf_loopback() == 1 assert engine._monitor_kind() == "avfoundation" assert engine.monitor_available is True command = engine._monitor_command() assert command is not None assert command[0] == "/opt/homebrew/bin/ffmpeg" assert command[command.index("-i") + 1] == ":1" assert "-f" in command and "avfoundation" in command assert command[-1] == "-" assert "pcm_s16le" in command assert command[command.index("-ar") + 1] == "44100" def test_avfoundation_never_autoselects_a_microphone(monkeypatch) -> None: engine = SpectrumEngine(ffmpeg_path="/opt/homebrew/bin/ffmpeg") engine._parec = None engine._darwin = True monkeypatch.delenv("SOMAFM_SPECTRUM_DEVICE", raising=False) monkeypatch.setattr( engine, "_avf_audio_devices", lambda: [(0, "MacBook Pro Microphone")] ) engine._avf_checked = False assert engine._avf_loopback() is None assert engine._monitor_kind() is None assert engine._monitor_command() is None def test_avfoundation_device_override_by_name_and_index(monkeypatch) -> None: devices = [(0, "MacBook Pro Microphone"), (1, "BlackHole 2ch"), (2, "Loopback Audio")] engine = SpectrumEngine(ffmpeg_path="/opt/homebrew/bin/ffmpeg") engine._parec = None engine._darwin = True monkeypatch.setattr(engine, "_avf_audio_devices", lambda: devices) monkeypatch.setenv("SOMAFM_SPECTRUM_DEVICE", "loopback audio") engine._avf_checked = False assert engine._avf_loopback() == 2 engine = SpectrumEngine(ffmpeg_path="/opt/homebrew/bin/ffmpeg") engine._parec = None engine._darwin = True monkeypatch.setattr(engine, "_avf_audio_devices", lambda: devices) monkeypatch.setenv("SOMAFM_SPECTRUM_DEVICE", "0") engine._avf_checked = False assert engine._avf_loopback() == 0 def test_avfoundation_listing_parser(monkeypatch) -> None: listing = ( "[AVFoundation indev @ 0x1] AVFoundation video devices:\n" "[AVFoundation indev @ 0x1] [0] Capture screen 0\n" "[AVFoundation indev @ 0x1] AVFoundation audio devices:\n" "[AVFoundation indev @ 0x1] [0] MacBook Pro Microphone\n" "[AVFoundation indev @ 0x1] [1] BlackHole 2ch\n" ) class Result: stdout = listing returncode = 1 def fake_run(*args, **kwargs): return Result() engine = SpectrumEngine(ffmpeg_path="/opt/homebrew/bin/ffmpeg") monkeypatch.setattr(spectrum_module.subprocess, "run", fake_run) assert engine._avf_audio_devices() == [ (0, "MacBook Pro Microphone"), (1, "BlackHole 2ch"), ] def test_dead_loopback_falls_back_instead_of_flat_bars(tmp_path) -> None: script = tmp_path / "silent-parec" script.write_text( "#!/usr/bin/env python3\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()