rnn_model_1.py

# -*- coding: utf-8 -*-
"""RNN Model 1.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1jTI_o_F1qAl2EsKcNj5shZc4v86Uubzl

# Preprocessing
"""

# Commented out IPython magic to ensure Python compatibility.
from google.colab import drive
drive.mount("/content/gdrive")
# %cd gdrive/Shareddrives/COS\ 401
# %cd dataset

import csv
import numpy
import os
import pandas as pd

import math
note_order = {'C0': 0, 'C#' : 1, 'Db': 1, 'C2': 2, 'D0': 2, 'D#': 3,
              'Eb': 3, 'E0': 4, 'E#': 5, 'Fb': 4, 'F0': 5, 'F#': 6,
              'Gb': 6, 'G0': 7, 'G#': 8, 'Ab': 8, 'A0': 9,
              'A#': 10, 'Bb': 10, 'B0': 11, 'B#': 0, 'Cb': 11, "rest": math.inf,
              'F2': 7, 'C2': 2, 'D2': 4, 'G2': 9, 'A2': 11, 'B-2': 9, 'B2': 1,
              'E2': 6, 'A-2': 7, 'C-2': 10, 'D-2': 0, 'E-2': 2, 'F-2': 3, "[]": -1}

def get_note_from_name(note_name):
  return note_order[note_name]


def get_standardized_note(note_name, key_fifths):
  if note_name == "rest":
    return -1
  note = get_note_from_name(note_name)
  return (note - 7*int(key_fifths) % 12) % 12

# standardize notes and chords in all train files

standardized_pds = []
for song in os.listdir("csv_train"):
  file_pd = pd.read_csv("csv_train/" + str(song))
  new_notes = []
  # print(song)

  for i in range(len(file_pd["note_root"])):
    new_notes.append(get_standardized_note(file_pd["note_root"][i], file_pd["key_fifths"][i]))

  file_pd["standardized_note"] = new_notes

  new_chords = []

  for i in range(len(file_pd["chord_root"])):
    new_chords.append(get_standardized_note(file_pd["chord_root"][i], file_pd["key_fifths"][i]))

  file_pd["standardized_chord"] = new_chords

  standardized_pds.append(file_pd)

# (note, chord, chord_type, octave)

import string
note_dataset = []
import random


for file_pd in standardized_pds:
  # note_dataset.append("13 major 13 13")
  notes_in_song = list(zip(file_pd["standardized_chord"], file_pd["chord_type"], file_pd["standardized_note"], file_pd["note_octave"]))
  for note in notes_in_song:
    # note_dataset.append([str(i) for i in note])
    
    final_note = ""
    for i in note:
      final_note += " " + str(i)
    note_dataset.append(final_note)

# import string

print(note_dataset[5])

from collections import Counter

a = Counter(note_dataset)

a["13 major 13 13"]

a.most_common

import tensorflow as tf

import numpy as np
import os
import time

vocab1 = set(note_dataset)
print(f'{len(vocab1)} unique characters')

"""# Process as Text

Much of the code for this was taken or adapted from:

https://www.tensorflow.org/text/tutorials/text_generation 

Since this was our first pass, we just used a standard language model using an RNN.
"""

ids_from_chars1 = tf.keras.layers.StringLookup(
    vocabulary=list(vocab1), mask_token=None)

ids_from_chars1(tf.convert_to_tensor(("13 major 13 13")))

chars_from_ids1 = tf.keras.layers.StringLookup(
    vocabulary=ids_from_chars1.get_vocabulary(), invert=True, mask_token=None)

chars1 = chars_from_ids1(1)
chars1

all_ids1 = ids_from_chars1(note_dataset)

ids_dataset1 = tf.data.Dataset.from_tensor_slices(all_ids1)

for ids in ids_dataset1.take(10):
    print(chars_from_ids1(ids).numpy().decode('utf-8'))

seq_length = 100
examples_per_epoch = len(note_dataset)//(seq_length+1)

"""The `batch` method lets you easily convert these individual characters to sequences of the desired size."""

sequences = ids_dataset1.batch(seq_length+1, drop_remainder=True)

for seq in sequences.take(1):
  print(chars_from_ids1(seq))

def split_input_target(sequence):
    input_text = sequence[:-1]
    target_text = sequence[1:]
    return input_text, target_text

dataset = sequences.map(split_input_target)

for input_example, target_example in dataset.take(1):
    print("Input :", chars_from_ids1(input_example).numpy())
    print("Target:", chars_from_ids1(target_example).numpy())

# Batch size
BATCH_SIZE = 64

# Buffer size to shuffle the dataset
# (TF data is designed to work with possibly infinite sequences,
# so it doesn't attempt to shuffle the entire sequence in memory. Instead,
# it maintains a buffer in which it shuffles elements).
BUFFER_SIZE = 10000

dataset = (
    dataset
    .shuffle(BUFFER_SIZE)
    .batch(BATCH_SIZE, drop_remainder=True)
    .prefetch(tf.data.experimental.AUTOTUNE))

