{ "cells": [ { "cell_type": "code", "execution_count": null, "id": "4d5d19aa", "metadata": {}, "outputs": [], "source": [ "import torch\n", "import torch.fft as fft\n", "import numpy as np\n", "import pyloudnorm as pyln\n", "\n", "def combine_width_octave(\n", " width_deltas, # array-like (can be signed)\n", " octave_totals, # array-like (>=0)\n", " weights=(0.5, 0.5), # (weight_width, weight_octave)\n", " method=\"auto\" # \"auto\" | \"minmax\" | \"robust\"\n", "):\n", " w = np.abs(np.asarray(width_deltas, float))\n", " o = np.asarray(octave_totals, float)\n", " n = len(w)\n", "\n", " def robust_norm(x):\n", " q1, q3 = np.percentile(x, 25), np.percentile(x, 75)\n", " iqr = q3 - q1\n", " if iqr <= 1e-12:\n", " return None # signal to fall back\n", " med = np.median(x)\n", " return (x - med) / iqr\n", "\n", " def minmax_norm(x):\n", " xmin, xmax = float(np.min(x)), float(np.max(x))\n", " rng = xmax - xmin\n", " if rng <= 1e-12:\n", " return np.zeros_like(x, dtype=float)\n", " return (x - xmin) / rng\n", "\n", " # Decide method\n", " if method == \"auto\":\n", " use_minmax = (n < 4) # with 2 clips, prefer min-max\n", " else:\n", " use_minmax = (method == \"minmax\")\n", "\n", " if not use_minmax:\n", " z_w, z_o = robust_norm(w), robust_norm(o)\n", " if z_w is None or z_o is None:\n", " use_minmax = True\n", "\n", " if use_minmax:\n", " n_w, n_o = minmax_norm(w), minmax_norm(o)\n", " combo = weights[0]*n_w + weights[1]*n_o # already 0..1\n", " score_0_100 = 100 * combo\n", " method_used = \"minmax\"\n", " else:\n", " combo = weights[0]*z_w + weights[1]*z_o # unbounded\n", " combo01 = minmax_norm(combo) # map to 0..1\n", " score_0_100 = 100 * combo01\n", " method_used = \"robust\"\n", "\n", " return score_0_100, method_used\n", "\n", "\n", "\n", "def third_octave_bands(sr, fmin=20.0, fmax=None):\n", " \"\"\"\n", " Compute 1/3-octave band center frequencies and edges.\n", " \"\"\"\n", " if fmax is None:\n", " fmax = sr / 2.0\n", "\n", " k = np.arange(-30, 30) # wide enough range\n", " f_center = 1000.0 * (2.0 ** (k / 3.0)) # ISO 1/3 octave centers\n", " f_center = f_center[(f_center >= fmin) & (f_center <= fmax)]\n", " \n", " f_lower = f_center / (2 ** (1/6))\n", " f_upper = f_center * (2 ** (1/6))\n", " return f_center, f_lower, f_upper\n", "\n", "\n", "def third_octave_response_db(waveform: torch.Tensor, sr: int):\n", " \"\"\"\n", " Compute 1/3 octave magnitude response in dB from waveform.\n", " \n", " Args:\n", " waveform (torch.Tensor): shape (n_samples,) or (1, n_samples)\n", " sr (int): sample rate\n", " \n", " Returns:\n", " freqs (np.ndarray): band center frequencies\n", " mags_db (torch.Tensor): band magnitudes in dB\n", " \"\"\"\n", " if waveform.ndim > 1:\n", " waveform = waveform.squeeze(0)\n", " \n", " n = waveform.numel()\n", " spec = fft.rfft(waveform)\n", " mag = torch.abs(spec) / n\n", " freqs = torch.fft.rfftfreq(n, d=1.0/sr)\n", "\n", " # Get bands\n", " f_center, f_lower, f_upper = third_octave_bands(sr)\n", " band_mags = []\n", " for fl, fu in zip(f_lower, f_upper):\n", " idx = (freqs >= fl) & (freqs < fu)\n", " if idx.any():\n", " band_mags.append(mag[idx].mean())\n", " else:\n", " band_mags.append(torch.tensor(0.0))\n", "\n", " band_mags = torch.stack(band_mags)\n", "\n", " # Convert to dB (avoid log(0))\n", " mags_db = 20 * torch.log10(band_mags + 1e-12)\n", " \n", " return f_center, mags_db\n", "\n", "def stereo_width(waveform: torch.Tensor):\n", " \"\"\"\n", " Compute the stereo width of a waveform.\n", " \"\"\"\n", " # can you implement this?\n", " # Assume waveform shape is (2, seq_len)\n", " if waveform.ndim != 2 or waveform.shape[0] != 2:\n", " raise ValueError(\"waveform must have shape (2, seq_len) for stereo width calculation\")\n", " left = waveform[0]\n", " right = waveform[1]\n", " # Compute correlation coefficient between L and R\n", " left = left - left.mean()\n", " right = right - right.mean()\n", " numerator = (left * right).mean()\n", " denominator = torch.sqrt((left ** 2).mean() * (right ** 2).mean()) + 1e-12\n", " corr = numerator / denominator\n", " # Stereo width: 0 = mono, 1 = fully wide (L and R uncorrelated), -1 = fully out of phase\n", " width = torch.sqrt(1 - corr ** 2)\n", " return width.item()\n", "\n", "def loudness(waveform: torch.Tensor, sample_rate: float):\n", " \"\"\"\n", " Compute the loudness of a waveform.\n", " \"\"\"\n", " meter = pyln.Meter(sample_rate)\n", " loudness = meter.integrated_loudness(waveform.permute(1, 0).numpy())\n", " return loudness\n", "\n", "\n", " \n", "import torch\n", "from typing import Optional, Dict, Any\n", "\n", "# Assumes third_octave_bands, third_octave_response_db, stereo_width\n", "# are defined exactly as in your snippet above.\n", "\n", "def analyze_audio_first_last(\n", " audio: torch.Tensor,\n", " sample_rate: int,\n", " segment_duration: int = 60,\n", ") -> Dict[str, Any]:\n", " \"\"\"\n", " Compute 1/3-octave response and stereo width for the first and last segments\n", " of a single audio tensor, and return their deltas.\n", "\n", " Args:\n", " audio: Tensor of shape (channels, samples) or (samples,)\n", " sample_rate: sample rate in Hz\n", " segment_duration: segment length in seconds for the first/last comparison\n", "\n", " Returns:\n", " {\n", " 'center_freqs': np.ndarray,\n", " 'third_octave_first': Tensor[dB],\n", " 'third_octave_last': Tensor[dB],\n", " 'delta_response': Tensor[dB], # first - last\n", " 'total_delta': Tensor[scalar], # L1 magnitude of delta_response\n", " 'stereo_width_first': float or None,\n", " 'stereo_width_last': float or None,\n", " 'stereo_width_delta': float or None, # first - last\n", " }\n", " \"\"\"\n", " if not isinstance(audio, torch.Tensor):\n", " raise TypeError(\"audio must be a torch.Tensor\")\n", " if audio.numel() == 0:\n", " raise ValueError(\"audio is empty\")\n", "\n", " # Determine segment length in samples (clip to available length)\n", " seg_len = min(audio.shape[-1], segment_duration * sample_rate)\n", "\n", " # Helper: mono mixdown for 1/3-octave analysis\n", " mono = audio.mean(dim=0) if audio.ndim > 1 else audio\n", " first_mono = mono[:seg_len]\n", " last_mono = mono[-seg_len:]\n", "\n", " # 1/3-octave responses (dB)\n", " center_freqs_first, oct_first = third_octave_response_db(first_mono, sample_rate)\n", " center_freqs_last, oct_last = third_octave_response_db(last_mono, sample_rate)\n", "\n", " # Centers should match; keep the first as canonical\n", " if len(center_freqs_first) != len(center_freqs_last) or (center_freqs_first != center_freqs_last).any():\n", " raise RuntimeError(\"Mismatched third-octave centers between first and last segments.\")\n", "\n", " delta = oct_first - oct_last\n", " total_delta = delta.abs().sum()\n", "\n", " # Stereo width (if stereo input)\n", " def maybe_width(x: torch.Tensor) -> Optional[float]:\n", " if x.ndim == 2 and x.shape[0] == 2 and x.shape[1] > 0:\n", " return stereo_width(x)\n", " return None\n", "\n", " first_full = audio[:, :seg_len] if audio.ndim == 2 else audio\n", " last_full = audio[:, -seg_len:] if audio.ndim == 2 else audio\n", "\n", " width_first = maybe_width(first_full)\n", " width_last = maybe_width(last_full)\n", " width_delta = None\n", " if (width_first is not None) and (width_last is not None):\n", " width_delta = float(width_first - width_last)\n", "\n", " # loudness\n", " loudness_first = loudness(first_full, sample_rate)\n", " loudness_last = loudness(last_full, sample_rate)\n", " loudness_delta = None\n", " if (loudness_first is not None) and (loudness_last is not None):\n", " loudness_delta = float(loudness_first - loudness_last)\n", "\n", " return {\n", " \"center_freqs\": center_freqs_first, # np.ndarray\n", " \"third_octave_first\": oct_first, # Tensor[dB]\n", " \"third_octave_last\": oct_last, # Tensor[dB]\n", " \"delta_response\": delta, # Tensor[dB]\n", " \"total_delta\": total_delta, # Tensor[scalar]\n", " \"stereo_width_first\": width_first, # float or None\n", " \"stereo_width_last\": width_last, # float or None\n", " \"stereo_width_delta\": width_delta, # float or None\n", " \"loudness_first\": loudness_first, # float or None\n", " \"loudness_last\": loudness_last, # float or None\n", " \"loudness_delta\": loudness_delta, # float or None\n", " }\n" ] }, { "cell_type": "code", "execution_count": null, "id": "7d55f859", "metadata": {}, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": null, "id": "b38e3c00", "metadata": {}, "outputs": [], "source": [ "# test some of the local data\n", "CODEC_FILEPATH = \"s3://suno-data/minz/models/dac_vae_tuned_25hz.pth\"\n", "\n", "from suno_utils.tasks.dac_vae_fixed_25hz import (\n", " preload_models as preload_codec_models,\n", " decode as codec_decode,\n", " encode as codec_encode,\n", " decode_stream_to_full_audio,\n", ")\n", "\n", "_ = preload_codec_models(CODEC_FILEPATH)\n" ] }, { "cell_type": "code", "execution_count": null, "id": "db2aa433", "metadata": {}, "outputs": [], "source": [ "import os\n", "#source_dir = \"/app2/suno/data/christian/outputs/v3-base-data-ctx-discogs-subset-t0\"\n", "#model_name = \"16n_25hz_v45_infill_shared_flow_resume_1_75m\"\n", "\n", "source_dir = \"/app2/suno/data/christian/outputs/v3-base-data-ctx-rs-t2\"\n", "model_name = \"4n_25hz_2b_flow_5e5_sft_t8_500k\"\n", "# get all the dirs in the source_dir\n", "dirs = os.listdir(source_dir)\n", "\n", "# get the dirs that start with \"2025-09-29\"\n", "dirs = [d for d in dirs if os.path.isdir(os.path.join(source_dir, d))]\n", "print(dirs)\n" ] }, { "cell_type": "code", "execution_count": null, "id": "65b6fd9b", "metadata": {}, "outputs": [], "source": [ "from suno_utils.audio import Audio\n", "\n", "# lets run this algo across all pairs in this data set\n", "for dirname in dirs[1:2]:\n", " stereo_width_deltas = []\n", " total_octave_deltas = []\n", " loudness_deltas = []\n", " for file_idx in [0, 1]:\n", " dir_path = os.path.join(source_dir, dirname)\n", " audio_filepath = os.path.join(dir_path, f\"{dirname}_{model_name}_{file_idx}.mp3\")\n", "\n", " audio = Audio.from_file(audio_filepath, n_channels=2)\n", "\n", " # convert this to float\n", " audio_tensor = torch.from_numpy(audio.array_float).float()\n", "\n", " results = analyze_audio_first_last(audio_tensor, 48000)\n", " #print(results)\n", " stereo_width_deltas.append(results['stereo_width_delta'])\n", " total_octave_deltas.append(results['total_delta'])\n", " loudness_deltas.append(results['loudness_delta'])\n", " print(\"stereo width delta: \", results['stereo_width_delta'])\n", " print(\"total octave delta: \", results['total_delta'])\n", " print(\"loudness delta: \", results['loudness_delta'])\n", " print()\n", " # compare the results\n", " combined_scores = combine_width_octave(stereo_width_deltas, total_octave_deltas)\n", " print(f\"combined scores: {combined_scores}\")\n" ] }, { "cell_type": "code", "execution_count": null, "id": "3dff4e31", "metadata": {}, "outputs": [], "source": [ "import numpy as np\n", "import matplotlib.pyplot as plt\n", "dir_idx = 12\n", "dir_path = os.path.join(source_dir, dirs[dir_idx])\n", "model_name = \"16n_25hz_v45_infill_shared_flow_resume_1_75m\"\n", "model_name = \"4n_25hz_2b_flow_5e5_sft_t8_500k\"\n", "\n", "# get all the files in the dir\n", "files = os.listdir(dir_path)\n", "# get the upsample vae and the origina_vae\n", "upsampled_audio_filepath = os.path.join(dir_path, f\"{dirs[dir_idx]}_{model_name}_0.mp3\")\n", "original_audio_filepath = os.path.join(dir_path, f\"{dirs[dir_idx]}_{model_name}_1.mp3\")\n", "\n", "upsampled_audio = Audio.from_file(upsampled_audio_filepath, n_channels=2)\n", "original_audio = Audio.from_file(original_audio_filepath, n_channels=2)\n", "\n", "# convert this to float\n", "upsampled_audio_tensor = torch.from_numpy(upsampled_audio.array_float).float()\n", "original_audio_tensor = torch.from_numpy(original_audio.array_float).float()\n", "print(upsampled_audio_tensor.shape)\n", "print(original_audio_tensor.shape)\n", "\n", "sample_rate = 48000\n", "\n", "results = compare_audio_deltas(upsampled_audio_tensor, original_audio_tensor)\n", "\n", "print(f\"a spectrum delta: {results['total_delta_a']}\")\n", "print(f\"b spectrum delta: {results['total_delta_b']}\")\n", "\n", "print(f\"a stereo width first: {results['stereo_width_first_a']}\")\n", "print(f\"a stereo width last: {results['stereo_width_last_a']}\")\n", "print(f\"a stereo width delta: {results['stereo_width_last_a'] - results['stereo_width_first_a']}\")\n", "print(f\"b stereo width first: {results['stereo_width_first_b']}\")\n", "print(f\"b stereo width last: {results['stereo_width_last_b']}\")\n", "print(f\"b stereo width delta: {results['stereo_width_last_b'] - results['stereo_width_first_b']}\")\n", "\n", "\n", "# plot the first 30s\n", "plt.figure(figsize=(6, 4))\n", "#plt.semilogx(center_freqs_upsampled, upsampled_audio_third_octave, label=\"upsampled_first_30s\", linewidth=2)\n", "#plt.semilogx(center_freqs_upsampled_last, upsampled_audio_third_octave_last, label=\"upsampled_last_30s\", linewidth=2)\n", "plt.semilogx(results['center_freqs_a'], results['delta_response_a'], label=\"a\", linewidth=2)\n", "#plt.semilogx(center_freqs_original, original_audio_third_octave[1], label=\"original_first_30s\", linewidth=2)\n", "#plt.semilogx(center_freqs_original_last, original_audio_third_octave_last[1], label=\"original_last_30s\", linewidth=2)\n", "plt.semilogx(results['center_freqs_b'], results['delta_response_b'], label=\"b\", linewidth=2)\n", "plt.grid(True, which=\"both\", ls=\"-\", alpha=0.2)\n", "plt.xlabel(\"Frequency (Hz)\")\n", "plt.ylabel(\"Magnitude (dB)\")\n", "plt.legend()\n", "plt.show()\n" ] }, { "cell_type": "code", "execution_count": null, "id": "6f7cd6e4", "metadata": {}, "outputs": [], "source": [ "upsampled_audio.play()\n", "original_audio.play()" ] }, { "cell_type": "code", "execution_count": null, "id": "08c8392f", "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "suno_diff", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.9" } }, "nbformat": 4, "nbformat_minor": 5 }