Adolfo Reyna
358 lines
16 KiB
Python
358 lines
16 KiB
Python
"""macOS default audio-device monitoring.
|
|
|
|
The monitor deliberately depends on a small backend protocol. Core Audio invokes
|
|
listeners on an arbitrary thread, while VoiceAgent consumes snapshots on its
|
|
asyncio loop; :class:`AudioDeviceMonitor` is the lifecycle and thread boundary
|
|
between those two worlds. The backend can be replaced by a deterministic fake
|
|
in unit tests.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import ctypes
|
|
import ctypes.util
|
|
import inspect
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Callable, Protocol
|
|
|
|
|
|
class DeviceChangeReason(str, Enum):
|
|
INITIAL = "initial"
|
|
DEFAULT_CHANGED = "default_changed"
|
|
DEVICE_ADDED = "device_added"
|
|
DEVICE_REMOVED = "device_removed"
|
|
DEVICE_RECONFIGURED = "device_reconfigured"
|
|
PROFILE_CHANGED = "profile_changed"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AudioDevice:
|
|
uid: str
|
|
name: str
|
|
can_input: bool
|
|
can_output: bool
|
|
transport: str = "unknown"
|
|
alive: bool = True
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AudioDeviceSnapshot:
|
|
generation: int
|
|
default_input_uid: str | None
|
|
default_output_uid: str | None
|
|
devices: dict[str, AudioDevice] = field(default_factory=dict)
|
|
reason: DeviceChangeReason = DeviceChangeReason.INITIAL
|
|
|
|
|
|
Listener = Callable[[], None]
|
|
|
|
|
|
class AudioDeviceBackend(Protocol):
|
|
"""The minimal Core Audio surface required by the monitor."""
|
|
|
|
def enumerate_devices(self) -> list[AudioDevice]: ...
|
|
def default_input_uid(self) -> str | None: ...
|
|
def default_output_uid(self) -> str | None: ...
|
|
def add_listener(self, callback: Listener) -> object: ...
|
|
def remove_listener(self, token: object) -> None: ...
|
|
|
|
|
|
class AudioDeviceMonitor:
|
|
"""Coalesces Core Audio notifications and dispatches snapshots on one loop."""
|
|
|
|
def __init__(self, backend: AudioDeviceBackend, *, debounce_seconds: float = 0.05,
|
|
loop: asyncio.AbstractEventLoop | None = None):
|
|
self._backend = backend
|
|
self._debounce_seconds = debounce_seconds
|
|
self._loop = loop
|
|
self._callback: Callable[[AudioDeviceSnapshot], object] | None = None
|
|
self._snapshot = AudioDeviceSnapshot(0, None, None)
|
|
self._generation = 0
|
|
self._listener_tokens: list[object] = []
|
|
self._debounce_handle: asyncio.TimerHandle | None = None
|
|
self._running = False
|
|
self._refresh_scheduled = False
|
|
self._refresh_task: asyncio.Task[None] | None = None
|
|
|
|
async def start(self, on_change: Callable[[AudioDeviceSnapshot], object]) -> None:
|
|
"""Register listeners once and publish an initial snapshot.
|
|
|
|
``on_change`` may be synchronous or return an awaitable. It is always
|
|
called on the loop used by this monitor, never on a Core Audio thread.
|
|
"""
|
|
if self._running:
|
|
return
|
|
self._loop = self._loop or asyncio.get_running_loop()
|
|
self._callback = on_change
|
|
self._running = True
|
|
try:
|
|
# Register all listeners before the initial read so a concurrent
|
|
# device change cannot be missed.
|
|
self._listener_tokens = [self._backend.add_listener(self._on_backend_event) for _ in range(1)]
|
|
await self._refresh(DeviceChangeReason.INITIAL)
|
|
except Exception:
|
|
await self.stop()
|
|
raise
|
|
|
|
async def stop(self) -> None:
|
|
"""Stop idempotently and ensure no callback can be scheduled afterward."""
|
|
self._running = False
|
|
if self._debounce_handle:
|
|
self._debounce_handle.cancel()
|
|
self._debounce_handle = None
|
|
current_task = asyncio.current_task()
|
|
if (self._refresh_task and not self._refresh_task.done()
|
|
and self._refresh_task is not current_task):
|
|
self._refresh_task.cancel()
|
|
try:
|
|
await self._refresh_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
self._refresh_task = None
|
|
tokens, self._listener_tokens = self._listener_tokens, []
|
|
for token in tokens:
|
|
self._backend.remove_listener(token)
|
|
self._callback = None
|
|
self._refresh_scheduled = False
|
|
|
|
def snapshot(self) -> AudioDeviceSnapshot:
|
|
return self._snapshot
|
|
|
|
def _on_backend_event(self) -> None:
|
|
"""Core Audio callback entry point; safe to call from any thread."""
|
|
if not self._running or not self._loop:
|
|
return
|
|
self._loop.call_soon_threadsafe(self._schedule_refresh)
|
|
|
|
def _schedule_refresh(self) -> None:
|
|
if not self._running or self._refresh_scheduled:
|
|
return
|
|
self._refresh_scheduled = True
|
|
if self._debounce_handle:
|
|
self._debounce_handle.cancel()
|
|
self._debounce_handle = self._loop.call_later(self._debounce_seconds, self._start_refresh)
|
|
|
|
def _start_refresh(self) -> None:
|
|
self._debounce_handle = None
|
|
self._refresh_scheduled = False
|
|
if self._running:
|
|
self._refresh_task = self._loop.create_task(self._refresh(self._infer_reason()))
|
|
self._refresh_task.add_done_callback(self._consume_refresh_failure)
|
|
|
|
@staticmethod
|
|
def _consume_refresh_failure(task: asyncio.Task[None]) -> None:
|
|
"""Retrieve scheduled refresh failures after cleanup has completed."""
|
|
if not task.cancelled():
|
|
task.exception()
|
|
|
|
def _infer_reason(self) -> DeviceChangeReason:
|
|
# The backend intentionally keeps the callback payload-free. The
|
|
# resulting snapshot is authoritative; callers can inspect UID/device
|
|
# differences. A generic device reconfiguration is safest here.
|
|
return DeviceChangeReason.DEFAULT_CHANGED
|
|
|
|
async def _refresh(self, reason: DeviceChangeReason) -> None:
|
|
if not self._running:
|
|
return
|
|
devices = {d.uid: d for d in self._backend.enumerate_devices() if d.uid and d.alive}
|
|
new_input = self._backend.default_input_uid()
|
|
new_output = self._backend.default_output_uid()
|
|
# An unavailable default is represented as None rather than a stale UID.
|
|
if new_input not in devices or not devices[new_input].can_input:
|
|
new_input = None
|
|
if new_output not in devices or not devices[new_output].can_output:
|
|
new_output = None
|
|
old = self._snapshot
|
|
if (old.default_input_uid == new_input and old.default_output_uid == new_output
|
|
and old.devices == devices and old.generation != 0):
|
|
return
|
|
if old.generation and reason != DeviceChangeReason.INITIAL:
|
|
old_uids, new_uids = set(old.devices), set(devices)
|
|
if old.default_input_uid != new_input or old.default_output_uid != new_output:
|
|
reason = DeviceChangeReason.DEFAULT_CHANGED
|
|
elif new_uids - old_uids:
|
|
reason = DeviceChangeReason.DEVICE_ADDED
|
|
elif old_uids - new_uids:
|
|
reason = DeviceChangeReason.DEVICE_REMOVED
|
|
elif any(old.devices[uid] != device for uid, device in devices.items()
|
|
if uid in old.devices):
|
|
reason = DeviceChangeReason.PROFILE_CHANGED
|
|
self._generation += 1
|
|
self._snapshot = AudioDeviceSnapshot(self._generation, new_input, new_output, devices, reason)
|
|
callback = self._callback
|
|
if callback and self._running:
|
|
try:
|
|
result = callback(self._snapshot)
|
|
if inspect.isawaitable(result):
|
|
await result
|
|
except Exception:
|
|
# A failed consumer must not leave a native listener active with
|
|
# an unusable callback. ``stop`` handles the current refresh
|
|
# task specially so this cleanup is safe from inside _refresh.
|
|
await self.stop()
|
|
raise
|
|
|
|
|
|
class MacOSCoreAudioBackend:
|
|
"""Core Audio backend hook.
|
|
|
|
PyObjC's CoreAudio listener ABI differs between macOS releases. Keeping the
|
|
native adapter behind this class lets packaging provide the matching adapter
|
|
without exposing it to the async monitor or its tests.
|
|
"""
|
|
|
|
def __init__(self, adapter):
|
|
self._adapter = adapter
|
|
|
|
def enumerate_devices(self) -> list[AudioDevice]:
|
|
return list(self._adapter.enumerate_devices())
|
|
|
|
def default_input_uid(self) -> str | None:
|
|
return self._adapter.default_input_uid()
|
|
|
|
def default_output_uid(self) -> str | None:
|
|
return self._adapter.default_output_uid()
|
|
|
|
def add_listener(self, callback: Listener) -> object:
|
|
return self._adapter.add_device_listener(callback)
|
|
|
|
def remove_listener(self, token: object) -> None:
|
|
self._adapter.remove_device_listener(token)
|
|
|
|
|
|
class NativeMacOSCoreAudioAdapter:
|
|
"""Native Core Audio adapter using stable device UIDs, never PortAudio IDs."""
|
|
|
|
_SYSTEM_OBJECT = 1
|
|
_GLOBAL = int.from_bytes(b"glob", "big")
|
|
_INPUT_SCOPE = int.from_bytes(b"inpt", "big")
|
|
_OUTPUT_SCOPE = int.from_bytes(b"outp", "big")
|
|
_DEFAULT_INPUT = int.from_bytes(b"dIn ", "big")
|
|
_DEFAULT_OUTPUT = int.from_bytes(b"dOut", "big")
|
|
_DEVICES = int.from_bytes(b"dev#", "big")
|
|
_UID = int.from_bytes(b"uid ", "big")
|
|
_NAME = int.from_bytes(b"lnam", "big")
|
|
_ALIVE = int.from_bytes(b"livn", "big")
|
|
_TRANSPORT = int.from_bytes(b"tran", "big")
|
|
_STREAMS = int.from_bytes(b"stm#", "big")
|
|
|
|
class _Address(ctypes.Structure):
|
|
_fields_ = [("selector", ctypes.c_uint32), ("scope", ctypes.c_uint32), ("element", ctypes.c_uint32)]
|
|
|
|
def __init__(self):
|
|
if sys.platform != "darwin":
|
|
raise RuntimeError("Core Audio is only available on macOS")
|
|
core_audio = ctypes.util.find_library("CoreAudio")
|
|
core_foundation = ctypes.util.find_library("CoreFoundation")
|
|
if not core_audio or not core_foundation:
|
|
raise RuntimeError("CoreAudio.framework is unavailable")
|
|
self._lib = ctypes.CDLL(core_audio)
|
|
self._cf = ctypes.CDLL(core_foundation)
|
|
self._listener_type = ctypes.CFUNCTYPE(ctypes.c_int32, ctypes.c_uint32, ctypes.c_uint32,
|
|
ctypes.POINTER(self._Address), ctypes.c_void_p)
|
|
self._lib.AudioObjectGetPropertyData.argtypes = [ctypes.c_uint32, ctypes.POINTER(self._Address), ctypes.c_uint32, ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint32), ctypes.c_void_p]
|
|
self._lib.AudioObjectGetPropertyData.restype = ctypes.c_int32
|
|
self._lib.AudioObjectGetPropertyDataSize.argtypes = [ctypes.c_uint32, ctypes.POINTER(self._Address), ctypes.c_uint32, ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint32)]
|
|
self._lib.AudioObjectGetPropertyDataSize.restype = ctypes.c_int32
|
|
self._lib.AudioObjectAddPropertyListener.argtypes = [ctypes.c_uint32, ctypes.POINTER(self._Address), self._listener_type, ctypes.c_void_p]
|
|
self._lib.AudioObjectRemovePropertyListener.argtypes = [ctypes.c_uint32, ctypes.POINTER(self._Address), self._listener_type, ctypes.c_void_p]
|
|
self._cf.CFStringGetCString.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_long, ctypes.c_uint32]
|
|
self._callbacks = {}
|
|
|
|
def _address(self, selector, scope=None):
|
|
return self._Address(selector, self._GLOBAL if scope is None else scope, 0)
|
|
|
|
def _value(self, object_id, selector, value_type, scope=None):
|
|
value, size = value_type(), ctypes.c_uint32(ctypes.sizeof(value_type))
|
|
status = self._lib.AudioObjectGetPropertyData(object_id, ctypes.byref(self._address(selector, scope)), 0, None, ctypes.byref(size), ctypes.byref(value))
|
|
if status:
|
|
raise OSError(f"AudioObjectGetPropertyData({selector}) failed: {status}")
|
|
return value.value
|
|
|
|
def _string(self, object_id, selector):
|
|
ref = self._value(object_id, selector, ctypes.c_void_p)
|
|
if not ref:
|
|
return ""
|
|
buffer = ctypes.create_string_buffer(1024)
|
|
if not self._cf.CFStringGetCString(ref, buffer, len(buffer), 0x08000100):
|
|
return ""
|
|
return buffer.value.decode("utf-8", "replace")
|
|
|
|
def _device_ids(self):
|
|
address, size = self._address(self._DEVICES), ctypes.c_uint32()
|
|
status = self._lib.AudioObjectGetPropertyDataSize(self._SYSTEM_OBJECT, ctypes.byref(address), 0, None, ctypes.byref(size))
|
|
if status:
|
|
raise OSError(f"Audio device enumeration failed: {status}")
|
|
devices = (ctypes.c_uint32 * (size.value // ctypes.sizeof(ctypes.c_uint32)))()
|
|
status = self._lib.AudioObjectGetPropertyData(self._SYSTEM_OBJECT, ctypes.byref(address), 0, None, ctypes.byref(size), devices)
|
|
if status:
|
|
raise OSError(f"Audio device enumeration failed: {status}")
|
|
return list(devices)
|
|
|
|
def _has_streams(self, device_id, scope):
|
|
address, size = self._address(self._STREAMS, scope), ctypes.c_uint32()
|
|
status = self._lib.AudioObjectGetPropertyDataSize(device_id, ctypes.byref(address), 0, None, ctypes.byref(size))
|
|
return not status and bool(size.value)
|
|
|
|
def enumerate_devices(self):
|
|
devices = []
|
|
for device_id in self._device_ids():
|
|
try:
|
|
uid = self._string(device_id, self._UID)
|
|
if not uid:
|
|
continue
|
|
devices.append(AudioDevice(uid, self._string(device_id, self._NAME) or uid,
|
|
self._has_streams(device_id, self._INPUT_SCOPE), self._has_streams(device_id, self._OUTPUT_SCOPE),
|
|
self._fourcc(self._value(device_id, self._TRANSPORT, ctypes.c_uint32)), bool(self._value(device_id, self._ALIVE, ctypes.c_uint32))))
|
|
except OSError:
|
|
continue
|
|
return devices
|
|
|
|
@staticmethod
|
|
def _fourcc(value):
|
|
return value.to_bytes(4, "big").decode("ascii", "replace").strip() or "unknown"
|
|
|
|
def default_input_uid(self):
|
|
return self._uid_for_id(self._value(self._SYSTEM_OBJECT, self._DEFAULT_INPUT, ctypes.c_uint32))
|
|
|
|
def default_output_uid(self):
|
|
return self._uid_for_id(self._value(self._SYSTEM_OBJECT, self._DEFAULT_OUTPUT, ctypes.c_uint32))
|
|
|
|
def _uid_for_id(self, device_id):
|
|
return self._string(device_id, self._UID) if device_id else None
|
|
|
|
def add_device_listener(self, callback):
|
|
addresses = [(self._SYSTEM_OBJECT, self._address(selector)) for selector in (self._DEFAULT_INPUT, self._DEFAULT_OUTPUT, self._DEVICES)]
|
|
addresses += [(device_id, self._address(selector, scope)) for device_id in self._device_ids() for selector, scope in ((self._ALIVE, None), (self._STREAMS, self._INPUT_SCOPE), (self._STREAMS, self._OUTPUT_SCOPE))]
|
|
native_callback = self._listener_type(lambda *_: (callback(), 0)[1])
|
|
registered = []
|
|
try:
|
|
for object_id, address in addresses:
|
|
status = self._lib.AudioObjectAddPropertyListener(object_id, ctypes.byref(address), native_callback, None)
|
|
if status:
|
|
raise OSError(f"AudioObjectAddPropertyListener failed: {status}")
|
|
registered.append((object_id, address))
|
|
except Exception:
|
|
for object_id, address in registered:
|
|
self._lib.AudioObjectRemovePropertyListener(object_id, ctypes.byref(address), native_callback, None)
|
|
raise
|
|
token = (native_callback, registered)
|
|
self._callbacks[id(token)] = token
|
|
return token
|
|
|
|
def remove_device_listener(self, token):
|
|
native_callback, addresses = token
|
|
for object_id, address in addresses:
|
|
self._lib.AudioObjectRemovePropertyListener(object_id, ctypes.byref(address), native_callback, None)
|
|
self._callbacks.pop(id(token), None)
|
|
|
|
|
|
def create_macos_audio_monitor(*, adapter=None, **kwargs) -> AudioDeviceMonitor:
|
|
"""Create a native Core Audio monitor (or a supplied test adapter)."""
|
|
if adapter is None:
|
|
adapter = NativeMacOSCoreAudioAdapter()
|
|
return AudioDeviceMonitor(MacOSCoreAudioBackend(adapter), **kwargs)
|