Utilities and Error Handling

def sleep(msec):
    """
    Sleep for specified milliseconds using PortAudio's timing.
    
    Parameters:
    - msec (float): Number of milliseconds to sleep
    
    Returns:
    None
    
    Notes:
    Uses PortAudio's Pa_Sleep() function which may provide more accurate
    timing than Python's time.sleep() on some platforms.
    """

def get_portaudio_version():
    """
    Get PortAudio library version information.
    
    Returns:
    tuple: (version_text, version_number)
    - version_text (str): Human-readable version string
    - version_number (int): Numeric version identifier
    """

class PortAudioError(Exception):
    """
    Exception raised for PortAudio-related errors.
    
    This exception is raised when PortAudio functions return error codes,
    typically indicating hardware problems, invalid parameters, or
    system resource issues.
    
    Attributes:
    - args[0] (str): Error message describing the specific problem
    """

class CallbackStop(Exception):
    """
    Exception to signal stream callback should stop gracefully.
    
    Raise this exception from within an audio callback function to
    stop the stream in a controlled manner. The stream will finish
    processing the current buffer and then stop.
    """

class CallbackAbort(Exception):
    """
    Exception to signal stream callback should abort immediately.
    
    Raise this exception from within an audio callback function to
    abort the stream immediately. This may cause audio dropouts but
    stops the stream as quickly as possible.
    """

class DeviceList(tuple):
    """
    Special tuple subclass containing device information.
    
    A list-like object that behaves like a tuple but provides special
    string representation for interactive display of device information.
    Each element is a device information dictionary.
    """

class CallbackFlags:
    """
    Status flags for stream callback functions.
    
    Provides information about buffer over-/underruns and stream conditions
    that occurred during callback processing.
    
    Attributes:
    - input_underflow (bool): Input buffer underflow occurred
    - input_overflow (bool): Input buffer overflow occurred
    - output_underflow (bool): Output buffer underflow occurred
    - output_overflow (bool): Output buffer overflow occurred
    - priming_output (bool): Stream is priming output buffers
    """
    
    def __init__(self, flags=0x0): ...
    
    input_underflow: bool
    input_overflow: bool
    output_underflow: bool
    output_overflow: bool
    priming_output: bool

import sounddevice as sd
import time

def compare_sleep_methods():
    """Compare sounddevice.sleep() with time.sleep() precision."""
    
    # Test sounddevice.sleep()
    start = time.time()
    sd.sleep(100)  # Sleep for 100 milliseconds
    sd_duration = time.time() - start
    
    # Test time.sleep()
    start = time.time()
    time.sleep(0.1)  # Sleep for 0.1 seconds (100 milliseconds)
    time_duration = time.time() - start
    
    print(f"sounddevice.sleep(100): {sd_duration:.6f} seconds")
    print(f"time.sleep(0.1): {time_duration:.6f} seconds")
    print(f"Target: 0.100000 seconds")

compare_sleep_methods()

import sounddevice as sd
import numpy as np

def play_with_gaps():
    """Play audio segments with precise gaps between them."""
    
    # Generate short audio beeps
    duration = 0.2  # 200ms beeps
    samplerate = 44100
    frequency = 800  # 800 Hz
    
    t = np.linspace(0, duration, int(samplerate * duration))
    beep = 0.3 * np.sin(2 * np.pi * frequency * t)
    
    # Play 5 beeps with 300ms gaps
    for i in range(5):
        print(f"Playing beep {i+1}")
        sd.play(beep, samplerate=samplerate, blocking=True)
        
        if i < 4:  # Don't sleep after the last beep
            print("Waiting...")
            sd.sleep(300)  # 300ms gap using PortAudio timing

play_with_gaps()

import sounddevice as sd

def show_version_info():
    """Display sounddevice and PortAudio version information."""
    
    # Get sounddevice version
    print(f"sounddevice version: {sd.__version__}")
    
    # Get PortAudio version
    version_text, version_number = sd.get_portaudio_version()
    print(f"PortAudio version: {version_text}")
    print(f"PortAudio version number: {version_number}")
    
    # Decode version number (example for PortAudio v19.7.0)
    major = (version_number >> 16) & 0xFF
    minor = (version_number >> 8) & 0xFF  
    patch = version_number & 0xFF
    print(f"PortAudio version breakdown: {major}.{minor}.{patch}")

show_version_info()

import sounddevice as sd
import numpy as np

def safe_audio_operation():
    """Demonstrate proper exception handling for audio operations."""
    
    try:
        # Attempt to use a specific device that might not exist
        devices = sd.query_devices()
        print("Available devices:")
        for i, device in enumerate(devices):
            print(f"  {i}: {device['name']}")
        
        # Try to record from device index 99 (likely doesn't exist)
        recording = sd.rec(frames=44100, samplerate=44100, device=99)
        sd.wait()
        
    except sd.PortAudioError as e:
        print(f"PortAudio error occurred: {e}")
        print("Falling back to default device...")
        
        # Fallback to default device
        try:
            recording = sd.rec(frames=44100, samplerate=44100)
            sd.wait()
            print("Recording successful with default device")
            
        except sd.PortAudioError as e2:
            print(f"Even default device failed: {e2}")
            return None
    
    except Exception as e:
        print(f"Unexpected error: {e}")
        return None
    
    return recording

