Aligned Stem Workflow

Aligned Stem Audio Production Workflow

This skill provides a resilient pattern for audio production that emphasizes incremental verification, fail-fast principles, and automatic duration alignment. Each major step produces verified outputs before proceeding, with explicit handling for stem duration mismatches using appropriate extension strategies.

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 extract target duration
3. Generate and verify each stem individually - One stem at a time with immediate verification
4. Detect and resolve duration mismatches - Apply appropriate extension strategy (zero-pad, loop, or crossfade)
5. Generate drum stem separately - Dedicated drum extension with rhythm patterns
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
• Duration alignment: Explicit detection and resolution of stem duration mismatches
• Adaptive extension: Choose appropriate strategy (zero-pad/loop/crossfade) based on stem type
• 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

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 actual 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, tolerance_sec=1.0):
    """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) > tolerance_sec:
        errors.append(f'duration: expected ~{expected_duration}s, got {info.duration}s')
    
    # Calculate duration discrepancy
    duration_diff = info.duration - expected_duration
    
    if errors:
        return {'success': False, 'error': '; '.join(errors), 'duration_diff': duration_diff}
    
    return {'success': True, 'info': info, 'duration_diff': duration_diff}

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

generated_stems = []
stem_info = {}  # Track duration discrepancies

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, TARGET_DURATION)
    
    if result['success']:
        print(f"✓ {stem_name} stem verified: {result['info'].duration:.2f}s @ {result['info'].samplerate}Hz")
        if abs(result['duration_diff']) > 0.1:
            print(f"  ⚠ Duration discrepancy: {result['duration_diff']:+.2f}s")
        generated_stems.append(filepath)
        stem_info[stem_name] = result
    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: Detect and Resolve Duration Mismatches

When stems have different durations, apply the appropriate extension strategy:

Strategy Selection Guidelines

| Strategy | Best For | Duration Gap | Sound Characteristic | |----------|----------|--------------|---------------------| | Zero-padding | Short gaps ( 0: # Need to EXTEND extendframes = int(durationdiff * sample_rate)

if strategy == 'auto': # Auto-select based on duration gap and stem type if durationdiff 1: # Apply crossfade at loop boundaries for seamless looping crossfadeframes = min(int(0.05 * samplerate), loopframes // 4) loopextension = np.zeros(extendframes, dtype=data.dtype)

for i in range(loopsneeded): start = i * loopframes end = min(start + loopframes, extendframes) actual_len = end - start

# Extract loop segment loopsegment = data[:actuallen].copy()

# Apply crossfade at boundaries if i > 0 and actuallen >= crossfadeframes 2: # Fade in from previous loop fadein = np.linspace(0, 1, crossfadeframes) loopsegment[:crossfadeframes] = fade_in

if i = crossfadeframes 2: # Fade out for next loop fadeout = np.linspace(1, 0, crossfadeframes) loopsegment[-crossfadeframes:] = fadeout

loopextension[start:end] = loopsegment

extendframesactual = len(loopextension) else: # Simple tiling loopextension = np.tile(data, loopsneeded)[:extendframes] extendframesactual = extend_frames

aligneddata = np.concatenate([data, loopextension[:extendframesactual]])

elif strategy == 'crossfade': # Extend using crossfade from the end of the source # Take last portion and crossfade it onto itself fadeduration = min(durationdiff 0.3, 2.0) # 30% of gap, max 2s fadeframes = int(fadeduration sample_rate)

if fadeframes >= len(data) // 2: # Source too short for crossfade, fall back to loop fadeframes = len(data) // 4

# Extract tail segment for extension tailsegment = data[-fadeframes:].copy()

# Create extended portion with crossfade extendedportion = np.zeros(extendframes, dtype=data.dtype)

if extendframes = 100: cflen = min(50, seglen // 4) if i > 0: fadein = np.linspace(0, 1, cflen) segment[:cflen] *= fade_in

extended_portion[start:end] = segment

aligneddata = np.concatenate([data, extendedportion])

else: return {'success': False, 'error': f'Unknown extension strategy: {strategy}'}

else: # Need to TRUNCATE truncateframes = int(abs(durationdiff) * samplerate) aligneddata = data[:len(data) - truncate_frames] strategy = 'truncate'

# Ensure proper data type and clip if subtype == 'FLOAT': aligneddata = aligneddata.astype(np.float32) elif subtype == 'PCM24': aligneddata = np.clip(aligneddata, -1, 1) (2*23 - 1) aligneddata = aligneddata.astype(np.int32) else: aligneddata = np.clip(aligned_data, -1, 1)

# Export aligned stem sf.write(outputfilepath, aligneddata, sample_rate, subtype=subtype, format='WAV')

return { 'success': True, 'strategy': strategy, 'sourceduration': sourceduration, 'targetduration': targetduration, 'durationdiff': durationdiff, 'alignedframes': len(aligneddata) }

Apply duration alignment to all stems

print("\n=== Aligning stem durations ===") aligned_stems = []

TARGETDURATION = refinfo['duration'] # Use reference as target

for stemname in STEMNAMES: inputfile = f'{stemname}stem.wav' outputfile = f'{stemname}aligned.wav'

# Determine strategy based on stem type if stemname in ['bass', 'drums']: strategy = 'loop' # Rhythmic elements loop well elif stemname in ['bridge', 'outro']: strategy = 'crossfade' # Sustained content benefits from crossfade else: strategy = 'auto' # Let the function decide

print(f"Aligning {stemname} (strategy: {strategy})...") result = alignstemduration(inputfile, outputfile, TARGETDURATION, strategy=strategy, samplerate=SAMPLERATE, subtype=SUBTYPE)

if result['success']: if result['strategy'] != 'none': print(f"✓ {stemname} aligned: {result['sourceduration']:.2f}s -> {result['targetduration']:.2f}s via {result['strategy']}") else: print(f"✓ {stemname} already aligned at {result['targetduration']:.2f}s") alignedstems.append(outputfile) else: print(f"✗ {stemname} alignment FAILED: {result['error']}") raise RuntimeError(f"Duration alignment failed for {stem_name}: {result['error']}")

print(f"\nAll {len(aligned_stems)} stems duration-aligned successfully")

## Step 5: Generate Drum Stem Separately

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

```python
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  len(master_audio):
            data = data[:len(master_audio)]
        elif len(data)  2.0:
        issues.append(f"Duration mismatch: expected ~{expected_duration}s, got {info.duration}s")
    
    if expected_sample_rate and info.samplerate != expected_sample_rate:
        issues.append(f"Sample rate mismatch: expected {expected_sample_rate}, got {info.samplerate}")
    
    # Check for clipping
    data, _ = sf.read(master_filepath)
    clip_ratio = np.sum(np.abs(data) >= 0.99) / len(data)
    if clip_ratio > 0.001:  # More than 0.1% clipped
        issues.append(f"Excessive clipping detected: {clip_ratio*100:.2f}% of samples at max level")
    
    # Check for silence
    rms = np.sqrt(np.mean(data**2))
    if rms < 0.01:
        issues.append(f"Audio too quiet: RMS level {rms:.4f}")
    
    success = len(issues) == 0
    
    return {
        'success': success,
        'issues': issues,
        'info': {
            'duration': info.duration,
            'sample_rate': info.samplerate,
            'channels': info.channels,
            'subtype': info.subtype,
            'clipping_ratio': clip_ratio,
            'rms_level': rms
        }
    }

print("\n=== Final verification ===")
final_result = final_verification(master_filepath, expected_duration=TARGET_DURATION, 
                                   expected_sample_rate=SAMPLE_RATE)

if final_result['success']:
    print("✓ All verification checks passed")
    print(f"  Master: {final_result['info']['duration']:.2f}s @ {final_result['info']['sample_rate']}Hz"

…

## 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.