Adaptive Stem Alignment

Adaptive Stem Alignment Workflow

This skill provides a resilient pattern for audio production that emphasizes incremental verification, fail-fast principles, and adaptive duration handling. Each major step produces verified outputs before proceeding, with explicit strategies for handling stems of different durations.

Overview

Follow these steps in strict order. Each step must complete successfully and pass verification before proceeding to the next:

1. Early timing calculation - Derive section transitions from BPM and duration first
2. Verify reference audio - Validate input file properties and establish target duration
3. Generate and verify each stem individually - One stem at a time with immediate verification
4. Generate drum stem separately - Dedicated drum extension with rhythm patterns
5. Align stem durations - Handle duration mismatches with adaptive extension strategies
6. Apply effects with verification - Process each stem and verify output
7. Export master track - Mix all verified stems
8. Archive and final verification - Package deliverables with comprehensive checks

Key Differences from Standard Workflow

• Incremental verification: Verify each stem immediately after generation, not just at the end
• Fail-fast approach: Stop and report errors at each step rather than accumulating failures
• Early timing: Calculate section transitions before any audio generation
• Separated drums: Drum stem generation is a distinct step with rhythm-specific processing
• Memory-efficient: Process stems individually to avoid large array operations that cause sandbox failures
• Adaptive duration handling: Explicit strategies for mismatched stem durations (zero-padding, looping, crossfade extension)
• Pre-mix alignment: Verify all stems match target duration before mixing

Step 1: Calculate Timing Parameters (Early)

Calculate all timing parameters before generating any audio. This ensures consistent timing across all stems:

def calculate_section_transitions(bpm, total_duration_sec, sections):
    """Calculate beat-aligned transition points for song sections."""
    beats_per_second = bpm / 60.0
    
    section_durations = {}
    cumulative_time = 0
    
    for section_name, beat_count in sections.items():
        duration = beat_count / beats_per_second
        section_durations[section_name] = {
            'start': cumulative_time,
            'end': cumulative_time + duration,
            'beats': beat_count,
            'start_beat': cumulative_time * beats_per_second
        }
        cumulative_time += duration
    
    return section_durations

# Configuration
BPM = 120
DURATION = 137
SECTIONS = {'intro': 16, 'verse': 32, 'chorus': 32, 'bridge': 16, 'outro': 16}

timing = calculate_section_transitions(BPM, DURATION, SECTIONS)
print("Timing calculated:")
for section, data in timing.items():
    print(f"  {section}: {data['start']:.2f}s - {data['end']:.2f}s ({data['beats']} beats)")

Step 2: Verify Reference Audio

Validate the reference file exists and has expected properties:

import soundfile as sf
import os

def verify_reference_file(filepath, expected_sample_rate=None, min_duration=None):
    """Verify reference audio file and return info dict."""
    if not os.path.exists(filepath):
        raise FileNotFoundError(f"Reference file not found: {filepath}")
    
    info = sf.info(filepath)
    errors = []
    
    if expected_sample_rate and info.samplerate != expected_sample_rate:
        errors.append(f"Sample rate mismatch: expected {expected_sample_rate}, got {info.samplerate}")
    
    if min_duration and info.duration = {min_duration}s, got {info.duration}s")
    
    if errors:
        raise ValueError(f"Reference file validation failed: {'; '.join(errors)}")
    
    print(f"Reference verified: {info.duration:.2f}s @ {info.samplerate}Hz, {info.channels}ch, {info.subtype}")
    return {
        'sample_rate': info.samplerate,
        'duration': info.duration,
        'channels': info.channels,
        'subtype': info.subtype
    }

# Verify reference
ref_info = verify_reference_file('reference.wav', expected_sample_rate=48000, min_duration=130)
TARGET_DURATION = ref_info['duration']  # Use reference duration as target

Step 3: Generate and Verify Each Stem Individually

Generate one stem at a time, verify it immediately before proceeding to the next:

import numpy as np