# Test safe audio operation
result = safe_audio_operation()
if result is not None:
    print(f"Recording completed: {len(result)} samples")

import sounddevice as sd
import numpy as np
import time

def controlled_callback_demo():
    """Demonstrate using callback exceptions to control stream behavior."""
    
    start_time = time.time()
    max_duration = 5.0  # Maximum recording duration
    
    recorded_data = []
    
    def recording_callback(indata, frames, time, status):
        """Callback that stops after maximum duration."""
        if status:
            print(f"Status: {status}")
        
        # Store the recorded data
        recorded_data.append(indata.copy())
        
        # Check if we should stop
        elapsed = time.time() - start_time
        if elapsed > max_duration:
            print("Maximum duration reached, stopping gracefully...")
            raise sd.CallbackStop()
        
        # Check for some error condition (example: too loud input)
        rms = np.sqrt(np.mean(indata**2))
        if rms > 0.8:  # If input is too loud
            print("Input too loud, aborting immediately!")
            raise sd.CallbackAbort()
    
    try:
        with sd.InputStream(callback=recording_callback, channels=1):
            print("Recording started. Speak normally (will stop after 5 seconds)")
            print("Speak loudly to trigger abort condition")
            
            # Keep the stream alive
            while True:
                sd.sleep(100)
                
    except sd.CallbackStop:
        print("Stream stopped gracefully via CallbackStop")
    except sd.CallbackAbort:
        print("Stream aborted immediately via CallbackAbort")
    except KeyboardInterrupt:
        print("Recording interrupted by user")
    
    if recorded_data:
        total_samples = sum(len(block) for block in recorded_data)
        print(f"Recorded {total_samples} samples in {len(recorded_data)} blocks")
    
    return recorded_data

# Run the controlled callback demonstration
recorded_blocks = controlled_callback_demo()

import sounddevice as sd
import numpy as np
import logging

# Configure logging for audio errors
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AudioManager:
    """Audio manager with comprehensive error handling."""
    
    def __init__(self): 
        self.fallback_devices = []
        self._discover_fallback_devices()
    
    def _discover_fallback_devices(self):
        """Discover available devices for fallback scenarios."""
        try:
            devices = sd.query_devices()
            # Find devices that support both input and output
            for i, device in enumerate(devices):
                if (device['max_input_channels'] > 0 and 
                    device['max_output_channels'] > 0):
                    self.fallback_devices.append(i)
            logger.info(f"Found {len(self.fallback_devices)} fallback devices")
        except sd.PortAudioError as e:
            logger.error(f"Failed to query devices: {e}")
    
    def safe_record(self, duration=5.0, samplerate=44100, device=None):
        """Record audio with automatic fallback on errors."""
        
        devices_to_try = [device] if device is not None else [None]
        devices_to_try.extend(self.fallback_devices)
        
        for attempt, dev in enumerate(devices_to_try):
            try:
                logger.info(f"Attempt {attempt + 1}: Using device {dev}")
                
                recording = sd.rec(
                    frames=int(duration * samplerate),
                    samplerate=samplerate,
                    channels=1,
                    device=dev
                )
                sd.wait()
                
                logger.info("Recording successful")
                return recording
                
            except sd.PortAudioError as e:
                logger.warning(f"Device {dev} failed: {e}")
                continue
            except Exception as e:
                logger.error(f"Unexpected error with device {dev}: {e}")
                continue
        
        logger.error("All devices failed")
        return None
    
    def get_system_info(self):
        """Get comprehensive system audio information."""
        info = {}
        
        try:
            # PortAudio version
            version_text, version_number = sd.get_portaudio_version()
            info['portaudio_version'] = version_text
            info['portaudio_version_number'] = version_number
            
            # sounddevice version
            info['sounddevice_version'] = sd.__version__
            
            # Device count
            devices = sd.query_devices()
            info['device_count'] = len(devices)
            info['input_devices'] = sum(1 for d in devices if d['max_input_channels'] > 0)
            info['output_devices'] = sum(1 for d in devices if d['max_output_channels'] > 0)
            
            # Host APIs
            hostapis = sd.query_hostapis()
            info['hostapi_count'] = len(hostapis)
            info['hostapi_names'] = [api['name'] for api in hostapis]
            
        except Exception as e:
            logger.error(f"Failed to get system info: {e}")
            info['error'] = str(e)
        
        return info

# Usage example
audio_manager = AudioManager()

# Get system information
system_info = audio_manager.get_system_info()
print("System Audio Information:")
for key, value in system_info.items():
    print(f"  {key}: {value}")

# Safe recording with fallback
recording = audio_manager.safe_record(duration=3.0)
if recording is not None:
    print(f"Successfully recorded {len(recording)} samples")
else:
    print("Recording failed on all available devices")