import tensorflow as tf import numpy as np import matplotlib.pyplot as plt import os import glob import pickle as pkl import datetime import music21 import itertools ################################################################################### # DataLoader ################################################################################### class ScoreDataset(tf.keras.utils.Sequence): def __init__(self, save_path='music_param.pkl', midi_paths=None, seq_len=32): if not midi_paths is None: self._build(midi_paths, seq_len) self.save(save_path) else: self.load(save_path) def _build(self, midi_paths, seq_len=32): self.seq_len = seq_len self.notes_list, self.durations_list = ScoreDataset.makeMusicData(midi_paths) notes_set = sorted(set(itertools.chain.from_iterable(self.notes_list))) # flatten 2D -> 1D, Unique, Sort durations_set = sorted(set(itertools.chain.from_iterable(self.durations_list))) self.note_to_index, self.index_to_note = ScoreDataset.createLookups(notes_set) self.duration_to_index, self.index_to_duration = ScoreDataset.createLookups(durations_set) self.c_notes = len(self.note_to_index) self.c_durations = len(self.duration_to_index) self.notes_index_list = ScoreDataset.convertIndex(self.notes_list, self.note_to_index) self.durations_index_list = ScoreDataset.convertIndex(self.durations_list, self.duration_to_index) self.n_music = len(self.notes_list) self.index = 0 self._build_tbl() @staticmethod def extractMidi(midi_path): notes, durations = [], [] score = music21.converter.parse(midi_path).chordify() for element in score.flat: if isinstance(element, music21.note.Note): # Note if element.isRest: # Rest notes.append(str(element.name)) # no pitch, then name only else: # note with pitch notes.append(str(element.nameWithOctave)) durations.append(element.duration.quaterLength) # 1/4 unit if isinstance(element, music21.chord.Chord): # chord contains multiple notes notes.append('.'.join(n.nameWithOctave for n in element.pitches)) # connect with '.' durations.append(element.duration.quarterLength) # 1/4 unit return notes, durations @staticmethod # [notes1, ..., notesN], [durations1, ..., durationsN] def makeMusicData(midi_paths): notes_list, durations_list = [], [] for path in midi_paths: notes, durations = ScoreDataset.extractMidi(path) # notes, durations notes_list.append(notes) durations_list.append(durations) return notes_list, durations_list @staticmethod def createLookups(names): # Lookup Table element_to_index = dict((element, idx) for idx, element in enumerate(names)) index_to_element = dict((idx, element) for idx, element in enumerate(names)) return element_to_index, index_to_element @staticmethod def convertIndex(data, element_to_index): return [ [ element_to_index[element] for element in x] for x in data] def getMidiStream(self, g_notes, g_durations): # [note_index, ...], [duration_index, ...] midi_stream = music21.stream.Stream() for note_idx, duration_idx in zip(g_notes, g_durations): note = self.index_to_note[note_idx] duration = self.index_to_duration[duration_idx] if ('.' in note): # chord notes_in_chord = note.split('.') chord_notes = [] for n_ in notes_in_chord: new_note = music21.note.Note(n_) new_note.duration = music21.duration.Duration(duration) new_note.storeInstrument = music21.instrument.Violoncello() chord_notes.append(new_note) new_chord = music21.chord.Chord(chord_notes) midi_stream.append(new_chord) elif note == 'rest': new_note = music21.note.Rest() new_note.duration = music21.duration.Duration(duration) new_note.storedInstrument = music21.instrument.Violoncello() midi_stream.append(new_note) else: new_note = music21.note.Note(note) new_note.duration = music21.duration.Duration(duration) new_note.storedInstrument = music21.instrument.Violoncello() midi_stream.append(new_note) return midi_stream def _build_tbl(self): a = [ len(x)-self.seq_len for x in self.notes_list ] # [int, int, ...] for i in range(1, len(a)): # cumulative frequency of data a[i] = a[i-1] + a[i] self.cumulative_freq = a #print(f'cumulative_freq: {self.cumulative_freq}') def searchTbl(self, index): index = index % self.__len__() low = 0 high = self.n_music - 1 for i in range(self.n_music): mid = (low + high) // 2 #print(f'{i}/{self.n_music}: {high} {low} {mid} {index}') if self.cumulative_freq[mid] > index: if mid == 0 or self.cumulative_freq[mid-1] <= index: return mid high = mid - 1 else: low = mid + 1 def __len__(self): return self.cumulative_freq[-1] def __getitem__(self, index): if isinstance(index, slice): start, stop, step = index.indices(self.__len__()) if start == None: start = 0 if stop == None: stop = self.__len__() if step == None: if start < stop: step = 1 elif start > stop: step = -1 else: step = 0 return self.__getitemList__(range(start, stop, step)) elif isinstance(index, int): return self.__getitemInt__(index) else: return self.__getitemList__(index) def __getitemList__(self, indices): x_notes, x_durations, y_notes, y_durations = [], [], [], [] for i in indices: [x_note, x_duration], [y_note, y_duration] = self.__getitemInt__(i) x_notes.append(x_note) x_durations.append(x_duration) y_notes.append(y_note) y_durations.append(y_duration) return (x_notes, x_durations), (y_notes, y_durations) def __getitemInt__(self, index): index = index % self.__len__() #print(f'index = {index} {self.__len__()}') tbl_idx = self.searchTbl(index) #print(f'tbl_idx = {tbl_idx}') tgt = index if (tbl_idx > 0): tgt -= self.cumulative_freq[tbl_idx - 1] #print(f'tgt = {tgt}') x_note = self.notes_index_list[tbl_idx][tgt: (tgt + self.seq_len)] y_note = self.notes_index_list[tbl_idx][tgt + self.seq_len] x_duration = self.durations_index_list[tbl_idx][tgt: (tgt + self.seq_len)] y_duration = self.durations_index_list[tbl_idx][tgt + self.seq_len] #ohv_y_note = tf.keras.utils.to_categorical(y_note, self.c_notes) #ohv_y_duration = tf.keras.utils.to_categorical(y_duration, self.c_durations) return (x_note, x_duration), (y_note, y_duration) def __next__(self): self.index += 1 return self.__getitem__(self.index-1) def save(self, filepath): dpath, fname = os.path.split(filepath) if not os.path.exists(dpath): os.makedirs(dpath) with open(filepath, 'wb') as f: pkl.dump([ self.seq_len, self.notes_list, self.durations_list, self.note_to_index, self.index_to_note, self.duration_to_index, self.index_to_duration, self.c_notes, self.c_durations, self.notes_index_list, self.durations_index_list, self.n_music, self.index, self.cumulative_freq ], f) def load(self, filepath): with open(filepath, 'rb') as f: params = pkl.load(f) [ self.seq_len, self.notes_list, self.durations_list, self.note_to_index, self.index_to_note, self.duration_to_index, self.index_to_duration, self.c_notes, self.c_durations, self.notes_index_list, self.durations_index_list, self.n_music, self.index, self.cumulative_freq ] = params ################################################################################### # Model ################################################################################### class LSTMMusic(): def __init__(self, c_notes, c_durations, seq_len = 32, optimizer='adam', learning_rate = 0.001, embed_size = 100, rnn_units = 256, use_attention = True, epochs = 0, losses = [], n_losses = [], d_losses = [], val_losses = [], val_n_losses = [], val_d_losses = [] ): self.c_notes = c_notes self.c_durations = c_durations self.seq_len = seq_len self.optimizer = optimizer self.learning_rate = learning_rate self.embed_size = embed_size self.rnn_units = rnn_units self.use_attention = use_attention self.epochs = epochs self.losses = losses self.n_losses = n_losses self.d_losses = d_losses self.val_losses = val_losses self.val_n_losses = val_n_losses self.val_d_losses = val_d_losses self.model, self.att_model = self._create_network(c_notes, c_durations, embed_size, rnn_units, use_attention) self.cce1 = tf.keras.losses.CategoricalCrossentropy(), self.cce2 = tf.keras.losses.CategoricalCrossentropy(), def _create_network( self, n_notes, n_durations, embed_size, rnn_units, use_attention ): notes_in = tf.keras.layers.Input(shape=(None,)) durations_in = tf.keras.layers.Input(shape=(None,)) x1 = tf.keras.layers.Embedding(n_notes, embed_size)(notes_in) x2 = tf.keras.layers.Embedding(n_durations, embed_size)(durations_in) x = tf.keras.layers.Concatenate()([x1, x2]) x = tf.keras.layers.LSTM(rnn_units, return_sequences=True)(x) # x = tf.keras.layers.Dropout(0.2)(x) if use_attention: x = tf.keras.layers.LSTM(rnn_units, return_sequences=True)(x) # x = tf.keras.layers.Dropout(0.2)(x) e = tf.keras.layers.Dense(1, activation='tanh')(x) e = tf.keras.layers.Reshape([-1])(e) # batch_size * N alpha = tf.keras.layers.Activation('softmax')(e) alpha_repeated = tf.keras.layers.Permute([2,1])(tf.keras.layers.RepeatVector(rnn_units)(alpha)) c = tf.keras.layers.Multiply()([x, alpha_repeated]) c = tf.keras.layers.Lambda(lambda xin: tf.keras.backend.sum(xin, axis=1), output_shape=(rnn_units,))(c) else: c = tf.keras.layers.LSTM(rnn_units)(x) #c = tf.keras.layers.Dropout(0.2)(c) notes_out = tf.keras.layers.Dense(n_notes, activation='softmax', name='pitch')(c) durations_out = tf.keras.layers.Dense(n_durations, activation='softmax', name='duration')(c) model = tf.keras.models.Model([notes_in, durations_in], [notes_out, durations_out]) if use_attention: att_model = tf.keras.models.Model([notes_in, durations_in], alpha) else: att_model = None return model, att_model def get_opti(self, learning_rate): if self.optimizer == 'adam': opti = tf.keras.optimizers.Adam(learning_rate=learning_rate, beta_1 = 0.5) elif self.optimizer == 'rmsprop': opti = tf.keras.optimizers.RMSprop(learning_rate=learning_rate) else: opti = tf.keras.optimizers.Adam(learning_rate=learning_rate) return opti def _compile(self): opti = self.get_opti(self.learning_rate) self.model.compile( loss=[ self.cce1, self.cce2 ], optimizer = tf.keras.optimizers.RMSprop(learning_rate = self.learning_rate) ) def train_with_fit( self, xs, # (x_notes,x_durations) ys, # (y_notes, y_durations) epochs=1, batch_size=32, run_folder='run/', shuffle=False, validation_data = None ): history = self.model.fit( xs, ys, initial_epoch = self.epochs, epochs = epochs, batch_size = batch_size, shuffle = shuffle, validation_data = validation_data ) self.epochs = epochs h = history.history self.losses += h['loss'] self.n_losses += h['pitch_loss'] self.d_losses += h['duration_loss'] if not validation_data is None: self.val_losses += h['val_loss'] self.val_n_losses += h['val_pitch_loss'] self.val_d_losses += h['val_duration_loss'] self.save(run_folder) self.save(run_folder, self.epochs) if validation_data is None: return self.losses, self.n_losses, self.d_losses else: return self.losses, self.n_losses, self.d_losses, self.val_losses, self.val_n_losses, self.val_d_losses, @tf.function def loss_fn(self, y_notes, y_durations, p_notes, p_durations): #print(y_notes.shape, p_notes.shape, y_durations.shape, p_durations.shape) #y_notes = np.array(y_notes, dtype='float32') #p_notes = np.array(p_notes, dtype='float32') n_loss = tf.keras.losses.CategoricalCrossentropy()(y_notes, p_notes) d_loss = tf.keras.losses.CategoricalCrossentropy()(y_durations, p_durations) #n_loss = self.cce1(y_notes, p_notes) #d_loss = self.cce2(y_durations, p_durations) loss = tf.add(n_loss, d_loss) return loss, n_loss, d_loss @tf.function def train_step( self, x_notes, x_durations, y_notes, y_durations, optimizer ): with tf.GradientTape() as tape: p_notes, p_durations = self.model([x_notes, x_durations]) loss, note_loss, duration_loss = self.loss_fn(y_notes, y_durations, p_notes, p_durations) variables = self.model.trainable_variables gradients = tape.gradient(loss, variables) optimizer.apply_gradients(zip(gradients, variables)) return loss, note_loss, duration_loss def train( self, xs, ys, epochs=1, batch_size=32, optimizer=tf.keras.optimizers.Adam(learning_rate=0.001), shuffle=False, run_folder='run/', print_step_interval = 100, save_epoch_interval = 100, validation_data = None ): start_time = datetime.datetime.now() x_notes_all, x_durations_all = xs y_notes_all, y_durations_all = ys steps = len(x_notes_all) // batch_size for epoch in range(self.epochs, epochs): indices = tf.range(len(x_notes_all), dtype=tf.int32) if shuffle: indices = tf.random.shuffle(indices) step_losses, step_n_losses, step_d_losses = [], [], [] for step in range(steps): start = batch_size * step end = start + batch_size idxs = indices[start:end] x_notes = x_notes_all[idxs] x_durations = x_durations_all[idxs] y_notes = np.array(y_notes_all[idxs]) y_durations = np.array(y_durations_all[idxs]) n_ = np.array(x_notes, dtype='float32') d_ = np.array(x_durations, dtype='float32') batch_loss, batch_n_loss, batch_d_loss = self.train_step( n_, d_, y_notes, y_durations, optimizer ) step_losses.append(np.mean(batch_loss)) step_n_losses.append(np.mean(batch_n_loss)) step_d_losses.append(np.mean(batch_d_loss)) elapsed_time = datetime.datetime.now() - start_time if (step+1) % print_step_interval == 0: print(f'{epoch+1}/{epochs} {step+1}/{steps} loss {step_losses[-1]:.3f} pitch_loss {step_n_losses[-1]:.3f} duration_loss {step_d_losses[-1]:.3f} {elapsed_time}') epoch_loss = np.mean(step_losses) epoch_n_loss = np.mean(step_n_losses) epoch_d_loss = np.mean(step_d_losses) self.losses.append(epoch_loss) self.n_losses.append(epoch_n_loss) self.d_losses.append(epoch_d_loss) val_str = '' if not validation_data is None: (val_x_notes, val_x_durations), (val_y_notes, val_y_durations) = validation_data p_notes, p_durations = self.model([val_x_notes, val_x_durations]) val_loss, val_n_loss, val_d_loss = self.loss_fn( val_y_notes, val_y_durations, p_notes, p_durations ) val_loss = np.mean(val_loss) val_n_loss = np.mean(val_n_loss) val_d_loss = np.mean(val_d_loss) self.val_losses.append(val_loss) self.val_n_losses.append(val_n_loss) self.val_d_losses.append(val_d_loss) val_str = f'val_loss {val_loss:.3f} val_pitch_loss {val_n_loss:.3f} val_duration_loss {val_d_loss:.3f}' self.epochs += 1 elapsed_time = datetime.datetime.now() - start_time print(f'{self.epochs}/{epochs} loss {epoch_loss:.3f} pitch_loss {epoch_n_loss:.3f} duration_loss {epoch_d_loss:.3f} {val_str} {elapsed_time}') if self.epochs % save_epoch_interval == 0: self.save(run_folder) self.save(run_folder, self.epochs) self.save(run_folder) self.save(run_folder, self.epochs) return self.losses, self.n_losses, self.d_losses, self.val_losses, self.val_n_losses, self.val_d_losses @staticmethod def sample_with_temperature(preds, temperature): if temperature == 0: return np.argmax(preds) else: preds = np.log(preds) / temperature exp_preds = np.exp(preds) preds = exp_preds / np.sum(exp_preds) return np.random.choice(len(preds), p=preds) def generate(self, s_notes, s_durations, count=64, note_temperature = 0.5, duration_temperature = 0.5): x_notes = np.array([s_notes], dtype='float32') x_durations = np.array([s_durations], dtype='float32') g_notes, g_durations = [], [] for i in range(count): p_notes, p_durations = self.model([x_notes, x_durations]) note = LSTMMusic.sample_with_temperature(p_notes[0], note_temperature) duration = LSTMMusic.sample_with_temperature(p_durations[0], duration_temperature) g_notes.append(note) g_durations.append(duration) x_notes = np.roll(x_notes, -1) x_durations = np.roll(x_durations, -1) x_notes[0,-1] = note x_durations[0,-1] = duration return g_notes, g_durations # [note_index, ...], [duration_index, ...] def save(self, folder, epoch=None): self.save_params(folder, epoch) self.save_weights(folder, epoch) @staticmethod def load(folder, epoch=None): params = LSTMMusic.load_params(folder, epoch) music = LSTMMusic(*params) music.load_weights(folder, epoch) return music def save_weights(self, run_folder, epoch=None): if epoch is None: self.save_model_weights(self.model, os.path.join(run_folder, 'weights/weights.h5')) self.save_model_weights(self.att_model, os.path.join(run_folder, 'weights/weights_att.h5')) else: self.save_model_weights(self.model, os.path.join(run_folder, f'weights/weights_{epoch}.h5')) self.save_model_weights(self.att_model, os.path.join(run_folder, f'weights/weights_att_{epoch}.h5')) def load_weights(self, run_folder, epoch=None): if epoch is None: self.load_model_weights(self.model, os.path.join(run_folder, 'weights/weights.h5')) self.load_model_weights(self.att_model, os.path.join(run_folder, 'weights/weights_att.h5')) else: self.load_model_weights(self.model, os.path.join(run_folder, f'weights/weights_{epoch}.h5')) self.load_model_weights(self.att_model, os.path.join(run_folder, f'weights/weights_att_{epoch}.h5')) def save_model_weights(self, model, filepath): dpath, fname = os.path.split(filepath) if dpath != '' and not os.path.exists(dpath): os.makedirs(dpath) model.save_weights(filepath) def load_model_weights(self, model, filepath): model.load_weights(filepath) def save_params(self, folder, epoch=None): if not os.path.exists(folder): os.makedirs(folder) if epoch is None: filepath = os.path.join(folder, 'params.pkl') else: filepath = os.path.join(folder, f'params_{epoch}.pkl') with open(filepath, 'wb') as f: pkl.dump([ self.c_notes, self.c_durations, self.seq_len, self.optimizer, self.learning_rate, self.embed_size, self.rnn_units, self.use_attention, self.epochs, self.losses, self.n_losses, self.d_losses, self.val_losses, self.val_n_losses, self.val_d_losses ], f) @staticmethod def load_params(folder, epoch=None): if epoch is None: filepath = os.path.join(folder, 'params.pkl') else: filepath = os.path.join(folder, f'params_{epoch}.pkl') with open(filepath, 'rb') as f: params = pkl.load(f) return params @staticmethod def plot_history(vals, labels): colors = ['red', 'blue', 'green', 'black', 'orange', 'pink', 'purple', 'olive', 'cyan'] n = len(vals) fig, ax = plt.subplots(1, 1, figsize=(12,6)) for i in range(n): ax.plot(vals[i], c=colors[i], label=labels[i]) ax.legend(loc='upper right') ax.set_xlabel('epochs') plt.show()