coreaudio_virtual_mic.py

#!/usr/bin/env python3
"""
CoreAudio Virtual Microphone - Pure Python Implementation
Creates aggregate audio devices programmatically using macOS CoreAudio API.

This uses PyObjC to interface with CoreAudio framework directly,
allowing creation of virtual microphones without BlackHole or any
external audio drivers.

Based on Apple's CoreAudio documentation:
https://developer.apple.com/documentation/coreaudio
"""

import subprocess
import time
from ctypes import *
from ctypes.util import find_library
import sounddevice as sd

# Try to use PyObjC if available
try:
    from Foundation import NSMutableDictionary, NSMutableArray
    from CoreAudio import *
    PYOBJC_AVAILABLE = True
except ImportError:
    PYOBJC_AVAILABLE = False
    print("PyObjC not available. Install with: pip install pyobjc-framework-CoreAudio")


class CoreAudioVirtualMic:
    """
    Create and manage virtual microphone using macOS CoreAudio API.

    This creates an aggregate device that combines:
    1. Built-in output (to receive audio from your app)
    2. Built-in input (optional, for monitoring)

    The aggregate device appears as a microphone input to other applications.
    """

    def __init__(self):
        self.aggregate_device_uid = "com.keypresser.virtualmic"
        self.aggregate_device_name = "KeyPresser Virtual Mic"
        self.created_device_uid = None

    def get_builtin_device_uids(self):
        """Get UIDs of built-in audio devices."""
        try:
            # Use system_profiler to get device information
            result = subprocess.run(
                ['system_profiler', 'SPAudioDataType', '-json'],
                capture_output=True,
                text=True,
                check=True
            )

            import json
            data = json.loads(result.stdout)

            builtin_output_uid = None
            builtin_input_uid = None

            # Parse audio data
            for item in data.get('SPAudioDataType', []):
                if '_items' in item:
                    for device in item['_items']:
                        name = device.get('_name', '').lower()

                        # Look for built-in devices
                        if 'built-in' in name or 'internal' in name:
                            # Try to get UID from various possible keys
                            uid = None
                            for key in ['coreaudio_device_uid', 'device_uid', '_uid']:
                                if key in device:
                                    uid = device[key]
                                    break

                            if uid:
                                if 'output' in name:
                                    builtin_output_uid = uid
                                elif 'input' in name or 'microphone' in name:
                                    builtin_input_uid = uid

            # Fallback: use sounddevice to find built-in devices
            if not builtin_output_uid or not builtin_input_uid:
                devices = sd.query_devices()
                for i, device in enumerate(devices):
                    name = device['name'].lower()
                    if 'built-in' in name:
                        # Use a simple UID based on device index
                        if device['max_output_channels'] > 0 and not builtin_output_uid:
                            builtin_output_uid = f"builtin-output-{i}"
                        if device['max_input_channels'] > 0 and not builtin_input_uid:
                            builtin_input_uid = f"builtin-input-{i}"

            return builtin_output_uid, builtin_input_uid

        except Exception as e:
            print(f"Error getting device UIDs: {e}")
            return None, None

    def create_aggregate_device_applescript(self):
        """
        Create aggregate device using AppleScript automation.
        This is more reliable than trying to use CoreAudio API directly.
        """

        output_uid, input_uid = self.get_builtin_device_uids()

        if not output_uid and not input_uid:
            print("Could not find built-in audio devices")
            return False

        # AppleScript to create aggregate device
        script = f'''
        set deviceName to "{self.aggregate_device_name}"

        tell application "System Events"
            -- Launch Audio MIDI Setup if not running
            if not (exists process "Audio MIDI Setup") then
                do shell script "open -g '/System/Applications/Utilities/Audio MIDI Setup.app'"
                delay 2
            end if

            tell process "Audio MIDI Setup"
                try
                    -- Try to use menu
                    click menu item "Create Aggregate Device" of menu 1 of menu bar item "File" of menu bar 1
                    delay 1

                    -- Name the device
                    keystroke deviceName
                    keystroke return
                    delay 1

                    return "success"
                on error errMsg
                    return "error: " & errMsg
                end try
            end tell
        end tell
        '''

        try:
            result = subprocess.run(
                ['osascript', '-e', script],
                capture_output=True,
                text=True,
                timeout=15
            )

            if 'success' in result.stdout.lower():
                print(f"✓ Created aggregate device: {self.aggregate_device_name}")

                # Close Audio MIDI Setup
                subprocess.run(
                    ['osascript', '-e', 'tell application "Audio MIDI Setup" to quit'],
                    timeout=5
                )

                time.sleep(2)
                return True
            else:
                print(f"Failed to create aggregate device: {result.stdout}")
                return False

        except Exception as e:
            print(f"Error creating aggregate device: {e}")
            return False

    def device_exists(self):
        """Check if aggregate device already exists."""
        try:
            devices = sd.query_devices()
            for device in devices:
                if self.aggregate_device_name in device['name']:
                    return True
            return False
        except Exception as e:
            print(f"Error checking device: {e}")
            return False

    def setup(self):
        """
        Set up virtual microphone.
        Returns True if successful.
        """
        print(f"Setting up virtual microphone: {self.aggregate_device_name}")

        # Check if already exists
        if self.device_exists():
            print("✓ Virtual microphone already exists")
            return True

        # Create using AppleScript
        print("Creating aggregate device...")
        success = self.create_aggregate_device_applescript()

        if success:
            # Verify it was created
            time.sleep(2)
            if self.device_exists():
                print("✓ Virtual microphone ready")
                return True
            else:
                print("⚠ Device created but not yet visible. Please restart the application.")
                return False
        else:
            print("✗ Failed to create virtual microphone")
            return False

    def cleanup(self):
        """Remove the created aggregate device."""
        script = f'''
        tell application "Audio MIDI Setup"
            try
                delete (every audio device whose name contains "{self.aggregate_device_name}")
                return "success"
            on error
                return "error"
            end try
        end tell
        '''

        try:
            result = subprocess.run(
                ['osascript', '-e', script],
                capture_output=True,
                text=True,
                timeout=10
            )
            return 'success' in result.stdout.lower()
        except Exception as e:
            print(f"Error cleaning up: {e}")
            return False


