beefcarver
48e480d58c
SYNC mode on macOS needed a virtual loopback driver (BlackHole, Loopback, Soundflower) routed through a Multi-Output Device, and shelled out to 'ffmpeg -f avfoundation' to read it. macOS 14.2 lets a process tap the system mix directly, so do that instead: CATapDescription -> AudioHardwareCreateProcessTap -> private aggregate device -> IOProc, straight through ctypes. No driver, no helper binary, no ffmpeg. The tap hands out native-rate interleaved float32, so the analyser can now retune its window and band edges when the capture rate differs from 44.1 kHz, and mono-downmixes the frames before the FFT. A tap that was never granted System Audio Recording permission is created happily but only ever delivers digital silence: treat that as a failed capture instead of flat bars, explain it in the status bar and fall back to the stream tap. A tap that renders nothing *yet* (playback still starting, or mpv pointed at another device) is not a permanent failure either - the stream tap covers for a minute, then the tap is retried. Drop the avfoundation device enumeration, the loopback-name heuristics and the microphone safeguards' tests, and document the new permission and its diagnostics instead.
327 lines
10 KiB
Python
327 lines
10 KiB
Python
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 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 ``somafm_tui.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(
|
|
"#!/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()
|