03_audio_audio_measureTransfer.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "e89a6ea3-3027-4513-bb6f-2dfc1d03b2b5",
   "metadata": {},
   "source": [
    "# BPSD: Audio-Audio Measure Transfer\n",
    "\n",
    "- Transfer manual measure annotations from Kempff recordings (WK64) to all other audio versions using a high-resolution approach from [1], MrMsDTW from [2], and the Sync Toolbox implementation from [3]\n",
    "- requirements: Sync Toolbox code (https://github.com/meinardmueller/synctoolbox)\n",
    "\n",
    "[1] Sebastian Ewert, Meinard Müller, and Peter Grosche. \"High resolution audio synchronization using chroma onset features.\" 2009 IEEE International Conference on Acoustics, Speech and Signal Processing. IEEE, 2009.  \n",
    "\n",
    "[2] Thomas Prätzlich, Jonathan Driedger, and Meinard Müller. \"Memory-restricted multiscale dynamic time warping.\" 2016 IEEE International Conference on Acoustics, Speech and Signal Processing (ICASSP). IEEE, 2016.\n",
    "\n",
    "[3] Meinard Müller et al. \"Sync Toolbox: A Python package for efficient, robust, and accurate music synchronization.\" Journal of Open Source Software 6.64 (2021): 3434.\n",
    "\n",
    "Johannes Zeitler (johannes.zeitler@audiolabs-erlangen.de), 2024"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "4f5c37a3-f191-42d0-8359-9b6f9a11d581",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import sys\n",
    "import os\n",
    "\n",
    "\n",
    "sys.path.append(\"path_to_synctoolbox\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "b8b23583-6271-4f44-83df-d7a12e576285",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import libtsm"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "c2ec9a98-a70c-44b5-8eeb-3707996e09d3",
   "metadata": {
    "pycharm": {
     "name": "#%%\n"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Loading some modules and defining some constants used later\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import librosa.display\n",
    "import matplotlib.pyplot as plt\n",
    "import IPython.display as ipd\n",
    "import scipy.interpolate\n",
    "from libfmp.b.b_plot import plot_signal, plot_chromagram\n",
    "from libfmp.c3.c3s2_dtw_plot import plot_matrix_with_points\n",
    "import libfmp.c2\n",
    "from tqdm.notebook import tqdm\n",
    "\n",
    "from synctoolbox.dtw.mrmsdtw import sync_via_mrmsdtw, sync_via_mrmsdtw_with_anchors, __split_features, __diagonal_warping_path\n",
    "from synctoolbox.dtw.utils import compute_optimal_chroma_shift, shift_chroma_vectors, make_path_strictly_monotonic, evaluate_synchronized_positions\n",
    "from synctoolbox.feature.chroma import pitch_to_chroma, quantize_chroma, quantized_chroma_to_CENS\n",
    "from synctoolbox.feature.dlnco import pitch_onset_features_to_DLNCO\n",
    "from synctoolbox.feature.pitch import audio_to_pitch_features\n",
    "from synctoolbox.feature.pitch_onset import audio_to_pitch_onset_features\n",
    "from synctoolbox.feature.utils import estimate_tuning\n",
    "\n",
    "from synctoolbox.feature.csv_tools import read_csv_to_df, df_to_pitch_features, df_to_pitch_onset_features\n",
    "\n",
    "\n",
    "from typing import List, Tuple, Optional\n",
    "\n",
    "%matplotlib inline\n",
    "\n",
    "Fs = 22050\n",
    "feature_rate = 50\n",
    "step_weights = np.array([1.5, 1.5, 2.0])\n",
    "threshold_rec = 10 ** 6\n",
    "\n",
    "figsize = (9, 3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "a2486c6b-db1d-4abc-a6f8-6db62885cbc4",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def read_csv_to_df(csv_filepath: str = '',\n",
    "                   csv_delimiter: str = ';') -> pd.DataFrame:\n",
    "    \"\"\"Reads .csv file containing symbolic music into a pandas DataFrame.\n",
    "    Column names are normalized to be lower case.\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    csv_filepath : str\n",
    "        Filepath to the .csv file.\n",
    "\n",
    "    csv_delimiter : str\n",
    "        Delimiter of the .csv file (default: ';')\n",
    "\n",
    "    Returns\n",
    "    -------\n",
    "    df : pd.Dataframe\n",
    "        Annotations in pandas Dataframe format.\n",
    "    \"\"\"\n",
    "\n",
    "    df = pd.read_csv(filepath_or_buffer=csv_filepath,\n",
    "                     delimiter=csv_delimiter)#, dtype=\"str\")#, index_col=0)\n",
    "    df.columns = df.columns.str.lower()\n",
    "\n",
    "    if 'pitch' in df.columns:\n",
    "        df['pitch'] = df['pitch'].astype(int)\n",
    "\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "c8079043-0614-4f96-ad91-495feb4fec0d",
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_features_from_annotation(df_annotation, feature_rate, visualize=True):\n",
    "    if not \"velocity\" in df_annotation.keys():\n",
    "        df_annotation[\"velocity\"] = 64\n",
    "\n",
    "    f_pitch = df_to_pitch_features(df_annotation, feature_rate=feature_rate)\n",
    "    f_chroma = pitch_to_chroma(f_pitch=f_pitch)\n",
    "    f_chroma_quantized = quantize_chroma(f_chroma=f_chroma)\n",
    "    if visualize:\n",
    "        plot_chromagram(f_chroma_quantized, title='Quantized chroma features - Annotation', Fs=feature_rate, figsize=(9, 3))\n",
    "    f_pitch_onset = df_to_pitch_onset_features(df_annotation)\n",
    "    f_DLNCO = pitch_onset_features_to_DLNCO(f_peaks=f_pitch_onset,\n",
    "                                            feature_rate=feature_rate,\n",
    "                                            feature_sequence_length=f_chroma_quantized.shape[1],\n",
    "                                            visualize=visualize)\n",
    "    \n",
    "    return f_chroma_quantized, f_DLNCO"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "fe03c89f-08e8-4e8f-b2ba-366786c4b5fa",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def iter_mods_deltaT(mods_list):\n",
    "    deltaT = 0\n",
    "    for _, mod_row in mods_list.iterrows():\n",
    "        yield deltaT, mod_row\n",
    "\n",
    "        mod = mod_row.modification\n",
    "\n",
    "        if mod == \"cut\":\n",
    "            cutStart = float(mod_row.start)\n",
    "            cutEnd = float(mod_row.end)\n",
    "            deltaT -= (cutEnd - cutStart)\n",
    "            \n",
    "        if mod == \"copy\":\n",
    "            copyStart = float(mod_row.start)\n",
    "            copyEnd = float(mod_row.end)\n",
    "            deltaT += (copyEnd - copyStart)\n",
    "        \n",
    "        if mod == \"silence\":\n",
    "            silence_duration = float(mod_row.silence_duration)\n",
    "            deltaT += silence_duration\n",
    "            \n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "ae365f54-0217-4cf1-aea0-7a923a6a08da",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def fill_added_silence(f_chroma, f_onsets, sonata, performer):\n",
    "    print(\"%s_%s.csv\"%(sonata, performer))\n",
    "    \n",
    "    is_silence=False\n",
    "\n",
    "    if \"%s_%s.csv\"%(sonata, performer) in os.listdir(ann_mod_dir):\n",
    "        print(\"found modification\")\n",
    "\n",
    "        mod_in = pd.read_csv(os.path.join(ann_mod_dir, \"%s_%s.csv\"%(sonata, performer)), sep=\";\")\n",
    "\n",
    "        if \"silence\" in list(mod_in.modification):\n",
    "            print(\"found silence\")\n",
    "            is_silence=True\n",
    "            \n",
    "\n",
    "            df_annotation = read_csv_to_df(os.path.join(score_csv_dir, sonata+\".csv\"), csv_delimiter=';')\n",
    "            quarterNoteOffset=[]\n",
    "            beat_fac = 1\n",
    "            curr_meas=0.\n",
    "            meas_offset=0\n",
    "\n",
    "            measuresIn = read_csv_to_df(os.path.join(\"../\", \"2_Annotations\", measures_dir, \"%s_%s.csv\"%(sonata, performer)), csv_delimiter=\";\")\n",
    "            measuresIn.sort_values(by=\"measure\", inplace=True)\n",
    "\n",
    "\n",
    "\n",
    "            measuresIn = pd.concat([pd.DataFrame({\"time\":[0, f_chroma.shape[1]/feature_rate], \"measure\":[0, int(max(measuresIn.measure))+1]}), measuresIn])\n",
    "\n",
    "            measuresIn.sort_values(by=\"measure\", inplace=True)\n",
    "\n",
    "            meas_start_time = []        \n",
    "            beat_fac_list = []\n",
    "            quarterNoteOffset_in_measure = []\n",
    "\n",
    "            for i, row in df_annotation.iterrows():           \n",
    "\n",
    "                # if new measure starts\n",
    "                if np.floor(row.start_meas) > curr_meas:\n",
    "                    curr_meas = np.floor(row.start_meas)\n",
    "                    meas_offset += beat_fac*4\n",
    "\n",
    "                meas_start_time.append(measuresIn.time[measuresIn.measure==curr_meas].item())\n",
    "\n",
    "                beat_fac = int(row.timesig.split(\"/\")[0])/int(row.timesig.split(\"/\")[1])\n",
    "                quarterNoteOffset.append( meas_offset+(row.start_meas-curr_meas)*beat_fac*4)\n",
    "                quarterNoteOffset_in_measure.append((row.start_meas-curr_meas)*beat_fac*4)\n",
    "\n",
    "                beat_fac_list.append(beat_fac)\n",
    "\n",
    "            df_annotation[\"quarternoteoffset\"] = quarterNoteOffset\n",
    "            df_annotation[\"quarternoteoffset\"] -= min(df_annotation[\"quarternoteoffset\"])\n",
    "\n",
    "            df_annotation[\"measstarttime\"] = meas_start_time\n",
    "            df_annotation[\"quarternoteoffsetinmeasure\"] = quarterNoteOffset_in_measure\n",
    "            #########################################################################\n",
    "\n",
    "            measure_duration = []\n",
    "            for _, row in df_annotation.iterrows():\n",
    "                rowMeas = int(row.start_meas)\n",
    "                measure_duration.append( measuresIn.time[measuresIn.measure==(rowMeas+1)].item() - measuresIn.time[measuresIn.measure==rowMeas].item())\n",
    "\n",
    "            df_annotation[\"beatfac\"] = beat_fac_list\n",
    "            df_annotation[\"secondspermeasure\"] = measure_duration\n",
    "            df_annotation[\"secondsperbeat\"] = df_annotation[\"secondspermeasure\"] / 4 / df_annotation[\"beatfac\"]\n",
    "\n",
    "            df_annotation[\"start\"] = df_annotation[\"measstarttime\"] + df_annotation[\"quarternoteoffsetinmeasure\"]*df_annotation[\"secondsperbeat\"]#df_annotation[\"offset\"]*secondsperquarter#df_annotation[\"secondsperquarter\"]  \n",
    "            df_annotation[\"duration\"] = df_annotation[\"duration_quarterlength\"]*df_annotation[\"secondsperbeat\"] - 0.01 #df_annotation[\"secondsperquarter\"]\n",
    "\n",
    "            df_annotation.duration[df_annotation.articulation == \"staccato\"] /= 2  #df_annotation.duration[df_annotation.articulation == \"staccato\"]/2\n",
    "            df_annotation[\"end\"] = df_annotation[\"start\"] + df_annotation[\"duration\"]\n",
    "            df_annotation[\"instrument\"] = [\"piano\" for _ in df_annotation.iterrows()]\n",
    "\n",
    "            f_chroma_quantized_annotation, f_DLNCO_annotation = get_features_from_annotation(df_annotation, feature_rate, visualize=False)\n",
    "\n",
    "\n",
    "            for deltaT, row in iter_mods_deltaT(mod_in):\n",
    "                if row.modification == \"silence\":\n",
    "\n",
    "                    iStart = int((row.insert_point + deltaT) * feature_rate)\n",
    "                    iEnd = int((row.insert_point + deltaT + row.silence_duration) * feature_rate)\n",
    "                    \n",
    "                    print(\"inserting score annotation from %.2fs - %.2fs (samples %i - %i)\"%(row.insert_point+deltaT, row.insert_point+deltaT+row.silence_duration, iStart, iEnd))\n",
    "\n",
    "                    fig, ax = plt.subplots(2,2, figsize=(10,6))\n",
    "                    ax[0,0].imshow(f_chroma[:,iStart:iEnd], aspect='auto', origin='lower', cmap='gray_r', interpolation=\"None\")\n",
    "                    ax[0,1].imshow(f_onsets[:,iStart:iEnd], aspect='auto', origin='lower', cmap='gray_r', interpolation=\"None\")\n",
    "                    \n",
    "                    ax[1,0].imshow(f_chroma_quantized_annotation[:,iStart:iEnd], aspect='auto', origin='lower', cmap='gray_r', interpolation=\"None\")\n",
    "                    ax[1,1].imshow(f_DLNCO_annotation[:,iStart:iEnd], aspect='auto', origin='lower', cmap='gray_r', interpolation=\"None\")\n",
    "                    \n",
    "                    ax[0,0].set_title(\"audio chroma\")\n",
    "                    ax[0,1].set_title(\"audio onsets\")\n",
    "                    ax[1,0].set_title(\"score chroma\")\n",
    "                    ax[1,0].set_title(\"score onsets\")\n",
    "                    plt.show()\n",
    "                    \n",
    "                    f_chroma[:,iStart:iEnd] = f_chroma_quantized_annotation[:,iStart:iEnd]\n",
    "                    f_onsets[:,iStart:iEnd] = f_DLNCO_annotation[:,iStart:iEnd]\n",
    "\n",
    "    return f_chroma, f_onsets, is_silence\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "61843c6d-84a9-4d04-97fb-413fe5d23dab",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "measures_dir = os.path.join(\"../\", \"2_Annotations\", \"ann_audio_measure\")\n",
    "audio_dir = os.path.join(\"../\", \"1_Audio\")\n",
    "\n",
    "all_sonatas = list(set([\"_\".join(f.split(\"_\")[:-1]) for f in os.listdir(audio_dir) if \".wav\" in f]))\n",
    "all_sonatas.sort()\n",
    "\n",
    "all_performers = list(set([(f.split(\".\")[0].split(\"_\")[-1]) for f in os.listdir(audio_dir) if \".wav\" in f]))\n",
    "all_performers.sort()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "7ccbcec1-d51e-449d-8de3-b2a27e400702",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "startEnd_dir = os.path.join(\"../\", \"2_Annotations\", \"ann_audio_startEnd\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "347cb768-3a1b-44c3-a132-0e28176d4adf",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Transcription features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "6bfadac2-c6a6-4acd-b4de-1715280ad38f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def pitch_onset_matrix_to_DLNCO(pitch_onset_matrix : np.ndarray,\n",
    "                                  feature_sequence_length: int,\n",
    "                                  feature_rate: int = 50,\n",
    "                                  midi_min: int = 21,\n",
    "                                  midi_max: int = 108,\n",
    "                                  log_compression_gamma: float = 10000.0,\n",
    "                                  chroma_norm_ord: int = 2,\n",
    "                                  LN_maxfilterlength_seconds: float = 0.8,\n",
    "                                  LN_maxfilterthresh: float = 0.1,\n",
    "                                  DLNCO_filtercoef: np.ndarray = np.sqrt(1 / np.arange(1, 11)),\n",
    "                                  visualize=False) -> np.ndarray:\n",
    "    \"\"\"Computes decaying locally adaptive normalized chroma onset (DLNCO) features from\n",
    "    a dictionary of peaks obtained e.g. by ``audio_to_pitch_onset_features``.\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    f_peaks : dict\n",
    "        A dictionary of onset peaks\n",
    "\n",
    "            * Each key corresponds to the midi pitch number\n",
    "\n",
    "            * Each value f_peaks[midi_pitch] is an array of doubles of size 2xN:\n",
    "\n",
    "                + First row give the positions of the peaks in milliseconds.\n",
    "\n",
    "                + Second row contains the corresponding magnitudes of the peaks.\n",
    "\n",
    "    feature_sequence_length : int\n",
    "        Desired length of the resulting feature sequence. This should be at least as long as the\n",
    "        position of the last peak in ``f_peaks``, but can be longer.\n",
    "\n",
    "    feature_rate : int\n",
    "        Desired features per second in the output representation\n",
    "\n",
    "    midi_min : int\n",
    "        Minimum MIDI pitch index (default: 21)\n",
    "\n",
    "    midi_max : int\n",
    "        Maximum MIDI pitch index (default: 108)\n",
    "\n",
    "    log_compression_gamma : float\n",
    "        Gamma factor of the log compression applied to peak magnitudes.\n",
    "        \n",
    "    chroma_norm_ord : int\n",
    "        Order of the norm used for chroma onset vectors.\n",
    "\n",
    "    LN_maxfilterlength_seconds : float\n",
    "        Length of the maximum filter applied for determining local norm of chroma onsets in seconds.\n",
    "\n",
    "    LN_maxfilterthresh : float\n",
    "        Minimum threshold for normalizing chroma onsets using local norm.\n",
    "\n",
    "    DLNCO_filtercoef : np.ndarray\n",
    "        Sequence of decay coefficients applied on normalized chroma onsets.\n",
    "\n",
    "    visualize : bool\n",
    "        Set `True` to visualize chroma onset features (Default: False)\n",
    "\n",
    "    Returns\n",
    "    -------\n",
    "    f_DLNCO : np.array [shape=(d_dlnco, N_dlnco)]\n",
    "        Decaying Locally adaptively Normalized Chroma Onset features\n",
    "    \"\"\"\n",
    "    \n",
    "    f_CO = pitch_to_chroma(pitch_onset_matrix).T\n",
    "\n",
    "    # No two ways to normalize F_CO: simply columnwise (f_N) or via local\n",
    "    # normalizing curve (f_LN)\n",
    "    f_N = np.zeros(feature_sequence_length)\n",
    "\n",
    "    for k in range(feature_sequence_length):\n",
    "        f_N[k] = np.linalg.norm(f_CO[k, :], chroma_norm_ord)\n",
    "\n",
    "    f_LN = np.array(f_N, copy=True)\n",
    "    f_left = np.array(f_N, copy=True)\n",
    "    f_right = np.array(f_N, copy=True)\n",
    "    LN_maxfilterlength_frames = int(LN_maxfilterlength_seconds * feature_rate)\n",
    "    if LN_maxfilterlength_frames % 2 == 1:\n",
    "        LN_maxfilterlength_frames -= 1\n",
    "    shift = int(np.floor((LN_maxfilterlength_frames) / 2))\n",
    "\n",
    "    # TODO improve with scipy.ndimage.maximum_filter\n",
    "    for s in range(shift):\n",
    "        f_left = np.roll(f_left, 1, axis=0)\n",
    "        f_left[0] = 0\n",
    "        f_right = np.roll(f_right, -1, axis=0)\n",
    "        f_right[-1] = 0\n",
    "        f_LN = np.max([f_left, f_LN, f_right], axis=0)\n",
    "\n",
    "    f_LN = np.maximum(f_LN, LN_maxfilterthresh)\n",
    "\n",
    "    # Compute f_NC0 (normalizing f_C0 using f_N)\n",
    "    # f_NCO = np.zeros((feature_sequence_length, 12))\n",
    "\n",
    "    # Compute f_LNC0 (normalizing f_C0 using f_LN)\n",
    "    f_LNCO = np.zeros((feature_sequence_length, 12))\n",
    "    for k in range(feature_sequence_length):\n",
    "        # f_NCO[k, :] = f_CO[k, :] / (f_N[k]) #+ eps)\n",
    "        f_LNCO[k, :] = f_CO[k, :] / f_LN[k]\n",
    "\n",
    "    # Compute f_DLNCO\n",
    "    f_DLNCO = np.zeros((feature_sequence_length, 12))\n",
    "\n",
    "    num_coef = DLNCO_filtercoef.size\n",
    "    for p_idx in range(12):\n",
    "        v_shift = np.array(f_LNCO[:, p_idx], copy=True)\n",
    "        v_help = np.zeros((feature_sequence_length, num_coef))\n",
    "\n",
    "        for n in range(num_coef):\n",
    "            v_help[:, n] = DLNCO_filtercoef[n] * v_shift\n",
    "            v_shift = np.roll(v_shift, 1)\n",
    "            v_shift[0] = 0\n",
    "\n",
    "        f_DLNCO[:, p_idx] = np.max(v_help, axis=1)\n",
    "\n",
    "    # visualization\n",
    "    if visualize:\n",
    "        plot_chromagram(X=f_CO.T, title='CO', colorbar=True, Fs=feature_rate, colorbar_aspect=50, figsize=(9, 3))\n",
    "        __visualize_LN_features(f_N, f_LN, feature_sequence_length, feature_rate)\n",
    "        plot_chromagram(X=f_LNCO.T, title='LNCO', colorbar=True, Fs=feature_rate, colorbar_aspect=50, figsize=(9, 3))\n",
    "        plot_chromagram(X=f_DLNCO.T, title='DLNCO', colorbar=True, Fs=feature_rate, colorbar_aspect=50, figsize=(9, 3))\n",
    "\n",
    "    f_DLNCO = f_DLNCO.T\n",
    "\n",
    "    return f_DLNCO"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "a47fa880-88aa-439e-93bf-5439e10bdba6",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def transcriptionFeatures_to_chroma_and_DLNCO(onsetsIn, framesIn, frame_rate_in, feature_rate, audio_len, Fs):\n",
    "    idx_interp = np.floor(np.arange(0, audio_len/Fs, 1/feature_rate)*frame_rate_in).astype(int)\n",
    "\n",
    "    onsetsIn_res = onsetsIn[:,idx_interp]\n",
    "    framesIn_res = framesIn[:,idx_interp]\n",
    "\n",
    "    f_frames_transcription = np.zeros((128, framesIn_res.shape[1]))\n",
    "    f_frames_transcription[21:109,:] = framesIn_res\n",
    "\n",
    "    f_onsets_transcription = np.zeros((128, onsetsIn_res.shape[1]))\n",
    "    f_onsets_transcription[21:109,:] = onsetsIn_res\n",
    "\n",
    "\n",
    "    f_chroma_transcriptions = pitch_to_chroma(f_pitch=f_frames_transcription)\n",
    "    f_chroma_quantized_transcription = quantize_chroma(f_chroma = f_chroma_transcriptions)\n",
    "\n",
    "    #plot_chromagram(f_chroma_quantized_transcription, title='Quantized chroma features - Transcription', Fs=feature_rate, figsize=(9,3))\n",
    "\n",
    "    f_DLNCO_transcription = pitch_onset_matrix_to_DLNCO(pitch_onset_matrix = f_onsets_transcription, feature_rate=feature_rate, \n",
    "                                                          feature_sequence_length=f_chroma_quantized_transcription.shape[1], visualize=False)\n",
    "\n",
    "    return f_chroma_quantized_transcription, f_DLNCO_transcription"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "948aca2c-5123-48b3-b69b-c33f17c97e84",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def sync_via_mrmsdtw_with_anchors(f_chroma1: np.ndarray,\n",
    "                                  f_chroma2: np.ndarray,\n",
    "                                  f_onset1: np.ndarray = None,\n",
    "                                  f_onset2: np.ndarray = None,\n",
    "                                  input_feature_rate: int = 50,\n",
    "                                  step_sizes: np.ndarray = np.array([[1, 0], [0, 1], [1, 1]], np.int32),\n",
    "                                  step_weights: np.ndarray = np.array([1.0, 1.0, 1.0], np.float64),\n",
    "                                  threshold_rec: int = 10000,\n",
    "                                  win_len_smooth: np.ndarray = np.array([201, 101, 21, 1]),\n",
    "                                  downsamp_smooth: np.ndarray = np.array([50, 25, 5, 1]),\n",
    "                                  verbose: bool = False,\n",
    "                                  dtw_implementation: str = 'synctoolbox',\n",
    "                                  normalize_chroma: bool = True,\n",
    "                                  chroma_norm_ord: int = 2,\n",
    "                                  chroma_norm_threshold: float = 0.001,\n",
    "                                  visualization_title: str = \"MrMsDTW result\",\n",
    "                                  anchor_pairs: List[Tuple] = None,\n",
    "                                  linear_inp_idx: List[int] = [],\n",
    "                                  alpha=0.5) -> np.ndarray:\n",
    "    \"\"\"Compute memory-restricted multi-scale DTW (MrMsDTW) using chroma and (optionally) onset features.\n",
    "        MrMsDTW is performed on multiple levels that get progressively finer, with rectangular constraint\n",
    "        regions defined by the alignment found on the previous, coarser level.\n",
    "        If onset features are provided, these are used on the finest level in addition to chroma\n",
    "        to provide higher synchronization accuracy.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        f_chroma1 : np.ndarray [shape=(12, N)]\n",
    "            Chroma feature matrix of the first sequence\n",
    "\n",
    "        f_chroma2 : np.ndarray [shape=(12, M)]\n",
    "            Chroma feature matrix of the second sequence\n",
    "\n",
    "        f_onset1 : np.ndarray [shape=(L, N)]\n",
    "            Onset feature matrix of the first sequence (optional, default: None)\n",
    "\n",
    "        f_onset2 : np.ndarray [shape=(L, M)]\n",
    "            Onset feature matrix of the second sequence (optional, default: None)\n",
    "\n",
    "        input_feature_rate: int\n",
    "            Input feature rate of the chroma features (default: 50)\n",
    "\n",
    "        step_sizes: np.ndarray\n",
    "            DTW step sizes (default: np.array([[1, 0], [0, 1], [1, 1]]))\n",
    "\n",
    "        step_weights: np.ndarray\n",
    "            DTW step weights (np.array([1.0, 1.0, 1.0]))\n",
    "\n",
    "        threshold_rec: int\n",
    "            Defines the maximum area that is spanned by the rectangle of two\n",
    "            consecutive elements in the alignment (default: 10000)\n",
    "\n",
    "        win_len_smooth : np.ndarray\n",
    "            Window lengths for chroma feature smoothing (default: np.array([201, 101, 21, 1]))\n",
    "\n",
    "        downsamp_smooth : np.ndarray\n",
    "            Downsampling factors (default: np.array([50, 25, 5, 1]))\n",
    "\n",
    "        verbose : bool\n",
    "            Set `True` for visualization (default: False)\n",
    "\n",
    "        dtw_implementation : str\n",
    "            DTW implementation, librosa or synctoolbox (default: synctoolbox)\n",
    "\n",
    "        normalize_chroma : bool\n",
    "            Set `True` to normalize input chroma features after each downsampling\n",
    "            and smoothing operation.\n",
    "\n",
    "        chroma_norm_ord: int\n",
    "            Order of chroma normalization, relevant if ``normalize_chroma`` is True.\n",
    "            (default: 2)\n",
    "\n",
    "        chroma_norm_threshold: float\n",
    "            If the norm falls below threshold for a feature vector, then the\n",
    "            normalized feature vector is set to be the unit vector. Relevant, if\n",
    "            ``normalize_chroma`` is True (default: 0.001)\n",
    "\n",
    "        visualization_title : str\n",
    "            Title for the visualization plots. Only relevant if 'verbose' is True\n",
    "            (default: \"MrMsDTW result\")\n",
    "\n",
    "        anchor_pairs: List[Tuple]\n",
    "            Anchor pairs given in seconds. Note that\n",
    "            * (0, 0) and (<audio-len1>, <audio-len2>) are not allowed.\n",
    "            * Anchors must be monotonously increasing.\n",
    "\n",
    "        linear_inp_idx: List[int]\n",
    "            List of the indices of intervals created by anchor pairs, for which\n",
    "            MrMsDTW shouldn't be run, e.g., if the interval only involves silence.\n",
    "\n",
    "            0        ap1        ap2        ap3\n",
    "            |         |          |          |\n",
    "            |  idx0   |   idx1   |  idx2    |  idx3 OR idx-1\n",
    "            |         |          |          |\n",
    "\n",
    "            Note that index -1 corresponds to the last interval, which begins with\n",
    "            the last anchor pair until the end of the audio files.\n",
    "\n",
    "        alpha: float\n",
    "            Coefficient for the Chroma cost matrix in the finest scale of the MrMsDTW algorithm.\n",
    "            C = alpha * C_Chroma + (1 - alpha) * C_act  (default: 0.5)\n",
    "\n",
    "        Returns\n",
    "        -------\n",
    "        wp : np.ndarray [shape=(2, T)]\n",
    "            Resulting warping path which indicates synchronized indices.\n",
    "    \"\"\"\n",
    "        \n",
    "    if anchor_pairs is None:\n",
    "        wp = sync_via_mrmsdtw(f_chroma1=f_chroma1,\n",
    "                              f_chroma2=f_chroma2,\n",
    "                              f_onset1=f_onset1,\n",
    "                              f_onset2=f_onset2,\n",
    "                              input_feature_rate=input_feature_rate,\n",
    "                              step_sizes=step_sizes,\n",
    "                              step_weights=step_weights,\n",
    "                              threshold_rec=threshold_rec,\n",
    "                              win_len_smooth=win_len_smooth,\n",
    "                              downsamp_smooth=downsamp_smooth,\n",
    "                              verbose=verbose,\n",
    "                              dtw_implementation=dtw_implementation,\n",
    "                              normalize_chroma=normalize_chroma,\n",
    "                              chroma_norm_ord=chroma_norm_ord,\n",
    "                              chroma_norm_threshold=chroma_norm_threshold,\n",
    "                              visualization_title=visualization_title,\n",
    "                              alpha=alpha)\n",
    "    else:\n",
    "        wp = None\n",
    "\n",
    "        if verbose:\n",
    "            print('Anchor points are given!')\n",
    "\n",
    "        # Add ending as the anchor point\n",
    "        if (anchor_pairs[-1][0] < f_chroma1.shape[1]/input_feature_rate) or (anchor_pairs[-1][1] < f_chroma2.shape[1]/input_feature_rate):\n",
    "            anchor_pairs.append((-1, -1))\n",
    "\n",
    "        prev_a1 = 0\n",
    "        prev_a2 = 0\n",
    "        \n",
    "        flag_quit = False\n",
    "\n",
    "        for idx, anchor_pair in enumerate(anchor_pairs):\n",
    "            cur_a1, cur_a2 = anchor_pair\n",
    "            \n",
    "            if cur_a1 == 0:\n",
    "                wp_cur = np.concatenate([-np.ones(int(cur_a2*input_feature_rate))[None,:],\n",
    "                                         np.arange(int(cur_a2*input_feature_rate))[None,:]],\n",
    "                                        axis=0)\n",
    "            elif cur_a2 == 0:\n",
    "                wp_cur = np.concatenate([np.arange(int(cur_a1*input_feature_rate))[None,:],\n",
    "                                        -np.ones(int(cur_a1*input_feature_rate))[None,:]],\n",
    "                                        axis=0)\n",
    "                \n",
    "            elif  (prev_a1 - f_chroma1.shape[1]/input_feature_rate) >= -1/input_feature_rate:\n",
    "                indices_2 = np.arange( f_chroma2.shape[1] - int(prev_a2*input_feature_rate))\n",
    "                \n",
    "                \n",
    "                wp_cur = np.concatenate([ np.ones_like(indices_2),\n",
    "                                          #int(prev_a1*input_feature_rate) + 1 + np.zeros_like(indices_2),\n",
    "                                          indices_2], axis=0)\n",
    "                \n",
    "                \n",
    "                flag_quit=True\n",
    "                \n",
    "                \n",
    "            elif  (prev_a2 - f_chroma2.shape[1]/input_feature_rate) >= -1/input_feature_rate:\n",
    "                indices_1 = np.arange( f_chroma1.shape[1] - int(prev_a1*input_feature_rate))\n",
    "                \n",
    "                wp_cur = np.concatenate([indices_1,                     \n",
    "                                        np.ones_like(indices_1),\n",
    "                                          ], axis=0)\n",
    "                \n",
    "                \n",
    "                flag_quit=True\n",
    "                \n",
    "            \n",
    "                \n",
    "            else:\n",
    "\n",
    "                # Split the features\n",
    "                f_chroma1_split, f_onset1_split, f_chroma2_split, f_onset2_split = __split_features(f_chroma1,\n",
    "                                                                                                    f_onset1,\n",
    "                                                                                                    f_chroma2,\n",
    "                                                                                                    f_onset2,\n",
    "                                                                                                    cur_a1,\n",
    "                                                                                                    cur_a2,\n",
    "                                                                                                    prev_a1,\n",
    "                                                                                                    prev_a2,\n",
    "                                                                                                    input_feature_rate)\n",
    "\n",
    "                if idx in linear_inp_idx or idx == len(anchor_pairs) - 1 and -1 in linear_inp_idx:\n",
    "                    # Generate a diagonal warping path, if the algorithm is not supposed to executed.\n",
    "                    # A typical scenario is the silence breaks which are enclosed by two anchor points.\n",
    "                    if verbose:\n",
    "                        print('A diagonal warping path is generated for the interval \\n\\t Feature sequence 1: %.2f - %.2f'\n",
    "                              '\\n\\t Feature sequence 2: %.2f - %.2f\\n' % (prev_a1, cur_a1, prev_a2, cur_a2))\n",
    "                    wp_cur = __diagonal_warping_path(f_chroma1_split, f_chroma2_split)\n",
    "\n",
    "                else:\n",
    "                    if verbose:\n",
    "                        if cur_a1 != -1 and cur_a2 != -1:\n",
    "                            print('MrMsDTW is applied for the interval \\n\\t Feature sequence 1: %.2f - %.2f'\n",
    "                                  '\\n\\t Feature sequence 2: %.2f - %.2f\\n' % (prev_a1, cur_a1, prev_a2, cur_a2))\n",
    "                        else:\n",
    "                            print('MrMsDTW is applied for the interval \\n\\t Feature sequence 1: %.2f - end'\n",
    "                                  '\\n\\t Feature sequence 2: %.2f - end\\n' % (prev_a1, prev_a2))\n",
    "                    wp_cur = sync_via_mrmsdtw(f_chroma1=f_chroma1_split,\n",
    "                                              f_chroma2=f_chroma2_split,\n",
    "                                              f_onset1=f_onset1_split,\n",
    "                                              f_onset2=f_onset2_split,\n",
    "                                              input_feature_rate=input_feature_rate,\n",
    "                                              step_sizes=step_sizes,\n",
    "                                              step_weights=step_weights,\n",
    "                                              threshold_rec=threshold_rec,\n",
    "                                              win_len_smooth=win_len_smooth,\n",
    "                                              downsamp_smooth=downsamp_smooth,\n",
    "                                              verbose=verbose,\n",
    "                                              dtw_implementation=dtw_implementation,\n",
    "                                              normalize_chroma=normalize_chroma,\n",
    "                                              chroma_norm_ord=chroma_norm_ord,\n",
    "                                              chroma_norm_threshold=chroma_norm_threshold,\n",
    "                                              alpha=alpha)\n",
    "\n",
    "            if wp is None:\n",
    "                wp = np.array(wp_cur, copy=True)\n",
    "\n",
    "            # Concatenate warping paths\n",
    "            else:\n",
    "                wp = np.concatenate([wp, wp_cur + wp[:, -1].reshape(2, 1) + 1], axis=1)\n",
    "                \n",
    "\n",
    "            prev_a1 = cur_a1\n",
    "            prev_a2 = cur_a2\n",
    "            \n",
    "            if flag_quit: \n",
    "                break\n",
    "\n",
    "        anchor_pairs.pop()\n",
    "\n",
    "    return wp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "07dcee01-cb97-4250-b8bb-b4c8dfbf6fb7",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import os\n",
    "from tqdm.notebook import tqdm"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0a1b32ee-cb4e-497f-a8e8-f4b0f40a0f62",
   "metadata": {},
   "source": [
    "### use features from finetuned transcriber\n",
    "\n",
    "also use chroma & onset features from annotation when a silence modification is added to the Kempff version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "2b951b8e-63be-4132-b0e6-bc3e0b342e04",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "measures_dir = os.path.join(\"../\", \"2_Annotations\", \"ann_audio_measure\")\n",
    "onset_frames_dir = os.path.expanduser(\"path_to_transcription_model\")\n",
    "startEnd_dir = os.path.join(\"../\", \"2_Annotations\", \"ann_audio_startEnd\")\n",
    "ann_mod_dir = os.path.join(\"../\", \"2_Annotations\", \"ann_audio_modifications\")\n",
    "score_csv_dir = os.path.join(\"../\", \"2_Annotations\", \"ann_score_note\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "53e65b75-45aa-4473-a655-f0f9828973f3",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "for sonata in all_sonatas:\n",
    "    print(\"Transferring %s\"%(sonata))\n",
    "    measuresIn_Kempff = read_csv_to_df(os.path.join(measures_dir, sonata+\"_WK64.csv\"))\n",
    "    \n",
    "    audio_1, _ = librosa.load(os.path.join(audio_dir, sonata+\"_WK64.wav\"), sr=Fs)    \n",
    "    \n",
    "    featuresIn_1 = np.load(os.path.join(onset_frames_dir, \"%s_%s.npz\"%(sonata, \"WK64\")))    \n",
    "    f_chroma_quantized_1, f_DLNCO_1 = transcriptionFeatures_to_chroma_and_DLNCO(onsetsIn = featuresIn_1[\"onset_pred\"].T,\n",
    "                                                                                framesIn = featuresIn_1[\"onset_pred\"].T,\n",
    "                                                                                frame_rate_in = featuresIn_1[\"sample_rate\"] / featuresIn_1[\"hop_length\"],\n",
    "                                                                                feature_rate = feature_rate, \n",
    "                                                                                audio_len = len(audio_1),\n",
    "                                                                                Fs=Fs)\n",
    "    \n",
    "    f_chroma_quantized_1, f_DLNCO_1, is_silence_1 = fill_added_silence(f_chroma=f_chroma_quantized_1, f_onsets=f_DLNCO_1, sonata=sonata, performer=\"WK64\")\n",
    "\n",
    "        \n",
    "    for performer in tqdm(all_performers):\n",
    "        print(\"processing %s, %s\"%(sonata, performer))\n",
    "        if performer == \"WK64\": continue        \n",
    "        audio_2, _ = librosa.load(os.path.join(audio_dir, sonata+\"_\"+performer+\".wav\"), sr=Fs)     \n",
    "        \n",
    "        \n",
    "        \n",
    "        featuresIn_2 = np.load(os.path.join(onset_frames_dir, \"%s_%s.npz\"%(sonata, performer)))    \n",
    "        f_chroma_quantized_2, f_DLNCO_2 = transcriptionFeatures_to_chroma_and_DLNCO(onsetsIn = featuresIn_2[\"onset_pred\"].T,\n",
    "                                                                                    framesIn = featuresIn_2[\"onset_pred\"].T,\n",
    "                                                                                    frame_rate_in = featuresIn_2[\"sample_rate\"] / featuresIn_2[\"hop_length\"],\n",
    "                                                                                    feature_rate = feature_rate, \n",
    "                                                                                    audio_len = len(audio_2),\n",
    "                                                                                    Fs=Fs)\n",
    "        \n",
    "        \n",
    "        startEnd = read_csv_to_df(os.path.join(startEnd_dir, \"%s_%s.csv\"%(sonata, performer)))\n",
    "\n",
    "        anchor_pairs = [ (measuresIn_Kempff.iloc[0].time, startEnd.iloc[0].start),\n",
    "                         (measuresIn_Kempff.iloc[-1].time, startEnd.iloc[0].end)]\n",
    "\n",
    "        wp = sync_via_mrmsdtw_with_anchors(f_chroma1=f_chroma_quantized_1, f_onset1=f_DLNCO_1, f_chroma2=f_chroma_quantized_2, f_onset2=f_DLNCO_2, input_feature_rate=feature_rate, \n",
    "                                           anchor_pairs=anchor_pairs, step_weights=step_weights, threshold_rec=threshold_rec, verbose=False)\n",
    "        wp = make_path_strictly_monotonic(wp)        \n",
    "\n",
    "        measuresOut = measuresIn_Kempff.copy()\n",
    "        measuresOut.time = scipy.interpolate.interp1d(wp[0] / feature_rate, wp[1] / feature_rate, kind='linear')(measuresOut.time)\n",
    "        \n",
    "        measuresOut.to_csv(os.path.join(measures_dir, sonata+\"_\"+performer+\".csv\"), sep=\";\", index=False)\n",
    "        print(\"saving to %s\"%(os.path.join(measures_dir, sonata+\"_\"+performer+\".csv\")))\n",
    "    \n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fa7e34c4-9aef-46cd-842a-b3ba32944c12",
   "metadata": {},
   "source": [
    "### refine measure transfer to match start/end annotations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "id": "3666fe5b-4a43-4833-8e2d-64167027e1f3",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "processing Beethoven_Op078-01\n"
     ]
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "99d0edf0fbe2438d87067ac48dc8ffae",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "  0%|          | 0/11 [00:00<?, ?it/s]"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "for sonata in all_sonatas:\n",
    "    print(\"processing %s\"%(sonata))\n",
    "    for performer in tqdm(all_performers):\n",
    "        \n",
    "        measuresIn = read_csv_to_df(os.path.join(measures_dir_transc, sonata+\"_\"+performer+\".csv\"))\n",
    "    \n",
    "        startEnd = read_csv_to_df(os.path.join(\"../\", \"2_Annotations\", \"ann_audio_startEnd\", \"%s_%s.csv\"%(sonata, performer)), csv_delimiter=\";\")\n",
    "        \n",
    "        measuresIn.sort_values(by=\"measure\", inplace=True)\n",
    "        measuresIn.iloc[0].time=startEnd.start\n",
    "        measuresIn.iloc[-1].time=startEnd.end\n",
    "        \n",
    "        measuresIn.to_csv(os.path.join(measures_dir_transc, sonata+\"_\"+performer+\".csv\"), sep=\";\", index=False)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}