def generate_stem(name, duration_sec, sample_rate, subtype='FLOAT', section_timing=None):
    """Generate a single stem with explicit sample type."""
    frames = int(duration_sec * sample_rate)
    t = np.linspace(0, duration_sec, frames)
    
    # Generate stem-specific content (customize per stem type)
    if name == 'bass':
        freq = 110  # A2
        audio_data = np.sin(2 * np.pi * freq * t) * 0.8
    elif name == 'guitars':
        freq = 440  # A4
        audio_data = np.sin(2 * np.pi * freq * t) * 0.6
    elif name == 'synths':
        freq = 880  # A5
        audio_data = np.sin(2 * np.pi * freq * t) * 0.5
    elif name == 'bridge':
        freq = 220  # A3
        audio_data = np.sin(2 * np.pi * freq * t) * 0.7
    else:
        audio_data = np.sin(2 * np.pi * 440 * t) * 0.5
    
    # Ensure proper data type
    if subtype == 'FLOAT':
        audio_data = audio_data.astype(np.float32)
    elif subtype == 'PCM_24':
        audio_data = np.clip(audio_data, -1, 1) * (2**23 - 1)
        audio_data = audio_data.astype(np.int32)
    
    filepath = f'{name}_stem.wav'
    sf.write(filepath, audio_data, sample_rate, subtype=subtype, format='WAV')
    
    return filepath, audio_data

def verify_stem(filepath, expected_sample_rate, expected_subtype, expected_duration):
    """Verify a single stem meets specifications."""
    if not os.path.exists(filepath):
        return {'success': False, 'error': f'File not found: {filepath}'}
    
    info = sf.info(filepath)
    errors = []
    
    if info.samplerate != expected_sample_rate:
        errors.append(f'sample_rate: expected {expected_sample_rate}, got {info.samplerate}')
    
    if info.subtype != expected_subtype:
        errors.append(f'subtype: expected {expected_subtype}, got {info.subtype}')
    
    if abs(info.duration - expected_duration) > 1.0:  # Allow 1s tolerance
        errors.append(f'duration: expected ~{expected_duration}s, got {info.duration}s')
    
    if errors:
        return {'success': False, 'error': '; '.join(errors)}
    
    return {'success': True, 'info': info}

# Generate stems one at a time with verification
SAMPLE_RATE = 48000
SUBTYPE = 'FLOAT'
STEM_NAMES = ['bass', 'guitars', 'synths', 'bridge']

generated_stems = []
stem_durations = {}  # Track actual durations for alignment step

for stem_name in STEM_NAMES:
    print(f"\n=== Generating {stem_name} stem ===")
    
    # Generate
    filepath, data = generate_stem(stem_name, DURATION, SAMPLE_RATE, subtype=SUBTYPE)
    
    # Verify immediately
    result = verify_stem(filepath, SAMPLE_RATE, SUBTYPE, DURATION)
    
    if result['success']:
        print(f"✓ {stem_name} stem verified: {result['info'].duration:.2f}s @ {result['info'].samplerate}Hz")
        generated_stems.append(filepath)
        stem_durations[stem_name] = result['info'].duration
    else:
        print(f"✗ {stem_name} stem FAILED: {result['error']}")
        raise RuntimeError(f"Stem generation failed for {stem_name}: {result['error']}")

print(f"\nAll {len(generated_stems)} stems generated and verified successfully")

Step 4: Generate Drum Stem Separately

Drums require different processing (rhythm patterns, percussion sounds):

