{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Consolidate midi datasets\n", "\n", "This notebook consolidates the midi datasets into a single dataset. It also serves as documentation for the midi datasets.\n", "\n", "It produces a metadata file pointing to the midi and audio files.\n", "\n", "## Datasets\n", "\n", "- Infinite synthetic MIDI\n", "- Mew\n", "\t- `/app2/suno/data/mew/scoretube`\n", "\t- `/app2/suno/data/mew/mmd_chunks`\n", "\t- Match midi instruments to stems.\n", "- A dataset of loosely paired real music with arrangements\n", "\t- [Slack](https://suno-main.slack.com/archives/C03A7347CSK/p1719682717375969)\n", "\t- 20k songs matched with high confidence\n", "\t- [The AI workspace that works for you. | Notion](https://www.notion.so/suno-ai/MooTube-187b01573ccf8065915ecb04af262d6c#187b01573ccf80de941bc4412c900eca)\n", "- Hooktheory melodys, 50k hooks only\n", "\t- [The AI workspace that works for you. | Notion](https://www.notion.so/suno-ai/hooktheory-91267f75401f4087b506f086abefc65f)\n", "- Trombone champ melodys. 4k full songs\n", "\t- [The AI workspace that works for you. | Notion](https://www.notion.so/suno-ai/Trombone-Champ-206b01573ccf8009aa1fe22f3440687c)\n", "- Piano Maestro\n", "\n", "\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# mew\n", "\n", "Victors's personal stash" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import os\n", "import glob\n", "import random\n", "\n", "from suno_utils.audio import Audio\n", "from suno_utils.audio.midi import Midi\n", "from tqdm import tqdm\n", "\n", "out_stem_dir = \"/app2/suno/data/victor/midi_stems\"\n", "\n", "\n", "class MidiPair:\n", " def __init__(self, midi_file: str, audio_file: str):\n", " self.midi_file = midi_file\n", " self.audio_file = audio_file\n", " self.midi = None\n", " self.audio = None\n", "\n", " def load_midi(self):\n", " if self.midi is not None:\n", " return self.midi\n", " self.midi = Midi.from_path(self.midi_file)\n", " return self.midi\n", "\n", " def load_audio(self):\n", " if self.audio is not None:\n", " return self.audio\n", " self.audio = Audio.from_file(self.audio_file)\n", " return self.audio\n", "\n", " def load_all(self):\n", " self.load_midi()\n", " self.load_audio()\n", "\n", " def __str__(self):\n", " return f\"MidiPair(midi_file={self.midi_file}, audio_file={self.audio_file})\"\n", "\n", " def __repr__(self):\n", " return self.__str__()\n", "\n", " def play(self):\n", " self.load_all()\n", " stereo_audio = self.midi.make_stereo_comparison(self.audio)\n", " stereo_audio.play()\n", "\n", "\n", "scoretube_dir = \"/app2/suno/data/victor/mew/scoretube\"\n", "\n", "\n", "def load_pairs(dir: str, audio_ext: str = \"mp3\", midi_ext: str = \"mid\"):\n", " midi_files = glob.glob(os.path.join(dir, \"**\", f\"*.{midi_ext}\"), recursive=True)\n", " audio_files = glob.glob(os.path.join(dir, \"**\", f\"*.{audio_ext}\"), recursive=True)\n", "\n", " # Create a mapping of base filenames to audio files\n", " audio_map = {}\n", " for audio_file in audio_files:\n", " base_name = os.path.splitext(os.path.basename(audio_file))[0]\n", " audio_map[base_name] = audio_file\n", "\n", " pairs = []\n", " for midi_file in midi_files:\n", " base_name = os.path.splitext(os.path.basename(midi_file))[0]\n", " if base_name in audio_map:\n", " pairs.append(MidiPair(midi_file, audio_map[base_name]))\n", "\n", " return pairs\n", "\n", "\n", "scoretube_pairs = load_pairs(scoretube_dir)\n", "print(f\"Loaded {len(scoretube_pairs)} scoretube pairs\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "mmd_dir = \"/app2/suno/data/victor/mew/mmd_chunks\"\n", "mmd_pairs = load_pairs(mmd_dir)\n", "print(f\"Loaded {len(mmd_pairs)} mmd pairs\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## mootube" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import json\n", "\n", "mootube_dir = \"/app2/suno/data/victor/mootube\"\n", "\n", "mootube_metas = []\n", "for line in open(\"/app2/suno/data/victor/mootube/score_ytm_clean.jsonl\"):\n", " meta = json.loads(line)\n", " mootube_metas.append(meta)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "mootube_pairs = []\n", "for line in open(\"/app2/suno/data/victor/mootube/score_ytm_clean.jsonl\"):\n", " meta = json.loads(line)\n", " midi_file = f\"/app2/suno/data/victor/mootube/midi/{meta['id']}.mid\"\n", " audio_file = f\"/app2/suno/data/victor/mootube/audio/{meta['matches'][0]['videoId']}.webm\"\n", " mootube_pairs.append(MidiPair(midi_file, audio_file))\n", "\n", "print(f\"Loaded {len(mootube_pairs)} moo tube pairs\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## maestro\n", "200 hours of piano" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "maestro_dir = \"/app2/suno/data/victor/maestro/maestro-v3.0.0\"\n", "maestro_pairs = load_pairs(maestro_dir, audio_ext=\"wav\", midi_ext=\"midi\")\n", "print(f\"Loaded {len(maestro_pairs)} maestro pairs\")\n", "\n", "# play random pair\n", "random_pair = random.choice(maestro_pairs)\n", "print(random_pair)\n", "# random_pair.play()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## trombone champ" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "trombone_champ_dir = \"/app2/suno/data/victor/trombone_champ\"\n", "trombone_champ_pairs = load_pairs(trombone_champ_dir, audio_ext=\"opus\")\n", "print(f\"Loaded {len(trombone_champ_pairs)} trombone champ pairs\")\n", "\n", "# play random pair\n", "random_pair = random.choice(trombone_champ_pairs)\n", "print(random_pair)\n", "# random_pair.play()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## hook theory" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "hook_theory_dir = \"/app2/suno/data/victor/hooktheory\"\n", "hooktheory_pairs = load_pairs(hook_theory_dir, audio_ext=\"opus\")\n", "print(f\"Loaded {len(hooktheory_pairs)} hook theory pairs\")\n", "\n", "# play random pair\n", "random_pair = random.choice(hooktheory_pairs)\n", "print(random_pair)\n", "# random_pair.play()\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Check stems" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# !du -sh /app2/suno/data/victor/midi_stems/" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# count number of stems in each dataset\n", "pairs_to_process = {\n", " # \"scoretube\": scoretube_pairs,\n", " \"trombone_champ\": trombone_champ_pairs,\n", " # \"hooktheory\": hooktheory_pairs,\n", " # \"mootube\": mootube_pairs,\n", " # \"mmd\": mmd_pairs,\n", "} # ignore maestro\n", "for dataset in pairs_to_process.keys():\n", " # count number of stems in each dataset in the dir\n", " dir = f\"{out_stem_dir}/{dataset}\"\n", " # count number of files in the dir\n", " num_files = len(os.listdir(dir))\n", " print(f\"{dataset}: {num_files} stemmed\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# pair up midis with stems\n", "def process_pair(args):\n", " pair, dataset = args\n", " # find stems\n", " id = pair.audio_file.split(\"/\")[-1].split(\".\")[0]\n", " stem_dir = f\"{out_stem_dir}/{dataset}/{id}\"\n", " if not os.path.exists(stem_dir):\n", " # print(f\"{id} not found\")\n", " return None\n", " # count number of files in the dir\n", " files = os.listdir(stem_dir)\n", " files = [stem_dir + \"/\" + file for file in files]\n", " assert len(files)\n", " return pair, files\n", "\n", "\n", "def pair_up_midis(dataset, pairs):\n", " from concurrent.futures import ThreadPoolExecutor\n", "\n", " with ThreadPoolExecutor(max_workers=20) as executor:\n", " args = [(pair, dataset) for pair in pairs]\n", " results = executor.map(process_pair, args)\n", " for result in results:\n", " if result is not None:\n", " yield result\n", "\n", "\n", "stem_pairs = []\n", "for dataset, pairs in pairs_to_process.items():\n", " print(f\"Processing {dataset}...\")\n", " stem_pairs.extend(list(tqdm(pair_up_midis(dataset, pairs), total=len(pairs))))\n", "\n", "print(f\"Found {len(stem_pairs)} stem pairs\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Count stems and stem types\n", "from collections import defaultdict\n", "import matplotlib.pyplot as plt\n", "\n", "stem_counts = defaultdict(int)\n", "stem_type_counts = defaultdict(int)\n", "\n", "for pair, files in stem_pairs:\n", " stem_counts[\"total_pairs\"] += 1\n", " for file in files:\n", " stem_counts[\"total_stems\"] += 1\n", " # Extract stem type from filename (assuming format like \"vocals.wav\", \"drums.wav\", etc.)\n", " stem_type = file.split(\"/\")[-1].split(\".\")[0]\n", " stem_type_counts[stem_type] += 1\n", "\n", "print(f\"Total pairs with stems: {stem_counts['total_pairs']}\")\n", "print(f\"Total stems: {stem_counts['total_stems']}\")\n", "print(\"\\nStem type distribution:\")\n", "for stem_type, count in sorted(stem_type_counts.items()):\n", " print(f\" {stem_type}: {count}\")\n", "\n", "# Plot stem type distribution\n", "plt.figure(figsize=(12, 6))\n", "sorted_items = sorted(stem_type_counts.items(), key=lambda x: x[1], reverse=True)\n", "stem_types = [item[0] for item in sorted_items]\n", "counts = [item[1] for item in sorted_items]\n", "\n", "plt.bar(stem_types, counts)\n", "plt.title(\"Stem Type Distribution\")\n", "plt.xlabel(\"Stem Type\")\n", "plt.ylabel(\"Count\")\n", "plt.xticks(rotation=45, ha=\"right\")\n", "plt.tight_layout()\n", "plt.show()\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Lets take one pair and align the midi to the stem\n", "\n", "pair, files = stem_pairs[0]\n", "midi = pair.load_midi()\n", "midi.show()\n", "stems = [Audio.from_file(file) for file in files]\n", "for stem in stems:\n", " stem.play()\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Fast DTW routines adapted for Python 3 and notebook use\n", "import numba\n", "import numpy as np\n", "\n", "\n", "@numba.jit(nopython=True)\n", "def band_mask(radius, mask):\n", " \"\"\"Construct band-around-diagonal mask (Sakoe-Chiba band). When\n", " ``mask.shape[0] != mask.shape[1]``, the radius will be expanded so that\n", " ``mask[-1, -1] = 1`` always.\n", "\n", " `mask` will be modified in place.\n", "\n", " Parameters\n", " ----------\n", " radius : float\n", " The band radius (1/2 of the width) will be\n", " ``int(radius*min(mask.shape))``.\n", " mask : np.ndarray\n", " Pre-allocated boolean matrix of zeros.\n", " \"\"\"\n", " nx, ny = mask.shape\n", " # The logic will be different depending on whether there are more rows\n", " # or columns in the mask. Coding it this way results in some code\n", " # duplication but it's the most efficient way with numba\n", " if nx < ny:\n", " # Calculate the radius in indices, rather than proportion\n", " radius = int(round(nx * radius))\n", " # Force radius to be at least one\n", " radius = 1 if radius == 0 else radius\n", " for i in range(nx):\n", " for j in range(ny):\n", " # If this i, j falls within the band\n", " if i - j + (nx - radius) < nx and j - i + (nx - radius) < ny:\n", " # Set the mask to 1 here\n", " mask[i, j] = 1\n", " # Same exact approach with ny/ny and i/j switched.\n", " else:\n", " radius = int(round(ny * radius))\n", " radius = 1 if radius == 0 else radius\n", " for i in range(nx):\n", " for j in range(ny):\n", " if j - i + (ny - radius) < ny and i - j + (ny - radius) < nx:\n", " mask[i, j] = 1\n", "\n", "\n", "@numba.jit(nopython=True)\n", "def dtw_core(dist_mat, add_pen, mul_pen, traceback):\n", " \"\"\"Core dynamic programming routine for DTW.\n", "\n", " `dist_mat` and `traceback` will be modified in-place.\n", "\n", " Parameters\n", " ----------\n", " dist_mat : np.ndarray\n", " Distance matrix to update with lowest-cost path to each entry.\n", " add_pen : int or float\n", " Additive penalty for non-diagonal moves.\n", " mul_pen : int or float\n", " Multiplicative penalty for non-diagonal moves.\n", " traceback : np.ndarray\n", " Matrix to populate with the lowest-cost traceback from each entry.\n", " \"\"\"\n", " # At each loop iteration, we are computing lowest cost to D[i + 1, j + 1]\n", " for i in range(dist_mat.shape[0] - 1):\n", " for j in range(dist_mat.shape[1] - 1):\n", " # Diagonal move (which has no penalty) is lowest\n", " if (\n", " dist_mat[i, j] <= mul_pen * dist_mat[i, j + 1] + add_pen\n", " and dist_mat[i, j] <= mul_pen * dist_mat[i + 1, j] + add_pen\n", " ):\n", " traceback[i + 1, j + 1] = 0\n", " dist_mat[i + 1, j + 1] += dist_mat[i, j]\n", " # Horizontal move (has penalty)\n", " elif (\n", " dist_mat[i, j + 1] <= dist_mat[i + 1, j]\n", " and mul_pen * dist_mat[i, j + 1] + add_pen <= dist_mat[i, j]\n", " ):\n", " traceback[i + 1, j + 1] = 1\n", " dist_mat[i + 1, j + 1] += mul_pen * dist_mat[i, j + 1] + add_pen\n", " # Vertical move (has penalty)\n", " elif (\n", " dist_mat[i + 1, j] <= dist_mat[i, j + 1]\n", " and mul_pen * dist_mat[i + 1, j] + add_pen <= dist_mat[i, j]\n", " ):\n", " traceback[i + 1, j + 1] = 2\n", " dist_mat[i + 1, j + 1] += mul_pen * dist_mat[i + 1, j] + add_pen\n", "\n", "\n", "@numba.jit(nopython=True)\n", "def dtw_core_masked(dist_mat, add_pen, mul_pen, traceback, mask):\n", " \"\"\"Core dynamic programming routine for DTW, with an index mask, so that\n", " the possible paths are constrained.\n", "\n", " `dist_mat` and `traceback` will be modified in-place.\n", "\n", " Parameters\n", " ----------\n", " dist_mat : np.ndarray\n", " Distance matrix to update with lowest-cost path to each entry.\n", " add_pen : int or float\n", " Additive penalty for non-diagonal moves.\n", " mul_pen : int or float\n", " Multiplicative penalty for non-diagonal moves.\n", " traceback : np.ndarray\n", " Matrix to populate with the lowest-cost traceback from each entry.\n", " mask : np.ndarray\n", " A boolean matrix, such that ``mask[i, j] == 1`` when the index ``i, j``\n", " should be allowed in the DTW path and ``mask[i, j] == 0`` otherwise.\n", " \"\"\"\n", " # At each loop iteration, we are computing lowest cost to D[i + 1, j + 1]\n", " for i in range(dist_mat.shape[0] - 1):\n", " for j in range(dist_mat.shape[1] - 1):\n", " # If this point is not reachable, set the cost to infinity\n", " if not mask[i, j] and not mask[i, j + 1] and not mask[i + 1, j]:\n", " dist_mat[i + 1, j + 1] = np.inf\n", " else:\n", " # Diagonal move (which has no penalty) is lowest, or is the\n", " # only valid move\n", " if (dist_mat[i, j] <= mul_pen * dist_mat[i, j + 1] + add_pen or not mask[i, j + 1]) and (\n", " dist_mat[i, j] <= mul_pen * dist_mat[i + 1, j] + add_pen or not mask[i + 1, j]\n", " ):\n", " traceback[i + 1, j + 1] = 0\n", " dist_mat[i + 1, j + 1] += dist_mat[i, j]\n", " # Horizontal move (has penalty)\n", " elif (dist_mat[i, j + 1] <= dist_mat[i + 1, j] or not mask[i + 1, j]) and (\n", " mul_pen * dist_mat[i, j + 1] + add_pen <= dist_mat[i, j] or not mask[i, j]\n", " ):\n", " traceback[i + 1, j + 1] = 1\n", " dist_mat[i + 1, j + 1] += mul_pen * dist_mat[i, j + 1] + add_pen\n", " # Vertical move (has penalty)\n", " elif (dist_mat[i + 1, j] <= dist_mat[i, j + 1] or not mask[i, j + 1]) and (\n", " mul_pen * dist_mat[i + 1, j] + add_pen <= dist_mat[i, j] or not mask[i, j]\n", " ):\n", " traceback[i + 1, j + 1] = 2\n", " dist_mat[i + 1, j + 1] += mul_pen * dist_mat[i + 1, j] + add_pen\n", "\n", "\n", "def dtw(\n", " distance_matrix, gully=1.0, additive_penalty=0.0, multiplicative_penalty=1.0, mask=None, inplace=True\n", "):\n", " \"\"\"Compute the dynamic time warping distance between two sequences given a\n", " distance matrix. The score is unnormalized.\n", "\n", " Parameters\n", " ----------\n", " distance_matrix : np.ndarray\n", " Distances between two sequences.\n", " gully : float\n", " Sequences must match up to this porportion of shorter sequence. Default\n", " 1., which means the entirety of the shorter sequence must be matched\n", " to part of the longer sequence.\n", " additive_penalty : int or float\n", " Additive penalty for non-diagonal moves. Default 0. means no penalty.\n", " multiplicative_penalty : int or float\n", " Multiplicative penalty for non-diagonal moves. Default 1. means no\n", " penalty.\n", " mask : np.ndarray\n", " A boolean matrix, such that ``mask[i, j] == 1`` when the index ``i, j``\n", " should be allowed in the DTW path and ``mask[i, j] == 0`` otherwise.\n", " If None (default), don't apply a mask - this is more efficient than\n", " providing a mask of all 1s.\n", " inplace : bool\n", " When ``inplace == True`` (default), `distance_matrix` will be modified\n", " in-place when computing path costs. When ``inplace == False``,\n", " `distance_matrix` will not be modified.\n", "\n", " Returns\n", " -------\n", " x_indices : np.ndarray\n", " Indices of the lowest-cost path in the first dimension of the distance\n", " matrix.\n", " y_indices : np.ndarray\n", " Indices of the lowest-cost path in the second dimension of the distance\n", " matrix.\n", " score : float\n", " DTW score of lowest cost path through the distance matrix, including\n", " penalties.\n", " \"\"\"\n", " if np.isnan(distance_matrix).any():\n", " raise ValueError(\"NaN values found in distance matrix.\")\n", " if not inplace:\n", " distance_matrix = distance_matrix.copy()\n", " # Pre-allocate path length matrix\n", " traceback = np.empty(distance_matrix.shape, np.uint8)\n", " # Don't use masked DTW routine if no mask was provided\n", " if mask is None:\n", " # Populate distance matrix with lowest cost path\n", " dtw_core(distance_matrix, additive_penalty, multiplicative_penalty, traceback)\n", " else:\n", " dtw_core_masked(distance_matrix, additive_penalty, multiplicative_penalty, traceback, mask)\n", " if gully < 1.0:\n", " # Allow the end of the path to start within gully percentage of the\n", " # smaller distance matrix dimension\n", " gully = int(gully * min(distance_matrix.shape))\n", " else:\n", " # When gully is 1 require matching the entirety of the smaller sequence\n", " gully = min(distance_matrix.shape) - 1\n", "\n", " # Find the indices of the smallest costs on the bottom and right edges\n", " i = np.argmin(distance_matrix[gully:, -1]) + gully\n", " j = np.argmin(distance_matrix[-1, gully:]) + gully\n", "\n", " # Choose the smaller cost on the two edges\n", " if distance_matrix[-1, j] > distance_matrix[i, -1]:\n", " j = distance_matrix.shape[1] - 1\n", " else:\n", " i = distance_matrix.shape[0] - 1\n", "\n", " # Score is the final score of the best path\n", " score = float(distance_matrix[i, j])\n", "\n", " # Pre-allocate the x and y path index arrays\n", " x_indices = np.zeros(sum(traceback.shape), dtype=np.int32)\n", " y_indices = np.zeros(sum(traceback.shape), dtype=np.int32)\n", " # Start the arrays from the end of the path\n", " x_indices[0] = i\n", " y_indices[0] = j\n", " # Keep track of path length\n", " n = 1\n", "\n", " # Until we reach an edge\n", " while i > 0 and j > 0:\n", " # If the tracback matrix indicates a diagonal move...\n", " if traceback[i, j] == 0:\n", " i = i - 1\n", " j = j - 1\n", " # Horizontal move...\n", " elif traceback[i, j] == 1:\n", " i = i - 1\n", " # Vertical move...\n", " elif traceback[i, j] == 2:\n", " j = j - 1\n", " # Add these indices into the path arrays\n", " x_indices[n] = i\n", " y_indices[n] = j\n", " n += 1\n", " # Reverse and crop the path index arrays\n", " x_indices = x_indices[:n][::-1]\n", " y_indices = y_indices[:n][::-1]\n", "\n", " return x_indices, y_indices, score\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import librosa\n", "import numpy as np\n", "import scipy.spatial.distance\n", "import copy\n", "# from dtw import dtw\n", "\n", "\n", "# Audio/CQT parameters\n", "FS = 22050\n", "NOTE_START = 36\n", "N_NOTES = 48\n", "HOP_LENGTH = 1024\n", "# DTW parameters\n", "GULLY = 0.96\n", "\n", "\n", "def plot_cqt(cqt: np.ndarray):\n", " \"\"\"Plot the CQT\"\"\"\n", " import matplotlib.pyplot as plt\n", "\n", " plt.figure(figsize=(12, 6))\n", " plt.imshow(cqt.T, aspect=\"auto\", origin=\"lower\", interpolation=\"nearest\")\n", " plt.colorbar(label=\"Magnitude (dB)\")\n", " plt.xlabel(\"Time (frames)\")\n", " plt.ylabel(\"MIDI Note\")\n", " plt.title(\"Constant-Q Transform\")\n", " plt.tight_layout()\n", " plt.show()\n", "\n", "\n", "def compute_cqt(audio: Audio):\n", " \"\"\"Compute the CQT and frame times for some audio data\"\"\"\n", " audio_data = audio.resample(FS).array_float\n", " # Compute CQT\n", " cqt = librosa.cqt(\n", " audio_data,\n", " sr=FS,\n", " fmin=librosa.midi_to_hz(NOTE_START),\n", " n_bins=N_NOTES,\n", " hop_length=HOP_LENGTH,\n", " tuning=0.0,\n", " )\n", " # Compute the time of each frame\n", " times = librosa.frames_to_time(np.arange(cqt.shape[1]), sr=FS, hop_length=HOP_LENGTH)\n", " # Compute log-amplitude\n", " cqt = librosa.amplitude_to_db(np.abs(cqt), ref=np.abs(cqt).max())\n", " # Plot the CQT\n", " # Normalize and return\n", " return librosa.util.normalize(cqt, axis=0).T, times\n", "\n", "\n", "def align_midi_to_audio(midi: Midi, audio: Audio, verbose: bool = False):\n", " \"\"\"Align MIDI to audio using DTW on CQT features\"\"\"\n", " # Compute the log-magnitude CQT of the audio data\n", " audio_cqt, audio_times = compute_cqt(audio)\n", "\n", " # Synthesize MIDI data\n", " midi_audio = midi.to_audio(sample_rate=FS)\n", "\n", " # Compute log-magnitude CQT of MIDI\n", " midi_cqt, midi_times = compute_cqt(midi_audio)\n", "\n", " # the midi might be transposed, find the best pitch shift\n", " best_score = float(\"inf\")\n", " best_pitch_shift = 0\n", " for pitch_shift in range(-12, 12 + 1, 12):\n", " midi_cqt_shifted = np.roll(midi_cqt, pitch_shift, axis=1)\n", " distance_matrix = scipy.spatial.distance.cdist(midi_cqt_shifted, audio_cqt, \"cosine\")\n", " p, q, score = dtw(\n", " distance_matrix,\n", " GULLY,\n", " np.median(distance_matrix),\n", " inplace=False,\n", " )\n", " score = score / len(p)\n", " score = score / distance_matrix[p.min() : p.max(), q.min() : q.max()].mean()\n", " # print(pitch_shift, score)\n", " if score < best_score:\n", " best_score = score\n", " best_pitch_shift = pitch_shift\n", "\n", " # Plot the CQT\n", " if verbose:\n", " # print(midi_cqt.shape, audio_cqt.shape)\n", " # midi_audio.play()\n", " plot_cqt(midi_cqt)\n", " # audio.play()\n", " plot_cqt(audio_cqt)\n", "\n", " midi = copy.deepcopy(midi)\n", " midi.pmidi.adjust_times(midi_times[p], audio_times[q])\n", " midi = midi.transpose(best_pitch_shift)\n", "\n", " # when the audio is silent, remove the notes from the midi\n", " # find silent frames\n", " silent_frames = np.max(audio_cqt, axis=1) < -0.8\n", " print(f\"Silent frames: {silent_frames.shape}, {silent_frames.mean()}\")\n", "\n", " def note_is_silent(note):\n", " start, end = note.start, note.end\n", " start_frame = int(start * FS / HOP_LENGTH)\n", " end_frame = int(end * FS / HOP_LENGTH)\n", " frames = silent_frames[start_frame:end_frame]\n", " if len(frames) == 0:\n", " return True\n", " return frames.mean() > 0.7\n", "\n", " # remove notes that overlap with >70% silence\n", " # for instrument in midi.pmidi.instruments:\n", " # instrument.notes = [note for note in instrument.notes if not note_is_silent(note)]\n", "\n", " return best_pitch_shift, best_score, midi\n", "\n", "\n", "# alignment = align_midi_to_audio(midi, midi.to_audio(sample_rate=FS))\n", "for stem in stems[0:2]:\n", " pitch_shift, score, aligned_midi = align_midi_to_audio(midi, stem)\n", " print(pitch_shift, score)\n", " # aligned_midi.show()\n", " aligned_midi.make_stereo_comparison(stem).play()" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def align_midi_to_best_stem(midi: Midi, stems: list[Audio], verbose: bool = False):\n", " results = [] # (pitch_shift, score, midi)\n", " for stem in tqdm(stems, disable=not verbose):\n", " results.append(align_midi_to_audio(midi, stem))\n", " argmin = np.argmin([score for _, score, _ in results])\n", " pitch_shift, score, aligned_midi = results[argmin]\n", " return pitch_shift, score, aligned_midi, stems[argmin]\n", "\n", "\n", "pair, files = random.choice(stem_pairs)\n", "midi = pair.load_midi()\n", "stems = [Audio.from_file(file) for file in files]\n", "full_audio = Audio.sum(stems)\n", "# full_audio.play()\n", "midi.make_stereo_comparison(full_audio).play()\n", "# pitch_shift, score, aligned_midi, best_stem = align_midi_to_best_stem(midi, stems)\n", "pitch_shift, score, aligned_midi, best_stem = align_midi_to_best_stem(midi, [full_audio])\n", "print(pitch_shift, score)\n", "aligned_midi.make_stereo_comparison(best_stem).play()\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 2 }