import torch import soundfile as sf import numpy as np import argparse import os import yaml import julius import sys currentdir = os.path.dirname(os.path.realpath(__file__)) sys.path.append(os.path.dirname(currentdir)) from networks import Dasp_Mastering_Style_Transfer, Effects_Encoder from modules.loss import AudioFeatureLoss, Loss from modules.data_normalization import Audio_Effects_Normalizer def convert_audio(wav: torch.Tensor, from_rate: float, to_rate: float, to_channels: int) -> torch.Tensor: """Convert audio to new sample rate and number of audio channels. """ wav = julius.resample_frac(wav, int(from_rate), int(to_rate)) wav = convert_audio_channels(wav, to_channels) return wav class MasteringStyleTransfer: def __init__(self, args): self.args = args self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # Load models self.effects_encoder = self.load_effects_encoder() self.mastering_converter = self.load_mastering_converter() self.fx_normalizer = Audio_Effects_Normalizer(precomputed_feature_path=args.fx_norm_feature_path, \ STEMS=['mixture'], \ EFFECTS=['eq', 'imager', 'loudness']) # Loss functions self.af_loss = AudioFeatureLoss( weights=ito_config['af_weights'], sample_rate=ito_config['sample_rate'], stem_separation=False, use_clap=False ) self.clap_loss = CLAPFeatureLoss(distance_fn='cosine') def load_effects_encoder(self): effects_encoder = Effects_Encoder(self.args.cfg_enc) reload_weights(effects_encoder, self.args.encoder_path, self.device) effects_encoder.to(self.device) effects_encoder.eval() return effects_encoder def load_mastering_converter(self): mastering_converter = Dasp_Mastering_Style_Transfer(num_features=2048, sample_rate=self.args.sample_rate, tgt_fx_names=['eq', 'distortion', 'multiband_comp', 'gain', 'imager', 'limiter'], model_type='tcn', config=self.args.cfg_converter, batch_size=1) reload_weights(mastering_converter, self.args.model_path, self.device) mastering_converter.to(self.device) mastering_converter.eval() return mastering_converter def get_reference_embedding(self, reference_tensor): with torch.no_grad(): reference_feature = self.effects_encoder(reference_tensor) return reference_feature def mastering_style_transfer(self, input_tensor, reference_feature): with torch.no_grad(): output_audio = self.mastering_converter(input_tensor, reference_feature) predicted_params = self.mastering_converter.get_last_predicted_params() return output_audio, predicted_params def inference_time_optimization(self, input_tensor, reference_tensor, ito_config, initial_reference_feature): fit_embedding = torch.nn.Parameter(initial_reference_feature) optimizer = getattr(torch.optim, ito_config['optimizer'])([fit_embedding], lr=ito_config['learning_rate']) min_loss = float('inf') min_loss_step = 0 all_results = [] for step in range(ito_config['num_steps']): optimizer.zero_grad() output_audio = self.mastering_converter(input_tensor, fit_embedding) current_params = self.mastering_converter.get_last_predicted_params() # Compute loss if ito_config['loss_function'] == 'AudioFeatureLoss': losses = self.af_loss(output_audio, reference_tensor) elif ito_config['loss_function'] == 'CLAPFeatureLoss': if ito_config['clap_target_type'] == 'Audio': target = ito_reference_tensor else: target = ito_config['clap_text_prompt'] losses = self.clap_loss(est_targets, target, self.args.sample_rate) total_loss = sum(losses.values()) if total_loss < min_loss: min_loss = total_loss.item() min_loss_step = step # Log top 5 parameter differences if step == 0: initial_params = current_params top_5_diff = self.get_top_n_diff_string(initial_params, current_params, top_n=5) log_entry = f"Step {step + 1}\n Loss: {total_loss.item():.4f}\n{top_5_diff}\n" all_results.append({ 'step': step + 1, 'loss': total_loss.item(), 'audio': output_audio.detach().cpu().numpy(), 'params': current_params, 'log': log_entry }) total_loss.backward() optimizer.step() return all_results, min_loss_step def preprocess_audio(self, audio, target_sample_rate=44100, normalize=False): sample_rate, data = audio # Normalize audio to -1 to 1 range if data.dtype == np.int16: data = data.astype(np.float32) / 32768.0 elif data.dtype == np.float32: data = np.clip(data, -1.0, 1.0) else: raise ValueError(f"Unsupported audio data type: {data.dtype}") # Ensure stereo channels if data.ndim == 1: data = np.stack([data, data]) elif data.ndim == 2: if data.shape[0] == 2: pass # Already in correct shape elif data.shape[1] == 2: data = data.T else: data = np.stack([data[:, 0], data[:, 0]]) # Duplicate mono channel else: raise ValueError(f"Unsupported audio shape: {data.shape}") # Resample if necessary if sample_rate != target_sample_rate: data = julius.resample_frac(torch.from_numpy(data), sample_rate, target_sample_rate).numpy() # Apply fx normalization for input audio during mastering style transfer if normalize: data = self.fx_normalizer.normalize_audio(data.T, 'mixture').T # Convert to torch tensor data_tensor = torch.FloatTensor(data).unsqueeze(0) return data_tensor.to(self.device) def process_audio(self, input_audio, reference_audio): input_tensor = self.preprocess_audio(input_audio, self.args.sample_rate, normalize=True) reference_tensor = self.preprocess_audio(reference_audio, self.args.sample_rate) reference_feature = self.get_reference_embedding(reference_tensor) output_audio, predicted_params = self.mastering_style_transfer(input_tensor, reference_feature) return output_audio, predicted_params, self.args.sample_rate, input_tensor def get_param_output_string(self, params): if params is None: return "No parameters available" param_mapper = { 'eq': { 'low_shelf_gain_db': ('Low Shelf Gain', 'dB', -20, 20), 'low_shelf_cutoff_freq': ('Low Shelf Cutoff', 'Hz', 20, 2000), 'low_shelf_q_factor': ('Low Shelf Q', '', 0.1, 5.0), 'band0_gain_db': ('Low-Mid Band Gain', 'dB', -20, 20), 'band0_cutoff_freq': ('Low-Mid Band Frequency', 'Hz', 80, 2000), 'band0_q_factor': ('Low-Mid Band Q', '', 0.1, 5.0), 'band1_gain_db': ('Mid Band Gain', 'dB', -20, 20), 'band1_cutoff_freq': ('Mid Band Frequency', 'Hz', 2000, 8000), 'band1_q_factor': ('Mid Band Q', '', 0.1, 5.0), 'band2_gain_db': ('High-Mid Band Gain', 'dB', -20, 20), 'band2_cutoff_freq': ('High-Mid Band Frequency', 'Hz', 8000, 12000), 'band2_q_factor': ('High-Mid Band Q', '', 0.1, 5.0), 'band3_gain_db': ('High Band Gain', 'dB', -20, 20), 'band3_cutoff_freq': ('High Band Frequency', 'Hz', 12000, 20000), 'band3_q_factor': ('High Band Q', '', 0.1, 5.0), 'high_shelf_gain_db': ('High Shelf Gain', 'dB', -20, 20), 'high_shelf_cutoff_freq': ('High Shelf Cutoff', 'Hz', 4000, 20000), 'high_shelf_q_factor': ('High Shelf Q', '', 0.1, 5.0), }, 'distortion': { 'drive_db': ('Drive', 'dB', 0, 8), 'parallel_weight_factor': ('Dry/Wet Mix', '%', 0, 100), }, 'multiband_comp': { 'low_cutoff': ('Low/Mid Crossover', 'Hz', 20, 1000), 'high_cutoff': ('Mid/High Crossover', 'Hz', 1000, 20000), 'parallel_weight_factor': ('Dry/Wet Mix', '%', 0, 100), 'low_shelf_comp_thresh': ('Low Band Comp Threshold', 'dB', -60, 0), 'low_shelf_comp_ratio': ('Low Band Comp Ratio', ': 1', 1, 20), 'low_shelf_exp_thresh': ('Low Band Exp Threshold', 'dB', -60, 0), 'low_shelf_exp_ratio': ('Low Band Exp Ratio', ': 1', 1, 20), 'low_shelf_at': ('Low Band Attack Time', 'ms', 5, 100), 'low_shelf_rt': ('Low Band Release Time', 'ms', 5, 100), 'mid_band_comp_thresh': ('Mid Band Comp Threshold', 'dB', -60, 0), 'mid_band_comp_ratio': ('Mid Band Comp Ratio', ': 1', 1, 20), 'mid_band_exp_thresh': ('Mid Band Exp Threshold', 'dB', -60, 0), 'mid_band_exp_ratio': ('Mid Band Exp Ratio', ': 1', 0, 1), 'mid_band_at': ('Mid Band Attack Time', 'ms', 5, 100), 'mid_band_rt': ('Mid Band Release Time', 'ms', 5, 100), 'high_shelf_comp_thresh': ('High Band Comp Threshold', 'dB', -60, 0), 'high_shelf_comp_ratio': ('High Band Comp Ratio', ': 1', 1, 20), 'high_shelf_exp_thresh': ('High Band Exp Threshold', 'dB', -60, 0), 'high_shelf_exp_ratio': ('High Band Exp Ratio', ': 1', 1, 20), 'high_shelf_at': ('High Band Attack Time', 'ms', 5, 100), 'high_shelf_rt': ('High Band Release Time', 'ms', 5, 100), }, 'gain': { 'gain_db': ('Output Gain', 'dB', -24, 24), }, 'imager': { 'width': ('Stereo Width', '', 0, 1), }, 'limiter': { 'threshold': ('Threshold', 'dB', -60, 0), 'at': ('Attack Time', 'ms', 5, 100), 'rt': ('Release Time', 'ms', 5, 100), }, } output = [] for fx_name, fx_params in params.items(): output.append(f"{fx_name.upper()}:") if isinstance(fx_params, dict): for param_name, param_value in fx_params.items(): if isinstance(param_value, torch.Tensor): param_value = param_value.item() print(f"fx name: {fx_name} param_name: {param_name}") if fx_name in param_mapper and param_name in param_mapper[fx_name]: friendly_name, unit, min_val, max_val = param_mapper[fx_name][param_name] if unit=='%': param_value = param_value * 100 current_content = f" {friendly_name}: {param_value:.2f} {unit}" if param_name=='mid_band_exp_ratio': current_content += f" (Range: {min_val}-{max_val})" output.append(current_content) else: output.append(f" {param_name}: {param_value:.2f}") else: # stereo imager width_percentage = fx_params.item() * 200 output.append(f" Stereo Width: {width_percentage:.2f}% (Range: 0-200%)") return "\n".join(output) def get_top_n_diff_string(self, initial_params, ito_params, top_n=5): if initial_params is None or ito_params is None: return "Cannot compare parameters" all_diffs = [] for fx_name in initial_params.keys(): if isinstance(initial_params[fx_name], dict): for param_name in initial_params[fx_name].keys(): initial_value = initial_params[fx_name][param_name] ito_value = ito_params[fx_name][param_name] param_range = self.mastering_converter.fx_processors[fx_name].param_ranges[param_name] normalized_diff = abs((ito_value - initial_value) / (param_range[1] - param_range[0])) all_diffs.append((fx_name, param_name, initial_value.item(), ito_value.item(), normalized_diff.item())) else: initial_value = initial_params[fx_name] ito_value = ito_params[fx_name] normalized_diff = abs(ito_value - initial_value) all_diffs.append((fx_name, 'width', initial_value.item(), ito_value.item(), normalized_diff.item())) top_diffs = sorted(all_diffs, key=lambda x: x[4], reverse=True)[:top_n] output = [f" Top {top_n} parameter differences (initial / ITO / normalized diff):"] for fx_name, param_name, initial_value, ito_value, normalized_diff in top_diffs: output.append(f" {fx_name.upper()} - {param_name}: {initial_value:.2f} / {ito_value:.2f} / {normalized_diff:.2f}") return "\n".join(output) def reload_weights(model, ckpt_path, device): checkpoint = torch.load(ckpt_path, map_location=device) from collections import OrderedDict new_state_dict = OrderedDict() for k, v in checkpoint["model"].items(): name = k[7:] # remove `module.` new_state_dict[name] = v model.load_state_dict(new_state_dict, strict=False)