def test_virtual_mic():
    """Test the CoreAudio virtual microphone creation."""
    print("=== CoreAudio Virtual Microphone Test ===\n")

    vm = CoreAudioVirtualMic()

    print("Step 1: Checking for existing devices...")
    if vm.device_exists():
        print(f"✓ Device '{vm.aggregate_device_name}' already exists")
    else:
        print("No existing device found")
        print("\nStep 2: Creating virtual microphone...")
        success = vm.setup()

        if success:
            print("\n✓ Setup complete!")
        else:
            print("\n✗ Setup failed")
            print("\nFallback: Manually create an aggregate device in Audio MIDI Setup")
            print("See VIRTUAL_MIC_SOLUTIONS.md for instructions")
            return

    print("\nStep 3: Listing audio devices...")
    devices = sd.query_devices()
    print("\nAvailable audio devices:")
    for i, device in enumerate(devices):
        if 'KeyPresser' in device['name'] or 'Built-in' in device['name']:
            print(f"  [{i}] {device['name']}")
            print(f"      Inputs: {device['max_input_channels']}, Outputs: {device['max_output_channels']}")

    print("\nHow to use:")
    print(f"1. In voice dictation apps, select '{vm.aggregate_device_name}' as microphone")
    print("2. Set system output to play through the aggregate device")
    print("3. Apps will capture your audio!")

    # Ask about cleanup
    cleanup = input("\nRemove the virtual microphone? (y/n): ")
    if cleanup.lower() == 'y':
        if vm.cleanup():
            print("✓ Cleanup complete")
        else:
            print("✗ Cleanup failed - remove manually in Audio MIDI Setup")


if __name__ == "__main__":
    test_virtual_mic()