Aligned Stem Workflow

def align_stem_duration(input_filepath, output_filepath, target_duration, strategy='auto', 
                        sample_rate=None, subtype='FLOAT', loop_seamless=True):
    """
    Align stem duration to target using appropriate strategy.
    
    Args:
        input_filepath: Path to source stem
        output_filepath: Path for aligned output
        target_duration: Target duration in seconds
        strategy: 'zero_pad', 'loop', 'crossfade', or 'auto'
        sample_rate: Sample rate (auto-detected if None)
        subtype: Audio subtype
        loop_seamless: Apply crossfade at loop boundaries if True
    
    Returns:
        dict with success status and alignment details
    """
    if not os.path.exists(input_filepath):
        return {'success': False, 'error': f'Input file not found: {input_filepath}'}
    
    # Load source
    data, sr = sf.read(input_filepath)
    if sample_rate is None:
        sample_rate = sr
    
    source_duration = len(data) / sample_rate
    duration_diff = target_duration - source_duration
    
    # If already aligned (within tolerance), just copy
    if abs(duration_diff) < 0.01:
        sf.write(output_filepath, data, sample_rate, subtype=subtype, format='WAV')
        return {'success': True, 'strategy': 'none', 'duration_diff': 0}
    
    if duration_diff > 0:
        # Need to EXTEND
        extend_frames = int(duration_diff * sample_rate)
        
        if strategy == 'auto':
            # Auto-select based on duration gap and stem type
            if duration_diff < 0.5:
                strategy = 'zero_pad'
            elif 'drum' in input_filepath or 'bass' in input_filepath:
                strategy = 'loop'
            else:
                strategy = 'crossfade'
        
        if strategy == 'zero_pad':
            # Append zeros
            padding = np.zeros(extend_frames, dtype=data.dtype)
            aligned_data = np.concatenate([data, padding])
            
        elif strategy == 'loop':
            # Loop the content
            loop_frames = len(data)
            loops_needed = int(np.ceil(extend_frames / loop_frames))
            
            if loop_seamless and loops_needed > 1:
                # Apply crossfade at loop boundaries for seamless looping
                crossfade_frames = min(int(0.05 * sample_rate), loop_frames // 4)
                loop_extension = np.zeros(extend_frames, dtype=data.dtype)
                
                for i in range(loops_needed):
                    start = i * loop_frames
                    end = min(start + loop_frames, extend_frames)
                    actual_len = end - start
                    
                    # Extract loop segment
                    loop_segment = data[:actual_len].copy()
                    
                    # Apply crossfade at boundaries
                    if i > 0 and actual_len >= crossfade_frames * 2:
                        # Fade in from previous loop
                        fade_in = np.linspace(0, 1, crossfade_frames)
                        loop_segment[:crossfade_frames] *= fade_in
                    
                    if i < loops_needed - 1 and actual_len >= crossfade_frames * 2:
                        # Fade out for next loop
                        fade_out = np.linspace(1, 0, crossfade_frames)
                        loop_segment[-crossfade_frames:] *= fade_out
                    
                    loop_extension[start:end] = loop_segment
                
                extend_frames_actual = len(loop_extension)
            else:
                # Simple tiling
                loop_extension = np.tile(data, loops_needed)[:extend_frames]
                extend_frames_actual = extend_frames
            
            aligned_data = np.concatenate([data, loop_extension[:extend_frames_actual]])
            
        elif strategy == 'crossfade':
            # Extend using crossfade from the end of the source
            # Take last portion and crossfade it onto itself
            fade_duration = min(duration_diff * 0.3, 2.0)  # 30% of gap, max 2s
            fade_frames = int(fade_duration * sample_rate)
            
            if fade_frames >= len(data) // 2:
                # Source too short for crossfade, fall back to loop
                fade_frames = len(data) // 4
            
            # Extract tail segment for extension
            tail_segment = data[-fade_frames:].copy()
            
            # Create extended portion with crossfade
            extended_portion = np.zeros(extend_frames, dtype=data.dtype)
            
            if extend_frames <= fade_frames:
                # Short extension: just crossfade tail onto itself
                fade_in = np.linspace(0, 1, extend_frames)
                extended_portion = tail_segment[:extend_frames] * fade_in
            else:
                # Longer extension: loop tail with crossfades
                loops = int(np.ceil(extend_frames / fade_frames))
                for i in range(loops):
                    start = i * fade_frames
                    end = min(start + fade_frames, extend_frames)
                    seg_len = end - start
                    
                    segment = tail_segment[:seg_len].copy()
                    
                    # Crossfade boundaries
                    if seg_len >= 100:
                        cf_len = min(50, seg_len // 4)
                        if i > 0:
                            fade_in = np.linspace(0, 1, cf_len)
                            segment[:cf_len] *= fade_in
                    
                    extended_portion[start:end] = segment
            
            aligned_data = np.concatenate([data, extended_portion])
        
        else:
            return {'success': False, 'error': f'Unknown extension strategy: {strategy}'}
        
    else:
        # Need to TRUNCATE
        truncate_frames = int(abs(duration_diff) * sample_rate)
        aligned_data = data[:len(data) - truncate_frames]
        strategy = 'truncate'
    
    # Ensure proper data type and clip
    if subtype == 'FLOAT':
        aligned_data = aligned_data.astype(np.float32)
    elif subtype == 'PCM_24':
        aligned_data = np.clip(aligned_data, -1, 1) * (2**23 - 1)
        aligned_data = aligned_data.astype(np.int32)
    else:
        aligned_data = np.clip(aligned_data, -1, 1)
    
    # Export aligned stem
    sf.write(output_filepath, aligned_data, sample_rate, subtype=subtype, format='WAV')
    
    return {
        'success': True, 
        'strategy': strategy,
        'source_duration': source_duration,
        'target_duration': target_duration,
        'duration_diff': duration_diff,
        'aligned_frames': len(aligned_data)
    }

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

TARGET_DURATION = ref_info['duration']  # Use reference as target

for stem_name in STEM_NAMES:
    input_file = f'{stem_name}_stem.wav'
    output_file = f'{stem_name}_aligned.wav'
    
    # Determine strategy based on stem type
    if stem_name in ['bass', 'drums']:
        strategy = 'loop'  # Rhythmic elements loop well
    elif stem_name in ['bridge', 'outro']:
        strategy = 'crossfade'  # Sustained content benefits from crossfade
    else:
        strategy = 'auto'  # Let the function decide
    
    print(f"Aligning {stem_name} (strategy: {strategy})...")
    result = align_stem_duration(input_file, output_file, TARGET_DURATION, 
                                  strategy=strategy, sample_rate=SAMPLE_RATE, subtype=SUBTYPE)
    
    if result['success']:
        if result['strategy'] != 'none':
            print(f"✓ {stem_name} aligned: {result['source_duration']:.2f}s -> {result['target_duration']:.2f}s via {result['strategy']}")
        else:
            print(f"✓ {stem_name} already aligned at {result['target_duration']:.2f}s")
        aligned_stems.append(output_file)
    else:
        print(f"✗ {stem_name} 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")

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 < frames:
            t = np.linspace(0, kick_duration, end_frame - start_frame)
            kick = np.exp(-5 * t) * np.sin(2 * np.pi * kick_freq * t)
            audio_data[start_frame:end_frame] += kick * 0.9
    
    # Simple snare pattern (every 2nd and 4th beat)
    snare_freq = 200
    snare_duration = 0.05
    snare_frames = int(snare_duration * sample_rate)
    
    for beat_time in np.arange(0, duration_sec, 2.0 / beats_per_second):
        start_frame = int((beat_time + 0.5 / beats_per_second) * sample_rate)
        end_frame = min(start_frame + snare_frames, frames)
        
        if start_frame < frames:
            t = np.linspace(0, snare_duration, end_frame - start_frame)
            snare = np.exp(-10 * t) * np.random.uniform(-1, 1, len(t)) * 0.5
            audio_data[start_frame:end_frame] += snare * 0.7
    
    audio_data = np.clip(audio_data, -1, 1)
    
    filepath = 'drums_stem.wav'
    sf.write(filepath, audio_data, sample_rate, subtype=subtype, format='WAV')
    
    return filepath, audio_data

print("\n=== Generating drum stem ===")
drums_filepath, drums_data = generate_drum_stem(TARGET_DURATION, SAMPLE_RATE, BPM, timing, subtype=SUBTYPE)

drums_result = verify_stem(drums_filepath, SAMPLE_RATE, SUBTYPE, TARGET_DURATION)
if drums_result['success']:
    print(f"✓ Drum stem verified: {drums_result['info'].duration:.2f}s @ {drums_result['info'].samplerate}Hz")
    # Align drums if needed (usually uses loop strategy)
    drums_aligned = 'drums_aligned.wav'
    drums_align_result = align_stem_duration(drums_filepath, drums_aligned, TARGET_DURATION,
                                              strategy='loop', sample_rate=SAMPLE_RATE, subtype=SUBTYPE)
    if drums_align_result['success']:
        aligned_stems.append(drums_aligned)
        print(f"✓ Drums aligned via {drums_align_result['strategy']}")
    else:
        print(f"✗ Drum alignment FAILED: {drums_align_result['error']}")
        raise RuntimeError(f"Drum alignment failed: {drums_align_result['error']}")
else:
    print(f"✗ Drum stem FAILED: {drums_result['error']}")
    raise RuntimeError(f"Drum stem generation failed: {drums_result['error']}")

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, tolerance_sec=2.0)
    return result, processed

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

for aligned_file in aligned_stems:
    stem_name = os.path.basename(aligned_file).replace('_aligned.wav', '')
    output_file = f'{stem_name}_processed.wav'
    
    print(f"Processing {stem_name}...")
    result, _ = apply_effects_and_verify(aligned_file, output_file, SAMPLE_RATE, SUBTYPE)
    
    if result['success']:
        print(f"✓ {stem_name} processed and verified")
        processed_stems.append(output_file)
    else:
        print(f"✗ {stem_name} processing FAILED: {result['error']}")
        raise RuntimeError(f"Effects processing failed for {stem_name}")

print(f"\nAll {len(processed_stems)} stems processed successfully")

def create_master_track(stem_files, output_filepath, sample_rate, subtype):
    """Create master track from verified stems."""
    # Load first stem to get dimensions
    first_data, sr = sf.read(stem_files[0])
    master_audio = np.zeros(len(first_data), dtype=np.float32)
    
    # Mix all stems with gain staging
    gain_per_stem = 0.4  # Prevent clipping with 5 stems
    
    for i, stem_file in enumerate(stem_files):
        print(f"  Mixing stem {i+1}/{len(stem_files)}: {stem_file}")
        data, sr = sf.read(stem_file)
        
        # Ensure same length as master
        if len(data) > len(master_audio):
            data = data[:len(master_audio)]
        elif len(data) < len(master_audio):
            # Pad with zeros if shorter (shouldn't happen after alignment)
            padding = np.zeros(len(master_audio) - len(data), dtype=np.float32)
            data = np.concatenate([data, padding])
        
        master_audio += data * gain_per_stem
    
    # Apply master bus processing
    master_audio = np.clip(master_audio, -1, 1)
    
    # Export master
    sf.write(output_filepath, master_audio, sample_rate, subtype=subtype, format='WAV')
    
    return output_filepath, master_audio

print("\n=== Creating master track ===")
master_filepath, master_data = create_master_track(processed_stems, 'master.wav', SAMPLE_RATE, SUBTYPE)

# Verify master
master_info = sf.info(master_filepath)
print(f"Master exported: {master_filepath}")
print(f"  Duration: {master_info.duration:.2f}s")
print(f"  Sample rate: {master_info.samplerate}Hz")
print(f"  Channels: {master_info.channels}")
print(f"  Subtype: {master_info.subtype}")

def create_archive(stem_files, master_file, output_archive='audio交付.zip'):
    """Create archive of all deliverables."""
    import zipfile
    
    all_files = stem_files + [master_file]
    
    with zipfile.ZipFile(output_archive, 'w', zipfile.ZIP_DEFLATED) as zipf:
        for filepath in all_files:
            if os.path.exists(filepath):
                zipf.write(filepath)
                print(f"  Added: {filepath}")
    
    return output_archive

def final_verification(master_filepath, expected_duration=None, expected_sample_rate=None):
    """Comprehensive final verification."""
    issues = []
    
    # Verify master file
    if not os.path.exists(master_filepath):
        return {'success': False, 'error': 'Master file not found'}
    
    info = sf.info(master_filepath)
    
    if expected_duration and abs(info.duration - expected_duration) > 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")
    print(f"  RMS level: {final_result['info']['rms_level']:.4f}")
    print(f"  Clipping: {final_result['info']['clipping_ratio']*100:.2f}%")
    
    # Create archive
    print("\n=== Creating archive ===")
    archive_path = create_archive(processed_stems, master_filepath, 'audio_deliverables.zip')
    print(f"✓ Archive created: {archive_path}")
else:
    print("✗ Verification failed:")
    for issue in final_result['issues']:
        print(f"  - {issue}")