audiodev.py

#!/usr/bin/env python3
"""
audiodev.py - inspect, identify, retune and reinitialize a single CoreAudio device
on macOS, without restarting coreaudiod (which disrupts every device and app).

No dependencies beyond a working python3: it calls the CoreAudio C API via ctypes.

Usage:
    ./audiodev.py list                      # every device: id, UID, transport, rate, state
    ./audiodev.py identify [-d SEL]         # speak through each match so you can hear which is which
    ./audiodev.py rate 48000 [-d SEL]
    ./audiodev.py reset [-d SEL] [-m hog|rate|default|all]

SEL (-d/--device) accepts any of:
    42                  a numeric AudioObjectID, exactly (see `list`)
    "uid:AppleHDA..."   a substring of the device UID  <- stable across reboots
    "Studio Display"    a substring of the device name
Name matching that hits more than one device is refused, and the candidates are
printed with their ids and UIDs - that is the case to solve with `identify`.

What "reset" does, since CoreAudio exposes no public "reinitialize" call:

  hog      Claim kAudioDevicePropertyHogMode (exclusive access), then release it.
           Taking it evicts every other client from THIS device; releasing it makes
           them re-open and re-negotiate. Closest thing to a per-device power cycle.
  rate     Bounce the nominal sample rate to another supported value and back,
           forcing the driver to rebuild its stream format and IO buffers.
  default  If the device is the current default output, point the system default at
           another device and back, so CoreAudio destroys and recreates the IOProc.

For HDMI/DisplayPort outputs, remember the audio endpoint rides on the display link:
if `list` reports the device alive and running but you hear nothing, the fault is
below CoreAudio and you need to relink the display, not reset the audio device.
"""

import argparse
import ctypes
import os
import struct
import subprocess
import sys
import time

ca = ctypes.CDLL("/System/Library/Frameworks/CoreAudio.framework/CoreAudio")
cf = ctypes.CDLL("/System/Library/Frameworks/CoreFoundation.framework/CoreFoundation")


def fcc(s):
    """CoreAudio four-char-code selector -> uint32. Note some codes contain a space."""
    return struct.unpack(">I", s.encode())[0]


kAudioObjectSystemObject = 1
kAudioHardwarePropertyDevices = fcc("dev#")
kAudioHardwarePropertyDefaultOutputDevice = fcc("dOut")
kAudioDevicePropertyNominalSampleRate = fcc("nsrt")
kAudioDevicePropertyAvailableNominalSampleRates = fcc("nsr#")
kAudioDevicePropertyStreams = fcc("stm#")
kAudioDevicePropertyHogMode = fcc("oink")
kAudioDevicePropertyDeviceIsAlive = fcc("livn")
kAudioDevicePropertyDeviceIsRunning = fcc("goin")
kAudioDevicePropertyDeviceUID = fcc("uid ")
kAudioDevicePropertyModelUID = fcc("muid")
kAudioDevicePropertyTransportType = fcc("tran")
kAudioObjectPropertyName = fcc("lnam")
kAudioObjectPropertyScopeGlobal = fcc("glob")
kAudioObjectPropertyScopeOutput = fcc("outp")
kAudioObjectPropertyElementMain = 0
kCFStringEncodingUTF8 = 0x08000100

TRANSPORTS = {
    fcc("bltn"): "built-in", fcc("usb "): "USB", fcc("hdmi"): "HDMI",
    fcc("dprt"): "DisplayPort", fcc("blue"): "Bluetooth", fcc("airp"): "AirPlay",
    fcc("virt"): "virtual", fcc("aggr"): "aggregate", fcc("pci "): "PCI",
    fcc("fire"): "FireWire", fcc("thun"): "Thunderbolt", fcc("unkn"): "unknown",
}


class Addr(ctypes.Structure):
    _fields_ = [("mSelector", ctypes.c_uint32),
                ("mScope", ctypes.c_uint32),
                ("mElement", ctypes.c_uint32)]


class AudioValueRange(ctypes.Structure):
    _fields_ = [("mMinimum", ctypes.c_double), ("mMaximum", ctypes.c_double)]


ca.AudioObjectGetPropertyDataSize.argtypes = [
    ctypes.c_uint32, ctypes.POINTER(Addr), ctypes.c_uint32,
    ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint32)]
