import numpy as np import librosa import mir_eval import torch import os idx2chord = ['C', 'C:min', 'C#', 'C#:min', 'D', 'D:min', 'D#', 'D#:min', 'E', 'E:min', 'F', 'F:min', 'F#', 'F#:min', 'G', 'G:min', 'G#', 'G#:min', 'A', 'A:min', 'A#', 'A#:min', 'B', 'B:min', 'N'] root_list = ['C', 'C#', 'D', 'D#', 'E', 'F', 'F#', 'G', 'G#', 'A', 'A#', 'B'] quality_list = ['min', 'maj', 'dim', 'aug', 'min6', 'maj6', 'min7', 'minmaj7', 'maj7', '7', 'dim7', 'hdim7', 'sus2', 'sus4'] def idx2voca_chord(): idx2voca_chord = {} idx2voca_chord[169] = 'N' idx2voca_chord[168] = 'X' for i in range(168): root = i // 14 root = root_list[root] quality = i % 14 quality = quality_list[quality] if i % 14 != 1: chord = root + ':' + quality else: chord = root idx2voca_chord[i] = chord return idx2voca_chord def audio_file_to_features(audio_file, config): original_wav, sr = librosa.load(audio_file, sr=config.mp3['song_hz'], mono=True) currunt_sec_hz = 0 while len(original_wav) > currunt_sec_hz + config.mp3['song_hz'] * config.mp3['inst_len']: start_idx = int(currunt_sec_hz) end_idx = int(currunt_sec_hz + config.mp3['song_hz'] * config.mp3['inst_len']) tmp = librosa.cqt(original_wav[start_idx:end_idx], sr=sr, n_bins=config.feature['n_bins'], bins_per_octave=config.feature['bins_per_octave'], hop_length=config.feature['hop_length']) if start_idx == 0: feature = tmp else: feature = np.concatenate((feature, tmp), axis=1) currunt_sec_hz = end_idx tmp = librosa.cqt(original_wav[currunt_sec_hz:], sr=sr, n_bins=config.feature['n_bins'], bins_per_octave=config.feature['bins_per_octave'], hop_length=config.feature['hop_length']) feature = np.concatenate((feature, tmp), axis=1) feature = np.log(np.abs(feature) + 1e-6) feature_per_second = config.mp3['inst_len'] / config.model['timestep'] song_length_second = len(original_wav)/config.mp3['song_hz'] return feature, feature_per_second, song_length_second # Audio files with format of wav and mp3 def get_audio_paths(audio_dir): return [os.path.join(root, fname) for (root, dir_names, file_names) in os.walk(audio_dir, followlinks=True) for fname in file_names if (fname.lower().endswith('.wav') or fname.lower().endswith('.mp3'))] def get_lab_paths(lab_dir): return [os.path.join(root, fname) for (root, dir_names, file_names) in os.walk(lab_dir, followlinks=True) for fname in file_names if (fname.lower().endswith('.lab'))] class metrics(): def __init__(self): super(metrics, self).__init__() self.score_metrics = ['root', 'thirds', 'triads', 'sevenths', 'tetrads', 'majmin', 'mirex'] self.score_list_dict = dict() for i in self.score_metrics: self.score_list_dict[i] = list() self.average_score = dict() def score(self, metric, gt_path, est_path): if metric == 'root': score = self.root_score(gt_path,est_path) elif metric == 'thirds': score = self.thirds_score(gt_path,est_path) elif metric == 'triads': score = self.triads_score(gt_path,est_path) elif metric == 'sevenths': score = self.sevenths_score(gt_path,est_path) elif metric == 'tetrads': score = self.tetrads_score(gt_path,est_path) elif metric == 'majmin': score = self.majmin_score(gt_path,est_path) elif metric == 'mirex': score = self.mirex_score(gt_path,est_path) else: raise NotImplementedError return score def root_score(self, gt_path, est_path): (ref_intervals, ref_labels) = mir_eval.io.load_labeled_intervals(gt_path) ref_labels = lab_file_error_modify(ref_labels) (est_intervals, est_labels) = mir_eval.io.load_labeled_intervals(est_path) est_intervals, est_labels = mir_eval.util.adjust_intervals(est_intervals, est_labels, ref_intervals.min(), ref_intervals.max(), mir_eval.chord.NO_CHORD, mir_eval.chord.NO_CHORD) (intervals, ref_labels, est_labels) = mir_eval.util.merge_labeled_intervals(ref_intervals, ref_labels, est_intervals, est_labels) durations = mir_eval.util.intervals_to_durations(intervals) comparisons = mir_eval.chord.root(ref_labels, est_labels) score = mir_eval.chord.weighted_accuracy(comparisons, durations) return score def thirds_score(self, gt_path, est_path): (ref_intervals, ref_labels) = mir_eval.io.load_labeled_intervals(gt_path) ref_labels = lab_file_error_modify(ref_labels) (est_intervals, est_labels) = mir_eval.io.load_labeled_intervals(est_path) est_intervals, est_labels = mir_eval.util.adjust_intervals(est_intervals, est_labels, ref_intervals.min(), ref_intervals.max(), mir_eval.chord.NO_CHORD, mir_eval.chord.NO_CHORD) (intervals, ref_labels, est_labels) = mir_eval.util.merge_labeled_intervals(ref_intervals, ref_labels, est_intervals, est_labels) durations = mir_eval.util.intervals_to_durations(intervals) comparisons = mir_eval.chord.thirds(ref_labels, est_labels) score = mir_eval.chord.weighted_accuracy(comparisons, durations) return score def triads_score(self, gt_path, est_path): (ref_intervals, ref_labels) = mir_eval.io.load_labeled_intervals(gt_path) ref_labels = lab_file_error_modify(ref_labels) (est_intervals, est_labels) = mir_eval.io.load_labeled_intervals(est_path) est_intervals, est_labels = mir_eval.util.adjust_intervals(est_intervals, est_labels, ref_intervals.min(), ref_intervals.max(), mir_eval.chord.NO_CHORD, mir_eval.chord.NO_CHORD) (intervals, ref_labels, est_labels) = mir_eval.util.merge_labeled_intervals(ref_intervals, ref_labels, est_intervals, est_labels) durations = mir_eval.util.intervals_to_durations(intervals) comparisons = mir_eval.chord.triads(ref_labels, est_labels) score = mir_eval.chord.weighted_accuracy(comparisons, durations) return score def sevenths_score(self, gt_path, est_path): (ref_intervals, ref_labels) = mir_eval.io.load_labeled_intervals(gt_path) ref_labels = lab_file_error_modify(ref_labels) (est_intervals, est_labels) = mir_eval.io.load_labeled_intervals(est_path) est_intervals, est_labels = mir_eval.util.adjust_intervals(est_intervals, est_labels, ref_intervals.min(), ref_intervals.max(), mir_eval.chord.NO_CHORD, mir_eval.chord.NO_CHORD) (intervals, ref_labels, est_labels) = mir_eval.util.merge_labeled_intervals(ref_intervals, ref_labels, est_intervals, est_labels) durations = mir_eval.util.intervals_to_durations(intervals) comparisons = mir_eval.chord.sevenths(ref_labels, est_labels) score = mir_eval.chord.weighted_accuracy(comparisons, durations) return score def tetrads_score(self, gt_path, est_path): (ref_intervals, ref_labels) = mir_eval.io.load_labeled_intervals(gt_path) ref_labels = lab_file_error_modify(ref_labels) (est_intervals, est_labels) = mir_eval.io.load_labeled_intervals(est_path) est_intervals, est_labels = mir_eval.util.adjust_intervals(est_intervals, est_labels, ref_intervals.min(), ref_intervals.max(), mir_eval.chord.NO_CHORD, mir_eval.chord.NO_CHORD) (intervals, ref_labels, est_labels) = mir_eval.util.merge_labeled_intervals(ref_intervals, ref_labels, est_intervals, est_labels) durations = mir_eval.util.intervals_to_durations(intervals) comparisons = mir_eval.chord.tetrads(ref_labels, est_labels) score = mir_eval.chord.weighted_accuracy(comparisons, durations) return score def majmin_score(self, gt_path, est_path): (ref_intervals, ref_labels) = mir_eval.io.load_labeled_intervals(gt_path) ref_labels = lab_file_error_modify(ref_labels) (est_intervals, est_labels) = mir_eval.io.load_labeled_intervals(est_path) est_intervals, est_labels = mir_eval.util.adjust_intervals(est_intervals, est_labels, ref_intervals.min(), ref_intervals.max(), mir_eval.chord.NO_CHORD, mir_eval.chord.NO_CHORD) (intervals, ref_labels, est_labels) = mir_eval.util.merge_labeled_intervals(ref_intervals, ref_labels, est_intervals, est_labels) durations = mir_eval.util.intervals_to_durations(intervals) comparisons = mir_eval.chord.majmin(ref_labels, est_labels) score = mir_eval.chord.weighted_accuracy(comparisons, durations) return score def mirex_score(self, gt_path, est_path): (ref_intervals, ref_labels) = mir_eval.io.load_labeled_intervals(gt_path) ref_labels = lab_file_error_modify(ref_labels) (est_intervals, est_labels) = mir_eval.io.load_labeled_intervals(est_path) est_intervals, est_labels = mir_eval.util.adjust_intervals(est_intervals, est_labels, ref_intervals.min(), ref_intervals.max(), mir_eval.chord.NO_CHORD, mir_eval.chord.NO_CHORD) (intervals, ref_labels, est_labels) = mir_eval.util.merge_labeled_intervals(ref_intervals, ref_labels, est_intervals, est_labels) durations = mir_eval.util.intervals_to_durations(intervals) comparisons = mir_eval.chord.mirex(ref_labels, est_labels) score = mir_eval.chord.weighted_accuracy(comparisons, durations) return score def lab_file_error_modify(ref_labels): for i in range(len(ref_labels)): if ref_labels[i][-2:] == ':4': ref_labels[i] = ref_labels[i].replace(':4', ':sus4') elif ref_labels[i][-2:] == ':6': ref_labels[i] = ref_labels[i].replace(':6', ':maj6') elif ref_labels[i][-4:] == ':6/2': ref_labels[i] = ref_labels[i].replace(':6/2', ':maj6/2') elif ref_labels[i] == 'Emin/4': ref_labels[i] = 'E:min/4' elif ref_labels[i] == 'A7/3': ref_labels[i] = 'A:7/3' elif ref_labels[i] == 'Bb7/3': ref_labels[i] = 'Bb:7/3' elif ref_labels[i] == 'Bb7/5': ref_labels[i] = 'Bb:7/5' elif ref_labels[i].find(':') == -1: if ref_labels[i].find('min') != -1: ref_labels[i] = ref_labels[i][:ref_labels[i].find('min')] + ':' + ref_labels[i][ref_labels[i].find('min'):] return ref_labels def root_majmin_score_calculation(valid_dataset, config, mean, std, device, model, model_type, verbose=False): valid_song_names = valid_dataset.song_names paths = valid_dataset.preprocessor.get_all_files() metrics_ = metrics() song_length_list = list() for path in paths: song_name, lab_file_path, mp3_file_path, _ = path if not song_name in valid_song_names: continue try: n_timestep = config.model['timestep'] feature, feature_per_second, song_length_second = audio_file_to_features(mp3_file_path, config) feature = feature.T feature = (feature - mean) / std time_unit = feature_per_second num_pad = n_timestep - (feature.shape[0] % n_timestep) feature = np.pad(feature, ((0, num_pad), (0, 0)), mode="constant", constant_values=0) num_instance = feature.shape[0] // n_timestep start_time = 0.0 lines = [] with torch.no_grad(): model.eval() feature = torch.tensor(feature, dtype=torch.float32).unsqueeze(0).to(device) for t in range(num_instance): if model_type == 'btc': encoder_output, _ = model.self_attn_layers(feature[:, n_timestep * t:n_timestep * (t + 1), :]) prediction, _ = model.output_layer(encoder_output) prediction = prediction.squeeze() elif model_type == 'cnn' or model_type =='crnn': prediction, _, _, _ = model(feature[:, n_timestep * t:n_timestep * (t + 1), :], torch.randint(config.model['num_chords'], (n_timestep,)).to(device)) for i in range(n_timestep): if t == 0 and i == 0: prev_chord = prediction[i].item() continue if prediction[i].item() != prev_chord: lines.append( '%.6f %.6f %s\n' % ( start_time, time_unit * (n_timestep * t + i), idx2chord[prev_chord])) start_time = time_unit * (n_timestep * t + i) prev_chord = prediction[i].item() if t == num_instance - 1 and i + num_pad == n_timestep: if start_time != time_unit * (n_timestep * t + i): lines.append( '%.6f %.6f %s\n' % ( start_time, time_unit * (n_timestep * t + i), idx2chord[prev_chord])) break pid = os.getpid() tmp_path = 'tmp_' + str(pid) + '.lab' with open(tmp_path, 'w') as f: for line in lines: f.write(line) root_majmin = ['root', 'majmin'] for m in root_majmin: metrics_.score_list_dict[m].append(metrics_.score(metric=m, gt_path=lab_file_path, est_path=tmp_path)) song_length_list.append(song_length_second) if verbose: for m in root_majmin: print('song name %s, %s score : %.4f' % (song_name, m, metrics_.score_list_dict[m][-1])) except: print('song name %s\' lab file error' % song_name) tmp = song_length_list / np.sum(song_length_list) for m in root_majmin: metrics_.average_score[m] = np.sum(np.multiply(metrics_.score_list_dict[m], tmp)) return metrics_.score_list_dict, song_length_list, metrics_.average_score def root_majmin_score_calculation_crf(valid_dataset, config, mean, std, device, pre_model, model, model_type, verbose=False): valid_song_names = valid_dataset.song_names paths = valid_dataset.preprocessor.get_all_files() metrics_ = metrics() song_length_list = list() for path in paths: song_name, lab_file_path, mp3_file_path, _ = path if not song_name in valid_song_names: continue try: n_timestep = config.model['timestep'] feature, feature_per_second, song_length_second = audio_file_to_features(mp3_file_path, config) feature = feature.T feature = (feature - mean) / std time_unit = feature_per_second num_pad = n_timestep - (feature.shape[0] % n_timestep) feature = np.pad(feature, ((0, num_pad), (0, 0)), mode="constant", constant_values=0) num_instance = feature.shape[0] // n_timestep start_time = 0.0 lines = [] with torch.no_grad(): model.eval() feature = torch.tensor(feature, dtype=torch.float32).unsqueeze(0).to(device) for t in range(num_instance): if (model_type == 'cnn') or (model_type == 'crnn') or (model_type == 'btc'): logits = pre_model(feature[:, n_timestep * t:n_timestep * (t + 1), :], torch.randint(config.model['num_chords'], (n_timestep,)).to(device)) prediction, _ = model(logits, torch.randint(config.model['num_chords'], (n_timestep,)).to(device)) else: raise NotImplementedError for i in range(n_timestep): if t == 0 and i == 0: prev_chord = prediction[i].item() continue if prediction[i].item() != prev_chord: lines.append( '%.6f %.6f %s\n' % ( start_time, time_unit * (n_timestep * t + i), idx2chord[prev_chord])) start_time = time_unit * (n_timestep * t + i) prev_chord = prediction[i].item() if t == num_instance - 1 and i + num_pad == n_timestep: if start_time != time_unit * (n_timestep * t + i): lines.append( '%.6f %.6f %s\n' % ( start_time, time_unit * (n_timestep * t + i), idx2chord[prev_chord])) break pid = os.getpid() tmp_path = 'tmp_' + str(pid) + '.lab' with open(tmp_path, 'w') as f: for line in lines: f.write(line) root_majmin = ['root', 'majmin'] for m in root_majmin: metrics_.score_list_dict[m].append(metrics_.score(metric=m, gt_path=lab_file_path, est_path=tmp_path)) song_length_list.append(song_length_second) if verbose: for m in root_majmin: print('song name %s, %s score : %.4f' % (song_name, m, metrics_.score_list_dict[m][-1])) except: print('song name %s\' lab file error' % song_name) tmp = song_length_list / np.sum(song_length_list) for m in root_majmin: metrics_.average_score[m] = np.sum(np.multiply(metrics_.score_list_dict[m], tmp)) return metrics_.score_list_dict, song_length_list, metrics_.average_score def large_voca_score_calculation(valid_dataset, config, mean, std, device, model, model_type, verbose=False): idx2voca = idx2voca_chord() valid_song_names = valid_dataset.song_names paths = valid_dataset.preprocessor.get_all_files() metrics_ = metrics() song_length_list = list() for path in paths: song_name, lab_file_path, mp3_file_path, _ = path if not song_name in valid_song_names: continue try: n_timestep = config.model['timestep'] feature, feature_per_second, song_length_second = audio_file_to_features(mp3_file_path, config) feature = feature.T feature = (feature - mean) / std time_unit = feature_per_second num_pad = n_timestep - (feature.shape[0] % n_timestep) feature = np.pad(feature, ((0, num_pad), (0, 0)), mode="constant", constant_values=0) num_instance = feature.shape[0] // n_timestep start_time = 0.0 lines = [] with torch.no_grad(): model.eval() feature = torch.tensor(feature, dtype=torch.float32).unsqueeze(0).to(device) for t in range(num_instance): if model_type == 'btc': encoder_output, _ = model.self_attn_layers(feature[:, n_timestep * t:n_timestep * (t + 1), :]) prediction, _ = model.output_layer(encoder_output) prediction = prediction.squeeze() elif model_type == 'cnn' or model_type =='crnn': prediction, _, _, _ = model(feature[:, n_timestep * t:n_timestep * (t + 1), :], torch.randint(config.model['num_chords'], (n_timestep,)).to(device)) for i in range(n_timestep): if t == 0 and i == 0: prev_chord = prediction[i].item() continue if prediction[i].item() != prev_chord: lines.append( '%.6f %.6f %s\n' % ( start_time, time_unit * (n_timestep * t + i), idx2voca[prev_chord])) start_time = time_unit * (n_timestep * t + i) prev_chord = prediction[i].item() if t == num_instance - 1 and i + num_pad == n_timestep: if start_time != time_unit * (n_timestep * t + i): lines.append( '%.6f %.6f %s\n' % ( start_time, time_unit * (n_timestep * t + i), idx2voca[prev_chord])) break pid = os.getpid() tmp_path = 'tmp_' + str(pid) + '.lab' with open(tmp_path, 'w') as f: for line in lines: f.write(line) for m in metrics_.score_metrics: metrics_.score_list_dict[m].append(metrics_.score(metric=m, gt_path=lab_file_path, est_path=tmp_path)) song_length_list.append(song_length_second) if verbose: for m in metrics_.score_metrics: print('song name %s, %s score : %.4f' % (song_name, m, metrics_.score_list_dict[m][-1])) except: print('song name %s\' lab file error' % song_name) tmp = song_length_list / np.sum(song_length_list) for m in metrics_.score_metrics: metrics_.average_score[m] = np.sum(np.multiply(metrics_.score_list_dict[m], tmp)) return metrics_.score_list_dict, song_length_list, metrics_.average_score def large_voca_score_calculation_crf(valid_dataset, config, mean, std, device, pre_model, model, model_type, verbose=False): idx2voca = idx2voca_chord() valid_song_names = valid_dataset.song_names paths = valid_dataset.preprocessor.get_all_files() metrics_ = metrics() song_length_list = list() for path in paths: song_name, lab_file_path, mp3_file_path, _ = path if not song_name in valid_song_names: continue try: n_timestep = config.model['timestep'] feature, feature_per_second, song_length_second = audio_file_to_features(mp3_file_path, config) feature = feature.T feature = (feature - mean) / std time_unit = feature_per_second num_pad = n_timestep - (feature.shape[0] % n_timestep) feature = np.pad(feature, ((0, num_pad), (0, 0)), mode="constant", constant_values=0) num_instance = feature.shape[0] // n_timestep start_time = 0.0 lines = [] with torch.no_grad(): model.eval() feature = torch.tensor(feature, dtype=torch.float32).unsqueeze(0).to(device) for t in range(num_instance): if (model_type == 'cnn') or (model_type == 'crnn') or (model_type == 'btc'): logits = pre_model(feature[:, n_timestep * t:n_timestep * (t + 1), :], torch.randint(config.model['num_chords'], (n_timestep,)).to(device)) prediction, _ = model(logits, torch.randint(config.model['num_chords'], (n_timestep,)).to(device)) else: raise NotImplementedError for i in range(n_timestep): if t == 0 and i == 0: prev_chord = prediction[i].item() continue if prediction[i].item() != prev_chord: lines.append( '%.6f %.6f %s\n' % ( start_time, time_unit * (n_timestep * t + i), idx2voca[prev_chord])) start_time = time_unit * (n_timestep * t + i) prev_chord = prediction[i].item() if t == num_instance - 1 and i + num_pad == n_timestep: if start_time != time_unit * (n_timestep * t + i): lines.append( '%.6f %.6f %s\n' % ( start_time, time_unit * (n_timestep * t + i), idx2voca[prev_chord])) break pid = os.getpid() tmp_path = 'tmp_' + str(pid) + '.lab' with open(tmp_path, 'w') as f: for line in lines: f.write(line) for m in metrics_.score_metrics: metrics_.score_list_dict[m].append(metrics_.score(metric=m, gt_path=lab_file_path, est_path=tmp_path)) song_length_list.append(song_length_second) if verbose: for m in metrics_.score_metrics: print('song name %s, %s score : %.4f' % (song_name, m, metrics_.score_list_dict[m][-1])) except: print('song name %s\' lab file error' % song_name) tmp = song_length_list / np.sum(song_length_list) for m in metrics_.score_metrics: metrics_.average_score[m] = np.sum(np.multiply(metrics_.score_list_dict[m], tmp)) return metrics_.score_list_dict, song_length_list, metrics_.average_score