Function audio_file_streamer

examples/08_stream_audio_file.py:88–156

Stream audio file in a loop

(audio_file_path, stereo=False)

def audio_file_streamer(audio_file_path, stereo=False):
    """Stream audio file in a loop"""
    from pydub import AudioSegment
    
    print(f"🎵 Loading audio file: {audio_file_path}")
    
    # Load audio file (supports MP3, WAV, etc.)
    audio = AudioSegment.from_file(audio_file_path)
    
    # Convert to 48kHz with specified channel count
    channels = 2 if stereo else 1
    audio = audio.set_frame_rate(48000).set_channels(channels)
    
    channel_mode = "stereo" if stereo else "mono"
    print(f"✓ Audio loaded ({channel_mode}):")
    print(f"  - Duration: {len(audio) / 1000:.2f} seconds")
    print(f"  - Sample rate: {audio.frame_rate} Hz")
    print(f"  - Channels: {audio.channels}")
    print(f"  - Sample width: {audio.sample_width} bytes")
    
    # Get raw audio data as int16 samples
    # For stereo, this array is interleaved: [L, R, L, R, L, R, ...]
    audio_samples = np.array(audio.get_array_of_samples(), dtype=np.int16)
    total_samples = len(audio_samples)
    
    # Track position in audio file (in terms of individual samples, not frames)
    current_position = 0
    
    def generate_audio(audio_frame):
        nonlocal current_position
        
        # Number of samples per frame PER CHANNEL
        num_samples_per_channel = audio_frame.samples
        
        # Total samples needed (for stereo: num_samples * 2, for mono: num_samples * 1)
        num_samples_needed = num_samples_per_channel * channels
        
        # Extract samples for this frame
        end_position = current_position + num_samples_needed
        
        # Handle looping
        if end_position >= total_samples:
            # Wrap around to beginning
            samples_before_end = total_samples - current_position
            samples_after_wrap = num_samples_needed - samples_before_end
            
            # Combine end + beginning
            frame_data = np.concatenate([
                audio_samples[current_position:],
                audio_samples[:samples_after_wrap]
            ])
            
            current_position = samples_after_wrap
        else:
            # Normal case: extract from current position
            frame_data = audio_samples[current_position:end_position]
            current_position = end_position