""" CAK Neural Audio Processor - Gradio Interface Interactive web interface for the CAK (Conditioning Aware Kernels) audio effect. Upload audio and adjust the texture control to apply the learned neural transformation. Supports both mono and stereo audio processing. Author: Austin Rockman (austin@gloame.ai) Date: October 2025 """ import torch import numpy as np import librosa import soundfile as sf import gradio as gr from pathlib import Path import json import tempfile # CAK model imports from cak_main_sandbox import SharedTextureDetector, TextureGenerator class AudioProcessor: """Audio processor for CAK with stereo support""" def __init__(self): self.device = torch.device('cuda' if torch.cuda.is_available() else 'mps' if torch.backends.mps.is_available() else 'cpu') print(f"Using device: {self.device}") # Audio params from training self.sample_rate = 44100 self.n_fft = 2048 self.hop_size = 512 self.win_size = 2048 # Load global normalization stats stats_path = 'norm_stats/global_normalization_stats.json' with open(stats_path, 'r') as f: self.global_stats = json.load(f) print(f"Loaded global stats: low={self.global_stats['global_low']:.3f}, " f"high={self.global_stats['global_high']:.3f}") # Create model architecture self.shared_detector = SharedTextureDetector() self.generator = TextureGenerator(self.shared_detector) # Load trained weights checkpoint_path = 'wgan_grain_output/final_wgan_grain.pt' if Path(checkpoint_path).exists(): checkpoint = torch.load(checkpoint_path, map_location=self.device, weights_only=False) self.generator.load_state_dict(checkpoint['generator']) self.shared_detector.load_state_dict(checkpoint['shared_detector']) # Set epoch for temperature if 'epoch' in checkpoint: self.generator.update_epoch(checkpoint['epoch']) print(f"Model loaded from: {checkpoint_path}") print(f"Scale parameter: {self.generator.cak.scale.item():.3f}") print(f"Temperature: {self.generator.cak.temperature:.1f}") print(f"Threshold: {self.generator.cak.threshold.item():.2f}") else: raise FileNotFoundError(f"No checkpoint found at {checkpoint_path}") # Move to device and set to eval mode self.shared_detector.to(self.device) self.generator.to(self.device) self.generator.eval() def normalize_stft(self, stft_mag): """STFT normalization matching training""" # Sqrt compression sqrt_mag = np.sqrt(stft_mag) # Log compression log_mag = np.log1p(sqrt_mag * self.global_stats['alpha']) # Normalize with global stats normalized = np.clip( (log_mag - self.global_stats['global_low']) / (self.global_stats['global_high'] - self.global_stats['global_low'] + 1e-8), 0, 1 ) # Gamma correction gamma_corrected = np.power(normalized, self.global_stats['gamma']) return gamma_corrected def denormalize_stft(self, normalized): """Reverse the STFT normalization""" normalized = np.clip(normalized, 0, 1) # Reverse gamma pre_gamma = np.power(normalized + 1e-8, 1.0 / self.global_stats['gamma']) # Reverse normalization log_mag = pre_gamma * (self.global_stats['global_high'] - self.global_stats['global_low']) + \ self.global_stats['global_low'] # Reverse log compression sqrt_mag = (np.exp(log_mag) - 1) / self.global_stats['alpha'] sqrt_mag = np.maximum(sqrt_mag, 0) # Reverse sqrt compression mag = sqrt_mag ** 2 mag = np.maximum(mag, 0) return mag def process_channel(self, audio_channel, texture_amount): """Process a single audio channel""" # Process in chunks (15 seconds, matching training) chunk_duration = 15.0 chunk_samples = int(chunk_duration * self.sample_rate) hop_samples = chunk_samples // 2 # 50% overlap chunks = [] for start in range(0, len(audio_channel) - chunk_samples + hop_samples, hop_samples): chunk = audio_channel[start:start + chunk_samples] if len(chunk) < chunk_samples: chunk = np.pad(chunk, (0, chunk_samples - len(chunk)), mode='constant') chunks.append((chunk, start)) # If audio is shorter than chunk_samples, process as single chunk if len(chunks) == 0: chunk = np.pad(audio_channel, (0, max(0, chunk_samples - len(audio_channel))), mode='constant') chunks = [(chunk, 0)] processed_chunks = [] for idx, (chunk, start_pos) in enumerate(chunks): # Compute complex STFT stft_complex = librosa.stft( chunk, n_fft=self.n_fft, hop_length=self.hop_size, win_length=self.win_size ) # Get magnitude and phase magnitude = np.abs(stft_complex) phase = np.angle(stft_complex) # Normalize magnitude normalized_mag = self.normalize_stft(magnitude) # Process with neural network with torch.no_grad(): mag_tensor = torch.FloatTensor(normalized_mag).unsqueeze(0).unsqueeze(0).to(self.device) texture_tensor = torch.FloatTensor([[texture_amount]]).to(self.device) # Apply CAK processing processed_mag, patterns = self.generator(mag_tensor, texture_tensor) processed_mag = processed_mag.cpu().numpy()[0, 0] # Denormalize back to linear magnitude processed_mag_linear = self.denormalize_stft(processed_mag) # Reconstruct complex STFT using original phase processed_stft = processed_mag_linear * np.exp(1j * phase) # Convert back to audio using iSTFT processed_chunk = librosa.istft( processed_stft, hop_length=self.hop_size, win_length=self.win_size, length=chunk_samples ) processed_chunks.append((processed_chunk, start_pos)) # Combine chunks with crossfade output_audio = self.combine_chunks(processed_chunks, len(audio_channel)) # Match RMS to original original_rms = np.sqrt(np.mean(audio_channel ** 2)) output_rms = np.sqrt(np.mean(output_audio ** 2)) if output_rms > 0: rms_ratio = original_rms / output_rms rms_ratio = np.clip(rms_ratio, 0.1, 10.0) output_audio *= rms_ratio # Soft clipping output_audio = np.tanh(output_audio * 0.95) / 0.95 return output_audio def combine_chunks(self, chunks, target_length): """Combine overlapping chunks with crossfade""" output = np.zeros(target_length) weights = np.zeros(target_length) for chunk, start_pos in chunks: if chunk.ndim > 1: chunk = chunk.flatten() end_pos = min(start_pos + len(chunk), target_length) chunk_len = end_pos - start_pos # Fade window window = np.ones(chunk_len) fade_len = min(self.hop_size // 2, chunk_len // 4) if start_pos > 0 and fade_len > 0: fade_in = np.linspace(0, 1, fade_len) window[:fade_len] = fade_in if end_pos < target_length and fade_len > 0: fade_out = np.linspace(1, 0, fade_len) window[-fade_len:] = fade_out output[start_pos:end_pos] += chunk[:chunk_len] * window weights[start_pos:end_pos] += window mask = weights > 0 output[mask] /= weights[mask] return output def process_audio(self, audio_path, texture_amount): """Process audio file with stereo support""" # Load audio (mono=False to preserve stereo if present) y, sr = librosa.load(audio_path, sr=self.sample_rate, mono=False) print(f"\n=== Processing Audio ===") print(f"Texture amount: {texture_amount}") print(f"Audio shape: {y.shape}") print(f"Sample rate: {sr}Hz") # Check if stereo or mono if y.ndim == 2: # Stereo print("Processing stereo audio...") left_channel = self.process_channel(y[0], texture_amount) right_channel = self.process_channel(y[1], texture_amount) output_audio = np.stack([left_channel, right_channel]) print(f"Output shape: {output_audio.shape} (stereo)") else: # Mono print("Processing mono audio...") output_audio = self.process_channel(y, texture_amount) print(f"Output shape: {output_audio.shape} (mono)") return output_audio, self.sample_rate def get_gate_value(self, texture_amount): """Get gate activation value for a given texture amount""" with torch.no_grad(): texture_tensor = torch.tensor([[texture_amount]]).to(self.device) gate_value = self.generator.cak.soft_gate(texture_tensor).item() return gate_value # Initialize processor globally print("Initializing CAK Audio Processor...") processor = AudioProcessor() print("Processor ready!") def process_audio_gradio(audio_file, texture_amount, progress=gr.Progress()): """Gradio interface function""" if audio_file is None: return None, None, "Please upload an audio file first." try: progress(0.1, desc="Loading audio...") # Process the audio progress(0.3, desc="Processing audio...") processed_audio, sr = processor.process_audio(audio_file, texture_amount) # Save to temporary files for Gradio progress(0.8, desc="Preparing output...") # Original audio original_audio, _ = librosa.load(audio_file, sr=sr, mono=False) original_temp = tempfile.NamedTemporaryFile(delete=False, suffix='.wav') # Handle stereo vs mono for original if original_audio.ndim == 2: sf.write(original_temp.name, original_audio.T, sr) else: sf.write(original_temp.name, original_audio, sr) # Processed audio processed_temp = tempfile.NamedTemporaryFile(delete=False, suffix='.wav') # Handle stereo vs mono for processed if processed_audio.ndim == 2: sf.write(processed_temp.name, processed_audio.T, sr) else: sf.write(processed_temp.name, processed_audio, sr) duration = len(original_audio.flatten()) / sr gate_value = processor.get_gate_value(texture_amount) status = f"✓ Processing complete! Duration: {duration:.2f}s | Gate value: {gate_value:.3f}" progress(1.0, desc="Done!") return original_temp.name, processed_temp.name, status except Exception as e: import traceback error_msg = f"Error processing audio: {str(e)}\n{traceback.format_exc()}" print(error_msg) return None, None, f"❌ Error: {str(e)}" def update_gate_display(texture_amount): """Update gate value display when slider moves""" gate_value = processor.get_gate_value(texture_amount) threshold = processor.generator.cak.threshold.item() if texture_amount < threshold: status = "minimal effect" elif texture_amount == 0: status = "identity (bypass)" else: status = "texture active" return f"Gate activation: {gate_value:.3f} ({status})" # Create Gradio interface with gr.Blocks(title="CAK Neural Audio Processor", theme=gr.themes.Monochrome()) as demo: gr.Markdown( """ # CAK Neural Audio Processor ### Conditioning Aware Kernels Upload an audio file and adjust the texture control to apply learned neural transformations. The model was trained on 200 samples using a single 3×3 convolutional kernel. **Supports both mono and stereo audio** • Best results with 44.1kHz WAV files """ ) with gr.Row(): with gr.Column(): audio_input = gr.Audio( label="Upload Audio", type="filepath", sources=["upload"] ) texture_slider = gr.Slider( minimum=0.0, maximum=2.0, value=0.0, step=0.01, label="Texture Amount", info="Control the intensity of the neural transformation" ) gate_display = gr.Textbox( label="Gate Status", value="Gate activation: 0.002 (identity)", interactive=False ) # Update gate display when slider moves texture_slider.change( fn=update_gate_display, inputs=[texture_slider], outputs=[gate_display] ) process_btn = gr.Button("Process Audio", variant="primary", size="lg") status_text = gr.Textbox( label="Status", value="Ready to process audio", interactive=False ) with gr.Row(): original_audio = gr.Audio( label="Original Audio", type="filepath", interactive=False ) processed_audio = gr.Audio( label="Processed Audio", type="filepath", interactive=False ) gr.Markdown( """ --- **Tips:** - Values < 0.3: Minimal effect (identity preservation) - Values ≥ 0.3: Progressive texture modulation - Use the audio players for A/B comparison - Download processed audio using the ⋮ menu in the audio player *Research paper: [CAK: Emergent Audio Effects from Minimal Deep Learning](https://arxiv.org/html/2508.02643v1)* """ ) # Connect the process button process_btn.click( fn=process_audio_gradio, inputs=[audio_input, texture_slider], outputs=[original_audio, processed_audio, status_text] ) # Launch the app if __name__ == "__main__": demo.launch()