<jupyter_start><jupyter_text>Generating Music with GANs  Introduction to Google ColabGoogle Colab is an online Jupyter notebook environment for machine learining education and research. With it, we can easily prototype and build machine learning models on the cloud. Google Colab provides a GPU or a TPU for faster training (compared to a CPU). See [official tutorial notebook](https://colab.research.google.com) and [official introductory video](https://www.youtube.com/watch?v=inN8seMm7UI) for more details.  Prerequisites<jupyter_code># Install packages
%%capture
!pip3 install torch torchvision pypianoroll matplotlib tqdm livelossplot termcolor
# Import packages
%%capture
from IPython.display import clear_output
from ipywidgets import interact, IntSlider
import os
import os.path
import random
import numpy as np
from matplotlib import pyplot as plt
import torch
import torchvision
import pypianoroll
from pypianoroll import Multitrack, Track
from tqdm import tqdm
from termcolor import colored
from livelossplot import PlotLosses
# Check GPU device
if not torch.cuda.is_available():
    print(colored('[Warnings]', 'red'), 'GPU device not found.',
          'Please go to "Edit >> Notebook settings >> Hardware accelerator"',
          'and choose "GPU".')<jupyter_output><empty_output><jupyter_text>Settings<jupyter_code># Hyperparameters
batch_size = 32
latent_dim = 128
n_steps = 50000

# Data
n_tracks = 5  # number of tracks
n_pitches = 72  # number of pitches
lowest_pitch = 24  # MIDI note number of the lowest pitch
n_samples_per_song = 8  # number of samples to extract from each song in the datset
n_measures = 4  # number of measures per sample
beat_resolution = 4  # temporal resolution of a beat (in timestep)
programs = [0, 0, 25, 33, 48]  # program number for each track
is_drums = [True, False, False, False, False]  # drum indicator for each track
track_names = ['Drums', 'Piano', 'Guitar', 'Bass', 'Strings']  # name of each track
tempo = 100

# Sampler settings
sample_interval = 500  # interval to run the sampler (in step)
n_samples = 4
measure_resolution = 4 * beat_resolution
if 24 % beat_resolution > 0:
    raise ValueError("beat_resolution must be a factor of 24 (the beat "
                     "resolution used in the source dataset)")
if len(programs) != len(is_drums) or len(programs) != len(track_names):
    raise ValueError("Lengths of programs, is_drums and track_names must be "
                     "the same")<jupyter_output><empty_output><jupyter_text>Download the Lakh Pianoroll Dataset<jupyter_code>%%bash
download () {
  confirm=$(wget --quiet --save-cookies /tmp/cookies.txt \
    --keep-session-cookies --no-check-certificate \
    "https://docs.google.com/uc?export=download&id=$2" -O- | sed -rn \
    's/.*confirm=([0-9A-Za-z_]+).*/\1\n/p')
  wget -q -O "$1" --load-cookies /tmp/cookies.txt \
    "https://docs.google.com/uc?export=download&confirm=$confirm&id=$2"
  rm -rf /tmp/cookies.txt
}
mkdir -p "data"
if [ ! -d "data/lpd_5/lpd_5_cleansed/" ]
then
  download "data/lpd_5_cleansed.tar.gz" "1XJ648WDMjRilbhs4hE3m099ZQIrJLvUB"
  tar zxf "data/lpd_5_cleansed.tar.gz" -C "data/"
fi
if [ ! -d "data/amg/" ]
then
  wget -q -O "data/id_lists_amg.tar.gz" \
    "https://drive.google.com/uc?id=1Rv1uAAkcebzYnmYdZYeaBkdK6Q2YtF7c&export=download"
  tar zxf "data/id_lists_amg.tar.gz" -C "data/"
fi
if [ ! -d "data/id_lists_lastfm/" ]
then
  wget -q -O "data/id_lists_lastfm.tar.gz" \
    "https://drive.google.com/uc?id=1mkmQjOifUHISszMsVTDHNuxd7H8Fo5Cs&export=download"
  tar zxf "data/id_lists_lastfm.tar.gz" -C "data/"
fi
dataset_root = "data/lpd_5/lpd_5_cleansed/"
id_list = []
for path in os.listdir("data/amg"):
    filepath = os.path.join("data/amg", path)
    if os.path.isfile(filepath):
        with open(filepath) as f:
            id_list.extend([line.rstrip() for line in f])
id_list = list(set(id_list))
def msd_id_to_dirs(msd_id):
    """Given an MSD ID, generate the path prefix.
    E.g. TRABCD12345678 -> A/B/C/TRABCD12345678"""
    return os.path.join(msd_id[2], msd_id[3], msd_id[4], msd_id)<jupyter_output><empty_output><jupyter_text>Pianoroll visualization<jupyter_code>song_dir = dataset_root + msd_id_to_dirs('TREVDFX128E07859E0') # 'TRQAOWZ128F93000A4', 'TREVDFX128E07859E0'
multitrack = Multitrack(song_dir + '/' + os.listdir(song_dir)[0])
fig, axs = multitrack[:, :12 * 96].plot()
fig.set_size_inches((16, 10))
for ax in axs:
    for x in range(96, 12 * 96, 96):     
        ax.axvline(x - 0.5, color='k', linestyle='-', linewidth=1)
plt.show()<jupyter_output><empty_output><jupyter_text>Collect training data<jupyter_code>data = []
# Iterate over all the songs in the ID list
for msd_id in id_list:
    # Load the multitrack as a pypianoroll.Multitrack instance
    song_dir = dataset_root + msd_id_to_dirs(msd_id)
    multitrack = Multitrack(song_dir + '/' + os.listdir(song_dir)[0])
    # Binarize the pianorolls
    multitrack.binarize()
    # Downsample the pianorolls (shape: n_timesteps x n_pitches)
    multitrack.downsample(24 // beat_resolution)
    # Get the stacked pianoroll (shape: n_timesteps x n_pitches x n_tracks)
    pianoroll = multitrack.get_stacked_pianoroll()
    # Get the target pitch range only
    pianoroll = pianoroll[:, lowest_pitch:lowest_pitch + n_pitches]
    # Calculate the total measures
    n_total_measures = multitrack.get_max_length() // measure_resolution
    candidate = n_total_measures - n_measures
    target_n_samples = min(n_total_measures // n_measures, n_samples_per_song)
    # Randomly select a number of phrases from the multitrack pianoroll
    for idx in np.random.choice(candidate, target_n_samples, False):
        start = idx * measure_resolution
        end = (idx + n_measures) * measure_resolution
        # Skip the samples where some track(s) has too few notes
        if (pianoroll.sum((0, 1)) < 10).any():
            continue
        data.append(pianoroll[start:end])
# Stack all the collected pianoroll segments into one big array
random.shuffle(data)
data = np.stack(data)
print("Successfully collect {} samples from {} songs".format(len(data), len(id_list)))
tracks = []
for idx, (program, is_drum) in enumerate(zip(programs, is_drums)):
    pianoroll = (np.pad(np.concatenate(data[:4])[..., idx], ((0, 0), (lowest_pitch, 128 - lowest_pitch - n_pitches)), 'constant'))
    tracks.append(Track(pianoroll, program, is_drum))
m = Multitrack(tracks=tracks, tempo=tempo, beat_resolution=beat_resolution)
fig, axs = m.plot()
fig.set_size_inches((16, 10))
for ax in axs:
    for x in range(measure_resolution, 4 * 4 * measure_resolution, measure_resolution):
        if x % (measure_resolution * 4) == 0:
            ax.axvline(x - 0.5, color='k')
        else:
            ax.axvline(x - 0.5, color='k', linestyle='-', linewidth=1)
plt.show()<jupyter_output><empty_output><jupyter_text>Training data loader function<jupyter_code>class MultitrackPianorollDataset(torch.utils.data.Dataset):
    """Multitrack pianoroll dataset."""
    def __init__(self, data, transform=None):
        self.data = data
        self.transform = transform

    def __len__(self):
        return len(self.data)

    def __getitem__(self, idx):
        if torch.is_tensor(idx):
            idx = idx.tolist()
        pianoroll = self.data[idx].astype(np.float32)
        if self.transform:
            pianoroll = self.transform(pianoroll)
        return pianoroll

class RandomTranspose():
    def __call__(self, pianoroll):
        shift = np.random.randint(12)
        return pianoroll[:, shift:n_pitches + shift]
def get_data_loader():
    """Download the MNIST dataset and create a data loader."""
    # Use both traning and test sets
    transform = torchvision.transforms.Compose([
        # RandomTranspose(),
        torchvision.transforms.ToTensor()])
    dataset = MultitrackPianorollDataset(data, transform=transform)
    data_loader = torch.utils.data.DataLoader(dataset, batch_size=batch_size,
                                              drop_last=True, shuffle=True)
    return data_loader<jupyter_output><empty_output><jupyter_text>Neural networks<jupyter_code>class LayerNorm(torch.nn.Module):
    """An implementation of Layer normalization that does not require size
    information. Copied from https://github.com/pytorch/pytorch/issues/1959."""
    def __init__(self, n_features, eps=1e-5, affine=True):
        super().__init__()
        self.n_features = n_features
        self.affine = affine
        self.eps = eps
        if self.affine:
            self.gamma = torch.nn.Parameter(torch.Tensor(n_features).uniform_())
            self.beta = torch.nn.Parameter(torch.zeros(n_features))

    def forward(self, x):
        shape = [-1] + [1] * (x.dim() - 1)
        mean = x.view(x.size(0), -1).mean(1).view(*shape)
        std = x.view(x.size(0), -1).std(1).view(*shape)
        y = (x - mean) / (std + self.eps)
        if self.affine:
            shape = [1, -1] + [1] * (x.dim() - 2)
            y = self.gamma.view(*shape) * y + self.beta.view(*shape)
        return y
class Generator(torch.nn.Module):
    """A convolutional neural network (CNN) based generator. The generator takes
    as input a latent vector and outputs a fake sample."""
    def __init__(self):
        super().__init__()
        self.transconv0 = torch.nn.Sequential(
            torch.nn.ConvTranspose3d(latent_dim, 512, (4, 1, 1), (4, 1, 1)),
            torch.nn.BatchNorm3d(512),
            torch.nn.ReLU())
        self.transconv1 = torch.nn.Sequential(
            torch.nn.ConvTranspose3d(512, 256, (1, 4, 1), (1, 4, 1)),
            torch.nn.BatchNorm3d(256),
            torch.nn.ReLU())
        self.transconv2 = torch.nn.Sequential(
            torch.nn.ConvTranspose3d(256, 128, (1, 1, 4), (1, 1, 4)),
            torch.nn.BatchNorm3d(128),
            torch.nn.ReLU())
        self.transconv3 = torch.nn.Sequential(
            torch.nn.ConvTranspose3d(128, 64, (1, 1, 3), (1, 1, 1)),
            torch.nn.BatchNorm3d(64),
            torch.nn.ReLU())
        self.transconv4 = torch.nn.ModuleList([
            torch.nn.Sequential(
                torch.nn.ConvTranspose3d(64, 16, (1, 4, 1), (1, 4, 1)),
                torch.nn.BatchNorm3d(16),
                torch.nn.Sigmoid()) for _ in range(n_tracks)])
        self.transconv5 = torch.nn.ModuleList([
            torch.nn.Sequential(
                torch.nn.ConvTranspose3d(16, 1, (1, 1, 12), (1, 1, 12)),
                torch.nn.BatchNorm3d(1),
                torch.nn.Sigmoid()) for _ in range(n_tracks)])

    def forward(self, x):
        """
        Input:
        Output:
        """
        x = x.view(-1, latent_dim, 1, 1, 1)
        x = self.transconv0(x)
        x = self.transconv1(x)
        x = self.transconv2(x)
        x = self.transconv3(x)
        x = [transconv(x) for transconv in self.transconv4]
        x = torch.cat([transconv(x_) for x_, transconv in zip(x, self.transconv5)], 1)
        x = x.view(-1, n_tracks, n_measures * measure_resolution, n_pitches)
        return x
class Discriminator(torch.nn.Module):
    """A convolutional neural network (CNN) based discriminator. The
    discriminator takes as input either a real sample (in the training data) or
    a fake sample (generated by the generator) and outputs a scalar indicating
    its authentity.
    """
    def __init__(self):
        super().__init__()
        self.conv0 = torch.nn.ModuleList([
            torch.nn.Sequential(
                torch.nn.Conv3d(1, 16, (1, 1, 12), (1, 1, 12)),
                LayerNorm(16),
                torch.nn.LeakyReLU()) for _ in range(n_tracks)])
        self.conv1 = torch.nn.ModuleList([
            torch.nn.Sequential(
                torch.nn.Conv3d(16, 16, (1, 4, 1), (1, 4, 1)),
                LayerNorm(16),
                torch.nn.LeakyReLU()) for _ in range(n_tracks)])
        self.conv2 = torch.nn.Sequential(
            torch.nn.Conv3d(16 * 5, 64, (1, 1, 3), (1, 1, 1)),
            LayerNorm(64),
            torch.nn.LeakyReLU())
        self.conv3 = torch.nn.Sequential(
            torch.nn.Conv3d(64, 128, (1, 1, 4), (1, 1, 4)),
            LayerNorm(128),
            torch.nn.LeakyReLU())
        self.conv4 = torch.nn.Sequential(
            torch.nn.Conv3d(128, 128, (1, 4, 1), (1, 4, 1)),
            LayerNorm(128),
            torch.nn.LeakyReLU())
        self.conv5 = torch.nn.Sequential(
            torch.nn.Conv3d(128, 256, (2, 1, 1), (1, 1, 1)),
            LayerNorm(256),
            torch.nn.LeakyReLU())
        self.conv6 = torch.nn.Sequential(
            torch.nn.Conv3d(256, 512, (3, 1, 1), (3, 1, 1)),
            LayerNorm(512),
            torch.nn.LeakyReLU())
        self.dense = torch.nn.Linear(512, 1)

    def forward(self, x):
        x = x.view(-1, n_tracks, n_measures, measure_resolution, n_pitches)
        x = [conv(x[:, [i]]) for i, conv in enumerate(self.conv0)]
        x = torch.cat([conv(x_) for x_, conv in zip(x, self.conv1)], 1)
        x = self.conv2(x)
        x = self.conv3(x)          
        x = self.conv4(x)
        x = self.conv5(x)
        x = self.conv6(x)
        x = x.view(-1, 512)
        x = self.dense(x)
        return x<jupyter_output><empty_output><jupyter_text>Training function<jupyter_code>def compute_gradient_penalty(discriminator, real_samples, fake_samples):
    """Compute the gradient penalty for regularization. Intuitively, the
    gradient penalty help stablize the magnitude of the gradients that the
    discriminator provides to the generator, and thus help stablize the training
    of the generator."""
    # Get random interpolations between real and fake samples
    alpha = torch.rand(real_samples.size(0), 1, 1, 1).cuda()
    interpolates = (alpha * real_samples + ((1 - alpha) * fake_samples))
    interpolates = interpolates.requires_grad_(True)
    # Get the discriminator output for the interpolations
    d_interpolates = discriminator(interpolates)
    # Get gradients w.r.t. the interpolations
    fake = torch.ones(real_samples.size(0), 1).cuda()
    gradients = torch.autograd.grad(outputs=d_interpolates, inputs=interpolates,
                                    grad_outputs=fake, create_graph=True,
                                    retain_graph=True, only_inputs=True)[0]
    # Compute gradient penalty
    gradients = gradients.view(gradients.size(0), -1)
    gradient_penalty = ((gradients.norm(2, dim=1) - 1) ** 2).mean()
    return gradient_penalty
def train_one_step(d_optimizer, g_optimizer, real_samples):
    """Train the networks for one step."""
    # Sample from the lantent distribution
    latent = torch.randn(batch_size, latent_dim)

    # Transfer data to GPU
    if torch.cuda.is_available():
        real_samples = real_samples.cuda()
        latent = latent.cuda()
    
    # === Train the discriminator ===
    # Reset cached gradients to zero
    discriminator.zero_grad()
    # Get discriminator outputs for the real samples
    prediction_real = discriminator(real_samples)
    # Compute the loss function
    # d_loss_real = torch.mean(torch.nn.functional.relu(1. - prediction_real))
    d_loss_real = -torch.mean(prediction_real)
    # Backpropagate the gradients
    d_loss_real.backward()
    
    # Generate fake samples with the generator
    fake_samples = generator(latent)
    # Get discriminator outputs for the fake samples
    prediction_fake_d = discriminator(fake_samples.detach())
    # Compute the loss function
    # d_loss_fake = torch.mean(torch.nn.functional.relu(1. + prediction_fake_d))
    d_loss_fake = torch.mean(prediction_fake_d)
    # Backpropagate the gradients
    d_loss_fake.backward()

    # Compute gradient penalty
    gradient_penalty = 10.0 * compute_gradient_penalty(
        discriminator, real_samples.data, fake_samples.data)
    # Backpropagate the gradients
    gradient_penalty.backward()

    # Update the weights
    d_optimizer.step()
    
    # === Train the generator ===
    # Reset cached gradients to zero
    generator.zero_grad()
    # Get discriminator outputs for the fake samples
    prediction_fake_g = discriminator(fake_samples)
    # Compute the loss function
    g_loss = -torch.mean(prediction_fake_g)
    # Backpropagate the gradients
    g_loss.backward()
    # Update the weights
    g_optimizer.step()

    return d_loss_real + d_loss_fake, g_loss<jupyter_output><empty_output><jupyter_text>Training preparation<jupyter_code># Create data loader
data_loader = get_data_loader()

# Create neural networks
discriminator = Discriminator()
generator = Generator()
print("Number of parameters in G: {}".format(sum(p.numel() for p in generator.parameters() if p.requires_grad)))
print("Number of parameters in D: {}".format(sum(p.numel() for p in discriminator.parameters() if p.requires_grad)))

# Create optimizers
d_optimizer = torch.optim.Adam(discriminator.parameters(), lr=0.001, 
                                betas=(0.5, 0.9))
g_optimizer = torch.optim.Adam(generator.parameters(), lr=0.001,
                                betas=(0.5, 0.9))

# Prepare the inputs for the sampler, which wil run during the training
sample_latent = torch.randn(n_samples, latent_dim)

# Transfer the neural nets and samples to GPU
if torch.cuda.is_available():
    discriminator = discriminator.cuda()
    generator = generator.cuda()
    sample_latent = sample_latent.cuda()

# Create an empty dictionary to sotre history samples
history_samples = {}

# Create a LiveLoss logger instance for monitoring
liveloss = PlotLosses(plot_extrema=False, skip_first=2,
                    #   series_fmt={'discriminator': 'd_{}', 'generator':'g_{}'},
                      metric2title={'loss': 'Training Losses'})

# Initialize step
step = 0<jupyter_output>Number of parameters in G: 966655
Number of parameters in D: 582753<jupyter_text>Training iterations<jupyter_code># Create a progress bar instance for monitoring
progress_bar = tqdm(total=n_steps, initial=step, ncols=120, mininterval=1)

# Start iterations
while step < n_steps + 1:
    # Iterate over the dataset
    for real_samples in data_loader:
        # Train the neural networks
        generator.train()
        d_loss, g_loss = train_one_step(d_optimizer, g_optimizer, real_samples)

        # Record smoothened loss values to LiveLoss logger
        if step > 0:
            running_d_loss = 0.05 * d_loss + 0.95 * running_d_loss
            running_g_loss = 0.05 * g_loss + 0.95 * running_g_loss
        else:
            running_d_loss, running_g_loss = 0.0, 0.0
        liveloss.update({'negative_critic_loss': -running_d_loss})
        # liveloss.update({'d_loss': running_d_loss, 'g_loss': running_g_loss})

        # Update losses to progress bar
        progress_bar.set_description_str(
            "(d_loss={: 8.6f}, g_loss={: 8.6f})".format(d_loss, g_loss))
        
        if step % sample_interval == 0:
            # Get generated samples
            generator.eval()
            samples = generator(sample_latent).cpu().detach().numpy()
            history_samples[step] = samples

            # Display loss curves
            clear_output(True)
            if step > 0:
                liveloss.draw()
            
            # Display generated samples
            samples = samples.transpose(1, 0, 2, 3).reshape(n_tracks, -1, n_pitches)
            tracks = []
            for idx, (program, is_drum, track_name) in enumerate(zip(programs, is_drums, track_names)):
                pianoroll = (np.pad(samples[idx] > 0.5,
                                    ((0, 0), (lowest_pitch, 128 - lowest_pitch - n_pitches)), 'constant'))
                tracks.append(Track(pianoroll, program, is_drum, name=track_name))
            m = Multitrack(tracks=tracks, tempo=tempo, beat_resolution=beat_resolution)
            fig, axs = m.plot()
            fig.set_size_inches((16, 8))
            for ax in axs:
                for x in range(measure_resolution, 4 * measure_resolution * n_measures
                
                , measure_resolution):
                    if x % (measure_resolution * 4) == 0:
                        ax.axvline(x - 0.5, color='k')
                    else:
                        ax.axvline(x - 0.5, color='k', linestyle='-', linewidth=1)
            plt.show()
            
        step += 1
        progress_bar.update(1)
        if step >= n_steps:
            break
# Show history
slider = IntSlider(min=0, max=n_steps, value=n_steps, step=sample_interval,
                   layout={'width': '500px'})
@interact
def plot(step=slider):
    tracks = []
    for idx, (program, is_drum, track_name) in enumerate(zip(programs, is_drums, track_names)):
        pianoroll = (np.pad(samples[idx] > 0.5,
                            ((0, 0), (lowest_pitch, 128 - lowest_pitch - n_pitches)), 'constant'))
        tracks.append(Track(pianoroll, program, is_drum, name=track_name))
    m = Multitrack(tracks=tracks, tempo=tempo, beat_resolution=beat_resolution)
    fig, axs = m.plot()
    fig.set_size_inches((16, 8))
    plt.show()
    plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11876190476190476, 'fraction_numerical': 0.025761904761904764, 'mean_word_length': 2.8732939874585024, 'pattern_counts': {'":': 0, '<': 42, '<?xml version=': 0, '>': 51, 'https://': 7, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '404394', 'license': 'MIT', 'n_tokens_mistral': 7410, 'n_tokens_neox': 6945, 'n_words': 1677, 'path': 'Music_Generation_with_GAN.ipynb', 'repo_name': 'danmackinlay/ISMIR2019', 'size': '21000'}