def generate_drum_stem(duration_sec, sample_rate, bpm, section_timing, subtype='FLOAT'):
    """Generate drum stem with rhythm patterns aligned to sections."""
    frames = int(duration_sec * sample_rate)
    audio_data = np.zeros(frames, dtype=np.float32)
    beats_per_second = bpm / 60.0
    
    # Simple kick drum pattern (every beat)
    kick_freq = 60
    kick_duration = 0.1
    kick_frames = int(kick_duration * sample_rate)
    
    for beat_time in np.arange(0, duration_sec, 1.0 / beats_per_second):
        start_frame = int(beat_time * sample_rate)
        end_frame = min(start_frame + kick_frames, frames)
        
        if start_frame  0:
                aligned_data[start:end] = data[:copy_len]
        
        # Apply crossfade at loop points to avoid clicks
        crossfade_frames = int(0.05 * sample_rate)  # 50ms crossfade
        if current_frames > crossfade_frames * 2:
            for i in range(1, loop_count):
                loop_start = i * current_frames
                if loop_start  cf_start:
                        fade_in = np.linspace(0, 1, cf_end - cf_start)
                        fade_out = np.linspace(1, 0, cf_end - cf_start)
                        aligned_data[cf_start:cf_end] = (
                            aligned_data[cf_start:cf_end] * fade_out +
                            np.roll(aligned_data[cf_start:cf_end], -current_frames) * fade_in
                        )
    
    elif strategy == 'crossfade':
        # Crossfade-based extension with smooth transition
        if current_frames  0:
                    fade_in = np.linspace(0, 1, min(fade_frames, extension_frames))
                    extension_data[:len(fade_in)] *= fade_in
            else:
                # Loop multiple times with crossfades
                loop_data = np.tile(data, (extension_frames // current_frames) + 2)[:extension_frames]
                
                # Apply fade-in to extension
                if fade_frames > 0:
                    fade_in = np.linspace(0, 1, fade_frames)
                    loop_data[:fade_frames] *= fade_in
                
                extension_data = loop_data
            
            # Concatenate with crossfade
            aligned_data = np.zeros(target_frames, dtype=data.dtype)
            aligned_data[:current_frames] = data
            
            # Crossfade region at junction
            if fade_frames > 0:
                junction_start = current_frames - fade_frames
                junction_end = min(current_frames + fade_frames, target_frames)
                
                if junction_end > junction_start:
                    crossfade_len = junction_end - junction_start
                    fade_out = np.linspace(1, 0, crossfade_len)
                    fade_in = np.linspace(0, 1, crossfade_len)
                    
                    aligned_data[junction_start:junction_end] = (
                        aligned_data[junction_start:junction_end] * fade_out +
                        extension_data[:crossfade_len] * fade_in
                    )
                else:
                    aligned_data[current_frames:current_frames + extension_frames] = extension_data
            else:
                aligned_data[current_frames:] = extension_data
        else:
            # Truncate with fade-out
            fade_frames = int(2.0 * sample_rate)
            aligned_data = data[:target_frames].copy()
            fade_start = max(0, target_frames - fade_frames)
            fade_curve = np.linspace(1, 0, target_frames - fade_start)
            aligned_data[fade_start:] *= fade_curve
    
    else:
        return {'success': False, 'error': f'Unknown strategy: {strategy}'}
    
    # Clip to prevent overload
    aligned_data = np.clip(aligned_data, -1, 1)
    
    # Export
    sf.write(output_filepath, aligned_data, sample_rate, subtype=subtype, format='WAV')
    
    # Verify
    result = verify_stem(output_filepath, sample_rate, subtype, target_duration)
    if result['success']:
        return {
            'success': True,
            'strategy': strategy,
            'original_duration': current_duration,
            'aligned_duration': result['info'].duration
        }
    else:
        return result

# Apply duration alignment to all stems
print("\n=== Aligning stem durations ===")
aligned_stems = []

for stem_name in STEM_NAMES:
    input_file = f'{stem_name}_stem.wav'
    output_file = f'{stem_name}_aligned.wav'
    
    # Determine stem type for strategy selection
    stem_type_map = {
        'bass': 'rhythmic',
        'guitars': 'melodic',
        'synths': 'ambient',
        'bridge': 'melodic'
    }
    stem_type = stem_type_map.get(stem_name, 'melodic')
    
    print(f"Aligning {stem_name} (type: {stem_type})...")
    result = align_stem_duration(
        input_file, output_file, TARGET_DURATION, SAMPLE_RATE,
        subtype=SUBTYPE, strategy='auto', stem_type=stem_type
    )
    
    if result['success']:
        if result['strategy'] != 'none':
            print(f"✓ {stem_name} aligned: {result['original_duration']:.2f}s → {result['aligned_duration']:.2f}s ({result['strategy']})")
        else:
            print(f"✓ {stem_name} already aligned")
        aligned_stems.append(output_file)
    else:
        print(f"✗ {stem_name} alignment FAILED: {result['error']}")
        raise RuntimeError(f"Stem alignment failed for {stem_name}: {result['error']}")

# Align drums separately
drums_aligned = 'drums_aligned.wav'
print(f"Aligning drums (type: percussion)...")
drums_result = align_stem_duration(
    'drums_stem.wav', drums_aligned, TARGET_DURATION, SAMPLE_RATE,
    subtype=SUBTYPE, strategy='auto', stem_type='percussion'
)
if drums_result['success']:
    if drums_result['strategy'] != 'none':
        print(f"✓ Drums aligned: {drums_result['original_duration']:.2f}s → {drums_result['aligned_duration']:.2f}s ({drums_result['strategy']})")
    else:
        print(f"✓ Drums already aligned")
    aligned_stems.append(drums_aligned)
else:
    raise RuntimeError(f"Drums alignment failed: {drums_result['error']}")

# Final duration verification - all stems must match
print("\n=== Verifying duration alignment ===")
final_durations = {}
for stem_file in aligned_stems:
    info = sf.info(stem_file)
    stem_name = os.path.basename(stem_file).replace('_aligned.wav', '')
    final_durations[stem_name] = info.duration
    duration_diff = abs(info.duration - TARGET_DURATION)
    
    if duration_diff > 0.5:
        print(f"✗ WARNING: {stem_name} duration mismatch: {info.duration:.2f}s vs target {TARGET_DURATION:.2f}s")
    else:
        print(f"✓ {stem_name}: {info.duration:.2f}s (Δ{duration_diff:.2f}s)")

max_duration_diff = max(abs(d - TARGET_DURATION) for d in final_durations.values())
if max_duration_diff > 0.5:
    raise RuntimeError(f"Duration alignment incomplete: max deviation {max_duration_diff:.2f}s exceeds tolerance")
print(f"\nAll stems aligned within tolerance (max deviation: {max_duration_diff:.2f}s)")

Step 6: Apply Effects with Verification

Process each stem and verify the output:

from scipy import signal

def apply_lowpass_filter(audio_data, sample_rate, cutoff_freq=8000):
    """Apply lowpass filter using scipy.signal."""
    nyquist = sample_rate / 2
    normalized_cutoff = cutoff_freq / nyquist
    b, a = signal.butter(4, normalized_cutoff, btype='low')
    return signal.filtfilt(b, a, audio_data)

def apply_effects_and_verify(input_filepath, output_filepath, sample_rate, subtype):
    """Apply effects to stem and verify output."""
    data, sr = sf.read(input_filepath)
    
    # Apply effects
    processed = apply_lowpass_filter(data, sr, cutoff_freq=8000)
    processed = np.clip(processed, -1, 1)
    
    # Export
    sf.write(output_filepath, processed, sample_rate, subtype=subtype, format='WAV')
    
    # Verify
    result = verify_stem(output_filepath, sample_rate, subtype, TARGET_DURATION)
    return result, processed

print("\n=== Applying effects to all stems ===")
processed_stems = []

for stem_name in STEM_NAMES:
    input_file = f'{stem_name}_aligned.wav'
    output_file = f'{stem_name}_processed.wav'
    
    print(f"Processing {stem_name}...")
    result, _ = apply_effects_and_verify(input_file, output_file, SAMPLE_RA

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [HKUDS](https://github.com/HKUDS)
- **Source:** [HKUDS/OpenSpace](https://github.com/HKUDS/OpenSpace)
- **License:** MIT

Install and usage instructions live in the source repository linked above.