dataset

# Length of the vocabulary in chars
vocab_size = len(vocab1)

# The embedding dimension
embedding_dim = 256

# Number of RNN units
rnn_units = 1024

class MyModel(tf.keras.Model):
  def __init__(self, vocab_size, embedding_dim, rnn_units):
    super().__init__(self)
    self.embedding = tf.keras.layers.Embedding(vocab_size, embedding_dim)
    self.gru = tf.keras.layers.GRU(rnn_units,
                                   return_sequences=True,
                                   return_state=True)
    self.dense = tf.keras.layers.Dense(vocab_size)

  def call(self, inputs, states=None, return_state=False, training=False):
    x = inputs
    x = self.embedding(x, training=training)
    if states is None:
      states = self.gru.get_initial_state(x)
    x, states = self.gru(x, initial_state=states, training=training)
    x = self.dense(x, training=training)

    if return_state:
      return x, states
    else:
      return x

model = MyModel(
    vocab_size=len(ids_from_chars1.get_vocabulary()),
    embedding_dim=embedding_dim,
    rnn_units=rnn_units)

for input_example_batch, target_example_batch in dataset.take(1):
    example_batch_predictions = model(input_example_batch)
    print(example_batch_predictions.shape, "# (batch_size, sequence_length, vocab_size)")

model.summary()

sampled_indices = tf.random.categorical(example_batch_predictions[0], num_samples=1)
sampled_indices = tf.squeeze(sampled_indices, axis=-1).numpy()

sampled_indices

print("Input:\n", chars_from_ids1(input_example_batch[0]).numpy())
print()
print("Next Char Predictions:\n", chars_from_ids1(sampled_indices).numpy())

loss = tf.losses.SparseCategoricalCrossentropy(from_logits=True)

example_batch_mean_loss = loss(target_example_batch, example_batch_predictions)
print("Prediction shape: ", example_batch_predictions.shape, " # (batch_size, sequence_length, vocab_size)")
print("Mean loss:        ", example_batch_mean_loss)

tf.exp(example_batch_mean_loss).numpy()

model.compile(optimizer='adam', loss=loss)

# Directory where the checkpoints will be saved
checkpoint_dir = './training_checkpoints'
# Name of the checkpoint files
checkpoint_prefix = os.path.join(checkpoint_dir, "ckpt_{epoch}")

checkpoint_callback = tf.keras.callbacks.ModelCheckpoint(
    filepath=checkpoint_prefix,
    save_weights_only=True)

EPOCHS = 30

history = model.fit(dataset, epochs=EPOCHS, callbacks=[checkpoint_callback])

class OneStep(tf.keras.Model):
  def __init__(self, model, chars_from_ids, ids_from_chars, temperature=1.0):
    super().__init__()
    self.temperature = temperature
    self.model = model
    self.chars_from_ids = chars_from_ids
    self.ids_from_chars = ids_from_chars

    # Create a mask to prevent "[UNK]" from being generated.
    skip_ids = self.ids_from_chars(['[UNK]'])[:, None]
    sparse_mask = tf.SparseTensor(
        # Put a -inf at each bad index.
        values=[-float('inf')]*len(skip_ids),
        indices=skip_ids,
        # Match the shape to the vocabulary
        dense_shape=[len(ids_from_chars.get_vocabulary())])
    self.prediction_mask = tf.sparse.to_dense(sparse_mask)

  @tf.function
  def generate_one_step(self, inputs, states=None):
    # Convert strings to token IDs.
    input_chars = tf.strings.unicode_split(inputs, 'UTF-8')
    input_ids = self.ids_from_chars(input_chars).to_tensor()

    # Run the model.
    # predicted_logits.shape is [batch, char, next_char_logits]
    predicted_logits, states = self.model(inputs=input_ids, states=states,
                                          return_state=True)
    # Only use the last prediction.
    predicted_logits = predicted_logits[:, -1, :]
    predicted_logits = predicted_logits/self.temperature
    # Apply the prediction mask: prevent "[UNK]" from being generated.
    predicted_logits = predicted_logits + self.prediction_mask

    # Sample the output logits to generate token IDs.
    predicted_ids = tf.random.categorical(predicted_logits, num_samples=1)
    predicted_ids = tf.squeeze(predicted_ids, axis=-1)

    # Convert from token ids to characters
    predicted_chars = self.chars_from_ids(predicted_ids)

    # Return the characters and model state.
    return predicted_chars, states

one_step_model = OneStep(model, chars_from_ids1, ids_from_chars1, temperature = 0.4)

"""Run it in a loop to generate some text. Looking at the generated text, you'll see the model knows when to capitalize, make paragraphs and imitates a Shakespeare-like writing vocabulary. With the small number of training epochs, it has not yet learned to form coherent sentences."""

start = time.time()
states = None
next_char = tf.constant(['0 major 0 5'])
result = [next_char]

for n in range(40):
  next_char, states = one_step_model.generate_one_step(next_char, states=states)
  result.append(next_char)