ca.AudioObjectGetPropertyData.argtypes = [
    ctypes.c_uint32, ctypes.POINTER(Addr), ctypes.c_uint32,
    ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint32), ctypes.c_void_p]
ca.AudioObjectSetPropertyData.argtypes = [
    ctypes.c_uint32, ctypes.POINTER(Addr), ctypes.c_uint32,
    ctypes.c_void_p, ctypes.c_uint32, ctypes.c_void_p]
cf.CFStringGetCString.argtypes = [ctypes.c_void_p, ctypes.c_char_p,
                                  ctypes.c_long, ctypes.c_uint32]
cf.CFRelease.argtypes = [ctypes.c_void_p]


def oserr(code):
    raw = struct.pack(">i", code)
    if all(32 <= b < 127 for b in raw):
        return "%d ('%s')" % (code, raw.decode("ascii"))
    return str(code)


def _addr(selector, scope=kAudioObjectPropertyScopeGlobal):
    return Addr(selector, scope, kAudioObjectPropertyElementMain)


def prop_size(obj, selector, scope=kAudioObjectPropertyScopeGlobal):
    a = _addr(selector, scope)
    size = ctypes.c_uint32(0)
    err = ca.AudioObjectGetPropertyDataSize(obj, ctypes.byref(a), 0, None, ctypes.byref(size))
    if err:
        raise OSError("CoreAudio error %s reading property size" % oserr(err))
    return size.value


def get_scalar(obj, selector, ctype, scope=kAudioObjectPropertyScopeGlobal):
    a = _addr(selector, scope)
    size = ctypes.c_uint32(ctypes.sizeof(ctype))
    val = ctype()
    err = ca.AudioObjectGetPropertyData(obj, ctypes.byref(a), 0, None,
                                        ctypes.byref(size), ctypes.byref(val))
    if err:
        raise OSError("CoreAudio error %s reading property" % oserr(err))
    return val.value


def set_scalar(obj, selector, value, scope=kAudioObjectPropertyScopeGlobal):
    a = _addr(selector, scope)
    err = ca.AudioObjectSetPropertyData(obj, ctypes.byref(a), 0, None,
                                        ctypes.sizeof(value), ctypes.byref(value))
    if err:
        raise OSError("CoreAudio error %s writing property" % oserr(err))


def get_array(obj, selector, ctype, scope=kAudioObjectPropertyScopeGlobal):
    nbytes = prop_size(obj, selector, scope)
    count = nbytes // ctypes.sizeof(ctype)
    if count == 0:
        return []
    buf = (ctype * count)()
    a = _addr(selector, scope)
    size = ctypes.c_uint32(nbytes)
    err = ca.AudioObjectGetPropertyData(obj, ctypes.byref(a), 0, None,
                                        ctypes.byref(size), ctypes.byref(buf))
    if err:
        raise OSError("CoreAudio error %s reading array property" % oserr(err))
    return list(buf)


def get_string(obj, selector):
    """Read a CFStringRef property and return it as a Python str."""
    a = _addr(selector)
    size = ctypes.c_uint32(ctypes.sizeof(ctypes.c_void_p))
    ref = ctypes.c_void_p()
    if ca.AudioObjectGetPropertyData(obj, ctypes.byref(a), 0, None,
                                     ctypes.byref(size), ctypes.byref(ref)):
        return ""
    buf = ctypes.create_string_buffer(1024)
    ok = cf.CFStringGetCString(ref, buf, 1024, kCFStringEncodingUTF8)
    cf.CFRelease(ref)
    return buf.value.decode("utf-8", "replace") if ok else ""


def device_name(dev):
    return get_string(dev, kAudioObjectPropertyName) or "<unnamed %d>" % dev


def device_uid(dev):
    return get_string(dev, kAudioDevicePropertyDeviceUID)


def transport(dev):
    try:
        return TRANSPORTS.get(get_scalar(dev, kAudioDevicePropertyTransportType,
                                         ctypes.c_uint32), "?")
    except OSError:
        return "?"


def has_output(dev):
    try:
        return prop_size(dev, kAudioDevicePropertyStreams, kAudioObjectPropertyScopeOutput) > 0
    except OSError:
        return False


