import os, glob, copy import torch from collections import defaultdict import partitura as pt import torch.nn.functional as F from scipy.interpolate import interp1d from scipy.stats import pearsonr, gaussian_kde, entropy from scipy.special import kl_div import numpy.lib.recfunctions as rfn import matplotlib.pyplot as plt import seaborn as sns import numpy as np import pandas as pd from prepare_data import * def tensor_pair_swap(x): if type(x) == list: x = np.array(x) # given batched x, swap the pairs from the first dimension permute_index = torch.arange(x.shape[0]).view(-1, 2)[:, [1, 0]].contiguous().view(-1) return x[permute_index] def get_batch_slice(batch, idx): """given a dictionary batch, get the sliced ditionary given idx""" return {k: v[idx] for k, v in batch.items()} TESTING_GROUP = [ '07935_seg0', # bethoven appasionata beginning '10713_seg0', # Jeux D'eau beginning '11579_seg3', # Chopin Ballade 2 transitioning passage '00129_seg0', # Rachmaninov etude tableaux no.5 eb minor 'ASAP_Schumann_Arabeske_seg0', # Schumann Arabeske beginning 'ASAP_Bach_Fugue_bwv_867_seg0', # Bach fugue 867 beginning 'ASAP_Mozart_Piano_Sonatas_12-1_seg0', # Mozart a minor beginning 'VIENNA422_Schubert_D783_no15_seg0', # Vienna422 schubert piece beginning #### second group of testing data: paired another performance of the test '07925_seg0', '10717_seg0', '11587_seg3', '00130_seg0', 'ASAP_Schumann_Arabeske_seg0.', # select another data example with the same snote_id_path 'ASAP_Bach_Fugue_bwv_867_seg0.', 'ASAP_Mozart_Piano_Sonatas_12-1_seg0.', 'VIENNA422_Schubert_D783_no15_seg0.', ] def split_train_valid(codec_data, select_num=58008, paired_input=False, data_root=None): """select the train and valid set according to the following criteria: - ASAP and VIENNA422 goes to testing set as they are better ground truths - split regarding to pieces.(same score path) """ codec_data_ = codec_data # codec_data_ = codec_data[list(map(lambda x: not (("ATEPP" in x['score_path']) and ("musicxml_cleaned.musicxml" in x['score_path'])), codec_data))] train_idx = int(len(codec_data_) * 0.85 - 1) assert (train_idx % 2 == 0) if not select_num: return codec_data_[:train_idx], codec_data_[train_idx:] selected_cd, unselected_cd = [], [] if paired_input: for idx in range(0, len(codec_data), 2): if len(selected_cd) > select_num: break cd, cd_ = codec_data[idx], codec_data[idx+1] if "ATEPP" not in cd['snote_id_path']: selected_cd.extend([cd, cd_]) else: unselected_cd.extend([cd, cd_]) else: for idx in range(0, len(codec_data_)): if len(selected_cd) > select_num: break cd = codec_data_[idx] if "ATEPP" not in cd['snote_id_path']: selected_cd.extend([cd]) else: unselected_cd.extend([cd]) # selected_cd, unselected_cd = defaultdict(list), [] # for cd in codec_data: # for name in TESTING_GROUP: # if (not selected_cd[name]) and name in cd['snote_id_path']: # selected_cd[name] = cd # break # else: # unselected_cd.append(cd) # np.random.shuffle(unselected_cd) train_set = unselected_cd[:train_idx] valid_set = selected_cd valid_set = np.array(valid_set)[list(map(lambda x: "mu" not in x['piece_name'], valid_set))] if data_root is not None: np.save(f"{data_root}/codec_N=200_mixup_train.npy", train_set) np.save(f"{data_root}/codec_N=200_mixup_test.npy", valid_set) return train_set, valid_set def load_transfer_pair(data_root, K=50000, N=200): """ returns: - transfer_pairs: list of tuple of codec - unpaired: list of single codec """ transfer_pairs = np.load(f"{data_root}/codec_N={N}_mixup_paired_K={K}.npy", allow_pickle=True) unpaired = np.load(f"{data_root}/codec_N={N}_mixup_unpaired_K={K}.npy", allow_pickle=True) return transfer_pairs, unpaired def group_same_seg(valid_set): """Group all valid_set by segment, return a dictionary""" valid_df = pd.DataFrame(list(valid_set)) # Group by 'grouptype' and collect indices grouped_indices = valid_df.groupby("snote_id_path").apply(lambda group: list(group.index)) indices_dict = grouped_indices.to_dict() indices_dict = {k.split("/")[-1][:-4]: v for k, v in indices_dict.items()} # {"ASAP_Bach_Fugue_bwv_854_seg2": [23, 766], ...} avg_repetition = sum([len(v) for k, v in indices_dict.items()]) / len(indices_dict) # 5.39 return indices_dict def plot_codec(codec1, codec2, ax0, ax1, fig): # plot the pred p_codec and label p_codec, on the given two axes p_im = ax0.imshow(codec1.T, aspect='auto', origin='lower') ax0.set_yticks([0, 1, 2, 3, 4]) ax0.set_yticklabels(["beat_period", "velocity", "timing", "articulation_log", "pedal"]) fig.colorbar(p_im, orientation='vertical', ax=ax0) s_im = ax1.imshow(codec2.T, aspect='auto', origin='lower') ax1.set_yticks([0, 1, 2, 3]) ax1.set_yticklabels(["beat_period", "velocity", "timing", "articulation_log"]) fig.colorbar(s_im, orientation='vertical', ax=ax1) return def animate_sampling(t_idx, fig, ax_flat, caxs, noise_list, total_timesteps): # noise_list: Tuple of (x_t, t), (x_t-1, t-1), ... (x_0, 0) # x_t (B, 1, T, F) # clearing figures to prevent slow down in each iteration.d fig.canvas.draw() for idx in range(len(noise_list[0][0])): # visualize 8 samples in the batch ax_flat[2*idx].cla() ax_flat[2*idx+1].cla() caxs[2*idx].cla() caxs[2*idx+1].cla() # roll_pred (1, T, F) im1 = ax_flat[2*idx].imshow(noise_list[0][0][idx][0].detach().T.cpu(), aspect='auto', origin='lower') im2 = ax_flat[2*idx+1].imshow(noise_list[1 + total_timesteps - t_idx][0][idx][0].T, aspect='auto', origin='lower') fig.colorbar(im1, cax=caxs[2*idx]) fig.colorbar(im2, cax=caxs[2*idx+1]) fig.suptitle(f't={t_idx}') row1_txt = ax_flat[0].text(-400,45,f'Gaussian N(0,1)') row2_txt = ax_flat[4].text(-300,45,'x_{t-1}') def compile_condition(s_codec, c_codec): """compile s_codec and c_codec into a joint condition Args: s_codec : (B, N, 4) c_codec : (B, N, 7) """ return torch.cat((s_codec, c_codec), dim=2) def apply_normalization(cd, mean, std, i, idx): # apply normalization for p codec in codec data return (cd['p_codec'][:, i] - mean) / std def dataset_normalization(train_set, valid_set): """ normalize the p_codec, across the dataset range. return mean and std for each column. """ codec_data = np.hstack([train_set, valid_set]) dataset_pc = np.vstack([cd['p_codec'] for cd in codec_data]) means, stds = [], [] codec_data_ = copy.deepcopy(codec_data) for i in range(5): mean = dataset_pc[:, i].mean() std = dataset_pc[:, i].std() for idx, cd in enumerate(codec_data): codec_data_[idx]['p_codec'][:, i] = apply_normalization(cd, mean, std, i, idx) means.append(float(mean)) stds.append(float(std)) # conversion for save into OmegaConf return codec_data_[:len(train_set)], codec_data_[len(train_set):], means, stds def p_codec_scale(p_codec, means, stds): # inverse of normalization applied on p_codec # p_codec: (B, N, 5) or (B, 1, N, 5) for i in range(5): p_codec[..., i] = p_codec[..., i] * stds[i] + means[i] return p_codec class Normalization(): """ This class is for normalizing the input batch by batch. The normalization used is min-max, two modes 'framewise' and 'imagewise' can be selected. In this paper, we found that 'imagewise' normalization works better than 'framewise' If framewise is used, then X must follow the shape of (B, F, T) """ def __init__(self, min, max, mode='imagewise'): if mode == 'framewise': def normalize(x): size = x.shape x_max = x.max(1, keepdim=True)[0] # Finding max values for each frame x_min = x.min(1, keepdim=True)[0] x_std = (x-x_min)/(x_max-x_min) # If there is a column with all zero, nan will occur x_std[torch.isnan(x_std)]=0 # Making nan to 0 x_scaled = x_std * (max - min) + min return x_scaled elif mode == 'imagewise': def normalize(x): # x_max = x.view(size[0], size[1]*size[2]).max(1, keepdim=True)[0] # x_min = x.view(size[0], size[1]*size[2]).min(1, keepdim=True)[0] x_max = x.flatten(1).max(1, keepdim=True)[0] x_min = x.flatten(1).min(1, keepdim=True)[0] x_max = x_max.unsqueeze(1) # Make it broadcastable x_min = x_min.unsqueeze(1) # Make it broadcastable x_std = (x-x_min)/(x_max-x_min) x_scaled = x_std * (max - min) + min x_scaled[torch.isnan(x_scaled)]=min # if piano roll is empty, turn them to min return x_scaled else: print(f'please choose the correct mode') self.normalize = normalize def __call__(self, x): return self.normalize(x) def render_midi_to_audio(midi_path, output_path=None): """The soundfont we used is the Essential Keys-sofrzando-v9.6 from https://sites.google.com/site/soundfonts4u/ """ if output_path == None: output_path = midi_path[:-4] + ".wav" os.system(f"fluidsynth -ni ../artifacts/Essential-Keys-sforzando-v9.6.sf2 {midi_path} -F {output_path} ") return if __name__ == "__main__": import random asap_mid = glob.glob("../Datasets/asap-dataset-alignment/**/*.mid", recursive=True) for am in random.choices(asap_mid, k=2): render_midi_to_audio(am, output_path=f"{am[35:-4]}.wav")