{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "**Chapter 15 – Processing Sequences Using RNNs and CNNs**" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "_This notebook contains all the sample code in chapter 15._" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "\n", " \n", "
\n", " Run in Google Colab\n", "
" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Setup" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "First, let's import a few common modules, ensure MatplotLib plots figures inline and prepare a function to save the figures. We also check that Python 3.5 or later is installed (although Python 2.x may work, it is deprecated so we strongly recommend you use Python 3 instead), as well as Scikit-Learn ≥0.20 and TensorFlow ≥2.0." ] }, { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [], "source": [ "# Python ≥3.5 is required\n", "import sys\n", "assert sys.version_info >= (3, 5)\n", "\n", "# Scikit-Learn ≥0.20 is required\n", "import sklearn\n", "assert sklearn.__version__ >= \"0.20\"\n", "\n", "try:\n", " # %tensorflow_version only exists in Colab.\n", " %tensorflow_version 2.x\n", " IS_COLAB = True\n", "except Exception:\n", " IS_COLAB = False\n", "\n", "# TensorFlow ≥2.0 is required\n", "import tensorflow as tf\n", "from tensorflow import keras\n", "assert tf.__version__ >= \"2.0\"\n", "\n", "if not tf.config.list_physical_devices('GPU'):\n", " print(\"No GPU was detected. LSTMs and CNNs can be very slow without a GPU.\")\n", " if IS_COLAB:\n", " print(\"Go to Runtime > Change runtime and select a GPU hardware accelerator.\")\n", "\n", "# Common imports\n", "import numpy as np\n", "import os\n", "from pathlib import Path\n", "\n", "# to make this notebook's output stable across runs\n", "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "# To plot pretty figures\n", "%matplotlib inline\n", "import matplotlib as mpl\n", "import matplotlib.pyplot as plt\n", "mpl.rc('axes', labelsize=14)\n", "mpl.rc('xtick', labelsize=12)\n", "mpl.rc('ytick', labelsize=12)\n", "\n", "# Where to save the figures\n", "PROJECT_ROOT_DIR = \".\"\n", "CHAPTER_ID = \"rnn\"\n", "IMAGES_PATH = os.path.join(PROJECT_ROOT_DIR, \"images\", CHAPTER_ID)\n", "os.makedirs(IMAGES_PATH, exist_ok=True)\n", "\n", "def save_fig(fig_id, tight_layout=True, fig_extension=\"png\", resolution=300):\n", " path = os.path.join(IMAGES_PATH, fig_id + \".\" + fig_extension)\n", " print(\"Saving figure\", fig_id)\n", " if tight_layout:\n", " plt.tight_layout()\n", " plt.savefig(path, format=fig_extension, dpi=resolution)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Basic RNNs" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Generate the Dataset" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [], "source": [ "def generate_time_series(batch_size, n_steps):\n", " freq1, freq2, offsets1, offsets2 = np.random.rand(4, batch_size, 1)\n", " time = np.linspace(0, 1, n_steps)\n", " series = 0.5 * np.sin((time - offsets1) * (freq1 * 10 + 10)) # wave 1\n", " series += 0.2 * np.sin((time - offsets2) * (freq2 * 20 + 20)) # + wave 2\n", " series += 0.1 * (np.random.rand(batch_size, n_steps) - 0.5) # + noise\n", " return series[..., np.newaxis].astype(np.float32)" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "\n", "n_steps = 50\n", "series = generate_time_series(10000, n_steps + 1)\n", "X_train, y_train = series[:7000, :n_steps], series[:7000, -1]\n", "X_valid, y_valid = series[7000:9000, :n_steps], series[7000:9000, -1]\n", "X_test, y_test = series[9000:, :n_steps], series[9000:, -1]" ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [], "source": [ "X_train.shape, y_train.shape" ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [], "source": [ "def plot_series(series, y=None, y_pred=None, x_label=\"$t$\", y_label=\"$x(t)$\"):\n", " plt.plot(series, \".-\")\n", " if y is not None:\n", " plt.plot(n_steps, y, \"bx\", markersize=10)\n", " if y_pred is not None:\n", " plt.plot(n_steps, y_pred, \"ro\")\n", " plt.grid(True)\n", " if x_label:\n", " plt.xlabel(x_label, fontsize=16)\n", " if y_label:\n", " plt.ylabel(y_label, fontsize=16, rotation=0)\n", " plt.hlines(0, 0, 100, linewidth=1)\n", " plt.axis([0, n_steps + 1, -1, 1])\n", "\n", "fig, axes = plt.subplots(nrows=1, ncols=3, sharey=True, figsize=(12, 4))\n", "for col in range(3):\n", " plt.sca(axes[col])\n", " plot_series(X_valid[col, :, 0], y_valid[col, 0],\n", " y_label=(\"$x(t)$\" if col==0 else None))\n", "save_fig(\"time_series_plot\")\n", "plt.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Computing Some Baselines" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Naive predictions (just predict the last observed value):" ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [], "source": [ "y_pred = X_valid[:, -1]\n", "np.mean(keras.losses.mean_squared_error(y_valid, y_pred))" ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [], "source": [ "plot_series(X_valid[0, :, 0], y_valid[0, 0], y_pred[0, 0])\n", "plt.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Linear predictions:" ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.Flatten(input_shape=[50, 1]),\n", " keras.layers.Dense(1)\n", "])\n", "\n", "model.compile(loss=\"mse\", optimizer=\"adam\")\n", "history = model.fit(X_train, y_train, epochs=20,\n", " validation_data=(X_valid, y_valid))" ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [], "source": [ "model.evaluate(X_valid, y_valid)" ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [], "source": [ "def plot_learning_curves(loss, val_loss):\n", " plt.plot(np.arange(len(loss)) + 0.5, loss, \"b.-\", label=\"Training loss\")\n", " plt.plot(np.arange(len(val_loss)) + 1, val_loss, \"r.-\", label=\"Validation loss\")\n", " plt.gca().xaxis.set_major_locator(mpl.ticker.MaxNLocator(integer=True))\n", " plt.axis([1, 20, 0, 0.05])\n", " plt.legend(fontsize=14)\n", " plt.xlabel(\"Epochs\")\n", " plt.ylabel(\"Loss\")\n", " plt.grid(True)\n", "\n", "plot_learning_curves(history.history[\"loss\"], history.history[\"val_loss\"])\n", "plt.show()" ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [], "source": [ "y_pred = model.predict(X_valid)\n", "plot_series(X_valid[0, :, 0], y_valid[0, 0], y_pred[0, 0])\n", "plt.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Using a Simple RNN" ] }, { "cell_type": "code", "execution_count": 12, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.SimpleRNN(1, input_shape=[None, 1])\n", "])\n", "\n", "optimizer = keras.optimizers.Adam(lr=0.005)\n", "model.compile(loss=\"mse\", optimizer=optimizer)\n", "history = model.fit(X_train, y_train, epochs=20,\n", " validation_data=(X_valid, y_valid))" ] }, { "cell_type": "code", "execution_count": 13, "metadata": {}, "outputs": [], "source": [ "model.evaluate(X_valid, y_valid)" ] }, { "cell_type": "code", "execution_count": 14, "metadata": {}, "outputs": [], "source": [ "plot_learning_curves(history.history[\"loss\"], history.history[\"val_loss\"])\n", "plt.show()" ] }, { "cell_type": "code", "execution_count": 15, "metadata": {}, "outputs": [], "source": [ "y_pred = model.predict(X_valid)\n", "plot_series(X_valid[0, :, 0], y_valid[0, 0], y_pred[0, 0])\n", "plt.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Deep RNNs" ] }, { "cell_type": "code", "execution_count": 16, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.SimpleRNN(20, return_sequences=True, input_shape=[None, 1]),\n", " keras.layers.SimpleRNN(20, return_sequences=True),\n", " keras.layers.SimpleRNN(1)\n", "])\n", "\n", "model.compile(loss=\"mse\", optimizer=\"adam\")\n", "history = model.fit(X_train, y_train, epochs=20,\n", " validation_data=(X_valid, y_valid))" ] }, { "cell_type": "code", "execution_count": 17, "metadata": {}, "outputs": [], "source": [ "model.evaluate(X_valid, y_valid)" ] }, { "cell_type": "code", "execution_count": 18, "metadata": {}, "outputs": [], "source": [ "plot_learning_curves(history.history[\"loss\"], history.history[\"val_loss\"])\n", "plt.show()" ] }, { "cell_type": "code", "execution_count": 19, "metadata": {}, "outputs": [], "source": [ "y_pred = model.predict(X_valid)\n", "plot_series(X_valid[0, :, 0], y_valid[0, 0], y_pred[0, 0])\n", "plt.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Make the second `SimpleRNN` layer return only the last output:" ] }, { "cell_type": "code", "execution_count": 20, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.SimpleRNN(20, return_sequences=True, input_shape=[None, 1]),\n", " keras.layers.SimpleRNN(20),\n", " keras.layers.Dense(1)\n", "])\n", "\n", "model.compile(loss=\"mse\", optimizer=\"adam\")\n", "history = model.fit(X_train, y_train, epochs=20,\n", " validation_data=(X_valid, y_valid))" ] }, { "cell_type": "code", "execution_count": 21, "metadata": {}, "outputs": [], "source": [ "model.evaluate(X_valid, y_valid)" ] }, { "cell_type": "code", "execution_count": 22, "metadata": {}, "outputs": [], "source": [ "plot_learning_curves(history.history[\"loss\"], history.history[\"val_loss\"])\n", "plt.show()" ] }, { "cell_type": "code", "execution_count": 23, "metadata": {}, "outputs": [], "source": [ "y_pred = model.predict(X_valid)\n", "plot_series(X_valid[0, :, 0], y_valid[0, 0], y_pred[0, 0])\n", "plt.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Forecasting Several Steps Ahead" ] }, { "cell_type": "code", "execution_count": 24, "metadata": {}, "outputs": [], "source": [ "np.random.seed(43) # not 42, as it would give the first series in the train set\n", "\n", "series = generate_time_series(1, n_steps + 10)\n", "X_new, Y_new = series[:, :n_steps], series[:, n_steps:]\n", "X = X_new\n", "for step_ahead in range(10):\n", " y_pred_one = model.predict(X[:, step_ahead:])[:, np.newaxis, :]\n", " X = np.concatenate([X, y_pred_one], axis=1)\n", "\n", "Y_pred = X[:, n_steps:]" ] }, { "cell_type": "code", "execution_count": 25, "metadata": {}, "outputs": [], "source": [ "Y_pred.shape" ] }, { "cell_type": "code", "execution_count": 26, "metadata": {}, "outputs": [], "source": [ "def plot_multiple_forecasts(X, Y, Y_pred):\n", " n_steps = X.shape[1]\n", " ahead = Y.shape[1]\n", " plot_series(X[0, :, 0])\n", " plt.plot(np.arange(n_steps, n_steps + ahead), Y[0, :, 0], \"ro-\", label=\"Actual\")\n", " plt.plot(np.arange(n_steps, n_steps + ahead), Y_pred[0, :, 0], \"bx-\", label=\"Forecast\", markersize=10)\n", " plt.axis([0, n_steps + ahead, -1, 1])\n", " plt.legend(fontsize=14)\n", "\n", "plot_multiple_forecasts(X_new, Y_new, Y_pred)\n", "save_fig(\"forecast_ahead_plot\")\n", "plt.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now let's use this model to predict the next 10 values. We first need to regenerate the sequences with 9 more time steps." ] }, { "cell_type": "code", "execution_count": 27, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "\n", "n_steps = 50\n", "series = generate_time_series(10000, n_steps + 10)\n", "X_train, Y_train = series[:7000, :n_steps], series[:7000, -10:, 0]\n", "X_valid, Y_valid = series[7000:9000, :n_steps], series[7000:9000, -10:, 0]\n", "X_test, Y_test = series[9000:, :n_steps], series[9000:, -10:, 0]" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now let's predict the next 10 values one by one:" ] }, { "cell_type": "code", "execution_count": 28, "metadata": {}, "outputs": [], "source": [ "X = X_valid\n", "for step_ahead in range(10):\n", " y_pred_one = model.predict(X)[:, np.newaxis, :]\n", " X = np.concatenate([X, y_pred_one], axis=1)\n", "\n", "Y_pred = X[:, n_steps:, 0]" ] }, { "cell_type": "code", "execution_count": 29, "metadata": {}, "outputs": [], "source": [ "Y_pred.shape" ] }, { "cell_type": "code", "execution_count": 30, "metadata": {}, "outputs": [], "source": [ "np.mean(keras.metrics.mean_squared_error(Y_valid, Y_pred))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Let's compare this performance with some baselines: naive predictions and a simple linear model:" ] }, { "cell_type": "code", "execution_count": 31, "metadata": {}, "outputs": [], "source": [ "Y_naive_pred = Y_valid[:, -1:]\n", "np.mean(keras.metrics.mean_squared_error(Y_valid, Y_naive_pred))" ] }, { "cell_type": "code", "execution_count": 32, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.Flatten(input_shape=[50, 1]),\n", " keras.layers.Dense(10)\n", "])\n", "\n", "model.compile(loss=\"mse\", optimizer=\"adam\")\n", "history = model.fit(X_train, Y_train, epochs=20,\n", " validation_data=(X_valid, Y_valid))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now let's create an RNN that predicts all 10 next values at once:" ] }, { "cell_type": "code", "execution_count": 33, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.SimpleRNN(20, return_sequences=True, input_shape=[None, 1]),\n", " keras.layers.SimpleRNN(20),\n", " keras.layers.Dense(10)\n", "])\n", "\n", "model.compile(loss=\"mse\", optimizer=\"adam\")\n", "history = model.fit(X_train, Y_train, epochs=20,\n", " validation_data=(X_valid, Y_valid))" ] }, { "cell_type": "code", "execution_count": 34, "metadata": {}, "outputs": [], "source": [ "np.random.seed(43)\n", "\n", "series = generate_time_series(1, 50 + 10)\n", "X_new, Y_new = series[:, :50, :], series[:, -10:, :]\n", "Y_pred = model.predict(X_new)[..., np.newaxis]" ] }, { "cell_type": "code", "execution_count": 35, "metadata": {}, "outputs": [], "source": [ "plot_multiple_forecasts(X_new, Y_new, Y_pred)\n", "plt.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now let's create an RNN that predicts the next 10 steps at each time step. That is, instead of just forecasting time steps 50 to 59 based on time steps 0 to 49, it will forecast time steps 1 to 10 at time step 0, then time steps 2 to 11 at time step 1, and so on, and finally it will forecast time steps 50 to 59 at the last time step. Notice that the model is causal: when it makes predictions at any time step, it can only see past time steps." ] }, { "cell_type": "code", "execution_count": 36, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "\n", "n_steps = 50\n", "series = generate_time_series(10000, n_steps + 10)\n", "X_train = series[:7000, :n_steps]\n", "X_valid = series[7000:9000, :n_steps]\n", "X_test = series[9000:, :n_steps]\n", "Y = np.empty((10000, n_steps, 10))\n", "for step_ahead in range(1, 10 + 1):\n", " Y[..., step_ahead - 1] = series[..., step_ahead:step_ahead + n_steps, 0]\n", "Y_train = Y[:7000]\n", "Y_valid = Y[7000:9000]\n", "Y_test = Y[9000:]" ] }, { "cell_type": "code", "execution_count": 37, "metadata": {}, "outputs": [], "source": [ "X_train.shape, Y_train.shape" ] }, { "cell_type": "code", "execution_count": 38, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.SimpleRNN(20, return_sequences=True, input_shape=[None, 1]),\n", " keras.layers.SimpleRNN(20, return_sequences=True),\n", " keras.layers.TimeDistributed(keras.layers.Dense(10))\n", "])\n", "\n", "def last_time_step_mse(Y_true, Y_pred):\n", " return keras.metrics.mean_squared_error(Y_true[:, -1], Y_pred[:, -1])\n", "\n", "model.compile(loss=\"mse\", optimizer=keras.optimizers.Adam(lr=0.01), metrics=[last_time_step_mse])\n", "history = model.fit(X_train, Y_train, epochs=20,\n", " validation_data=(X_valid, Y_valid))" ] }, { "cell_type": "code", "execution_count": 39, "metadata": {}, "outputs": [], "source": [ "np.random.seed(43)\n", "\n", "series = generate_time_series(1, 50 + 10)\n", "X_new, Y_new = series[:, :50, :], series[:, 50:, :]\n", "Y_pred = model.predict(X_new)[:, -1][..., np.newaxis]" ] }, { "cell_type": "code", "execution_count": 40, "metadata": {}, "outputs": [], "source": [ "plot_multiple_forecasts(X_new, Y_new, Y_pred)\n", "plt.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Deep RNN with Batch Norm" ] }, { "cell_type": "code", "execution_count": 41, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.SimpleRNN(20, return_sequences=True, input_shape=[None, 1]),\n", " keras.layers.BatchNormalization(),\n", " keras.layers.SimpleRNN(20, return_sequences=True),\n", " keras.layers.BatchNormalization(),\n", " keras.layers.TimeDistributed(keras.layers.Dense(10))\n", "])\n", "\n", "model.compile(loss=\"mse\", optimizer=\"adam\", metrics=[last_time_step_mse])\n", "history = model.fit(X_train, Y_train, epochs=20,\n", " validation_data=(X_valid, Y_valid))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Deep RNNs with Layer Norm" ] }, { "cell_type": "code", "execution_count": 42, "metadata": {}, "outputs": [], "source": [ "from tensorflow.keras.layers import LayerNormalization" ] }, { "cell_type": "code", "execution_count": 43, "metadata": {}, "outputs": [], "source": [ "class LNSimpleRNNCell(keras.layers.Layer):\n", " def __init__(self, units, activation=\"tanh\", **kwargs):\n", " super().__init__(**kwargs)\n", " self.state_size = units\n", " self.output_size = units\n", " self.simple_rnn_cell = keras.layers.SimpleRNNCell(units,\n", " activation=None)\n", " self.layer_norm = LayerNormalization()\n", " self.activation = keras.activations.get(activation)\n", " def get_initial_state(self, inputs=None, batch_size=None, dtype=None):\n", " if inputs is not None:\n", " batch_size = tf.shape(inputs)[0]\n", " dtype = inputs.dtype\n", " return [tf.zeros([batch_size, self.state_size], dtype=dtype)]\n", " def call(self, inputs, states):\n", " outputs, new_states = self.simple_rnn_cell(inputs, states)\n", " norm_outputs = self.activation(self.layer_norm(outputs))\n", " return norm_outputs, [norm_outputs]" ] }, { "cell_type": "code", "execution_count": 44, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.RNN(LNSimpleRNNCell(20), return_sequences=True,\n", " input_shape=[None, 1]),\n", " keras.layers.RNN(LNSimpleRNNCell(20), return_sequences=True),\n", " keras.layers.TimeDistributed(keras.layers.Dense(10))\n", "])\n", "\n", "model.compile(loss=\"mse\", optimizer=\"adam\", metrics=[last_time_step_mse])\n", "history = model.fit(X_train, Y_train, epochs=20,\n", " validation_data=(X_valid, Y_valid))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Creating a Custom RNN Class" ] }, { "cell_type": "code", "execution_count": 45, "metadata": {}, "outputs": [], "source": [ "class MyRNN(keras.layers.Layer):\n", " def __init__(self, cell, return_sequences=False, **kwargs):\n", " super().__init__(**kwargs)\n", " self.cell = cell\n", " self.return_sequences = return_sequences\n", " self.get_initial_state = getattr(\n", " self.cell, \"get_initial_state\", self.fallback_initial_state)\n", " def fallback_initial_state(self, inputs):\n", " return [tf.zeros([self.cell.state_size], dtype=inputs.dtype)]\n", " @tf.function\n", " def call(self, inputs):\n", " states = self.get_initial_state(inputs)\n", " n_steps = tf.shape(inputs)[1]\n", " if self.return_sequences:\n", " sequences = tf.TensorArray(inputs.dtype, size=n_steps)\n", " outputs = tf.zeros(shape=[n_steps, self.cell.output_size], dtype=inputs.dtype)\n", " for step in tf.range(n_steps):\n", " outputs, states = self.cell(inputs[:, step], states)\n", " if self.return_sequences:\n", " sequences = sequences.write(step, outputs)\n", " if self.return_sequences:\n", " return sequences.stack()\n", " else:\n", " return outputs" ] }, { "cell_type": "code", "execution_count": 46, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " MyRNN(LNSimpleRNNCell(20), return_sequences=True,\n", " input_shape=[None, 1]),\n", " MyRNN(LNSimpleRNNCell(20), return_sequences=True),\n", " keras.layers.TimeDistributed(keras.layers.Dense(10))\n", "])\n", "\n", "model.compile(loss=\"mse\", optimizer=\"adam\", metrics=[last_time_step_mse])\n", "history = model.fit(X_train, Y_train, epochs=20,\n", " validation_data=(X_valid, Y_valid))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# LSTMs" ] }, { "cell_type": "code", "execution_count": 47, "metadata": { "scrolled": true }, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.LSTM(20, return_sequences=True, input_shape=[None, 1]),\n", " keras.layers.LSTM(20, return_sequences=True),\n", " keras.layers.TimeDistributed(keras.layers.Dense(10))\n", "])\n", "\n", "model.compile(loss=\"mse\", optimizer=\"adam\", metrics=[last_time_step_mse])\n", "history = model.fit(X_train, Y_train, epochs=20,\n", " validation_data=(X_valid, Y_valid))" ] }, { "cell_type": "code", "execution_count": 48, "metadata": {}, "outputs": [], "source": [ "model.evaluate(X_valid, Y_valid)" ] }, { "cell_type": "code", "execution_count": 49, "metadata": {}, "outputs": [], "source": [ "plot_learning_curves(history.history[\"loss\"], history.history[\"val_loss\"])\n", "plt.show()" ] }, { "cell_type": "code", "execution_count": 50, "metadata": {}, "outputs": [], "source": [ "np.random.seed(43)\n", "\n", "series = generate_time_series(1, 50 + 10)\n", "X_new, Y_new = series[:, :50, :], series[:, 50:, :]\n", "Y_pred = model.predict(X_new)[:, -1][..., np.newaxis]" ] }, { "cell_type": "code", "execution_count": 51, "metadata": { "scrolled": true }, "outputs": [], "source": [ "plot_multiple_forecasts(X_new, Y_new, Y_pred)\n", "plt.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# GRUs" ] }, { "cell_type": "code", "execution_count": 52, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.GRU(20, return_sequences=True, input_shape=[None, 1]),\n", " keras.layers.GRU(20, return_sequences=True),\n", " keras.layers.TimeDistributed(keras.layers.Dense(10))\n", "])\n", "\n", "model.compile(loss=\"mse\", optimizer=\"adam\", metrics=[last_time_step_mse])\n", "history = model.fit(X_train, Y_train, epochs=20,\n", " validation_data=(X_valid, Y_valid))" ] }, { "cell_type": "code", "execution_count": 53, "metadata": {}, "outputs": [], "source": [ "model.evaluate(X_valid, Y_valid)" ] }, { "cell_type": "code", "execution_count": 54, "metadata": {}, "outputs": [], "source": [ "plot_learning_curves(history.history[\"loss\"], history.history[\"val_loss\"])\n", "plt.show()" ] }, { "cell_type": "code", "execution_count": 55, "metadata": {}, "outputs": [], "source": [ "np.random.seed(43)\n", "\n", "series = generate_time_series(1, 50 + 10)\n", "X_new, Y_new = series[:, :50, :], series[:, 50:, :]\n", "Y_pred = model.predict(X_new)[:, -1][..., np.newaxis]" ] }, { "cell_type": "code", "execution_count": 56, "metadata": { "scrolled": true }, "outputs": [], "source": [ "plot_multiple_forecasts(X_new, Y_new, Y_pred)\n", "plt.show()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Using One-Dimensional Convolutional Layers to Process Sequences" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "```\n", "1D conv layer with kernel size 4, stride 2, VALID padding:\n", "\n", " |-----2-----| |-----5---...------| |-----23----|\n", " |-----1-----| |-----4-----| ... |-----22----|\n", " |-----0----| |-----3-----| |---...|-----21----|\n", "X: 0 1 2 3 4 5 6 7 8 9 10 11 12 ... 42 43 44 45 46 47 48 49\n", "Y: 1 2 3 4 5 6 7 8 9 10 11 12 13 ... 43 44 45 46 47 48 49 50\n", " /10 11 12 13 14 15 16 17 18 19 20 21 22 ... 52 53 54 55 56 57 58 59\n", "\n", "Output:\n", "\n", "X: 0/3 2/5 4/7 6/9 8/11 10/13 .../43 42/45 44/47 46/49\n", "Y: 4/13 6/15 8/17 10/19 12/21 14/23 .../53 46/55 48/57 50/59\n", "```" ] }, { "cell_type": "code", "execution_count": 57, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.Conv1D(filters=20, kernel_size=4, strides=2, padding=\"valid\",\n", " input_shape=[None, 1]),\n", " keras.layers.GRU(20, return_sequences=True),\n", " keras.layers.GRU(20, return_sequences=True),\n", " keras.layers.TimeDistributed(keras.layers.Dense(10))\n", "])\n", "\n", "model.compile(loss=\"mse\", optimizer=\"adam\", metrics=[last_time_step_mse])\n", "history = model.fit(X_train, Y_train[:, 3::2], epochs=20,\n", " validation_data=(X_valid, Y_valid[:, 3::2]))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## WaveNet" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "```\n", "C2 /\\ /\\ /\\ /\\ /\\ /\\ /\\ /\\ /\\ /\\ /\\ /\\ /\\.../\\ /\\ /\\ /\\ /\\ /\\\n", " \\ / \\ / \\ / \\ / \\ / \\ / \\ / \\ / \\ / \\\n", " / \\ / \\ / \\ / \\\n", "C1 /\\ /\\ /\\ /\\ /\\ /\\ /\\ /\\ /\\ /\\ /\\ /\\ /.../\\ /\\ /\\ /\\ /\\ /\\ /\\\n", "X: 0 1 2 3 4 5 6 7 8 9 10 11 12 ... 43 44 45 46 47 48 49\n", "Y: 1 2 3 4 5 6 7 8 9 10 11 12 13 ... 44 45 46 47 48 49 50\n", " /10 11 12 13 14 15 16 17 18 19 20 21 22 ... 53 54 55 56 57 58 59\n", "```" ] }, { "cell_type": "code", "execution_count": 58, "metadata": {}, "outputs": [], "source": [ "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "model = keras.models.Sequential()\n", "model.add(keras.layers.InputLayer(input_shape=[None, 1]))\n", "for rate in (1, 2, 4, 8) * 2:\n", " model.add(keras.layers.Conv1D(filters=20, kernel_size=2, padding=\"causal\",\n", " activation=\"relu\", dilation_rate=rate))\n", "model.add(keras.layers.Conv1D(filters=10, kernel_size=1))\n", "model.compile(loss=\"mse\", optimizer=\"adam\", metrics=[last_time_step_mse])\n", "history = model.fit(X_train, Y_train, epochs=20,\n", " validation_data=(X_valid, Y_valid))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Here is the original WaveNet defined in the paper: it uses Gated Activation Units instead of ReLU and parametrized skip connections, plus it pads with zeros on the left to avoid getting shorter and shorter sequences:" ] }, { "cell_type": "code", "execution_count": 59, "metadata": {}, "outputs": [], "source": [ "class GatedActivationUnit(keras.layers.Layer):\n", " def __init__(self, activation=\"tanh\", **kwargs):\n", " super().__init__(**kwargs)\n", " self.activation = keras.activations.get(activation)\n", " def call(self, inputs):\n", " n_filters = inputs.shape[-1] // 2\n", " linear_output = self.activation(inputs[..., :n_filters])\n", " gate = keras.activations.sigmoid(inputs[..., n_filters:])\n", " return self.activation(linear_output) * gate" ] }, { "cell_type": "code", "execution_count": 60, "metadata": {}, "outputs": [], "source": [ "def wavenet_residual_block(inputs, n_filters, dilation_rate):\n", " z = keras.layers.Conv1D(2 * n_filters, kernel_size=2, padding=\"causal\",\n", " dilation_rate=dilation_rate)(inputs)\n", " z = GatedActivationUnit()(z)\n", " z = keras.layers.Conv1D(n_filters, kernel_size=1)(z)\n", " return keras.layers.Add()([z, inputs]), z" ] }, { "cell_type": "code", "execution_count": 61, "metadata": {}, "outputs": [], "source": [ "keras.backend.clear_session()\n", "np.random.seed(42)\n", "tf.random.set_seed(42)\n", "\n", "n_layers_per_block = 3 # 10 in the paper\n", "n_blocks = 1 # 3 in the paper\n", "n_filters = 32 # 128 in the paper\n", "n_outputs = 10 # 256 in the paper\n", "\n", "inputs = keras.layers.Input(shape=[None, 1])\n", "z = keras.layers.Conv1D(n_filters, kernel_size=2, padding=\"causal\")(inputs)\n", "skip_to_last = []\n", "for dilation_rate in [2**i for i in range(n_layers_per_block)] * n_blocks:\n", " z, skip = wavenet_residual_block(z, n_filters, dilation_rate)\n", " skip_to_last.append(skip)\n", "z = keras.activations.relu(keras.layers.Add()(skip_to_last))\n", "z = keras.layers.Conv1D(n_filters, kernel_size=1, activation=\"relu\")(z)\n", "Y_proba = keras.layers.Conv1D(n_outputs, kernel_size=1, activation=\"softmax\")(z)\n", "\n", "model = keras.models.Model(inputs=[inputs], outputs=[Y_proba])" ] }, { "cell_type": "code", "execution_count": 62, "metadata": {}, "outputs": [], "source": [ "model.compile(loss=\"mse\", optimizer=\"adam\", metrics=[last_time_step_mse])\n", "history = model.fit(X_train, Y_train, epochs=2,\n", " validation_data=(X_valid, Y_valid))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "In this chapter we explored the fundamentals of RNNs and used them to process sequences (namely, time series). In the process we also looked at other ways to process sequences, including CNNs. In the next chapter we will use RNNs for Natural Language Processing, and we will learn more about RNNs (bidirectional RNNs, stateful vs stateless RNNs, Encoder–Decoders, and Attention-augmented Encoder-Decoders). We will also look at the Transformer, an Attention-only architecture." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Exercise solutions" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 1. to 8." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "See Appendix A." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 9. Tackling the SketchRNN Dataset" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "_Exercise: Train a classification model for the SketchRNN dataset, available in TensorFlow Datasets._" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "The dataset is not available in TFDS yet, the [pull request](https://github.com/tensorflow/datasets/pull/361) is still work in progress. Luckily, the data is conveniently available as TFRecords, so let's download it (it might take a while, as it's about 1 GB large, with 3,450,000 training sketches and 345,000 test sketches):" ] }, { "cell_type": "code", "execution_count": 63, "metadata": {}, "outputs": [], "source": [ "DOWNLOAD_ROOT = \"http://download.tensorflow.org/data/\"\n", "FILENAME = \"quickdraw_tutorial_dataset_v1.tar.gz\"\n", "filepath = keras.utils.get_file(FILENAME,\n", " DOWNLOAD_ROOT + FILENAME,\n", " cache_subdir=\"datasets/quickdraw\",\n", " extract=True)" ] }, { "cell_type": "code", "execution_count": 64, "metadata": {}, "outputs": [], "source": [ "quickdraw_dir = Path(filepath).parent\n", "train_files = sorted([str(path) for path in quickdraw_dir.glob(\"training.tfrecord-*\")])\n", "eval_files = sorted([str(path) for path in quickdraw_dir.glob(\"eval.tfrecord-*\")])" ] }, { "cell_type": "code", "execution_count": 65, "metadata": {}, "outputs": [], "source": [ "train_files" ] }, { "cell_type": "code", "execution_count": 66, "metadata": {}, "outputs": [], "source": [ "eval_files" ] }, { "cell_type": "code", "execution_count": 67, "metadata": {}, "outputs": [], "source": [ "with open(quickdraw_dir / \"eval.tfrecord.classes\") as test_classes_file:\n", " test_classes = test_classes_file.readlines()\n", " \n", "with open(quickdraw_dir / \"training.tfrecord.classes\") as train_classes_file:\n", " train_classes = train_classes_file.readlines()" ] }, { "cell_type": "code", "execution_count": 68, "metadata": {}, "outputs": [], "source": [ "assert train_classes == test_classes\n", "class_names = [name.strip().lower() for name in train_classes]" ] }, { "cell_type": "code", "execution_count": 69, "metadata": {}, "outputs": [], "source": [ "sorted(class_names)" ] }, { "cell_type": "code", "execution_count": 70, "metadata": {}, "outputs": [], "source": [ "def parse(data_batch):\n", " feature_descriptions = {\n", " \"ink\": tf.io.VarLenFeature(dtype=tf.float32),\n", " \"shape\": tf.io.FixedLenFeature([2], dtype=tf.int64),\n", " \"class_index\": tf.io.FixedLenFeature([1], dtype=tf.int64)\n", " }\n", " examples = tf.io.parse_example(data_batch, feature_descriptions)\n", " flat_sketches = tf.sparse.to_dense(examples[\"ink\"])\n", " sketches = tf.reshape(flat_sketches, shape=[tf.size(data_batch), -1, 3])\n", " lengths = examples[\"shape\"][:, 0]\n", " labels = examples[\"class_index\"][:, 0]\n", " return sketches, lengths, labels" ] }, { "cell_type": "code", "execution_count": 71, "metadata": {}, "outputs": [], "source": [ "def quickdraw_dataset(filepaths, batch_size=32, shuffle_buffer_size=None,\n", " n_parse_threads=5, n_read_threads=5, cache=False):\n", " dataset = tf.data.TFRecordDataset(filepaths,\n", " num_parallel_reads=n_read_threads)\n", " if cache:\n", " dataset = dataset.cache()\n", " if shuffle_buffer_size:\n", " dataset = dataset.shuffle(shuffle_buffer_size)\n", " dataset = dataset.batch(batch_size)\n", " dataset = dataset.map(parse, num_parallel_calls=n_parse_threads)\n", " return dataset.prefetch(1)" ] }, { "cell_type": "code", "execution_count": 72, "metadata": {}, "outputs": [], "source": [ "train_set = quickdraw_dataset(train_files, shuffle_buffer_size=10000)\n", "valid_set = quickdraw_dataset(eval_files[:5])\n", "test_set = quickdraw_dataset(eval_files[5:])" ] }, { "cell_type": "code", "execution_count": 73, "metadata": {}, "outputs": [], "source": [ "for sketches, lengths, labels in train_set.take(1):\n", " print(\"sketches =\", sketches)\n", " print(\"lengths =\", lengths)\n", " print(\"labels =\", labels)" ] }, { "cell_type": "code", "execution_count": 74, "metadata": {}, "outputs": [], "source": [ "def draw_sketch(sketch, label=None):\n", " origin = np.array([[0., 0., 0.]])\n", " sketch = np.r_[origin, sketch]\n", " stroke_end_indices = np.argwhere(sketch[:, -1]==1.)[:, 0]\n", " coordinates = np.cumsum(sketch[:, :2], axis=0)\n", " strokes = np.split(coordinates, stroke_end_indices + 1)\n", " title = class_names[label.numpy()] if label is not None else \"Try to guess\"\n", " plt.title(title)\n", " plt.plot(coordinates[:, 0], -coordinates[:, 1], \"y:\")\n", " for stroke in strokes:\n", " plt.plot(stroke[:, 0], -stroke[:, 1], \".-\")\n", " plt.axis(\"off\")\n", "\n", "def draw_sketches(sketches, lengths, labels):\n", " n_sketches = len(sketches)\n", " n_cols = 4\n", " n_rows = (n_sketches - 1) // n_cols + 1\n", " plt.figure(figsize=(n_cols * 3, n_rows * 3.5))\n", " for index, sketch, length, label in zip(range(n_sketches), sketches, lengths, labels):\n", " plt.subplot(n_rows, n_cols, index + 1)\n", " draw_sketch(sketch[:length], label)\n", " plt.show()\n", "\n", "for sketches, lengths, labels in train_set.take(1):\n", " draw_sketches(sketches, lengths, labels)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Most sketches are composed of less than 100 points:" ] }, { "cell_type": "code", "execution_count": 75, "metadata": {}, "outputs": [], "source": [ "lengths = np.concatenate([lengths for _, lengths, _ in train_set.take(1000)])\n", "plt.hist(lengths, bins=150, density=True)\n", "plt.axis([0, 200, 0, 0.03])\n", "plt.xlabel(\"length\")\n", "plt.ylabel(\"density\")\n", "plt.show()" ] }, { "cell_type": "code", "execution_count": 76, "metadata": {}, "outputs": [], "source": [ "def crop_long_sketches(dataset, max_length=100):\n", " return dataset.map(lambda inks, lengths, labels: (inks[:, :max_length], labels))\n", "\n", "cropped_train_set = crop_long_sketches(train_set)\n", "cropped_valid_set = crop_long_sketches(valid_set)\n", "cropped_test_set = crop_long_sketches(test_set)" ] }, { "cell_type": "code", "execution_count": 77, "metadata": {}, "outputs": [], "source": [ "model = keras.models.Sequential([\n", " keras.layers.Conv1D(32, kernel_size=5, strides=2, activation=\"relu\"),\n", " keras.layers.BatchNormalization(),\n", " keras.layers.Conv1D(64, kernel_size=5, strides=2, activation=\"relu\"),\n", " keras.layers.BatchNormalization(),\n", " keras.layers.Conv1D(128, kernel_size=3, strides=2, activation=\"relu\"),\n", " keras.layers.BatchNormalization(),\n", " keras.layers.LSTM(128, return_sequences=True),\n", " keras.layers.LSTM(128),\n", " keras.layers.Dense(len(class_names), activation=\"softmax\")\n", "])\n", "optimizer = keras.optimizers.SGD(lr=1e-2, clipnorm=1.)\n", "model.compile(loss=\"sparse_categorical_crossentropy\",\n", " optimizer=optimizer,\n", " metrics=[\"accuracy\", \"sparse_top_k_categorical_accuracy\"])\n", "history = model.fit(cropped_train_set, epochs=2,\n", " validation_data=cropped_valid_set)" ] }, { "cell_type": "code", "execution_count": 78, "metadata": {}, "outputs": [], "source": [ "y_test = np.concatenate([labels for _, _, labels in test_set])\n", "y_probas = model.predict(test_set)" ] }, { "cell_type": "code", "execution_count": 79, "metadata": {}, "outputs": [], "source": [ "np.mean(keras.metrics.sparse_top_k_categorical_accuracy(y_test, y_probas))" ] }, { "cell_type": "code", "execution_count": 80, "metadata": {}, "outputs": [], "source": [ "n_new = 10\n", "Y_probas = model.predict(sketches)\n", "top_k = tf.nn.top_k(Y_probas, k=5)\n", "for index in range(n_new):\n", " plt.figure(figsize=(3, 3.5))\n", " draw_sketch(sketches[index])\n", " plt.show()\n", " print(\"Top-5 predictions:\".format(index + 1))\n", " for k in range(5):\n", " class_name = class_names[top_k.indices[index, k]]\n", " proba = 100 * top_k.values[index, k]\n", " print(\" {}. {} {:.3f}%\".format(k + 1, class_name, proba))\n", " print(\"Answer: {}\".format(class_names[labels[index].numpy()]))" ] }, { "cell_type": "code", "execution_count": 81, "metadata": {}, "outputs": [], "source": [ "model.save(\"my_sketchrnn\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 10. Bach Chorales\n", "_Exercise: Download the [Bach chorales](https://homl.info/bach) dataset and unzip it. It is composed of 382 chorales composed by Johann Sebastian Bach. Each chorale is 100 to 640 time steps long, and each time step contains 4 integers, where each integer corresponds to a note's index on a piano (except for the value 0, which means that no note is played). Train a model—recurrent, convolutional, or both—that can predict the next time step (four notes), given a sequence of time steps from a chorale. Then use this model to generate Bach-like music, one note at a time: you can do this by giving the model the start of a chorale and asking it to predict the next time step, then appending these time steps to the input sequence and asking the model for the next note, and so on. Also make sure to check out [Google's Coconet model](https://homl.info/coconet), which was used for a nice [Google doodle about Bach](https://www.google.com/doodles/celebrating-johann-sebastian-bach)._\n", "\n" ] }, { "cell_type": "code", "execution_count": 82, "metadata": {}, "outputs": [], "source": [ "DOWNLOAD_ROOT = \"https://github.com/ageron/handson-ml2/raw/master/datasets/jsb_chorales/\"\n", "FILENAME = \"jsb_chorales.tgz\"\n", "filepath = keras.utils.get_file(FILENAME,\n", " DOWNLOAD_ROOT + FILENAME,\n", " cache_subdir=\"datasets/jsb_chorales\",\n", " extract=True)" ] }, { "cell_type": "code", "execution_count": 83, "metadata": {}, "outputs": [], "source": [ "jsb_chorales_dir = Path(filepath).parent\n", "train_files = sorted(jsb_chorales_dir.glob(\"train/chorale_*.csv\"))\n", "valid_files = sorted(jsb_chorales_dir.glob(\"valid/chorale_*.csv\"))\n", "test_files = sorted(jsb_chorales_dir.glob(\"test/chorale_*.csv\"))" ] }, { "cell_type": "code", "execution_count": 84, "metadata": {}, "outputs": [], "source": [ "import pandas as pd\n", "\n", "def load_chorales(filepaths):\n", " return [pd.read_csv(filepath).values.tolist() for filepath in filepaths]\n", "\n", "train_chorales = load_chorales(train_files)\n", "valid_chorales = load_chorales(valid_files)\n", "test_chorales = load_chorales(test_files)" ] }, { "cell_type": "code", "execution_count": 85, "metadata": {}, "outputs": [], "source": [ "train_chorales[0]" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Notes range from 36 (C1 = C on octave 1) to 81 (A5 = A on octave 5), plus 0 for silence:" ] }, { "cell_type": "code", "execution_count": 86, "metadata": {}, "outputs": [], "source": [ "notes = set()\n", "for chorales in (train_chorales, valid_chorales, test_chorales):\n", " for chorale in chorales:\n", " for chord in chorale:\n", " notes |= set(chord)\n", "\n", "n_notes = len(notes)\n", "min_note = min(notes - {0})\n", "max_note = max(notes)\n", "\n", "assert min_note == 36\n", "assert max_note == 81" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Let's write a few functions to listen to these chorales (you don't need to understand the details here, and in fact there are certainly simpler ways to do this, for example using MIDI players, but I just wanted to have a bit of fun writing a synthesizer):" ] }, { "cell_type": "code", "execution_count": 87, "metadata": {}, "outputs": [], "source": [ "from IPython.display import Audio\n", "\n", "def notes_to_frequencies(notes):\n", " # Frequency doubles when you go up one octave; there are 12 semi-tones\n", " # per octave; Note A on octave 4 is 440 Hz, and it is note number 69.\n", " return 2 ** ((np.array(notes) - 69) / 12) * 440\n", "\n", "def frequencies_to_samples(frequencies, tempo, sample_rate):\n", " note_duration = 60 / tempo # the tempo is measured in beats per minutes\n", " # To reduce click sound at every beat, we round the frequencies to try to\n", " # get the samples close to zero at the end of each note.\n", " frequencies = np.round(note_duration * frequencies) / note_duration\n", " n_samples = int(note_duration * sample_rate)\n", " time = np.linspace(0, note_duration, n_samples)\n", " sine_waves = np.sin(2 * np.pi * frequencies.reshape(-1, 1) * time)\n", " # Removing all notes with frequencies ≤ 9 Hz (includes note 0 = silence)\n", " sine_waves *= (frequencies > 9.).reshape(-1, 1)\n", " return sine_waves.reshape(-1)\n", "\n", "def chords_to_samples(chords, tempo, sample_rate):\n", " freqs = notes_to_frequencies(chords)\n", " freqs = np.r_[freqs, freqs[-1:]] # make last note a bit longer\n", " merged = np.mean([frequencies_to_samples(melody, tempo, sample_rate)\n", " for melody in freqs.T], axis=0)\n", " n_fade_out_samples = sample_rate * 60 // tempo # fade out last note\n", " fade_out = np.linspace(1., 0., n_fade_out_samples)**2\n", " merged[-n_fade_out_samples:] *= fade_out\n", " return merged\n", "\n", "def play_chords(chords, tempo=160, amplitude=0.1, sample_rate=44100, filepath=None):\n", " samples = amplitude * chords_to_samples(chords, tempo, sample_rate)\n", " if filepath:\n", " from scipy.io import wavfile\n", " samples = (2**15 * samples).astype(np.int16)\n", " wavfile.write(filepath, sample_rate, samples)\n", " return display(Audio(filepath))\n", " else:\n", " return display(Audio(samples, rate=sample_rate))" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now let's listen to a few chorales:" ] }, { "cell_type": "code", "execution_count": 88, "metadata": {}, "outputs": [], "source": [ "for index in range(3):\n", " play_chords(train_chorales[index])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Divine! :)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "In order to be able to generate new chorales, we want to train a model that can predict the next chord given all the previous chords. If we naively try to predict the next chord in one shot, predicting all 4 notes at once, we run the risk of getting notes that don't go very well together (believe me, I tried). It's much better and simpler to predict one note at a time. So we will need to preprocess every chorale, turning each chord into an arpegio (i.e., a sequence of notes rather than notes played simultaneuously). So each chorale will be a long sequence of notes (rather than chords), and we can just train a model that can predict the next note given all the previous notes. We will use a sequence-to-sequence approach, where we feed a window to the neural net, and it tries to predict that same window shifted one time step into the future.\n", "\n", "We will also shift the values so that they range from 0 to 46, where 0 represents silence, and values 1 to 46 represent notes 36 (C1) to 81 (A5).\n", "\n", "And we will train the model on windows of 128 notes (i.e., 32 chords).\n", "\n", "Since the dataset fits in memory, we could preprocess the chorales in RAM using any Python code we like, but I will demonstrate here how to do all the preprocessing using tf.data (there will be more details about creating windows using tf.data in the next chapter)." ] }, { "cell_type": "code", "execution_count": 89, "metadata": {}, "outputs": [], "source": [ "def create_target(batch):\n", " X = batch[:, :-1]\n", " Y = batch[:, 1:] # predict next note in each arpegio, at each step\n", " return X, Y\n", "\n", "def preprocess(window):\n", " window = tf.where(window == 0, window, window - min_note + 1) # shift values\n", " return tf.reshape(window, [-1]) # convert to arpegio\n", "\n", "def bach_dataset(chorales, batch_size=32, shuffle_buffer_size=None,\n", " window_size=32, window_shift=16, cache=True):\n", " def batch_window(window):\n", " return window.batch(window_size + 1)\n", "\n", " def to_windows(chorale):\n", " dataset = tf.data.Dataset.from_tensor_slices(chorale)\n", " dataset = dataset.window(window_size + 1, window_shift, drop_remainder=True)\n", " return dataset.flat_map(batch_window)\n", "\n", " chorales = tf.ragged.constant(chorales, ragged_rank=1)\n", " dataset = tf.data.Dataset.from_tensor_slices(chorales)\n", " dataset = dataset.flat_map(to_windows).map(preprocess)\n", " if cache:\n", " dataset = dataset.cache()\n", " if shuffle_buffer_size:\n", " dataset = dataset.shuffle(shuffle_buffer_size)\n", " dataset = dataset.batch(batch_size)\n", " dataset = dataset.map(create_target)\n", " return dataset.prefetch(1)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now let's create the training set, the validation set and the test set:" ] }, { "cell_type": "code", "execution_count": 90, "metadata": {}, "outputs": [], "source": [ "train_set = bach_dataset(train_chorales, shuffle_buffer_size=1000)\n", "valid_set = bach_dataset(valid_chorales)\n", "test_set = bach_dataset(test_chorales)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now let's create the model:\n", "\n", "* We could feed the note values directly to the model, as floats, but this would probably not give good results. Indeed, the relationships between notes are not that simple: for example, if you replace a C3 with a C4, the melody will still sound fine, even though these notes are 12 semi-tones apart (i.e., one octave). Conversely, if you replace a C3 with a C\\#3, it's very likely that the chord will sound horrible, despite these notes being just next to each other. So we will use an `Embedding` layer to convert each note to a small vector representation (see Chapter 16 for more details on embeddings). We will use 5-dimensional embeddings, so the output of this first layer will have a shape of `[batch_size, window_size, 5]`.\n", "* We will then feed this data to a small WaveNet-like neural network, composed of a stack of 4 `Conv1D` layers with doubling dilation rates. We will intersperse these layers with `BatchNormalization` layers for faster better convergence.\n", "* Then one `LSTM` layer to try to capture long-term patterns.\n", "* And finally a `Dense` layer to produce the final note probabilities. It will predict one probability for each chorale in the batch, for each time step, and for each possible note (including silence). So the output shape will be `[batch_size, window_size, 47]`." ] }, { "cell_type": "code", "execution_count": 91, "metadata": {}, "outputs": [], "source": [ "n_embedding_dims = 5\n", "\n", "model = keras.models.Sequential([\n", " keras.layers.Embedding(input_dim=n_notes, output_dim=n_embedding_dims,\n", " input_shape=[None]),\n", " keras.layers.Conv1D(32, kernel_size=2, padding=\"causal\", activation=\"relu\"),\n", " keras.layers.BatchNormalization(),\n", " keras.layers.Conv1D(48, kernel_size=2, padding=\"causal\", activation=\"relu\", dilation_rate=2),\n", " keras.layers.BatchNormalization(),\n", " keras.layers.Conv1D(64, kernel_size=2, padding=\"causal\", activation=\"relu\", dilation_rate=4),\n", " keras.layers.BatchNormalization(),\n", " keras.layers.Conv1D(96, kernel_size=2, padding=\"causal\", activation=\"relu\", dilation_rate=8),\n", " keras.layers.BatchNormalization(),\n", " keras.layers.LSTM(256, return_sequences=True),\n", " keras.layers.Dense(n_notes, activation=\"softmax\")\n", "])\n", "\n", "model.summary()" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now we're ready to compile and train the model!" ] }, { "cell_type": "code", "execution_count": 92, "metadata": {}, "outputs": [], "source": [ "optimizer = keras.optimizers.Nadam(lr=1e-3)\n", "model.compile(loss=\"sparse_categorical_crossentropy\", optimizer=optimizer,\n", " metrics=[\"accuracy\"])\n", "model.fit(train_set, epochs=20, validation_data=valid_set)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "I have not done much hyperparameter search, so feel free to iterate on this model now and try to optimize it. For example, you could try removing the `LSTM` layer and replacing it with `Conv1D` layers. You could also play with the number of layers, the learning rate, the optimizer, and so on." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Once you're satisfied with the performance of the model on the validation set, you can save it and evaluate it one last time on the test set:" ] }, { "cell_type": "code", "execution_count": 93, "metadata": {}, "outputs": [], "source": [ "model.save(\"my_bach_model.h5\")\n", "model.evaluate(test_set)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**Note:** There's no real need for a test set in this exercise, since we will perform the final evaluation by just listening to the music produced by the model. So if you want, you can add the test set to the train set, and train the model again, hopefully getting a slightly better model." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now let's write a function that will generate a new chorale. We will give it a few seed chords, it will convert them to arpegios (the format expected by the model), and use the model to predict the next note, then the next, and so on. In the end, it will group the notes 4 by 4 to create chords again, and return the resulting chorale." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**Warning**: `model.predict_classes(X)` is deprecated. It is replaced with `np.argmax(model.predict(X), axis=-1)`." ] }, { "cell_type": "code", "execution_count": 94, "metadata": {}, "outputs": [], "source": [ "def generate_chorale(model, seed_chords, length):\n", " arpegio = preprocess(tf.constant(seed_chords, dtype=tf.int64))\n", " arpegio = tf.reshape(arpegio, [1, -1])\n", " for chord in range(length):\n", " for note in range(4):\n", " #next_note = model.predict_classes(arpegio)[:1, -1:]\n", " next_note = np.argmax(model.predict(arpegio), axis=-1)[:1, -1:]\n", " arpegio = tf.concat([arpegio, next_note], axis=1)\n", " arpegio = tf.where(arpegio == 0, arpegio, arpegio + min_note - 1)\n", " return tf.reshape(arpegio, shape=[-1, 4])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "To test this function, we need some seed chords. Let's use the first 8 chords of one of the test chorales (it's actually just 2 different chords, each played 4 times):" ] }, { "cell_type": "code", "execution_count": 95, "metadata": {}, "outputs": [], "source": [ "seed_chords = test_chorales[2][:8]\n", "play_chords(seed_chords, amplitude=0.2)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Now we are ready to generate our first chorale! Let's ask the function to generate 56 more chords, for a total of 64 chords, i.e., 16 bars (assuming 4 chords per bar, i.e., a 4/4 signature):" ] }, { "cell_type": "code", "execution_count": 96, "metadata": {}, "outputs": [], "source": [ "new_chorale = generate_chorale(model, seed_chords, 56)\n", "play_chords(new_chorale)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "This approach has one major flaw: it is often too conservative. Indeed, the model will not take any risk, it will always choose the note with the highest score, and since repeating the previous note generally sounds good enough, it's the least risky option, so the algorithm will tend to make notes last longer and longer. Pretty boring. Plus, if you run the model multiple times, it will always generate the same melody.\n", "\n", "So let's spice things up a bit! Instead of always picking the note with the highest score, we will pick the next note randomly, according to the predicted probabilities. For example, if the model predicts a C3 with 75% probability, and a G3 with a 25% probability, then we will pick one of these two notes randomly, with these probabilities. We will also add a `temperature` parameter that will control how \"hot\" (i.e., daring) we want the system to feel. A high temperature will bring the predicted probabilities closer together, reducing the probability of the likely notes and increasing the probability of the unlikely ones." ] }, { "cell_type": "code", "execution_count": 97, "metadata": {}, "outputs": [], "source": [ "def generate_chorale_v2(model, seed_chords, length, temperature=1):\n", " arpegio = preprocess(tf.constant(seed_chords, dtype=tf.int64))\n", " arpegio = tf.reshape(arpegio, [1, -1])\n", " for chord in range(length):\n", " for note in range(4):\n", " next_note_probas = model.predict(arpegio)[0, -1:]\n", " rescaled_logits = tf.math.log(next_note_probas) / temperature\n", " next_note = tf.random.categorical(rescaled_logits, num_samples=1)\n", " arpegio = tf.concat([arpegio, next_note], axis=1)\n", " arpegio = tf.where(arpegio == 0, arpegio, arpegio + min_note - 1)\n", " return tf.reshape(arpegio, shape=[-1, 4])" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Let's generate 3 chorales using this new function: one cold, one medium, and one hot (feel free to experiment with other seeds, lengths and temperatures). The code saves each chorale to a separate file. You can run these cells over an over again until you generate a masterpiece!\n", "\n", "**Please share your most beautiful generated chorale with me on Twitter @aureliengeron, I would really appreciate it! :))**" ] }, { "cell_type": "code", "execution_count": 98, "metadata": { "scrolled": true }, "outputs": [], "source": [ "new_chorale_v2_cold = generate_chorale_v2(model, seed_chords, 56, temperature=0.8)\n", "play_chords(new_chorale_v2_cold, filepath=\"bach_cold.wav\")" ] }, { "cell_type": "code", "execution_count": 99, "metadata": {}, "outputs": [], "source": [ "new_chorale_v2_medium = generate_chorale_v2(model, seed_chords, 56, temperature=1.0)\n", "play_chords(new_chorale_v2_medium, filepath=\"bach_medium.wav\")" ] }, { "cell_type": "code", "execution_count": 100, "metadata": {}, "outputs": [], "source": [ "new_chorale_v2_hot = generate_chorale_v2(model, seed_chords, 56, temperature=1.5)\n", "play_chords(new_chorale_v2_hot, filepath=\"bach_hot.wav\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Lastly, you can try a fun social experiment: send your friends a few of your favorite generated chorales, plus the real chorale, and ask them to guess which one is the real one!" ] }, { "cell_type": "code", "execution_count": 101, "metadata": {}, "outputs": [], "source": [ "play_chords(test_chorales[2][:64], filepath=\"bach_test_4.wav\")" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.7.9" }, "nav_menu": {}, "toc": { "navigate_menu": true, "number_sections": true, "sideBar": true, "threshold": 6, "toc_cell": false, "toc_section_display": "block", "toc_window_display": false } }, "nbformat": 4, "nbformat_minor": 4 }