def all_devices():
    return get_array(kAudioObjectSystemObject, kAudioHardwarePropertyDevices, ctypes.c_uint32)


def output_devices():
    return [d for d in all_devices() if has_output(d)]


def default_output():
    return get_scalar(kAudioObjectSystemObject,
                      kAudioHardwarePropertyDefaultOutputDevice, ctypes.c_uint32)


def set_default_output(dev):
    set_scalar(kAudioObjectSystemObject, kAudioHardwarePropertyDefaultOutputDevice,
               ctypes.c_uint32(dev))


def current_rate(dev):
    return get_scalar(dev, kAudioDevicePropertyNominalSampleRate, ctypes.c_double)


def set_rate(dev, rate):
    set_scalar(dev, kAudioDevicePropertyNominalSampleRate, ctypes.c_double(float(rate)))


def supported_rates(dev):
    out = []
    for r in get_array(dev, kAudioDevicePropertyAvailableNominalSampleRates, AudioValueRange):
        out.append(r.mMinimum)
        if r.mMaximum != r.mMinimum:
            out.append(r.mMaximum)
    return sorted(set(out))


def hog_owner(dev):
    return get_scalar(dev, kAudioDevicePropertyHogMode, ctypes.c_int32)


def fmt(rate):
    return "%g kHz" % (rate / 1000.0)


def describe(dev):
    return "id=%-4d %-30s [%s] uid=%s" % (dev, device_name(dev), transport(dev), device_uid(dev))


def matching(selector, outputs_only=False):
    """Resolve a -d selector to a list of devices. Empty selector = all (or all outputs)."""
    pool = output_devices() if outputs_only else all_devices()
    if not selector:
        return pool
    if selector.isdigit():
        want = int(selector)
        return [d for d in pool if d == want]
    if selector.lower().startswith("uid:"):
        needle = selector[4:].lower()
        return [d for d in pool if needle in device_uid(d).lower()]
    needle = selector.lower()
    return [d for d in pool if needle in device_name(d).lower() or needle in device_uid(d).lower()]


def pick_device(selector):
    if not selector:
        return default_output()
    hits = matching(selector)
    if not hits:
        sys.exit("No device matching %r. Try: ./audiodev.py list" % selector)
    if len(hits) > 1:
        print("%r matches %d devices:" % (selector, len(hits)), file=sys.stderr)
        for d in hits:
            print("  " + describe(d), file=sys.stderr)
        sys.exit("\nThey are indistinguishable by name. Run:\n"
                 "    ./audiodev.py identify -d %r\n"
                 "to hear which is which, then select the one you want by its id or "
                 "by 'uid:<fragment>' (the UID survives reboots; the numeric id does not)."
                 % selector)
    return hits[0]


def cmd_list(args):
    default = default_output()
    for dev in all_devices():
        try:
            rate = current_rate(dev)
        except OSError:
            continue
        flags = []
        if dev == default:
            flags.append("default-out")
        hog = hog_owner(dev)
        if hog not in (-1, 0):
            flags.append("HOGGED by pid %d" % hog)
        for sel, label in ((kAudioDevicePropertyDeviceIsAlive, "alive"),
                           (kAudioDevicePropertyDeviceIsRunning, "running")):
            try:
                if get_scalar(dev, sel, ctypes.c_uint32):
                    flags.append(label)
            except OSError:
                pass
        try:
            avail = ", ".join(fmt(r) for r in supported_rates(dev))
        except OSError:
            avail = "?"
        print("%s %s" % (describe(dev), "(" + ", ".join(flags) + ")" if flags else ""))
        print("      %-4s  now %s   supports: %s"
              % ("out" if has_output(dev) else "in", fmt(rate), avail))


def cmd_identify(args):
    cands = [d for d in matching(args.device, outputs_only=True)]
    if not cands:
        sys.exit("No output device matching %r." % (args.device or "<all>"))
    print("Speaking through each candidate in turn. Listen for which monitor talks.\n")
    for n, dev in enumerate(cands, 1):
        print("  %d. %s" % (n, describe(dev)))
        phrase = "This is device number %d" % n
        try:
            subprocess.run(["say", "-a", str(dev), "-r", "170", phrase],
                           check=True, timeout=20)
        except subprocess.CalledProcessError:
            print("     (could not play through this device - it may be unavailable)")
        except FileNotFoundError:
            sys.exit("'say' not found; cannot play a test phrase.")
        time.sleep(0.5)
    print("\nNote the number you heard, then use that entry's id for this session,\n"
          "or its 'uid:<fragment>' to select the same monitor after a reboot.")


