diff --git a/README.md b/README.md index 02d1aa0..4e6f851 100644 --- a/README.md +++ b/README.md @@ -22,10 +22,10 @@ implemented here from the public SomaFM endpoints and open protocols. spectrum; or a full-height oscilloscope. Three modes (`v`). - **Synced spectrum, not a guess** — by default the analyser records your actual audio output, so the bars follow what you hear (`SYNC`): the - PulseAudio/PipeWire monitor via `parec` on Linux, an `ffmpeg`/avfoundation - loopback input (BlackHole, Loopback, Soundflower, …) on macOS. Without an - output capture it falls back to a muted second `ffmpeg` decode of the stream - (`FFT`, can drift), then to a synthesized display (`SIM`). + PulseAudio/PipeWire monitor via `parec` on Linux, and a native Core Audio + process tap on macOS (no driver, no BlackHole, nothing to install). Without + an output capture it falls back to a muted second `ffmpeg` decode of the + stream (`FFT`, can drift), then to a synthesized display (`SIM`). - **Real artwork on iTerm2, Kitty and WezTerm** — Kitty uses the kitty graphics protocol; iTerm2 (including on macOS) and WezTerm use their native iTerm2 inline-image protocol. Images occupy exactly the ON AIR art cell box @@ -65,11 +65,14 @@ Python **3.10+** and a few command-line audio tools on `$PATH`. | `python3` | runtime (3.10 or newer; no exact version pin) | | [`uv`](https://docs.astral.sh/uv/) | creates the venv and installs every Python dependency from `uv.lock` | | `mpv` | playback and JSON IPC control | -| `ffmpeg` | muted stream decode for the fallback analyser, and the macOS output capture | +| `ffmpeg` | muted stream decode for the fallback analyser | | `parec` / `pactl` *(Linux)* | synced monitor capture for the spectrum (`pulseaudio-utils`; works with PipeWire) | -| a loopback audio device *(macOS, optional)* | synced output capture for the spectrum: BlackHole, Loopback, Soundflower, … | | D-Bus session bus *(Linux)* | MPRIS desktop media integration (`playerctl` is handy for testing) | +On macOS the spectrum capture is part of the app: it uses the Core Audio +process-tap API (macOS 14.2 or newer), so there is no loopback driver to +install. + On macOS the media integration needs nothing extra: mpv is what the system's Now Playing session (Control Center, media keys) talks to. @@ -110,33 +113,20 @@ the synced spectrum needs `parec`/`pactl`. brew install mpv ffmpeg uv # ffplay ships with ffmpeg ``` -That is enough to run: playback, artwork in iTerm2/Kitty/WezTerm, the muted -`ffmpeg` spectrum tap and the macOS Now Playing integration all work out of the -box. The first capture attempt makes macOS ask for **microphone** access for -whatever terminal you use — that permission is what lets `ffmpeg` read the -loopback device. +That is enough to run: playback, artwork in iTerm2/Kitty/WezTerm, the synced +Core Audio spectrum tap and the macOS Now Playing integration all work out of +the box. -For a *synced* spectrum (`SYNC` instead of `FFT`) macOS needs an output -loopback, because CoreAudio does not let an ordinary process record the -speakers directly: +The first time the analyser starts it asks macOS for **System Audio +Recording** permission — a one-off prompt for your terminal (iTerm2, Kitty, +WezTerm, …). Approve it under *System Settings → Privacy & Security → System +Audio Recording*; no loopback driver, Multi-Output Device or audio routing is +needed. If the permission is denied the tap only ever hands out silence, the +analyser notices and falls back to `FFT` instead of showing dead bars. -```bash -brew install blackhole-2ch # or Loopback / Soundflower -``` - -Then either - -1. create a **Multi-Output Device** in *Audio MIDI Setup* containing both your - speakers/headphones and *BlackHole 2ch*, and select it as the system output, - or -2. set *BlackHole 2ch* as the output and monitor the audio elsewhere. - -The app auto-detects loopback-style inputs (BlackHole, Loopback, Soundflower, -“Multi-Output”, “Aggregate”) and never grabs a plain microphone by accident. -Pick one explicitly with `SOMAFM_SPECTRUM_DEVICE` (name or avfoundation index, -see `ffmpeg -f avfoundation -list_devices true -i ""`). If the loopback turns -out to be silent (nothing routed into it), the analyser notices after a few -seconds and falls back to `FFT` instead of showing dead bars. +The tap records the **default output device**, which is also what mpv uses; if +you point mpv elsewhere with `SOMAFM_MPV_ARGS='--audio-device=...'` the +analyser falls back to `FFT`. Terminal image rendering uses [Pillow](https://python-pillow.org/) and is implemented directly (Kitty TGP + iTerm2 OSC 1337) — no terminal probing. @@ -248,12 +238,16 @@ spectrum and the station list stay in sync. The status bar shows engine, and uses `SIGSTOP`/`SIGCONT` when falling back to `ffplay`. On macOS it also asks mpv to publish Now Playing and maps the media keys onto `script-message`s that come back over the same IPC socket. -- `somafm_tui/spectrum.py` — 44.1 kHz mono s16le analyser. Prefers capturing - the real output: `parec` (PulseAudio/PipeWire monitor) on Linux, or - `ffmpeg -f avfoundation` on a loopback input on macOS (SYNC). Otherwise a +- `somafm_tui/spectrum.py` — analyser (44.1 kHz mono s16le from `parec`, or + the device's native float32 rate from the macOS tap). Prefers capturing the + real output: `parec` (PulseAudio/PipeWire monitor) on Linux, or a native + Core Audio process tap on macOS (SYNC, see `macos_tap.py`). Otherwise a muted low-latency `ffmpeg` decode of the stream (FFT). Hann-windowed 2048-point FFT, log band grouping, auto-gain, fast attack/decay smoothing, peak hold and waveform downsampling. +- `somafm_tui/macos_tap.py` — the macOS output capture, in pure `ctypes` + against Core Audio: `CATapDescription` → `AudioHardwareCreateProcessTap` → + private aggregate device → IOProc. No driver, no helper binary, no ffmpeg. - `somafm_tui/mediasession.py` — picks the desktop media backend (MPRIS on Linux, Now Playing on macOS) so the app has one platform-neutral interface. - `somafm_tui/mpris.py` — `dbus-next` MPRIS service in its own thread with a @@ -303,11 +297,11 @@ same dotfiles/config work everywhere; `XDG_CONFIG_HOME`, `XDG_CACHE_HOME` and either, the UI still runs and tells you. Extra mpv args can be injected with `SOMAFM_MPV_ARGS` (e.g. `SOMAFM_MPV_ARGS='--audio-device=...'`, which is the macOS way to choose an output device). -- **Spectrum says `FFT` instead of `SYNC`?** `parec`/`pactl` are unavailable, - the monitor device failed, or you forced the stream tap. The bars are still - real audio, but the second stream can drift relative to playback. Use - `--spectrum-source monitor` to force sync, or `SOMAFM_SPECTRUM_DEVICE=` - to pick a specific monitor/source. `SIM` means the bars are synthesized. +- **Spectrum says `FFT` instead of `SYNC`?** The monitor is unavailable or + failed, or you forced the stream tap. The bars are still real audio, but the + second stream can drift relative to playback. Use `--spectrum-source + monitor` to force sync, or `SOMAFM_SPECTRUM_DEVICE=` on Linux to pick + a specific monitor. `SIM` means the bars are synthesized. - **Diagnose artwork:** `uv run somafm-tui --diagnose-images` prints the detected protocol, widget class, terminal env, dependency versions and cache path. If `protocol` is `None`, force it with `SOMAFM_IMAGES=tgp` (Kitty) or @@ -315,12 +309,11 @@ same dotfiles/config work everywhere; `XDG_CONFIG_HOME`, `XDG_CACHE_HOME` and - **`SYNC` but the bars are flat?** mpv may be playing to a different sink (or directly to ALSA). Point `SOMAFM_SPECTRUM_DEVICE` at that sink's monitor, or use `--spectrum-source stream`. -- **macOS: spectrum stays `FFT`?** Install a loopback (BlackHole/Loopback) and - route your speakers through a Multi-Output Device, then check - `ffmpeg -f avfoundation -list_devices true -i ""` lists it. The first capture - triggers a macOS microphone-permission prompt; grant it to your terminal - (System Settings → Privacy & Security → Microphone), otherwise ffmpeg cannot - open the device and the analyser stays on `FFT`. +- **macOS: spectrum stays `FFT`?** Grant your terminal **System Audio + Recording** permission (*System Settings → Privacy & Security → System Audio + Recording*), then restart the app; the status bar explains when the tap is + silent because of this. Also check that mpv is playing to the default output + device, since that is what the tap records. - **macOS: media keys do nothing?** Another app currently owns the Now Playing session, or the app was started with `--no-mpris`, or the audio backend fell back to `ffplay` (mpv is required for Now Playing). The status bar shows diff --git a/somafm_tui/__main__.py b/somafm_tui/__main__.py index 8b211c7..51369dd 100644 --- a/somafm_tui/__main__.py +++ b/somafm_tui/__main__.py @@ -55,7 +55,7 @@ def _build_parser() -> argparse.ArgumentParser: choices=SOURCE_MODES, default="auto", help="spectrum audio source: auto (monitor→stream→sim), monitor " - "(parec on Linux, an avfoundation loopback device on macOS), " + "(parec on Linux, a Core Audio process tap on macOS), " "stream (muted second ffmpeg decode), or sim", ) parser.add_argument( diff --git a/somafm_tui/app.py b/somafm_tui/app.py index 4ac5627..44851da 100644 --- a/somafm_tui/app.py +++ b/somafm_tui/app.py @@ -576,6 +576,10 @@ class SomaFMApp(App[None]): frame = self.spectrum.snapshot() self.query_one(SpectrumPanel).sync(self.visualizer, frame.label) + notice = self.spectrum.take_notice() + if notice: + self._set_notice(notice, 10) + if self.tick_count % 4 == 0: self.query_one(StagePanel).refresh_stage() self.query_one(HeaderBar).refresh() diff --git a/somafm_tui/macos_tap.py b/somafm_tui/macos_tap.py new file mode 100644 index 0000000..803312c --- /dev/null +++ b/somafm_tui/macos_tap.py @@ -0,0 +1,576 @@ +"""Native macOS system-audio capture via Core Audio process taps. + +macOS 14.2 added *process taps* to the HAL: a process may ask ``coreaudiod`` +for a copy of the audio that is being rendered, without a kernel extension, a +virtual loopback (BlackHole, Loopback, Soundflower, …), a Multi-Output Device +or a helper binary. The tap is driven here straight through Core Audio's C +API with :mod:`ctypes`, so the only dependency is macOS itself. + +Capture is a three step dance, all of it native: + +1. a ``CATapDescription`` (global stereo mix-down, unmuted) is turned into a + tap object with ``AudioHardwareCreateProcessTap``; +2. the tap is wrapped in a *private* aggregate device so it can be read like + any other input device; +3. an IOProc on that aggregate device delivers interleaved ``float32`` frames + from the HAL's real-time thread, which are buffered for the analyser. + +Tapping other applications' audio is privacy sensitive: the first attempt +makes macOS ask for **System Audio Recording** permission (System Settings ▸ +Privacy & Security). Until that is granted the tap is created successfully +but delivers digital silence, so callers must treat a silent tap as a failed +capture. + +The module is importable everywhere; :func:`available` is ``False`` off macOS +or on systems older than the tap API. +""" + +from __future__ import annotations + +import ctypes +import ctypes.util +import sys +import threading +import time +import uuid +from collections import deque +from typing import Any + +__all__ = ["SystemAudioTap", "TapError", "available"] + +#: ``kAudioObjectPropertyScopeGlobal`` / ``kAudioObjectPropertyElementMain``. +_SCOPE_GLOBAL = int.from_bytes(b"glob", "big") +_ELEMENT_MAIN = 0 + +#: Tap object property selectors (``CoreAudio/AudioHardware.h``). +_TAP_UID = int.from_bytes(b"tuid", "big") +_TAP_FORMAT = int.from_bytes(b"tfmt", "big") + +#: ``kAudioFormatFlagIsFloat | kAudioFormatFlagIsPacked``. +_FORMAT_LPCM = int.from_bytes(b"lpcm", "big") +_FLAG_FLOAT = 0x1 + +#: Aggregate-device description keys (``CoreAudio/AudioHardware.h``). +_AGGREGATE_UID = "uid" +_AGGREGATE_NAME = "name" +_AGGREGATE_PRIVATE = "private" +_AGGREGATE_STACKED = "stacked" +_AGGREGATE_TAPS = "taps" +_AGGREGATE_TAP_AUTOSTART = "tapautostart" +_SUB_TAP_UID = "uid" +_SUB_TAP_DRIFT = "drift" + +_COREAUDIO = "/System/Library/Frameworks/CoreAudio.framework/CoreAudio" +_COREFOUNDATION = "/System/Library/Frameworks/CoreFoundation.framework/CoreFoundation" +_UTF8 = 0x08000100 + +#: Seconds of audio kept in the hand-off buffer before the oldest is dropped. +_BUFFER_SECONDS = 1.0 + + +class TapError(RuntimeError): + """The native audio tap could not be created.""" + + +class _PropertyAddress(ctypes.Structure): + _fields_ = [ + ("mSelector", ctypes.c_uint32), + ("mScope", ctypes.c_uint32), + ("mElement", ctypes.c_uint32), + ] + + +class _StreamFormat(ctypes.Structure): + """``AudioStreamBasicDescription``.""" + + _fields_ = [ + ("mSampleRate", ctypes.c_double), + ("mFormatID", ctypes.c_uint32), + ("mFormatFlags", ctypes.c_uint32), + ("mBytesPerPacket", ctypes.c_uint32), + ("mFramesPerPacket", ctypes.c_uint32), + ("mBytesPerFrame", ctypes.c_uint32), + ("mChannelsPerFrame", ctypes.c_uint32), + ("mBitsPerChannel", ctypes.c_uint32), + ("mReserved", ctypes.c_uint32), + ] + + +class _AudioBuffer(ctypes.Structure): + _fields_ = [ + ("mNumberChannels", ctypes.c_uint32), + ("mDataByteSize", ctypes.c_uint32), + ("mData", ctypes.c_void_p), + ] + + +class _AudioBufferList(ctypes.Structure): + # The HAL hands us a variably sized struct; mNumberBuffers says how many + # entries are valid. The array is oversized so indexing stays in bounds + # for the (few) buffers a tap aggregate ever reports. + _fields_ = [ + ("mNumberBuffers", ctypes.c_uint32), + ("mBuffers", _AudioBuffer * 8), + ] + + +_IOProc = ctypes.CFUNCTYPE( + ctypes.c_int32, + ctypes.c_uint32, # inDevice + ctypes.c_void_p, # inNow + ctypes.POINTER(_AudioBufferList), # inInputData + ctypes.c_void_p, # inInputTime + ctypes.c_void_p, # outOutputData + ctypes.c_void_p, # inOutputTime + ctypes.c_void_p, # inClientData +) + +_loaded: dict[str, Any] = {} +_load_lock = threading.Lock() + + +def _frameworks() -> tuple[ctypes.CDLL, ctypes.CDLL, ctypes.CDLL]: + """Load (and cache) CoreAudio, CoreFoundation and the Objective-C runtime.""" + with _load_lock: + if not _loaded: + ca = ctypes.CDLL(_COREAUDIO) + cf = ctypes.CDLL(_COREFOUNDATION) + objc = ctypes.CDLL(ctypes.util.find_library("objc") or "libobjc.A.dylib") + _configure(ca, cf, objc) + _loaded.update(ca=ca, cf=cf, objc=objc) + return _loaded["ca"], _loaded["cf"], _loaded["objc"] + + +def _signature(lib: ctypes.CDLL, name: str, arguments: list, result: Any) -> None: + """Declare a C function, skipping it when this macOS does not have it.""" + function = getattr(lib, name, None) + if function is None: + return + function.argtypes = arguments + function.restype = result + + +def _configure(ca: ctypes.CDLL, cf: ctypes.CDLL, objc: ctypes.CDLL) -> None: + _signature( + ca, + "AudioObjectGetPropertyData", + [ + ctypes.c_uint32, + ctypes.POINTER(_PropertyAddress), + ctypes.c_uint32, + ctypes.c_void_p, + ctypes.POINTER(ctypes.c_uint32), + ctypes.c_void_p, + ], + ctypes.c_int32, + ) + _signature( + ca, + "AudioHardwareCreateProcessTap", + [ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint32)], + ctypes.c_int32, + ) + _signature(ca, "AudioHardwareDestroyProcessTap", [ctypes.c_uint32], ctypes.c_int32) + _signature( + ca, + "AudioHardwareCreateAggregateDevice", + [ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint32)], + ctypes.c_int32, + ) + _signature( + ca, "AudioHardwareDestroyAggregateDevice", [ctypes.c_uint32], ctypes.c_int32 + ) + _signature( + ca, + "AudioDeviceCreateIOProcID", + [ctypes.c_uint32, _IOProc, ctypes.c_void_p, ctypes.POINTER(ctypes.c_void_p)], + ctypes.c_int32, + ) + _signature(ca, "AudioDeviceDestroyIOProcID", [ctypes.c_uint32, ctypes.c_void_p], ctypes.c_int32) + _signature(ca, "AudioDeviceStart", [ctypes.c_uint32, ctypes.c_void_p], ctypes.c_int32) + _signature(ca, "AudioDeviceStop", [ctypes.c_uint32, ctypes.c_void_p], ctypes.c_int32) + + _signature( + cf, + "CFStringCreateWithCString", + [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_uint32], + ctypes.c_void_p, + ) + _signature( + cf, + "CFDictionaryCreateMutable", + [ctypes.c_void_p, ctypes.c_long, ctypes.c_void_p, ctypes.c_void_p], + ctypes.c_void_p, + ) + _signature( + cf, + "CFDictionarySetValue", + [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p], + None, + ) + _signature( + cf, + "CFArrayCreate", + [ctypes.c_void_p, ctypes.POINTER(ctypes.c_void_p), ctypes.c_long, ctypes.c_void_p], + ctypes.c_void_p, + ) + _signature(cf, "CFRelease", [ctypes.c_void_p], None) + + _signature(objc, "objc_getClass", [ctypes.c_char_p], ctypes.c_void_p) + _signature(objc, "sel_registerName", [ctypes.c_char_p], ctypes.c_void_p) + + +def _sender(result: Any, arguments: tuple[Any, ...]): + """A typed ``objc_msgSend`` for one method signature (arm64 needs this).""" + key = f"msg:{result}:{arguments}" + objc = _frameworks()[2] + with _load_lock: + sender = _loaded.get(key) + if sender is None: + prototype = ctypes.CFUNCTYPE( + result, ctypes.c_void_p, ctypes.c_void_p, *arguments + ) + sender = prototype(("objc_msgSend", objc)) + _loaded[key] = sender + return sender + + +def _selector(name: bytes) -> int: + return int(_frameworks()[2].sel_registerName(name) or 0) + + +def _send_object(receiver: int, name: bytes, *arguments: int) -> int: + sender = _sender(ctypes.c_void_p, (ctypes.c_void_p,) * len(arguments)) + return int(sender(receiver, _selector(name), *arguments) or 0) + + +def _send_flag(receiver: int, name: bytes, value: bool) -> None: + _sender(None, (ctypes.c_bool,))(receiver, _selector(name), value) + + +def _send_integer(receiver: int, name: bytes, value: int) -> None: + _sender(None, (ctypes.c_long,))(receiver, _selector(name), value) + + +def _osstatus(code: int) -> str: + """Render an ``OSStatus`` as its four-character code when it has a printable one.""" + raw = (int(code) & 0xFFFFFFFF).to_bytes(4, "big") + if all(32 <= byte < 127 for byte in raw): + return f"{raw.decode()!r}" + return str(int(code)) + + +def available() -> bool: + """True when this machine can create Core Audio process taps.""" + if sys.platform != "darwin": + return False + try: + ca = _frameworks()[0] + except OSError: + return False + return hasattr(ca, "AudioHardwareCreateProcessTap") + + +class SystemAudioTap: + """Records what the default output device is playing (no driver required). + + ``start()`` raises :class:`TapError` when the tap cannot be created. + Buffers arrive from the HAL as ``float32``; :meth:`read` hands them to the + caller as raw little-endian PCM bytes. + """ + + def __init__(self) -> None: + self.sample_rate = 48000 + self.channels = 2 + self.bytes_per_frame = 8 + self._tag = str(uuid.uuid4()).upper() + self._tap_id: int | None = None + self._device_id: int | None = None + self._io_proc: ctypes.c_void_p | None = None + self._callback: Any = None + self._buffers: deque[bytes] = deque() + self._buffered = 0 + self._limit = int(self.sample_rate * self.bytes_per_frame * _BUFFER_SECONDS) + self._running = False + self._cond = threading.Condition() + + # ----- lifecycle ---------------------------------------------------- # + + def start(self) -> None: + if self._running: + return + if not available(): + raise TapError("Core Audio process taps are unavailable on this system") + + ca, cf, _ = _frameworks() + owned: list[int] = [] + + def cfstr(text: str) -> int: + ref = int(cf.CFStringCreateWithCString(None, text.encode(), _UTF8) or 0) + if not ref: + raise TapError("could not allocate a Core Foundation string") + owned.append(ref) + return ref + + try: + description, nsuuid = self._describe(cfstr) + owned.extend((description, nsuuid)) + + tap_id = ctypes.c_uint32(0) + status = ca.AudioHardwareCreateProcessTap(description, ctypes.byref(tap_id)) + if status != 0: + raise TapError( + f"AudioHardwareCreateProcessTap failed ({_osstatus(status)})" + ) + self._tap_id = tap_id.value + + self._read_format(ca) + uid = self._read_uid(ca) + if not uid: + raise TapError("could not read the tap UID") + owned.append(uid) + + self._build_device(ca, cf, cfstr, uid, owned) + self._start_io(ca) + except Exception: + self._teardown() + raise + finally: + for ref in owned: + cf.CFRelease(ref) + + with self._cond: + self._running = True + self._cond.notify_all() + + def stop(self) -> None: + self._teardown() + with self._cond: + self._running = False + self._buffers.clear() + self._buffered = 0 + self._cond.notify_all() + + @property + def running(self) -> bool: + return self._running + + # ----- capture ------------------------------------------------------ # + + def read(self, timeout: float | None = 1.0) -> bytes | None: + """Return the next buffer of PCM, or ``None`` on timeout / after stop.""" + deadline = None if timeout is None else time.monotonic() + timeout + with self._cond: + while not self._buffers: + if not self._running: + return None + if deadline is None: + self._cond.wait(0.5) + continue + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + self._cond.wait(remaining) + data = self._buffers.popleft() + self._buffered -= len(data) + return data + + # ----- Core Audio plumbing ------------------------------------------ # + + def _describe(self, cfstr) -> tuple[int, int]: + """Build the ``CATapDescription`` for a global, unmuted stereo tap.""" + objc = _frameworks()[2] + cls = objc.objc_getClass(b"CATapDescription") + if not cls: + raise TapError("CATapDescription is unavailable on this system") + + # "…but exclude these processes": an empty list means everything the + # output device plays. + empty = _empty_array() + try: + description = _send_object( + _send_object(int(cls), b"alloc"), + b"initStereoGlobalTapButExcludeProcesses:", + empty, + ) + finally: + # The new CATapDescription has copied the array. + _frameworks()[1].CFRelease(empty) + if not description: + raise TapError("could not create a CATapDescription") + + nsuuid = _send_object( + _send_object(int(objc.objc_getClass(b"NSUUID")), b"alloc"), + b"initWithUUIDString:", + cfstr(self._tag), + ) + if not nsuuid: + raise TapError("could not create the tap UUID") + _send_object(description, b"setUUID:", nsuuid) + _send_object(description, b"setName:", cfstr("somafm-tui")) + _send_flag(description, b"setPrivate:", True) + # 0 == CATapUnmuted: the audio we record keeps playing as usual. + _send_integer(description, b"setMuteBehavior:", 0) + return description, nsuuid + + def _read_uid(self, ca: ctypes.CDLL) -> int: + uid = ctypes.c_void_p(0) + size = ctypes.c_uint32(ctypes.sizeof(ctypes.c_void_p)) + if _get_property(ca, self._tap_id, _TAP_UID, uid, size) != 0: + return 0 + return int(uid.value or 0) + + def _read_format(self, ca: ctypes.CDLL) -> None: + fmt = _StreamFormat() + size = ctypes.c_uint32(ctypes.sizeof(fmt)) + if _get_property(ca, self._tap_id, _TAP_FORMAT, fmt, size) != 0: + raise TapError("could not read the tap's stream format") + if ( + fmt.mFormatID != _FORMAT_LPCM + or fmt.mBitsPerChannel != 32 + or not fmt.mFormatFlags & _FLAG_FLOAT + ): + raise TapError( + "unsupported tap format " + f"({fmt.mFormatID.to_bytes(4, 'big')!r}, {fmt.mBitsPerChannel} bit)" + ) + self.sample_rate = round(fmt.mSampleRate) or 48000 + self.channels = int(fmt.mChannelsPerFrame) or 1 + self.bytes_per_frame = int(fmt.mBytesPerFrame or 4 * self.channels) + self._limit = int(self.sample_rate * self.bytes_per_frame * _BUFFER_SECONDS) + + def _build_device( + self, ca: ctypes.CDLL, cf: ctypes.CDLL, cfstr, uid: int, owned: list[int] + ) -> None: + tap_entry = int(cf.CFDictionaryCreateMutable(None, 0, None, None) or 0) + if not tap_entry: + raise TapError("could not allocate the tap description dictionary") + owned.append(tap_entry) + cf.CFDictionarySetValue(tap_entry, cfstr(_SUB_TAP_UID), uid) + cf.CFDictionarySetValue(tap_entry, cfstr(_SUB_TAP_DRIFT), _boolean(True)) + entries = (ctypes.c_void_p * 1)(tap_entry) + tap_list = int(cf.CFArrayCreate(None, entries, 1, None) or 0) + if not tap_list: + raise TapError("could not allocate the tap list") + owned.append(tap_list) + + aggregate = int(cf.CFDictionaryCreateMutable(None, 0, None, None) or 0) + if not aggregate: + raise TapError("could not allocate the aggregate description") + owned.append(aggregate) + for key, value in ( + (_AGGREGATE_UID, cfstr(str(uuid.uuid4()).upper())), + (_AGGREGATE_NAME, cfstr("somafm-tui tap")), + (_AGGREGATE_PRIVATE, _boolean(True)), + (_AGGREGATE_STACKED, _boolean(False)), + (_AGGREGATE_TAPS, tap_list), + (_AGGREGATE_TAP_AUTOSTART, _boolean(True)), + ): + cf.CFDictionarySetValue(aggregate, cfstr(key), value) + + device_id = ctypes.c_uint32(0) + status = ca.AudioHardwareCreateAggregateDevice( + aggregate, ctypes.byref(device_id) + ) + if status != 0: + raise TapError( + f"AudioHardwareCreateAggregateDevice failed ({_osstatus(status)})" + ) + self._device_id = device_id.value + + def _start_io(self, ca: ctypes.CDLL) -> None: + self._callback = _IOProc(self._on_audio) + io_proc = ctypes.c_void_p() + status = ca.AudioDeviceCreateIOProcID( + self._device_id, self._callback, None, ctypes.byref(io_proc) + ) + if status != 0: + raise TapError(f"AudioDeviceCreateIOProcID failed ({_osstatus(status)})") + self._io_proc = io_proc + status = ca.AudioDeviceStart(self._device_id, io_proc) + if status != 0: + raise TapError(f"AudioDeviceStart failed ({_osstatus(status)})") + + def _teardown(self) -> None: + device = self._device_id + io_proc = self._io_proc + tap = self._tap_id + self._device_id = None + self._io_proc = None + self._tap_id = None + self._callback = None + try: + ca = _frameworks()[0] + except OSError: + return + if device is not None: + if io_proc is not None: + ca.AudioDeviceStop(device, io_proc) + ca.AudioDeviceDestroyIOProcID(device, io_proc) + ca.AudioHardwareDestroyAggregateDevice(device) + if tap is not None: + ca.AudioHardwareDestroyProcessTap(tap) + + # ----- real-time thread --------------------------------------------- # + + def _on_audio( + self, + _device: int, + _now: int, + input_data: ctypes.POINTER(_AudioBufferList), + _input_time: int, + _output_data: int, + _output_time: int, + _client: int, + ) -> int: + try: + chunk = _copy_buffers(input_data) + except Exception: # never let an exception escape into the HAL + return 0 + if not chunk: + return 0 + with self._cond: + self._buffers.append(chunk) + self._buffered += len(chunk) + while self._buffered > self._limit and self._buffers: + self._buffered -= len(self._buffers.popleft()) + self._cond.notify_all() + return 0 + + +def _empty_array() -> int: + cf = _frameworks()[1] + return int(cf.CFArrayCreate(None, None, 0, None) or 0) + + +def _boolean(value: bool) -> int: + cf = _frameworks()[1] + name = "kCFBooleanTrue" if value else "kCFBooleanFalse" + return int(ctypes.c_void_p.in_dll(cf, name).value or 0) + + +def _get_property( + ca: ctypes.CDLL, + obj: int | None, + selector: int, + target: Any, + size: ctypes.c_uint32, +) -> int: + if obj is None: + return -1 + address = _PropertyAddress(selector, _SCOPE_GLOBAL, _ELEMENT_MAIN) + return ca.AudioObjectGetPropertyData( + obj, ctypes.byref(address), 0, None, ctypes.byref(size), ctypes.byref(target) + ) + + +def _copy_buffers(input_data: ctypes.POINTER(_AudioBufferList)) -> bytes: + if not input_data: + return b"" + block = input_data.contents + pieces = [] + for index in range(min(int(block.mNumberBuffers), 8)): + buffer = block.mBuffers[index] + if buffer.mData and buffer.mDataByteSize: + pieces.append(ctypes.string_at(buffer.mData, buffer.mDataByteSize)) + return b"".join(pieces) diff --git a/somafm_tui/spectrum.py b/somafm_tui/spectrum.py index 7804f9f..00df02f 100644 --- a/somafm_tui/spectrum.py +++ b/somafm_tui/spectrum.py @@ -8,11 +8,11 @@ That capture is platform specific: * **Linux** — ``parec`` records the PulseAudio/PipeWire monitor of the default sink (or ``SOMAFM_SPECTRUM_DEVICE``). -* **macOS** — ``ffmpeg -f avfoundation`` records a loopback input device. A - virtual loopback such as BlackHole/Loopback/Soundflower (used as part of a - Multi-Output Device) is auto-detected; ``SOMAFM_SPECTRUM_DEVICE`` picks one - by index or by name. Microphones are deliberately never auto-selected. - The first capture makes macOS ask for microphone permission. +* **macOS** — a native Core Audio *process tap* records the default output + device (see :mod:`somafm_tui.macos_tap`). Nothing to install: no virtual + loopback, no Multi-Output Device, no helper binary. macOS asks for + **System Audio Recording** permission the first time; while it is missing + the tap delivers silence, which is treated as a failed capture. If no output capture is available we fall back to a muted second ffmpeg decode of the stream (status ``FFT``; it can drift relative to playback), and finally @@ -24,7 +24,6 @@ from __future__ import annotations import contextlib import math import os -import re import select import shutil import subprocess @@ -35,6 +34,7 @@ from dataclasses import dataclass import numpy as np +from . import macos_tap from .api import USER_AGENT SAMPLE_RATE = 44100 @@ -46,18 +46,20 @@ MIN_FREQ = 35.0 MAX_FREQ = 16000.0 SOURCE_MODES = ("auto", "monitor", "stream", "sim") -#: ~46 ms of all-zero audio per block; give a loopback ~6 s to prove itself. +#: ~46 ms of all-zero audio per block; give an output capture ~6 s to prove +#: itself before falling back to the stream tap. SILENT_MONITOR_BLOCKS = 130 -#: Device names (lower-case) that look like an output loopback on macOS. -LOOPBACK_HINTS = ( - "blackhole", - "loopback", - "soundflower", - "existential audio", - "multi-output", - "aggregate", -) +#: Frames per analysis block for the macOS tap (float32, ~43 ms at 48 kHz). +TAP_BLOCK_FRAMES = 2048 + +#: Give up (for now) on the tap when the output renders nothing for this long. +#: macOS only drives a tap while the output device is playing, so this is also +#: the window playback gets to start before the stream tap takes over. +TAP_IDLE_SECONDS = 10.0 + +#: How long the stream tap covers for us before the tap is tried again. +TAP_RETRY_SECONDS = 60.0 @dataclass(frozen=True) @@ -102,8 +104,6 @@ class SpectrumEngine: self._parec = shutil.which("parec") self._pactl = shutil.which("pactl") self._darwin = sys.platform == "darwin" - self._avf_checked = False - self._avf_device: int | None = None self._window = np.hanning(fft_size).astype(np.float32) self._groups = self._make_groups() @@ -119,7 +119,9 @@ class SpectrumEngine: self._proc: subprocess.Popen[bytes] | None = None self._capture_kind: str | None = None self._monitor_failed = False + self._tap_retry_at = 0.0 self._monitor_device_cache: str | None = None + self._notice: str | None = None self._lock = threading.Lock() self._stop = threading.Event() self._thread: threading.Thread | None = None @@ -173,6 +175,12 @@ class SpectrumEngine: status=self._status, ) + def take_notice(self) -> str | None: + """One-shot message for the UI (e.g. macOS audio-capture permission).""" + with self._lock: + notice, self._notice = self._notice, None + return notice + def stop(self) -> None: self._stop.set() self._kill_proc() @@ -217,20 +225,23 @@ class SpectrumEngine: proc.wait(timeout=1) def _use_monitor(self) -> bool: - if self._monitor_failed: + if self._monitor_failed or self.source_mode in ("sim", "stream"): return False - if self.source_mode == "sim": + kind = self._monitor_kind() + if kind is None: return False - if self.source_mode == "stream": - return False - return self._monitor_kind() is not None + if kind != "tap": + return True + # The tap needs the default output to actually render something; when + # it does not, the stream tap covers us until the retry deadline. + return time.monotonic() >= self._tap_retry_at def _monitor_kind(self) -> str | None: """Name of the output-capture backend available here, if any.""" if self._parec: return "parec" - if self._darwin and self._ffmpeg and self._avf_loopback() is not None: - return "avfoundation" + if self._darwin and macos_tap.available(): + return "tap" return None def _monitor_device(self) -> str: @@ -260,120 +271,18 @@ class SpectrumEngine: return self._monitor_device_cache def _monitor_command(self) -> list[str] | None: - kind = self._monitor_kind() - if kind == "parec": - return [ - self._parec or "parec", - f"--device={self._monitor_device()}", - "--format=s16le", - "--rate=44100", - "--channels=1", - "--raw", - "--latency-msec=30", - "--client-name=somafm-tui", - ] - if kind == "avfoundation": - device = self._avf_loopback() - if device is None: - return None - return [ - self._ffmpeg or "ffmpeg", - "-hide_banner", - "-loglevel", - "error", - "-nostdin", - "-f", - "avfoundation", - "-i", - # avfoundation wants "video:audio"; we only record audio. - f":{device}", - "-ac", - "1", - "-ar", - str(self.sample_rate), - "-f", - "s16le", - "-acodec", - "pcm_s16le", - "-", - ] - return None - - # ----- macOS: avfoundation loopback ---------------------------------- # - - def _avf_audio_devices(self) -> list[tuple[int, str]]: - """List ``(index, name)`` of the audio inputs ffmpeg can see. - - The first call may be slow (ffmpeg enumerates CoreAudio devices) and - macOS may show a microphone permission prompt. - """ - if not self._ffmpeg: - return [] - try: - result = subprocess.run( - [ - self._ffmpeg, - "-hide_banner", - "-f", - "avfoundation", - "-list_devices", - "true", - "-i", - "", - ], - stdin=subprocess.DEVNULL, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - timeout=8.0, - check=False, - ) - except (OSError, subprocess.SubprocessError): - return [] - devices: list[tuple[int, str]] = [] - in_audio = False - for line in result.stdout.splitlines(): - if "AVFoundation audio devices" in line: - in_audio = True - continue - if "AVFoundation video devices" in line: - in_audio = False - continue - if not in_audio: - continue - match = re.search(r"\[(\d+)\]\s+(.+?)\s*$", line) - if match: - devices.append((int(match.group(1)), match.group(2))) - return devices - - def _avf_loopback(self) -> int | None: - """Pick the avfoundation input device used for the sync capture. - - A virtual loopback (BlackHole, Loopback, Soundflower, …) is preferred. - Microphones are never chosen automatically -- recording the room is - not what the bars are supposed to show. - """ - if self._avf_checked: - return self._avf_device - self._avf_checked = True - devices = self._avf_audio_devices() - override = os.environ.get("SOMAFM_SPECTRUM_DEVICE", "").strip() - if override: - if override.isdigit(): - self._avf_device = int(override) - return self._avf_device - needle = override.casefold() - for index, name in devices: - if needle in name.casefold(): - self._avf_device = index - return index + if self._monitor_kind() != "parec": return None - for index, name in devices: - haystack = name.casefold() - if any(hint in haystack for hint in LOOPBACK_HINTS): - self._avf_device = index - return index - return None + return [ + self._parec or "parec", + f"--device={self._monitor_device()}", + "--format=s16le", + "--rate=44100", + "--channels=1", + "--raw", + "--latency-msec=30", + "--client-name=somafm-tui", + ] def _spawn( self, command: list[str], kind: str @@ -418,24 +327,20 @@ class SpectrumEngine: continue if self._use_monitor(): - command = self._monitor_command() - proc = self._spawn(command, "monitor") if command else None - if proc is not None: - self._capture( - proc, url, stop_on_url_change=False, status="monitor" - ) - self._stop_proc(proc) - if self._stop.is_set(): - break - if self._read_url() is None: - continue - # The monitor device failed (or vanished). Use the stream - # tap for the rest of the session. - self._monitor_failed = True - with self._lock: - self._status = "sim" + outcome = self._capture_monitor(url) + if outcome == "done": + continue + if outcome == "retry": + # Nothing was rendered (yet): try the tap again later. + self._tap_retry_at = time.monotonic() + TAP_RETRY_SECONDS + else: + # Device gone, or the macOS audio-capture permission was + # never granted: do not retry in this session. + self._monitor_failed = True + if self._stop.is_set(): + break + if self._read_url() is None: continue - self._monitor_failed = True if not self._ffmpeg: self._simulate_for(url, 0.05) @@ -498,6 +403,92 @@ class SpectrumEngine: "-", ] + def _capture_monitor(self, url: str) -> str: + """Capture the real audio output. + + Returns ``"done"`` when playback stopped, ``"failed"`` when the + monitor is unusable, and ``"retry"`` when it merely had nothing to + capture yet. + """ + if self._monitor_kind() == "tap": + return self._capture_tap() + command = self._monitor_command() + proc = self._spawn(command, "monitor") if command else None + if proc is None: + return "failed" + try: + self._capture(proc, url, stop_on_url_change=False, status="monitor") + finally: + self._stop_proc(proc) + if self._stop.is_set() or self._read_url() is None: + return "done" + return "failed" + + def _capture_tap(self) -> str: + """Capture the native macOS output tap.""" + tap = macos_tap.SystemAudioTap() + try: + tap.start() + except macos_tap.TapError as exc: + self._notify(f"macOS audio capture failed ({exc}) — using FFT") + return "failed" + + with self._lock: + self._capture_kind = "monitor" + self._status = "monitor" + try: + self._retune(tap.sample_rate) + return self._pump_tap(tap) + finally: + tap.stop() + with self._lock: + if self._capture_kind == "monitor": + self._capture_kind = None + + def _pump_tap(self, tap: macos_tap.SystemAudioTap) -> str: + """Feed tap audio to the analyser until playback stops or it fails.""" + channels = max(1, tap.channels) + block_bytes = TAP_BLOCK_FRAMES * max(4, tap.bytes_per_frame) + buffer = bytearray() + silent_blocks = 0 + last_audio = time.monotonic() + while not self._stop.is_set(): + if self._read_url() is None: + return "done" + chunk = tap.read(0.5) + now = time.monotonic() + if chunk is None: + # A tap only ticks while the output device renders something; + # give playback a moment to start before giving up on it. + if now - last_audio >= TAP_IDLE_SECONDS: + self._notify( + "nothing is playing on the default output device — " + "using the stream tap (FFT)" + ) + return "retry" + continue + last_audio = now + buffer.extend(chunk) + # Fixed-size blocks keep attack/decay smoothing tied to time. + while len(buffer) >= block_bytes: + block = bytes(buffer[:block_bytes]) + del buffer[:block_bytes] + # A tap that was never granted permission is created happily + # but only ever hands out digital silence; detect that here. + if not block.strip(b"\x00"): + silent_blocks += 1 + if silent_blocks >= SILENT_MONITOR_BLOCKS: + self._notify( + "macOS is not letting the app record audio — allow " + "'System Audio Recording' for your terminal in " + "System Settings, then restart (using FFT)" + ) + return "failed" + continue + silent_blocks = 0 + self.feed_audio(block, channels, "monitor") + return "done" + def _capture( self, proc: subprocess.Popen[bytes], @@ -544,10 +535,11 @@ class SpectrumEngine: while len(buffer) >= READ_BYTES: block = bytes(buffer[:READ_BYTES]) del buffer[:READ_BYTES] - # A loopback device that nobody routes into delivers endless - # digital silence. SYNC bars that never move are worse than - # the (slightly drifting) stream tap, so declare the monitor - # dead and let the caller fall back. + # A monitor that was never granted permission (macOS) or that + # nobody routes into (PulseAudio) delivers endless digital + # silence. SYNC bars that never move are worse than the + # (slightly drifting) stream tap, so give up and let the + # caller fall back. if status == "monitor": silent_blocks = silent_blocks + 1 if block.count(0) == len(block) else 0 if silent_blocks >= SILENT_MONITOR_BLOCKS: @@ -573,6 +565,36 @@ class SpectrumEngine: return self._process(samples.astype(np.float32) / 32768.0, status) + def feed_audio(self, data: bytes, channels: int = 1, status: str = "monitor") -> None: + """Feed interleaved little-endian ``float32`` PCM (the macOS tap).""" + usable = len(data) - (len(data) % 4) + if usable <= 0: + return + try: + samples = np.frombuffer(data[:usable], dtype=" 1: + frames = samples.size - (samples.size % channels) + samples = samples[:frames].reshape(-1, channels).mean(axis=1) + self._process(np.ascontiguousarray(samples, dtype=np.float32), status) + + def _retune(self, sample_rate: int) -> None: + """Rebuild the FFT window and band edges for a different capture rate.""" + rate = int(sample_rate) or self.sample_rate + if rate == self.sample_rate: + return + self.sample_rate = rate + self._window = np.hanning(self.fft_size).astype(np.float32) + self._groups = self._make_groups() + + def _notify(self, message: str) -> None: + with self._lock: + self._notice = message + def _process(self, samples: np.ndarray, status: str = "live") -> None: size = self.fft_size if samples.size >= size: diff --git a/somafm_tui/widgets/help.py b/somafm_tui/widgets/help.py index 7884f1a..4df9eee 100644 --- a/somafm_tui/widgets/help.py +++ b/somafm_tui/widgets/help.py @@ -53,10 +53,11 @@ ABOUT_LINUX = ( ABOUT_MACOS = ( ( - "The spectrum follows your audio output (SYNC) when a loopback device " - "such as BlackHole is available for ffmpeg/avfoundation; otherwise it " - "falls back to a muted ffmpeg stream tap (FFT, can drift) or a " - "synthesized display (SIM)." + "The spectrum captures your Mac's real audio output (SYNC) with a " + "native Core Audio process tap — nothing to install. macOS asks for " + "System Audio Recording permission the first time; without it the " + "analyser falls back to a muted ffmpeg stream tap (FFT, can drift) " + "or a synthesized display (SIM)." ), ( "mpv publishes what is playing to the macOS Now Playing session, so " diff --git a/tests/test_macos_tap.py b/tests/test_macos_tap.py new file mode 100644 index 0000000..60b2dc8 --- /dev/null +++ b/tests/test_macos_tap.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import pytest + +from somafm_tui import macos_tap + + +def test_osstatus_decodes_four_character_codes() -> None: + # 'who?' is kAudioHardwareBadObjectError. + assert macos_tap._osstatus(int.from_bytes(b"who?", "big")) == "'who?'" + assert macos_tap._osstatus(-50) == "-50" + + +def test_start_raises_without_the_native_api(monkeypatch) -> None: + monkeypatch.setattr(macos_tap, "available", lambda: False) + tap = macos_tap.SystemAudioTap() + with pytest.raises(macos_tap.TapError): + tap.start() + assert tap.running is False + tap.stop() # tearing down a tap that never started is safe + + +def test_read_returns_none_when_not_running() -> None: + tap = macos_tap.SystemAudioTap() + assert tap.read(0.0) is None + + +def test_uses_no_external_helpers() -> None: + # The tap must never shell out: no helper binary, no ffmpeg, no driver. + import inspect + + source = inspect.getsource(macos_tap).casefold() + assert "subprocess" not in source + assert "avfoundation" not in source diff --git a/tests/test_spectrum.py b/tests/test_spectrum.py index fb7590a..0d4cf6f 100644 --- a/tests/test_spectrum.py +++ b/tests/test_spectrum.py @@ -29,6 +29,55 @@ def write_pcm_script(path: Path) -> None: 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(" 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() @@ -140,91 +189,118 @@ def test_stream_tap_pipeline(tmp_path) -> None: 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 +def test_macos_tap_is_the_sync_monitor(monkeypatch) -> None: + engine = SpectrumEngine(ffmpeg_path="") + engine._parec = None 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" + 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" - 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" + 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_avfoundation_never_autoselects_a_microphone(monkeypatch) -> None: - engine = SpectrumEngine(ffmpeg_path="/opt/homebrew/bin/ffmpeg") +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.delenv("SOMAFM_SPECTRUM_DEVICE", raising=False) + 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( - 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" + spectrum_module.macos_tap, + "SystemAudioTap", + lambda: FakeTap(error=spectrum_module.macos_tap.TapError("boom")), ) - 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"), - ] + 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_dead_loopback_falls_back_instead_of_flat_bars(tmp_path) -> None: +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"