{
"cells": [
{
"cell_type": "code",
"execution_count": 4,
"metadata": {},
"outputs": [],
"source": [
"import torch\n",
"import math\n",
"import torchaudio\n",
"\n",
"\n",
"def apply_freq_boost(\n",
" audio: torch.Tensor,\n",
" sample_rate: float,\n",
" freq_hz: float = 1000.0,\n",
" boost_db: float = 10.0,\n",
"):\n",
" # Design peaking filter with Q=0.707\n",
" w0 = 2 * math.pi * freq_hz / sample_rate\n",
" alpha = math.sin(w0) / (2 * 0.707)\n",
" A = 10 ** (boost_db / 40.0)\n",
"\n",
" b0 = 1 + alpha * A\n",
" b1 = -2 * math.cos(w0)\n",
" b2 = 1 - alpha * A\n",
" a0 = 1 + alpha / A\n",
" a1 = -2 * math.cos(w0)\n",
" a2 = 1 - alpha / A\n",
"\n",
" # Normalize coefficients by a0\n",
" b0 = b0 / a0\n",
" b1 = b1 / a0\n",
" b2 = b2 / a0\n",
" a1 = a1 / a0\n",
" a2 = a2 / a0\n",
" a0 = 1.0\n",
"\n",
" return torchaudio.functional.biquad(audio, b0, b1, b2, a0, a1, a2)\n"
]
},
{
"cell_type": "code",
"execution_count": 21,
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"import glob\n",
"import torchaudio\n",
"from tqdm import tqdm\n",
"\n",
"\n",
"def process_audio_pairs(base_dir: str):\n",
" \"\"\"\n",
" Process audio pairs from different test directories and return paired data.\n",
"\n",
" Args:\n",
" base_dir: Base directory containing the test folders\n",
"\n",
" Returns:\n",
" List of tuples containing (audio_pairs, test_type)\n",
" \"\"\"\n",
" # Test types and their corresponding directories\n",
" test_types = {\n",
" \"codec_cycled\": \"codec_cycle_test\",\n",
" \"diff_cycled\": \"diffusion_cycle_test\",\n",
" \"gen_pref\": \"dpo_diffusion_test_set\",\n",
" }\n",
"\n",
" def load_and_process_audio(filepath: str):\n",
" \"\"\"Load and preprocess audio file.\"\"\"\n",
" try:\n",
" x, sr = torchaudio.load(filepath)\n",
" if sr != 48000:\n",
" x = torchaudio.functional.resample(x, sr, 48000)\n",
" return x[:, : 48000 * 120] # Trim to 120 seconds\n",
" except Exception as e:\n",
" print(f\"Error processing {filepath}: {e}\")\n",
" return None\n",
"\n",
" def find_pairs(filepaths):\n",
" \"\"\"Find and process audio pairs from a list of file paths.\"\"\"\n",
" pairs = []\n",
" processed_files = set()\n",
"\n",
" for filepath in tqdm(filepaths):\n",
" if filepath in processed_files:\n",
" continue\n",
"\n",
" if \"a_input\" in filepath:\n",
" base_name = filepath.split(\"a_input_\")[0]\n",
" else:\n",
" base_name = filepath.split(\"_input_\")[0].split(\"_cycled_\")[0]\n",
" pref_true = \"pref=True\" in filepath\n",
" pref_false = \"pref=False\" in filepath\n",
"\n",
" # Load first audio file\n",
" x = load_and_process_audio(filepath)\n",
" if x is None:\n",
" continue\n",
"\n",
" # Find matching pair\n",
" for other_path in filepaths:\n",
" if base_name in other_path and other_path != filepath:\n",
" y = load_and_process_audio(other_path)\n",
" if y is None:\n",
" continue\n",
"\n",
" # Add pair with correct label based on preference\n",
" if pref_true:\n",
" pairs.append((x, y, \"a\"))\n",
" elif pref_false:\n",
" pairs.append((y, x, \"b\"))\n",
"\n",
" processed_files.add(filepath)\n",
" processed_files.add(other_path)\n",
" break\n",
"\n",
" return pairs\n",
"\n",
" # Process all test types\n",
" results = []\n",
" for test_type, folder in test_types.items():\n",
" dir_path = os.path.join(base_dir, folder)\n",
" print(f\"Processing {test_type} from {dir_path}\")\n",
" filepaths = glob.glob(os.path.join(dir_path, \"*.mp3\"))\n",
" pairs = find_pairs(filepaths)\n",
" print(f\"Found {len(pairs)} pairs for {test_type}\")\n",
" results.append((pairs, test_type))\n",
"\n",
" return results"
]
},
{
"cell_type": "code",
"execution_count": 22,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Processing codec_cycled from /home/christian/audio/ear-bench/codec_cycle_test\n"
]
},
{
"name": "stderr",
"output_type": "stream",
"text": [
"100%|██████████| 110/110 [00:17<00:00, 6.21it/s]\n"
]
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"Found 55 pairs for codec_cycled\n",
"Processing diff_cycled from /home/christian/audio/ear-bench/diffusion_cycle_test\n"
]
},
{
"name": "stderr",
"output_type": "stream",
"text": [
"100%|██████████| 110/110 [00:19<00:00, 5.73it/s]\n"
]
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"Found 55 pairs for diff_cycled\n",
"Processing gen_pref from /home/christian/audio/ear-bench/dpo_diffusion_test_set\n"
]
},
{
"name": "stderr",
"output_type": "stream",
"text": [
"100%|██████████| 188/188 [00:48<00:00, 3.91it/s]\n"
]
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"Found 94 pairs for gen_pref\n"
]
}
],
"source": [
"results = process_audio_pairs(\"/home/christian/audio/ear-bench\")"
]
},
{
"cell_type": "code",
"execution_count": 9,
"metadata": {},
"outputs": [
{
"data": {
"text/html": [
"\n",
" \n",
" "
],
"text/plain": [
""
]
},
"metadata": {},
"output_type": "display_data"
},
{
"data": {
"text/html": [
"\n",
" \n",
" "
],
"text/plain": [
""
]
},
"metadata": {},
"output_type": "display_data"
}
],
"source": [
"x, sr = torchaudio.load(\"/home/christian/audio/reference-audio-wav/02 Dreams.wav\")\n",
"\n",
"x = x[:, :480000]\n",
"\n",
"x_boosted = apply_freq_boost(x, sr, freq_hz=10000.0, boost_db=20.0)\n",
"\n",
"import IPython\n",
"IPython.display.display(IPython.display.Audio(x.numpy(), rate=sr))\n",
"IPython.display.display(IPython.display.Audio(x_boosted.numpy(), rate=sr))\n",
"\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": []
}
],
"metadata": {
"kernelspec": {
"display_name": "suno_env",
"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.10.9"
}
},
"nbformat": 4,
"nbformat_minor": 2
}