def cmd_rate(args):
    dev = pick_device(args.device)
    before = current_rate(dev)
    set_rate(dev, args.value)
    print("%s: %s -> %s" % (device_name(dev), fmt(before), fmt(current_rate(dev))))


def reset_hog(dev, verbose):
    owner = hog_owner(dev)
    if owner not in (-1, 0) and owner != os.getpid():
        print("  hog:     already held exclusively by pid %d - skipping" % owner)
        return False
    try:
        set_scalar(dev, kAudioDevicePropertyHogMode, ctypes.c_int32(os.getpid()))
    except OSError as e:
        print("  hog:     could not take exclusive access (%s)" % e)
        return False
    if verbose:
        print("  hog:     took exclusive access, other clients evicted")
    time.sleep(0.6)
    try:
        set_scalar(dev, kAudioDevicePropertyHogMode, ctypes.c_int32(-1))
    except OSError as e:
        print("  hog:     WARNING - failed to release exclusive access (%s)" % e)
        return False
    print("  hog:     claimed and released exclusive access")
    return True


def reset_rate(dev, verbose):
    try:
        rates = supported_rates(dev)
        original = current_rate(dev)
    except OSError as e:
        print("  rate:    could not read rates (%s)" % e)
        return False
    others = [r for r in rates if r != original]
    if not others:
        print("  rate:    only %s supported - nothing to bounce" % fmt(original))
        return False
    scratch = min(others, key=lambda r: abs(r - original))
    try:
        set_rate(dev, scratch)
        time.sleep(0.8)
        set_rate(dev, original)
    except OSError as e:
        print("  rate:    bounce failed (%s)" % e)
        return False
    print("  rate:    bounced %s -> %s -> %s" % (fmt(original), fmt(scratch), fmt(original)))
    return True


def reset_default(dev, verbose):
    if default_output() != dev:
        print("  default: not the default output - skipping")
        return False
    others = [d for d in output_devices() if d != dev]
    if not others:
        print("  default: no other output device to bounce through - skipping")
        return False
    try:
        set_default_output(others[0])
        time.sleep(0.5)
        set_default_output(dev)
    except OSError as e:
        print("  default: bounce failed (%s)" % e)
        return False
    print("  default: bounced via %s" % device_name(others[0]))
    return True


def cmd_reset(args):
    dev = pick_device(args.device)
    print("Resetting: %s" % describe(dev))
    methods = ["hog", "rate", "default"] if args.method == "all" else [args.method]
    table = {"hog": reset_hog, "rate": reset_rate, "default": reset_default}
    for m in methods:
        table[m](dev, args.verbose)
    try:
        alive = bool(get_scalar(dev, kAudioDevicePropertyDeviceIsAlive, ctypes.c_uint32))
        print("Now at %s, alive=%s" % (fmt(current_rate(dev)), alive))
    except OSError as e:
        print("Device did not come back cleanly: %s" % e)


def main():
    p = argparse.ArgumentParser(description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("-v", "--verbose", action="store_true")
    sub = p.add_subparsers(dest="cmd", required=True)

    sub.add_parser("list", help="show every device: id, UID, transport, rate, state")

    pi = sub.add_parser("identify", help="speak through each matching output device")
    pi.add_argument("-d", "--device", default="")

    pr = sub.add_parser("rate", help="set the nominal sample rate")
    pr.add_argument("value", type=float)
    pr.add_argument("-d", "--device", default="")

    ps = sub.add_parser("reset", help="reinitialize one device")
    ps.add_argument("-d", "--device", default="")
    ps.add_argument("-m", "--method", choices=["hog", "rate", "default", "all"], default="all")

    args = p.parse_args()
    {"list": cmd_list, "identify": cmd_identify,
     "rate": cmd_rate, "reset": cmd_reset}[args.cmd](args)


if __name__ == "__main__":
    main()