result1 = tf.strings.join(result)
end = time.time()
print(result1[0].numpy().decode('utf-8'), '\n\n' + '_'*80)
print('\nRun time:', end - start)

"""# Play the Music"""

# result which is a list
# split each string in result
# first is chord, chordtype, then note then octave

chords = []
chord_types = []
notes = []
octaves = []
measures = []
duration = []
count = 0
measure = 0
for note_tf in result:
  note_str = note_tf[0].numpy().decode('utf-8')
  splitup = note_str.split(" ")
  if splitup[0] in ["", " "]:
    splitup.pop(0)
  chords.append(int(splitup[0]))
  chord_types.append(splitup[1])
  notes.append(int(splitup[2]))
  octaves.append(int(splitup[3]))
  duration.append(4)
  if count % 4 == 0:
    measure += 1
  count += 1
  measures.append(measure)



new_pd = pd.DataFrame(list(zip(chords, chord_types, notes, octaves, duration, measures)),
                      columns = ["standardized_chord", "chord_type", "standardized_note", "note_octave", 
                      "note_duration", "measure"])

new_pd.head()

new_pd

"""We used Google Magenta for music playback (not for learning). The installation and import statements were taken from:

https://colab.research.google.com/notebooks/magenta/hello_magenta/hello_magenta.ipynb

"""

#at test {"output": "ignore"}
print('Installing dependencies...')

!apt-get update -qq && apt-get install -qq libfluidsynth1 fluid-soundfont-gm build-essential libasound2-dev libjack-dev
!pip install -qU pyfluidsynth pretty_midi

!pip install -qU magenta

# Hack to allow python to pick up the newly-installed fluidsynth lib. 
# This is only needed for the hosted Colab environment.
import ctypes.util
orig_ctypes_util_find_library = ctypes.util.find_library
def proxy_find_library(lib):
  if lib == 'fluidsynth':
    return 'libfluidsynth.so.1'
  else:
    return orig_ctypes_util_find_library(lib)
ctypes.util.find_library = proxy_find_library

print('Importing libraries and defining some helper functions...')
from google.colab import files

import magenta
import note_seq
import tensorflow

print('🎉 Done!')
print(magenta.__version__)
print(tensorflow.__version__)

# from google.colab import drive
# drive.mount("/content/gdrive/")
# %cd gdrive/Shareddrives/COS\ 401
# %cd dataset

from google.colab import files

import magenta
import note_seq
import tensorflow

from note_seq.protobuf import music_pb2

CHORDS_KEY = {'major': [4, 7], 'minor':[3, 7], 'dominant':[4, 7, 10],
          'major-seventh': [4, 7, 11], 'minor-seventh': [3, 7, 9],
          'major-sixth':[4, 7, 9], 'minor-sixth':[3, 7, 9], 
          'major-ninth':[4, 7, 10, 14], 'minor-ninth': [3, 7, 9, 14],
          'power': [5], 'half-diminished': [3, 5], 
          'suspended_fourth': [5, 7], 'dominant-13th': [3, 5, 7, 9],
          'dominant-ninth': [4, 7, 9, 14], 'nan': [], '[]': []}

DURATION_SCALE = 0.1
def play_song(file_pd):
  song_song = music_pb2.NoteSequence()
  current_measure = 1
  current_chord_duration = 0
  current_time = 0
  for i in range(len(file_pd["standardized_note"])):
    if file_pd["standardized_note"][i] == 13:
      continue
    # add chord
    if file_pd["measure"][i] > current_measure:
      current_measure += 1
      chord_duration = current_chord_duration
      current_chord_duration = 0
      chord_start_pitch = file_pd["standardized_chord"][i-1] + 12*4
      chord_pitches = [chord_start_pitch]
      if file_pd["chord_type"][i-1] in CHORDS_KEY.keys():
        for j in CHORDS_KEY[file_pd["chord_type"][i-1]]:
          chord_pitches.append(chord_start_pitch + j)
      else:
        print("chord not registered yet")
        print(file_pd["chord_type"][i-1])
      for pitch1 in chord_pitches:
        song_song.notes.add(pitch=pitch1, start_time= current_time - chord_duration, end_time = current_time - 0.001, velocity=60)
      
    # add note
    pitch = file_pd["standardized_note"][i] + 12*(file_pd["note_octave"][i] + 1)
    note_duration = file_pd["note_duration"][i] * DURATION_SCALE
    current_time += note_duration
    current_chord_duration += note_duration
    if file_pd["standardized_note"][i] != -1:
      song_song.notes.add(pitch=pitch, start_time= current_time - note_duration, end_time = current_time, velocity=60, instrument = 2, program = 5)
      pass
  
  song_song.total_time = current_time
  song_song.tempos.add(qpm=60);

  note_seq.plot_sequence(song_song)
  # This is a colab utility method that plays a NoteSequence.
  note_seq.play_sequence(song_song,synth=note_seq.fluidsynth)

play_song(new_pd)