[1]AI-engineering-Keras.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# AI ENGINEERING IBM COURSE TEXTBOOK - KERAS"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import keras\n",
    "import tensorflow as tf\n",
    "from keras.models import Sequential\n",
    "from keras.layers import Dense\n",
    "from keras.utils import to_categorical\n",
    "from keras.models import load_model\n",
    "from tensorflow.keras.models import Model \n",
    "from tensorflow.keras.layers import BatchNormalization\n",
    "from tensorflow.keras.layers import Dropout\n",
    "from tensorflow.keras.layers import Softmax\n",
    "from keras.layers.convolutional import Conv2D # to add convolutional layers\n",
    "from keras.layers.convolutional import MaxPooling2D # to add pooling layers\n",
    "from keras.layers import Flatten # to flatten data for fully connected layers\n",
    "from PIL import Image, ImageDraw\n",
    "from tensorflow.keras.preprocessing.image import ImageDataGenerator, load_img, img_to_array\n",
    "from tensorflow.keras.utils import array_to_img\n",
    "from tensorflow.keras.applications import VGG16\n",
    "from tensorflow.keras.layers import Input, Conv2D, Conv2DTranspose, UpSampling2D\n",
    "from tensorflow.keras.layers import LayerNormalization\n",
    "from tensorflow.keras.callbacks import EarlyStopping\n",
    "from tensorflow.keras.datasets import mnist\n",
    "import keras_tuner as kt\n",
    "import gym\n",
    "from collections import deque"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Initializing a network without framework"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "n = 2 # number of inputs\n",
    "num_hidden_layers = 2 # number of hidden layers\n",
    "m = [2, 2] # number of nodes in each hidden layer\n",
    "num_nodes_output = 1 # number of nodes in the output layer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np # import the Numpy library\n",
    "\n",
    "def initialize_network(num_inputs, num_hidden_layers, num_nodes_hidden, num_nodes_output):\n",
    "    \n",
    "    num_nodes_previous = num_inputs # number of nodes in the previous layer\n",
    "\n",
    "    network = {}\n",
    "    \n",
    "    # loop through each layer and randomly initialize the weights and biases associated with each layer\n",
    "    for layer in range(num_hidden_layers + 1):\n",
    "        \n",
    "        if layer == num_hidden_layers:\n",
    "            layer_name = 'output' # name last layer in the network output\n",
    "            num_nodes = num_nodes_output\n",
    "        else:\n",
    "            layer_name = 'layer_{}'.format(layer + 1) # otherwise give the layer a number\n",
    "            num_nodes = num_nodes_hidden[layer] \n",
    "        \n",
    "        # initialize weights and bias for each node\n",
    "        network[layer_name] = {}\n",
    "        for node in range(num_nodes):\n",
    "            node_name = 'node_{}'.format(node+1)\n",
    "            network[layer_name][node_name] = {\n",
    "                'weights': np.around(np.random.uniform(size=num_nodes_previous), decimals=2),\n",
    "                'bias': np.around(np.random.uniform(size=1), decimals=2),\n",
    "            }\n",
    "    \n",
    "        num_nodes_previous = num_nodes\n",
    "\n",
    "    return network # return the network\n",
    "\n",
    "def compute_weighted_sum(inputs, weights, bias):\n",
    "    # Convert inputs and weights to numpy arrays\n",
    "    inputs = np.array(inputs)\n",
    "    weights = np.array(weights)\n",
    "    return np.sum(inputs * weights) + bias\n",
    "\n",
    "#node activation using sigmoid function\n",
    "def node_activation(weighted_sum):\n",
    "    return 1.0 / (1.0 + np.exp(-1 * weighted_sum))\n",
    "\n",
    "#automate all process with this function\n",
    "def forward_propagate(network, inputs):\n",
    "    \n",
    "    layer_inputs = list(inputs) # start with the input layer as the input to the first hidden layer\n",
    "    \n",
    "    for layer in network:\n",
    "        \n",
    "        layer_data = network[layer]\n",
    "        \n",
    "        layer_outputs = [] \n",
    "        for layer_node in layer_data:\n",
    "        \n",
    "            node_data = layer_data[layer_node]\n",
    "        \n",
    "            # compute the weighted sum and the output of each node at the same time \n",
    "            node_output = node_activation(compute_weighted_sum(layer_inputs, node_data['weights'], node_data['bias']))\n",
    "            layer_outputs.append(np.around(node_output[0], decimals=4))\n",
    "            \n",
    "        if layer != 'output':\n",
    "            print('The outputs of the nodes in hidden layer number {} is {}'.format(layer.split('_')[1], layer_outputs))\n",
    "    \n",
    "        layer_inputs = layer_outputs # set the output of this layer to be the input to next layer\n",
    "\n",
    "    network_predictions = layer_outputs\n",
    "    return network_predictions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "#example of using compute_weighted_sum function with network dictionary\n",
    "node_weights = small_network['layer_1']['node_1']['weights']\n",
    "node_bias = small_network['layer_1']['node_1']['bias']\n",
    "\n",
    "weighted_sum = compute_weighted_sum(inputs, node_weights, node_bias)\n",
    "print('The weighted sum at the first node in the hidden layer is {}'.format(np.around(weighted_sum[0], decimals=4)))\n",
    "\n",
    "#and/or\n",
    "\n",
    "node_output  = node_activation(compute_weighted_sum(inputs, node_weights, node_bias))\n",
    "print('The output of the first node in the hidden layer is {}'.format(np.around(node_output[0], decimals=4)))\n",
    "\n",
    "#create random inputs\n",
    "inputs_01 = np.around(np.random.uniform(size=10), decimals=2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.2675\n",
      "0.5664790559676278\n"
     ]
    }
   ],
   "source": [
    "weighted_sum = compute_weighted_sum([0.5, -0.35], [0.55, 0.45], 0.15)\n",
    "print(weighted_sum)\n",
    "output = node_activation(weighted_sum)\n",
    "print(output)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Regression with Keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define regression model\n",
    "def regression_model():\n",
    "    # create model\n",
    "    model = Sequential()\n",
    "    model.add(Dense(50, activation='relu', input_shape=(n_cols,)))\n",
    "    model.add(Dense(50, activation='relu'))\n",
    "    model.add(Dense(1))\n",
    "    \n",
    "    # compile model\n",
    "    model.compile(optimizer='adam', loss='mean_squared_error')\n",
    "    return model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# build the model\n",
    "model = regression_model()\n",
    "\n",
    "# fit the model\n",
    "model.fit(predictors_norm, target, validation_split=0.3, epochs=100, verbose=2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Classification with Keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define classification model\n",
    "def classification_model():\n",
    "    # create model\n",
    "    model = Sequential()\n",
    "    model.add(Dense(num_pixels, activation='relu', input_shape=(num_pixels,)))\n",
    "    model.add(Dense(100, activation='relu'))\n",
    "    model.add(Dense(num_classes, activation='softmax'))\n",
    "    \n",
    "    # compile model\n",
    "    model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy'])\n",
    "    return model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# import the data\n",
    "from keras.datasets import mnist\n",
    "\n",
    "# read the data\n",
    "(X_train, y_train), (X_test, y_test) = mnist.load_data()\n",
    "\n",
    "X_train.shape\n",
    "#(60000, 28, 28)\n",
    "\n",
    "plt.imshow(X_train[0])\n",
    "\n",
    "#With conventional neural networks, we cannot feed in the image as input as is. So we need to flatten the images \n",
    "#into one-dimensional vectors, each of size 1 x (28 x 28) = 1 x 784.\n",
    "# flatten images into one-dimensional vector\n",
    "\n",
    "num_pixels = X_train.shape[1] * X_train.shape[2] # find size of one-dimensional vector\n",
    "\n",
    "X_train = X_train.reshape(X_train.shape[0], num_pixels).astype('float32') # flatten training images\n",
    "X_test = X_test.reshape(X_test.shape[0], num_pixels).astype('float32') # flatten test images\n",
    "\n",
    "#Since pixel values can range from 0 to 255, let's normalize the vectors to be between 0 and 1.\n",
    "# normalize inputs from 0-255 to 0-1\n",
    "X_train = X_train / 255\n",
    "X_test = X_test / 255\n",
    "\n",
    "#Finally, before we start building our model, remember that for classification we need to divide our target variable \n",
    "#into categories. We use the to_categorical function from the Keras Utilities package.\n",
    "# one hot encode outputs\n",
    "y_train = to_categorical(y_train)\n",
    "y_test = to_categorical(y_test)\n",
    "\n",
    "num_classes = y_test.shape[1]\n",
    "print(num_classes)\n",
    "\n",
    "# build the model\n",
    "model = classification_model()\n",
    "\n",
    "# fit the model (!!!training model can take up to 20 minutes!!!)\n",
    "model.fit(X_train, y_train, validation_data=(X_test, y_test), epochs=10, verbose=2)\n",
    "\n",
    "# evaluate the model\n",
    "scores = model.evaluate(X_test, y_test, verbose=0)\n",
    "\n",
    "#save model\n",
    "model.save('classification_model.h5')\n",
    "\n",
    "#load saved model\n",
    "pretrained_model = load_model('classification_model.h5')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Convolutional Neural Networks with Keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def convolutional_model():\n",
    "    \n",
    "    # create model\n",
    "    model = Sequential()\n",
    "    model.add(Conv2D(16, (5, 5), strides=(1, 1), activation='relu', input_shape=(28, 28, 1)))\n",
    "    model.add(MaxPooling2D(pool_size=(2, 2), strides=(2, 2)))\n",
    "    \n",
    "    model.add(Flatten())\n",
    "    model.add(Dense(100, activation='relu'))\n",
    "    model.add(Dense(num_classes, activation='softmax'))\n",
    "    \n",
    "    # compile model\n",
    "    model.compile(optimizer='adam', loss='categorical_crossentropy',  metrics=['accuracy'])\n",
    "    return model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# build the model\n",
    "model = convolutional_model()\n",
    "\n",
    "# fit the model\n",
    "model.fit(X_train, y_train, validation_data=(X_test, y_test), epochs=10, batch_size=200, verbose=2)\n",
    "\n",
    "# evaluate the model\n",
    "scores = model.evaluate(X_test, y_test, verbose=0)\n",
    "print(\"Accuracy: {} \\n Error: {}\".format(scores[1], 100-scores[1]*100))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Functional API in Keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create input layer. Assume that inputs are of vector size 20\n",
    "input_layer = Input(shape=(20,))\n",
    "print(input_layer)\n",
    "\n",
    "#add hidden layers\n",
    "hidden_layer1 = Dense(64, activation='relu')(input_layer) #creates a dense (fully connected) layer with 64 units and ReLU activation function.\n",
    "hidden_layer2 = Dense(64, activation='relu')(hidden_layer1)\n",
    "\n",
    "#define output layer\n",
    "output_layer = Dense(1, activation='sigmoid')(hidden_layer2) #creates a dense layer with 1 unit and a sigmoid activation function, suitable for binary classification.\n",
    "\n",
    "#create model and give summary\n",
    "model = Model(inputs=input_layer, outputs=output_layer)\n",
    "model.summary()\n",
    "\n",
    "#Before training the model, you need to compile it. You will specify the loss function, optimizer, and evaluation metrics.\n",
    "model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])\n",
    "\n",
    "#train model (example)\n",
    "import numpy as np \n",
    "X_train = np.random.rand(1000, 20) \n",
    "y_train = np.random.randint(2, size=(1000, 1)) \n",
    "model.fit(X_train, y_train, epochs=10, batch_size=32)\n",
    "\n",
    "# Example test data (in practice, use real dataset) \n",
    "loss, accuracy = model.evaluate(X_test, y_test)\n",
    "print(f'Test loss: {loss}')\n",
    "print(f'Test accuracy: {accuracy}')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dropout and Batch Normalization\n",
    "\n",
    "Before we proceed with the practice exercise, let's briefly discuss two important techniques often used to improve the performance of neural networks: **Dropout Layers** and **Batch Normalization**.\n",
    "\n",
    "#### Dropout Layers\n",
    "\n",
    "Dropout is a regularization technique that helps prevent overfitting in neural networks. During training, Dropout randomly sets a fraction of input units to zero at each update cycle. This prevents the model from becoming overly reliant on any specific neurons, which encourages the network to learn more robust features that generalize better to unseen data.\n",
    "\n",
    "**Key points:**\n",
    "- Dropout is only applied during training, not during inference.\n",
    "- The dropout rate is a hyperparameter that determines the fraction of neurons to drop.\n",
    "\n",
    "\n",
    "#### Batch Normalization\n",
    "\n",
    "Batch Normalization is a technique used to improve the training stability and speed of neural networks. It normalizes the output of a previous layer by re-centering and re-scaling the data, which helps in stabilizing the learning process. By reducing the internal covariate shift (the changes in the distribution of layer inputs), batch normalization allows the model to use higher learning rates, which often speeds up convergence.\n",
    "\n",
    "**Key Points:**\n",
    "\n",
    "- Batch normalization works by normalizing the inputs to each layer to have a mean of zero and a variance of one.\n",
    "- It is applied during both training and inference, although its behavior varies slightly between the two phases.\n",
    "- Batch normalization layers also introduce two learnable parameters that allow the model to scale and - shift the normalized output, which helps in restoring the model's representational power."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#example of adding dropout layer in Keras\n",
    "from tensorflow.keras.layers import Dropout, Dense, Input\n",
    "from tensorflow.keras.models import Model\n",
    "\n",
    "# Define the input layer\n",
    "input_layer = Input(shape=(20,))\n",
    "\n",
    "# Add a hidden layer\n",
    "hidden_layer = Dense(64, activation='relu')(input_layer)\n",
    "\n",
    "# Add a Dropout layer\n",
    "dropout_layer = Dropout(rate=0.5)(hidden_layer)\n",
    "\n",
    "# Add another hidden layer after Dropout\n",
    "hidden_layer2 = Dense(64, activation='relu')(dropout_layer)\n",
    "\n",
    "# Define the output layer\n",
    "output_layer = Dense(1, activation='sigmoid')(hidden_layer2)\n",
    "\n",
    "# Create the model\n",
    "model = Model(inputs=input_layer, outputs=output_layer)\n",
    "\n",
    "# Summary of the model\n",
    "model.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#example of adding batch normalization in Keras\n",
    "from tensorflow.keras.layers import BatchNormalization, Dense, Input\n",
    "from tensorflow.keras.models import Model\n",
    "\n",
    "# Define the input layer\n",
    "input_layer = Input(shape=(20,))\n",
    "\n",
    "# Add a hidden layer\n",
    "hidden_layer = Dense(64, activation='relu')(input_layer)\n",
    "\n",
    "# Add a BatchNormalization layer\n",
    "batch_norm_layer = BatchNormalization()(hidden_layer)\n",
    "\n",
    "# Add another hidden layer after BatchNormalization\n",
    "hidden_layer2 = Dense(64, activation='relu')(batch_norm_layer)\n",
    "\n",
    "# Define the output layer\n",
    "output_layer = Dense(1, activation='sigmoid')(hidden_layer2)\n",
    "\n",
    "# Create the model\n",
    "model = Model(inputs=input_layer, outputs=output_layer)\n",
    "\n",
    "# Summary of the model\n",
    "model.summary()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Example of full model using Keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create input layer. Assume that inputs are of vector size 20\n",
    "input_layer = Input(shape=(20,))\n",
    "\n",
    "#add hidden layer\n",
    "hidden_layer1 = Dense(64, activation='relu')(input_layer)\n",
    "\n",
    "#add batch layer\n",
    "batch_norm_layer1 = BatchNormalization()(hidden_layer1)\n",
    "\n",
    "#add dropout layer\n",
    "dropout_layer1 = Dropout(rate=0.5)(batch_norm_layer1)\n",
    "\n",
    "#add hidden layer\n",
    "hidden_layer2 = Dense(64, activation='relu')(dropout_layer1)\n",
    "\n",
    "#add batch layer\n",
    "batch_norm_layer2 = BatchNormalization()(hidden_layer2)\n",
    "\n",
    "#add dropout layer\n",
    "dropout_layer2 = Dropout(rate=0.5)(batch_norm_layer2)\n",
    "\n",
    "#define output layer\n",
    "output_layer = Dense(1, activation='sigmoid')(dropout_layer2)\n",
    "\n",
    "#recreate model\n",
    "model = Model(inputs=input_layer, outputs=output_layer)\n",
    "\n",
    "#model summary\n",
    "model.summary()\n",
    "\n",
    "#recompile model\n",
    "model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])\n",
    "\n",
    "#train model\n",
    "model.fit(X_train, y_train, epochs=10, batch_size=32)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating Custom Layers and Models"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class CustomDenseLayer(Layer):\n",
    "    def __init__(self, units=32):\n",
    "        super(CustomDenseLayer, self).__init__()\n",
    "        self.units = units\n",
    "\n",
    "    def build(self, input_shape):\n",
    "        self.w = self.add_weight(shape=(input_shape[-1], self.units),\n",
    "                                 initializer='random_normal',\n",
    "                                 trainable=True)\n",
    "        self.b = self.add_weight(shape=(self.units,),\n",
    "                                 initializer='zeros',\n",
    "                                 trainable=True)\n",
    "    def call(self, inputs):\n",
    "        return tf.nn.relu(tf.matmul(inputs, self.w) + self.b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tensorflow.keras.layers import Softmax\n",
    "\n",
    "# Define the model with Softmax in the output layer\n",
    "model = Sequential([\n",
    "    CustomDenseLayer(128),\n",
    "    CustomDenseLayer(10),  # Hidden layer with ReLU activation\n",
    "    Softmax()              # Output layer with Softmax activation for multi-class classification\n",
    "])\n",
    "\n",
    "#The Softmax activation function is used in the output layer for multi-class classification tasks, ensuring the model \n",
    "# outputs probabilities that sum up to 1 for each class, which aligns with categorical cross-entropy as the loss function. \n",
    "# This adjustment ensures the model is optimized correctly for multi-class classification.\n",
    "\n",
    "model.compile(optimizer='adam', loss='categorical_crossentropy')\n",
    "print(\"Model summary before building:\")\n",
    "model.summary()\n",
    "\n",
    "# Build the model to show parameters\n",
    "model.build((1000, 20))\n",
    "print(\"\\nModel summary after building:\")\n",
    "model.summary()\n",
    "\n",
    "# Visualize the model architecture\n",
    "plot_model(model, to_file=\"model_plot.png\", show_shapes=True, show_layer_names=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Advanced Data Augmentation with Keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from PIL import Image, ImageDraw\n",
    "\n",
    "# Create a blank white image\n",
    "image = Image.new('RGB', (224, 224), color = (255, 255, 255))\n",
    "\n",
    "# Draw a red square\n",
    "draw = ImageDraw.Draw(image)\n",
    "draw.rectangle([(50, 50), (174, 174)], fill=(255, 0, 0))\n",
    "\n",
    "# Save the image\n",
    "image.save('sample.jpg')\n",
    "\n",
    "# Load a sample image \n",
    "img_path = 'sample.jpg' \n",
    "img = load_img(img_path) \n",
    "x = img_to_array(img) \n",
    "x = np.expand_dims(x, axis=0) \n",
    "\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "from tensorflow.keras.preprocessing.image import ImageDataGenerator, load_img, img_to_array\n",
    "\n",
    "# Load the sample image\n",
    "img_path = 'sample.jpg'\n",
    "img = load_img(img_path)\n",
    "x = img_to_array(img)\n",
    "x = np.expand_dims(x, axis=0)\n",
    "\n",
    "# Create an instance of ImageDataGenerator with basic augmentations\n",
    "datagen = ImageDataGenerator(\n",
    "    rotation_range=40,\n",
    "    width_shift_range=0.2,\n",
    "    height_shift_range=0.2,\n",
    "    shear_range=0.2,\n",
    "    zoom_range=0.2,\n",
    "    horizontal_flip=True,\n",
    "    fill_mode='nearest'\n",
    ")\n",
    "\n",
    "# Generate batches of augmented images\n",
    "i = 0\n",
    "for batch in datagen.flow(x, batch_size=1):\n",
    "    plt.figure(i)\n",
    "    imgplot = plt.imshow(batch[0].astype('uint8'))\n",
    "    i += 1\n",
    "    if i % 4 == 0:\n",
    "        break\n",
    "\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create an instance of ImageDataGenerator with normalization options\n",
    "datagen = ImageDataGenerator(\n",
    "    featurewise_center=True,\n",
    "    featurewise_std_normalization=True,\n",
    "    samplewise_center=True,\n",
    "    samplewise_std_normalization=True\n",
    ")\n",
    "\n",
    "# Load the sample image again and fit the generator (normally done on the training set)\n",
    "datagen.fit(x)\n",
    "\n",
    "# Generate batches of normalized images\n",
    "i = 0\n",
    "for batch in datagen.flow(x, batch_size=1):\n",
    "    plt.figure(i)\n",
    "    imgplot = plt.imshow(batch[0].astype('uint8'))\n",
    "    i += 1\n",
    "    if i % 4 == 0:\n",
    "        break\n",
    "\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define a custom data augmentation function\n",
    "def add_random_noise(image):\n",
    "    noise = np.random.normal(0, 0.5, image.shape)\n",
    "    return image + noise\n",
    "\n",
    "# Create an instance of ImageDataGenerator with the custom augmentation\n",
    "datagen = ImageDataGenerator(preprocessing_function=add_random_noise)\n",
    "\n",
    "# Generate batches of augmented images with noise\n",
    "i = 0\n",
    "for batch in datagen.flow(x, batch_size=1):\n",
    "    plt.figure(i)\n",
    "    imgplot = plt.imshow(batch[0].astype('uint8'))\n",
    "    i += 1\n",
    "    if i % 4 == 0:\n",
    "        break\n",
    "\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Visualizing multiple augmented versions of the same image\n",
    "plt.figure(figsize=(10, 10))\n",
    "for i, batch in enumerate(datagen.flow(x, batch_size=1)):\n",
    "    if i >= 4:  # Show only 4 versions\n",
    "        break\n",
    "    plt.subplot(2, 2, i+1)\n",
    "    plt.imshow(batch[0].astype('uint8'))\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tensorflow.keras.utils import array_to_img\n",
    "\n",
    "# Create an instance of ImageDataGenerator with normalization options\n",
    "datagen = ImageDataGenerator(\n",
    "    featurewise_center=True,\n",
    "    featurewise_std_normalization=True,\n",
    "    samplewise_center=True,\n",
    "    samplewise_std_normalization=True\n",
    ")\n",
    "\n",
    "# Load the sample image again and fit the generator (normally done on the training set)\n",
    "datagen.fit(x_train)\n",
    "\n",
    "# Generate batches of normalized images\n",
    "i = 0\n",
    "for batch in datagen.flow(x_train, batch_size=1):\n",
    "    plt.figure(i)\n",
    "    imgplot = plt.imshow(array_to_img(batch[0]))\n",
    "    plt.title(f'Normalized Image {i + 1}')\n",
    "    i += 1\n",
    "    if i % 4 == 0:\n",
    "        break\n",
    "\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transfer Learning Implementation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load the VGG16 model pre-trained on ImageNet\n",
    "base_model = VGG16(weights='imagenet', include_top=False, input_shape=(224, 224, 3))\n",
    "\n",
    "# Freeze the base model layers\n",
    "for layer in base_model.layers:\n",
    "    layer.trainable = False\n",
    "\n",
    "# Create a new model and add the base model and new layers\n",
    "model = Sequential([\n",
    "    base_model,\n",
    "    Flatten(),\n",
    "    Dense(256, activation='relu'),\n",
    "    Dense(1, activation='sigmoid')  # Change to the number of classes you have\n",
    "])\n",
    "\n",
    "# Compile the model\n",
    "model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])\n",
    "\n",
    "# Create directories if they don't exist\n",
    "os.makedirs('sample_data/class_a', exist_ok=True)\n",
    "os.makedirs('sample_data/class_b', exist_ok=True)\n",
    "\n",
    "# Create 10 sample images for each class\n",
    "for i in range(10):\n",
    "    # Create a blank white image for class_a\n",
    "    img = Image.fromarray(np.ones((224, 224, 3), dtype=np.uint8) * 255)\n",
    "    img.save(f'sample_data/class_a/img_{i}.jpg')\n",
    "\n",
    "    # Create a blank black image for class_b\n",
    "    img = Image.fromarray(np.zeros((224, 224, 3), dtype=np.uint8))\n",
    "    img.save(f'sample_data/class_b/img_{i}.jpg')\n",
    "\n",
    "print(\"Sample images created in 'sample_data/'\")\n",
    "\n",
    "# Load and preprocess the dataset\n",
    "train_datagen = ImageDataGenerator(rescale=1./255)\n",
    "train_generator = train_datagen.flow_from_directory(\n",
    "    'sample_data',\n",
    "    target_size=(224, 224),\n",
    "    batch_size=32,\n",
    "    class_mode='binary'\n",
    ")\n",
    "\n",
    "# Verify if the generator has loaded images correctly\n",
    "print(f\"Found {train_generator.samples} images belonging to {train_generator.num_classes} classes.\")\n",
    "\n",
    "# Train the model\n",
    "if train_generator.samples > 0:\n",
    "    model.fit(train_generator, epochs=10)\n",
    "\n",
    "# Unfreeze the top layers of the base model \n",
    "\n",
    "for layer in base_model.layers[-4:]:\n",
    "    layer.trainable = True \n",
    "\n",
    "# Compile the model again \n",
    "model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy']) \n",
    "\n",
    "# Train the model again \n",
    "model.fit(train_generator, epochs=10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Validation generator and loss by epochs graph"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load and preprocess the dataset\n",
    "train_datagen = ImageDataGenerator(rescale=1./255, validation_split=0.2)\n",
    "train_generator = train_datagen.flow_from_directory(\n",
    "    'sample_data',\n",
    "    target_size=(224, 224),\n",
    "    batch_size=32,\n",
    "    class_mode='binary',\n",
    "    subset='training'\n",
    ")\n",
    "\n",
    "validation_generator = train_datagen.flow_from_directory(\n",
    "    'sample_data',\n",
    "    target_size=(224, 224),\n",
    "    batch_size=32,\n",
    "    class_mode='binary',\n",
    "    subset='validation'\n",
    ")\n",
    "\n",
    "#train the model with the validation data\n",
    "history = model.fit(train_generator, epochs=10, validation_data=validation_generator)\n",
    "\n",
    "# Plot training and validation loss\n",
    "plt.plot(history.history['loss'], label='Training Loss')\n",
    "plt.plot(history.history['val_loss'], label='Validation Loss')\n",
    "plt.title('Training and Validation Loss')\n",
    "plt.xlabel('Epochs')\n",
    "plt.ylabel('Loss')\n",
    "plt.legend()\n",
    "plt.show()\n",
    "\n",
    "# Evaluate the fine-tuned model on the test set\n",
    "test_loss, test_accuracy = model.evaluate(test_generator)\n",
    "print(f'Test Accuracy: {test_accuracy * 100:.2f}%')\n",
    "print(f'Test Loss: {test_loss:.4f}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Hyperparameter tuning\n",
    "\n",
    "Hyperparameter tuning is a crucial step in enhancing the performance of deep learning models. Below are some best practices for optimizing hyperparameters effectively:\n",
    "<br>\n",
    "\n",
    "Use Grid Search for Small Spaces: If the number of hyperparameters and their possible values is small, grid search can be a useful approach. It systematically explores all combinations but can become computationally expensive for larger search spaces.\n",
    "<br>\n",
    "\n",
    "Random search for larger spaces: In larger search spaces, random search is more efficient than grid search, as it randomly samples configurations. While it's less exhaustive, it often identifies a good set of hyperparameters faster.\n",
    "<br>\n",
    "\n",
    "Leverage bayesian optimization: Bayesian optimization methods, such as those used in Keras Tuner, model the model's performance as a function of the hyperparameters. This approach focuses the search on regions of the space that are more likely to yield better results.\n",
    "<br>\n",
    "\n",
    "Use early stopping: To avoid overfitting and reduce computational cost, apply early stopping during hyperparameter tuning. This halts training when performance on a validation set plateaus or declines.\n",
    "<br>\n",
    "\n",
    "Tune one parameter at a time: When starting, adjust one hyperparameter at a time (e.g., learning rate, batch size, or number of layers) while keeping the others constant. Once you have a clearer understanding, you can start tuning multiple parameters simultaneously.\n",
    "<br>\n",
    "\n",
    "Prioritize learning rate: The learning rate is often the most critical hyperparameter. Begin by optimizing the learning rate before focusing on others, such as the dropout rate or optimizer type.\n",
    "<br>\n",
    "\n",
    "Monitor overfitting: Employ techniques such as cross-validation and tracking validation loss to detect overfitting during hyperparameter tuning. Regularization methods such as L2 regularization or dropout can help mitigate overfitting.\n",
    "<br>\n",
    "\n",
    "By following these best practices, you can optimize your model's performance and avoid common pitfalls in hyperparameter tuning."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Practical Application of Transpose Convolution (Increasing quality of images)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_layer = Input(shape=(28, 28, 1))\n",
    "\n",
    "conv_layer = Conv2D(filters=32, kernel_size=(3, 3), activation='relu', padding='same')(input_layer) \n",
    "\n",
    "transpose_conv_layer = Conv2DTranspose(filters=1, kernel_size=(3, 3), activation='sigmoid', padding='same')(conv_layer)\n",
    "\n",
    "model = Model(inputs=input_layer, outputs=transpose_conv_layer)\n",
    "\n",
    "model.compile(optimizer='adam', loss='mean_squared_error', metrics=['accuracy'])\n",
    "\n",
    "history = model.fit(X_train, y_train, epochs=10, batch_size=32, validation_split=0.2)\n",
    "\n",
    "# Predict on test data \n",
    "y_pred = model.predict(X_test) \n",
    "\n",
    "# Plot some sample images \n",
    "\n",
    "n = 10 # Number of samples to display \n",
    "\n",
    "plt.figure(figsize=(20, 4))\n",
    "\n",
    "for i in range(n): \n",
    "\n",
    "    # Display original \n",
    "    ax = plt.subplot(2, n, i + 1) \n",
    "    plt.imshow(X_test[i].reshape(28, 28), cmap='gray')\n",
    "    plt.title(\"Original\") \n",
    "    plt.axis('off') \n",
    "    # Display reconstruction \n",
    "    ax = plt.subplot(2, n, i + 1 + n) \n",
    "    plt.imshow(y_pred[i].reshape(28, 28), cmap='gray')\n",
    "    plt.title(\"Reconstructed\")\n",
    "    plt.axis('off')\n",
    "\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Building Advanced Transformers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# MultiHeadSelfAttention class implements the self-attention mechanism with multiple attention heads.\n",
    "class MultiHeadSelfAttention(Layer): \n",
    "\n",
    "    def __init__(self, embed_dim, num_heads=8): \n",
    "        # Initialize the MultiHeadSelfAttention layer with specified embedding dimensions and number of heads\n",
    "        super(MultiHeadSelfAttention, self).__init__()\n",
    "        \n",
    "        # Set embedding dimension and number of attention heads\n",
    "        self.embed_dim = embed_dim \n",
    "        self.num_heads = num_heads \n",
    "        \n",
    "        # Projection dimension per head, calculated by dividing embedding dimension by number of heads\n",
    "        self.projection_dim = embed_dim // num_heads \n",
    "        \n",
    "        # Layers for dense projections of query, key, and value tensors\n",
    "        self.query_dense = Dense(embed_dim) \n",
    "        self.key_dense = Dense(embed_dim) \n",
    "        self.value_dense = Dense(embed_dim) \n",
    "        \n",
    "        # Layer to combine the heads' output into a single tensor\n",
    "        self.combine_heads = Dense(embed_dim) \n",
    "\n",
    "    def attention(self, query, key, value): \n",
    "        # Calculate the attention score as a dot product of query and key matrices\n",
    "        score = tf.matmul(query, key, transpose_b=True) \n",
    "        \n",
    "        # Scale the score by the square root of the key dimension to stabilize gradients\n",
    "        dim_key = tf.cast(tf.shape(key)[-1], tf.float32) \n",
    "        scaled_score = score / tf.math.sqrt(dim_key) \n",
    "        \n",
    "        # Apply softmax to obtain attention weights\n",
    "        weights = tf.nn.softmax(scaled_score, axis=-1) \n",
    "        \n",
    "        # Compute the weighted sum of the value vectors using the attention weights\n",
    "        output = tf.matmul(weights, value) \n",
    "        return output, weights \n",
    "\n",
    "    def split_heads(self, x, batch_size): \n",
    "        # Split the input tensor into multiple heads and reshape it\n",
    "        x = tf.reshape(x, (batch_size, -1, self.num_heads, self.projection_dim)) \n",
    "        \n",
    "        # Transpose the tensor to shape it as (batch_size, num_heads, seq_len, projection_dim)\n",
    "        return tf.transpose(x, perm=[0, 2, 1, 3]) \n",
    "\n",
    "    def call(self, inputs): \n",
    "        # Obtain the batch size from the input tensor\n",
    "        batch_size = tf.shape(inputs)[0] \n",
    "        \n",
    "        # Pass inputs through dense layers to generate query, key, and value matrices\n",
    "        query = self.query_dense(inputs) \n",
    "        key = self.key_dense(inputs) \n",
    "        value = self.value_dense(inputs) \n",
    "        \n",
    "        # Split query, key, and value tensors into multiple heads\n",
    "        query = self.split_heads(query, batch_size) \n",
    "        key = self.split_heads(key, batch_size) \n",
    "        value = self.split_heads(value, batch_size) \n",
    "        \n",
    "        # Compute attention output and weights using the attention mechanism\n",
    "        attention, _ = self.attention(query, key, value) \n",
    "        \n",
    "        # Transpose attention output back to shape (batch_size, seq_len, num_heads, projection_dim)\n",
    "        attention = tf.transpose(attention, perm=[0, 2, 1, 3]) \n",
    "        \n",
    "        # Concatenate heads' output into a single tensor of shape (batch_size, seq_len, embed_dim)\n",
    "        concat_attention = tf.reshape(attention, (batch_size, -1, self.embed_dim)) \n",
    "        \n",
    "        # Pass concatenated output through a dense layer to combine heads\n",
    "        output = self.combine_heads(concat_attention) \n",
    "        return output "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TransformerBlock class implements a transformer block layer, combining multi-head self-attention and a feed-forward network.\n",
    "class TransformerBlock(Layer): \n",
    "\n",
    "    def __init__(self, embed_dim, num_heads, ff_dim, rate=0.1): \n",
    "        # Initialize the TransformerBlock with embedding dimensions, number of heads, feed-forward dimension, and dropout rate\n",
    "        super(TransformerBlock, self).__init__()\n",
    "        \n",
    "        # Multi-head self-attention layer\n",
    "        self.att = MultiHeadSelfAttention(embed_dim, num_heads)\n",
    "        \n",
    "        # Feed-forward network (FFN) with two dense layers:\n",
    "        # First layer expands to `ff_dim` and applies ReLU activation\n",
    "        # Second layer reduces back to `embed_dim` without activation\n",
    "        self.ffn = tf.keras.Sequential([ \n",
    "            Dense(ff_dim, activation=\"relu\"), \n",
    "            Dense(embed_dim), \n",
    "        ])\n",
    "        \n",
    "        # Layer normalization applied after attention and feed-forward layers\n",
    "        self.layernorm1 = LayerNormalization(epsilon=1e-6) \n",
    "        self.layernorm2 = LayerNormalization(epsilon=1e-6) \n",
    "        \n",
    "        # Dropout layers to prevent overfitting during training\n",
    "        self.dropout1 = Dropout(rate) \n",
    "        self.dropout2 = Dropout(rate) \n",
    "\n",
    "    def call(self, inputs, training):\n",
    "        # Apply multi-head self-attention to the inputs\n",
    "        attn_output = self.att(inputs)\n",
    "        \n",
    "        # Apply dropout to attention output (if training)\n",
    "        attn_output = self.dropout1(attn_output, training=training)\n",
    "        \n",
    "        # Add residual connection and normalize after attention\n",
    "        out1 = self.layernorm1(inputs + attn_output)\n",
    "        \n",
    "        # Pass normalized output through feed-forward network (FFN)\n",
    "        ffn_output = self.ffn(out1)\n",
    "        \n",
    "        # Apply dropout to FFN output (if training)\n",
    "        ffn_output = self.dropout2(ffn_output, training=training)\n",
    "        \n",
    "        # Add residual connection and normalize after FFN, returning final output\n",
    "        return self.layernorm2(out1 + ffn_output)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# EncoderLayer class implements an encoder layer for a transformer model, \n",
    "# combining multi-head self-attention and a feed-forward network with residual connections.\n",
    "class EncoderLayer(Layer): \n",
    "\n",
    "    def __init__(self, embed_dim, num_heads, ff_dim, rate=0.1): \n",
    "        # Initialize the EncoderLayer with embedding dimensions, number of heads, feed-forward dimension, and dropout rate\n",
    "        super(EncoderLayer, self).__init__()\n",
    "        \n",
    "        # Multi-head self-attention layer\n",
    "        self.att = MultiHeadSelfAttention(embed_dim, num_heads)\n",
    "        \n",
    "        # Feed-forward network (FFN) with two dense layers:\n",
    "        # First layer expands to `ff_dim` and applies ReLU activation\n",
    "        # Second layer reduces back to `embed_dim` without activation\n",
    "        self.ffn = tf.keras.Sequential([ \n",
    "            Dense(ff_dim, activation=\"relu\"), \n",
    "            Dense(embed_dim), \n",
    "        ])\n",
    "        \n",
    "        # Layer normalization applied after attention and feed-forward layers\n",
    "        self.layernorm1 = LayerNormalization(epsilon=1e-6) \n",
    "        self.layernorm2 = LayerNormalization(epsilon=1e-6) \n",
    "        \n",
    "        # Dropout layers to prevent overfitting during training\n",
    "        self.dropout1 = Dropout(rate) \n",
    "        self.dropout2 = Dropout(rate) \n",
    "\n",
    "    def call(self, inputs, training): \n",
    "        # Apply multi-head self-attention to the inputs\n",
    "        attn_output = self.att(inputs)\n",
    "        \n",
    "        # Apply dropout to attention output (if training)\n",
    "        attn_output = self.dropout1(attn_output, training=training)\n",
    "        \n",
    "        # Add residual connection and normalize after attention\n",
    "        out1 = self.layernorm1(inputs + attn_output)\n",
    "        \n",
    "        # Pass normalized output through feed-forward network (FFN)\n",
    "        ffn_output = self.ffn(out1)\n",
    "        \n",
    "        # Apply dropout to FFN output (if training)\n",
    "        ffn_output = self.dropout2(ffn_output, training=training)\n",
    "        \n",
    "        # Add residual connection and normalize after FFN, returning final output\n",
    "        return self.layernorm2(out1 + ffn_output)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Example usage \n",
    "embed_dim = 128 \n",
    "num_heads = 8 \n",
    "ff_dim = 512 \n",
    "num_layers = 4 \n",
    "\n",
    "transformer_encoder = TransformerEncoder(num_layers, embed_dim, num_heads, ff_dim) \n",
    "inputs = tf.random.uniform((1, 100, embed_dim)) \n",
    "outputs = transformer_encoder(inputs, training=False)  # Use keyword argument for 'training' \n",
    "print(outputs.shape)  # Should print (1, 100, 128)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define the necessary parameters \n",
    "embed_dim = 128\n",
    "num_heads = 8\n",
    "ff_dim = 512\n",
    "num_layers = 4\n",
    "\n",
    "# Define the Transformer Encoder \n",
    "transformer_encoder = TransformerEncoder(num_layers, embed_dim, num_heads, ff_dim) \n",
    "\n",
    "# Build the model \n",
    "input_shape = (X.shape[1], X.shape[2]) \n",
    "inputs = tf.keras.Input(shape=input_shape) \n",
    "\n",
    "# Project the inputs to the embed_dim \n",
    "x = tf.keras.layers.Dense(embed_dim)(inputs) \n",
    "encoder_outputs = transformer_encoder(x) \n",
    "flatten = tf.keras.layers.Flatten()(encoder_outputs) \n",
    "outputs = tf.keras.layers.Dense(1)(flatten) \n",
    "model = tf.keras.Model(inputs, outputs) \n",
    "\n",
    "# Compile the model \n",
    "model.compile(optimizer='adam', loss='mse') \n",
    "\n",
    "# Summary of the model \n",
    "model.summary()\n",
    "\n",
    "# Train the model\n",
    "model.fit(X, Y, epochs=20, batch_size=32)\n",
    "\n",
    "# Make predictions \n",
    "predictions = model.predict(X) \n",
    "predictions = scaler.inverse_transform(predictions) \n",
    " \n",
    "# Plot the predictions\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "plt.plot(data, label='True Data')\n",
    "plt.plot(np.arange(time_step, time_step + len(predictions)), predictions, label='Predictions')\n",
    "plt.xlabel('Time')\n",
    "plt.ylabel('Stock Price')\n",
    "plt.legend()\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Implementing Transformers for Text Generation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load the dataset \n",
    "path_to_file = get_file('shakespeare.txt', 'https://storage.googleapis.com/download.tensorflow.org/data/shakespeare.txt') \n",
    "text = open(path_to_file, 'rb').read().decode(encoding='utf-8') \n",
    "\n",
    "# Preview the dataset \n",
    "print(text[:1000])\n",
    "\n",
    "# Preprocess the dataset \n",
    "vocab_size = 10000\n",
    "seq_length = 100\n",
    "\n",
    "# Adapt TextVectorization to full text\n",
    "vectorizer = TextVectorization(max_tokens=vocab_size, output_mode='int')\n",
    "text_ds = tf.data.Dataset.from_tensor_slices([text]).batch(1)\n",
    "vectorizer.adapt(text_ds)\n",
    "\n",
    "# Vectorize the text\n",
    "vectorized_text = vectorizer([text])[0]\n",
    "print(\"Vectorized text shape:\", vectorized_text.shape)\n",
    "print(\"First 10 vectorized tokens:\", vectorized_text.numpy()[:10])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This function takes a long text and breaks it down into smaller input-target sequences.\n",
    "# Each sequence is `seq_length` characters long, helping the model learn to predict the next character.\n",
    "def create_sequences(text, seq_length): \n",
    "    input_seqs = []   # List to hold our input sequences\n",
    "    target_seqs = []  # List to hold our target (or next-character) sequences\n",
    "    \n",
    "    # Loop over the text to create sequences of specified length\n",
    "    # Why? This lets the model learn patterns over a fixed sequence length.\n",
    "    for i in range(len(text) - seq_length): \n",
    "        # Create an input sequence by taking `seq_length` characters from the text, starting from index i\n",
    "        input_seq = text[i:i + seq_length]\n",
    "        \n",
    "        # Create the target sequence, which is the next character after each input sequence\n",
    "        target_seq = text[i + 1:i + seq_length + 1] \n",
    "        \n",
    "        # Add both input and target sequences to their respective lists\n",
    "        input_seqs.append(input_seq) \n",
    "        target_seqs.append(target_seq)\n",
    "    \n",
    "    # Convert the lists to numpy arrays, a format that is easier to work with for ML frameworks\n",
    "    return np.array(input_seqs), np.array(target_seqs) \n",
    "\n",
    "# Generate sequences using the function, from vectorized text data.\n",
    "# `X` holds input sequences; `Y` holds the target sequences the model should learn to predict.\n",
    "X, Y = create_sequences(vectorized_text.numpy(), seq_length)\n",
    "\n",
    "# Check how many sequences were created, so we know the function worked\n",
    "print(\"Number of sequences generated:\", len(X))\n",
    "\n",
    "# Print a sample sequence to understand what the data looks like\n",
    "print(\"Sample input sequence:\", X[0] if len(X) > 0 else \"No sequences generated\")\n",
    "\n",
    "# Ensure that both `X` and `Y` contain data. If they are empty, there was an issue in sequence generation.\n",
    "assert X.size > 0, \"Input data X is empty\" \n",
    "assert Y.size > 0, \"Target data Y is empty\" \n",
    "\n",
    "# Convert the numpy arrays to TensorFlow tensors, as required for model training in TensorFlow\n",
    "X = tf.convert_to_tensor(X) \n",
    "Y = tf.convert_to_tensor(Y) \n",
    "\n",
    "# Print the shapes of `X` and `Y` to confirm they are in the expected dimensions\n",
    "print(\"Shape of X:\", X.shape) \n",
    "print(\"Shape of Y:\", Y.shape)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import necessary layers and Model from TensorFlow Keras for building a Transformer model\n",
    "from tensorflow.keras.layers import Embedding, MultiHeadAttention, Dense, LayerNormalization, Dropout\n",
    "from tensorflow.keras.models import Model\n",
    "\n",
    "# TransformerBlock class defines a single Transformer block layer\n",
    "# Each block consists of multi-head attention and feed-forward layers with normalization and dropout\n",
    "class TransformerBlock(tf.keras.layers.Layer):\n",
    "    def __init__(self, embed_dim, num_heads, ff_dim, rate=0.1):\n",
    "        super(TransformerBlock, self).__init__()\n",
    "        \n",
    "        # Multi-head self-attention layer, allowing the model to focus on different parts of input simultaneously\n",
    "        self.att = MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)\n",
    "        \n",
    "        # Feed-forward network with two dense layers\n",
    "        # The first layer expands to ff_dim with ReLU activation, and the second layer compresses back to embed_dim\n",
    "        self.ffn = tf.keras.Sequential([\n",
    "            Dense(ff_dim, activation=\"relu\"),\n",
    "            Dense(embed_dim),\n",
    "        ])\n",
    "        \n",
    "        # Layer normalization to stabilize and improve training\n",
    "        self.layernorm1 = LayerNormalization(epsilon=1e-6)\n",
    "        self.layernorm2 = LayerNormalization(epsilon=1e-6)\n",
    "        \n",
    "        # Dropout layers to reduce overfitting during training\n",
    "        self.dropout1 = Dropout(rate)\n",
    "        self.dropout2 = Dropout(rate)\n",
    "\n",
    "    def call(self, inputs, training=False):\n",
    "        # Calculate attention output and apply dropout\n",
    "        attn_output = self.att(inputs, inputs)\n",
    "        attn_output = self.dropout1(attn_output, training=training)\n",
    "        \n",
    "        # Add residual connection and normalize the result\n",
    "        out1 = self.layernorm1(inputs + attn_output)\n",
    "        \n",
    "        # Apply feed-forward network and dropout\n",
    "        ffn_output = self.ffn(out1)\n",
    "        ffn_output = self.dropout2(ffn_output, training=training)\n",
    "        \n",
    "        # Add residual connection and normalize the final output\n",
    "        return self.layernorm2(out1 + ffn_output)\n",
    "\n",
    "# TransformerModel class builds the entire Transformer model\n",
    "class TransformerModel(Model):  # Model is now properly imported\n",
    "    def __init__(self, vocab_size, embed_dim, num_heads, ff_dim, num_layers, seq_length):\n",
    "        super(TransformerModel, self).__init__()\n",
    "        \n",
    "        # Embedding layer converts token indices into dense vectors of size embed_dim\n",
    "        self.embedding = Embedding(vocab_size, embed_dim)\n",
    "        \n",
    "        # Positional encoding to give the model information about the position of each token in the sequence\n",
    "        self.pos_encoding = self.positional_encoding(seq_length, embed_dim)\n",
    "        \n",
    "        # Stack of Transformer blocks (num_layers indicates how many Transformer blocks to use)\n",
    "        self.transformer_blocks = [TransformerBlock(embed_dim, num_heads, ff_dim) for _ in range(num_layers)]\n",
    "        \n",
    "        # Final dense layer that maps to vocab_size to predict the next token\n",
    "        self.dense = Dense(vocab_size)\n",
    "\n",
    "    def positional_encoding(self, seq_length, embed_dim):\n",
    "        # This function calculates a fixed positional encoding for the input sequence\n",
    "        # Why? Unlike RNNs, the Transformer has no notion of order, so positional encoding adds this information.\n",
    "        angle_rads = self.get_angles(np.arange(seq_length)[:, np.newaxis], np.arange(embed_dim)[np.newaxis, :], embed_dim)\n",
    "        \n",
    "        # Apply sin to even indices in the angle array and cos to odd indices\n",
    "        angle_rads[:, 0::2] = np.sin(angle_rads[:, 0::2])\n",
    "        angle_rads[:, 1::2] = np.cos(angle_rads[:, 1::2])\n",
    "        \n",
    "        # Add a new axis for batch compatibility and cast to float32\n",
    "        pos_encoding = angle_rads[np.newaxis, ...]\n",
    "        return tf.cast(pos_encoding, dtype=tf.float32)\n",
    "\n",
    "    def get_angles(self, pos, i, embed_dim):\n",
    "        # Calculate the angles for positional encoding based on position `pos` and embedding index `i`\n",
    "        # Formula: angle = pos / (10000^(2i/dim)), where `i` is even/odd for sin/cos use\n",
    "        angle_rates = 1 / np.power(10000, (2 * (i // 2)) / np.float32(embed_dim))\n",
    "        return pos * angle_rates\n",
    "\n",
    "    def call(self, inputs, training=False):\n",
    "        # Forward pass of the model\n",
    "        seq_len = tf.shape(inputs)[1]  # Get the actual sequence length of the input\n",
    "        \n",
    "        # Embed the inputs and add positional encoding to preserve token order\n",
    "        x = self.embedding(inputs)\n",
    "        x += self.pos_encoding[:, :seq_len, :]\n",
    "        \n",
    "        # Pass input through each Transformer block in the stack\n",
    "        for transformer_block in self.transformer_blocks:\n",
    "            x = transformer_block(x, training=training)\n",
    "        \n",
    "        # Final dense layer outputs probabilities over the vocabulary for the next token prediction\n",
    "        output = self.dense(x)\n",
    "        return output"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Hyperparameters \n",
    "embed_dim = 256 \n",
    "num_heads = 4 \n",
    "ff_dim = 512 \n",
    "num_layers = 4 \n",
    "\n",
    "# Build the Transformer model \n",
    "model = TransformerModel(vocab_size, embed_dim, num_heads, ff_dim, num_layers, seq_length)\n",
    "\n",
    "# Provide input shape to build the model by passing a dummy input with maxval specified\n",
    "_ = model(tf.random.uniform((1, seq_length), maxval=vocab_size, dtype=tf.int32))\n",
    "\n",
    "# Compile the model \n",
    "model.compile(optimizer='adam', loss='sparse_categorical_crossentropy')\n",
    "\n",
    "# Summary of the model \n",
    "model.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import necessary libraries for training visualization\n",
    "import matplotlib.pyplot as plt\n",
    "from tensorflow.keras.callbacks import EarlyStopping\n",
    "\n",
    "# Early stopping callback to stop training if the loss doesn't improve\n",
    "early_stopping = EarlyStopping(monitor='loss', patience=2, restore_best_weights=True)\n",
    "\n",
    "# Train the transformer model on the full input and target sequences\n",
    "history = model.fit(X, Y, epochs=20, batch_size=32, callbacks=[early_stopping])\n",
    "\n",
    "# Plot training loss to monitor model performance over epochs\n",
    "plt.plot(history.history['loss'])\n",
    "plt.xlabel('Epoch')\n",
    "plt.ylabel('Loss')\n",
    "plt.title('Training Loss')\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def generate_text(model, start_string, num_generate=100, temperature=1.0):\n",
    "    # Convert the starting text (start_string) into numbers the model can understand\n",
    "    input_eval = vectorizer([start_string]).numpy()\n",
    "\n",
    "    # Ensure input is the right length for the model\n",
    "    if input_eval.shape[1] < seq_length:\n",
    "        # If input is too short, pad it with zeros to match the required length\n",
    "        padding = np.zeros((1, seq_length - input_eval.shape[1]))\n",
    "        input_eval = np.concatenate((padding, input_eval), axis=1)\n",
    "    elif input_eval.shape[1] > seq_length:\n",
    "        # If input is too long, truncate it to match the sequence length\n",
    "        input_eval = input_eval[:, -seq_length:]\n",
    "\n",
    "    input_eval = tf.convert_to_tensor(input_eval)  # Convert to tensor for model compatibility\n",
    "    \n",
    "    # Prepare an empty list to store the generated words\n",
    "    text_generated = []\n",
    "\n",
    "    # Loop for the number of words we want to generate\n",
    "    for i in range(num_generate):\n",
    "        # Get predictions from the model based on the current input\n",
    "        predictions = model(input_eval)\n",
    "\n",
    "        # Get rid of batch dimension, keeping just the predictions for this one sequence\n",
    "        predictions = predictions[0]\n",
    "\n",
    "        # Adjust the randomness level of predictions using temperature\n",
    "        predictions = predictions / temperature\n",
    "\n",
    "        # Make sure predictions have the right shape\n",
    "        predictions = tf.expand_dims(predictions, 0)\n",
    "\n",
    "        # Choose the next word based on predictions with randomness (temperature) applied\n",
    "        predicted_id = tf.random.categorical(predictions, num_samples=1)[0, 0].numpy()\n",
    "\n",
    "        # Update the input for the next prediction by adding the newly predicted word\n",
    "        input_eval = np.append(input_eval.numpy(), [[predicted_id]], axis=1)\n",
    "        input_eval = input_eval[:, -seq_length:]  # Keep only the last `seq_length` tokens\n",
    "        input_eval = tf.convert_to_tensor(input_eval)\n",
    "\n",
    "        # Add the predicted word to our list of generated words\n",
    "        text_generated.append(vectorizer.get_vocabulary()[predicted_id])\n",
    "\n",
    "    # Return the starting text along with all the new words\n",
    "    return start_string + ' ' + ' '.join(text_generated)\n",
    "\n",
    "# Example usage to generate text with more coherent output by setting temperature to 0.7\n",
    "start_string = \"To be, or not to be\"\n",
    "generated_text = generate_text(model, start_string, temperature=0.7)\n",
    "print(generated_text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import pandas as pd\n",
    "import tensorflow as tf\n",
    "from sklearn.preprocessing import MinMaxScaler\n",
    "from sklearn.model_selection import train_test_split\n",
    "from tensorflow.keras.layers import Layer, Dense, LayerNormalization, Dropout, Input\n",
    "from tensorflow.keras.callbacks import EarlyStopping\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "# Define a learning rate scheduler  \n",
    "def scheduler(epoch, lr):  \n",
    "    # This function adjusts the learning rate at each epoch\n",
    "    # If the epoch is a multiple of 10 (and not zero), reduce the learning rate by half\n",
    "    if epoch % 10 == 0 and epoch != 0:  \n",
    "        lr = lr * 0.5  \n",
    "    return lr  \n",
    "\n",
    "# Create a callback to use the learning rate scheduler\n",
    "callback = tf.keras.callbacks.LearningRateScheduler(scheduler)  \n",
    "\n",
    "# Train the model with the learning rate scheduler\n",
    "# Note: Replace `X` and `Y` with your input (X) and target (Y) data, and ensure `model` is defined\n",
    "history = model.fit(X, Y, epochs=20, batch_size=64, callbacks=[callback])  \n",
    "\n",
    "# Plot the training loss to visualize model performance over epochs\n",
    "plt.plot(history.history['loss'])  \n",
    "plt.xlabel('Epoch')  \n",
    "plt.ylabel('Loss')  \n",
    "plt.title('Training Loss with Learning Rate Scheduler')  \n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Building Autoencoders"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np \n",
    "from tensorflow.keras.datasets import mnist \n",
    "\n",
    "# Load the dataset \n",
    "(x_train, _), (x_test, _) = mnist.load_data() \n",
    "\n",
    "# Normalize the pixel values \n",
    "x_train = x_train.astype('float32') / 255. \n",
    "x_test = x_test.astype('float32') / 255. \n",
    "\n",
    "# Flatten the images \n",
    "x_train = x_train.reshape((len(x_train), np.prod(x_train.shape[1:]))) \n",
    "x_test = x_test.reshape((len(x_test), np.prod(x_test.shape[1:])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tensorflow.keras.models import Model \n",
    "from tensorflow.keras.layers import Input, Dense \n",
    "\n",
    "# Encoder \n",
    "input_layer = Input(shape=(784,)) \n",
    "encoded = Dense(64, activation='relu')(input_layer) \n",
    "\n",
    "# Bottleneck \n",
    "bottleneck = Dense(32, activation='relu')(encoded) \n",
    "\n",
    "# Decoder \n",
    "decoded = Dense(64, activation='relu')(bottleneck) \n",
    "output_layer = Dense(784, activation='sigmoid')(decoded) \n",
    "\n",
    "# Autoencoder model \n",
    "autoencoder = Model(input_layer, output_layer) \n",
    "\n",
    "# Compile the model \n",
    "autoencoder.compile(optimizer='adam', loss='binary_crossentropy') \n",
    "\n",
    "# Summary of the model \n",
    "autoencoder.summary()\n",
    "\n",
    "# Fit the model\n",
    "autoencoder.fit(\n",
    "    x_train, x_train,  \n",
    "    epochs=25,  \n",
    "    batch_size=256,  \n",
    "    shuffle=True,  \n",
    "    validation_data=(x_test, x_test)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt \n",
    "\n",
    "# Predict the test data \n",
    "reconstructed = autoencoder.predict(x_test) \n",
    "\n",
    "# Visualize the results \n",
    "n = 10  # Number of digits to display \n",
    "plt.figure(figsize=(20, 4)) \n",
    "\n",
    "for i in range(n): \n",
    "    # Display original \n",
    "    ax = plt.subplot(2, n, i + 1) \n",
    "    plt.imshow(x_test[i].reshape(28, 28)) \n",
    "    plt.gray() \n",
    "    ax.get_xaxis().set_visible(False) \n",
    "    ax.get_yaxis().set_visible(False) \n",
    "\n",
    "    # Display reconstruction \n",
    "    ax = plt.subplot(2, n, i + 1 + n) \n",
    "    plt.imshow(reconstructed[i].reshape(28, 28)) \n",
    "    plt.gray() \n",
    "    ax.get_xaxis().set_visible(False) \n",
    "    ax.get_yaxis().set_visible(False) \n",
    "\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Unfreeze the top layers of the encoder\n",
    "for layer in autoencoder.layers[-4:]: \n",
    "    layer.trainable = True \n",
    "\n",
    "# Compile the model again\n",
    "autoencoder.compile(optimizer='adam', loss='binary_crossentropy') \n",
    "\n",
    "# Train the model again\n",
    "autoencoder.fit(x_train, x_train,  \n",
    "                epochs=10,  \n",
    "                batch_size=256,  \n",
    "                shuffle=True,  \n",
    "                validation_data=(x_test, x_test))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "# Add noise to the data\n",
    "noise_factor = 0.5\n",
    "x_train_noisy = x_train + noise_factor * np.random.normal(loc=0.0, scale=1.0, size=x_train.shape)\n",
    "x_test_noisy = x_test + noise_factor * np.random.normal(loc=0.0, scale=1.0, size=x_test.shape)\n",
    "x_train_noisy = np.clip(x_train_noisy, 0., 1.)\n",
    "x_test_noisy = np.clip(x_test_noisy, 0., 1.)\n",
    "\n",
    "# Train the autoencoder with noisy data\n",
    "autoencoder.fit(\n",
    "    x_train_noisy, x_train,\n",
    "    epochs=20,\n",
    "    batch_size=512,\n",
    "    shuffle=True,\n",
    "    validation_data=(x_test_noisy, x_test)\n",
    ")\n",
    "\n",
    "# Denoise the test images\n",
    "reconstructed_noisy = autoencoder.predict(x_test_noisy)\n",
    "\n",
    "# Visualize the results\n",
    "n = 10  # Number of digits to display\n",
    "plt.figure(figsize=(20, 6))\n",
    "for i in range(n):\n",
    "    # Display noisy images\n",
    "    ax = plt.subplot(3, n, i + 1)\n",
    "    plt.imshow(x_test_noisy[i].reshape(28, 28))\n",
    "    plt.gray()\n",
    "    ax.get_xaxis().set_visible(False)\n",
    "    ax.get_yaxis().set_visible(False)\n",
    "    \n",
    "    # Display denoised images\n",
    "    ax = plt.subplot(3, n, i + 1 + n)\n",
    "    plt.imshow(reconstructed_noisy[i].reshape(28, 28))\n",
    "    plt.gray()\n",
    "    ax.get_xaxis().set_visible(False)\n",
    "    ax.get_yaxis().set_visible(False)\n",
    "\n",
    "    # Display original images\n",
    "    ax = plt.subplot(3, n, i + 1 + 2 * n)\n",
    "    plt.imshow(x_test[i].reshape(28, 28))\n",
    "    plt.gray()\n",
    "    ax.get_xaxis().set_visible(False)\n",
    "    ax.get_yaxis().set_visible(False)\n",
    "\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Visualize encoder features\n",
    "import matplotlib.pyplot as plt \n",
    "\n",
    "# Extract the encoder part of the autoencoder \n",
    "encoder_model = Model(input_layer, bottleneck) \n",
    "\n",
    "# Encode the test data \n",
    "encoded_imgs = encoder_model.predict(x_test) \n",
    "\n",
    "# Visualize the first two dimensions of the encoded features \n",
    "plt.figure(figsize=(10, 8)) \n",
    "plt.scatter(encoded_imgs[:, 0], encoded_imgs[:, 1], c='blue', alpha=0.5) \n",
    "plt.title('Encoded Features - First Two Dimensions') \n",
    "plt.xlabel('Encoded Feature 1') \n",
    "plt.ylabel('Encoded Feature 2') \n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Implementing Diffusion Models"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import tensorflow as tf\n",
    "from tensorflow.keras.datasets import mnist\n",
    "from tensorflow.keras.layers import Input, Conv2D, Flatten, Dense, Reshape, Conv2DTranspose\n",
    "from tensorflow.keras.models import Model\n",
    "from tensorflow.keras.callbacks import EarlyStopping\n",
    "\n",
    "# Load the data set  \n",
    "(x_train, _), (x_test, _) = mnist.load_data()\n",
    "\n",
    "# Normalize the pixel values  \n",
    "x_train = x_train.astype('float32') / 255.\n",
    "x_test = x_test.astype('float32') / 255.\n",
    "\n",
    "# Expand dimensions to match the input shape (28, 28, 1)  \n",
    "x_train = np.expand_dims(x_train, axis=-1)\n",
    "x_test = np.expand_dims(x_test, axis=-1)\n",
    "\n",
    "# Add noise to the data\n",
    "noise_factor = 0.5\n",
    "x_train_noisy = x_train + noise_factor * np.random.normal(loc=0.0, scale=1.0, size=x_train.shape)\n",
    "x_test_noisy = x_test + noise_factor * np.random.normal(loc=0.0, scale=1.0, size=x_test.shape)\n",
    "\n",
    "# Clip the values to be within the range [0, 1]\n",
    "x_train_noisy = np.clip(x_train_noisy, 0., 1.)\n",
    "x_test_noisy = np.clip(x_test_noisy, 0., 1.)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define the diffusion model architecture with reduced complexity\n",
    "input_layer = Input(shape=(28, 28, 1))\n",
    "x = Conv2D(16, (3, 3), activation='relu', padding='same')(input_layer)  # Reduced filters\n",
    "x = Conv2D(32, (3, 3), activation='relu', padding='same')(x)  # Reduced filters\n",
    "x = Flatten()(x)\n",
    "x = Dense(64, activation='relu')(x)  # Reduced size\n",
    "x = Dense(28*28*32, activation='relu')(x)  # Reduced size\n",
    "x = Reshape((28, 28, 32))(x)\n",
    "x = Conv2DTranspose(32, (3, 3), activation='relu', padding='same')(x)  # Reduced filters\n",
    "x = Conv2DTranspose(16, (3, 3), activation='relu', padding='same')(x)  # Reduced filters\n",
    "output_layer = Conv2D(1, (3, 3), activation='sigmoid', padding='same')(x)\n",
    "diffusion_model = Model(input_layer, output_layer)\n",
    "\n",
    "# Compile the model with mixed precision and a different loss function\n",
    "diffusion_model.compile(optimizer='adam', loss='mean_squared_error')  # Using MSE for regression tasks\n",
    "\n",
    "# Summary of the optimized model\n",
    "diffusion_model.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Cache and prefetch the data using TensorFlow data pipelines for faster loading\n",
    "train_dataset = tf.data.Dataset.from_tensor_slices((x_train_noisy, x_train))\n",
    "train_dataset = train_dataset.cache().batch(64).prefetch(tf.data.AUTOTUNE)  # Reduced batch size\n",
    "\n",
    "val_dataset = tf.data.Dataset.from_tensor_slices((x_test_noisy, x_test))\n",
    "val_dataset = val_dataset.cache().batch(64).prefetch(tf.data.AUTOTUNE)  # Reduced batch size"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Implement early stopping based on validation loss\n",
    "early_stopping = EarlyStopping(monitor='val_loss', patience=2, restore_best_weights=True)\n",
    "\n",
    "# Train the model with early stopping and smaller batch size\n",
    "diffusion_model.fit(\n",
    "    train_dataset,\n",
    "    epochs=3,\n",
    "    shuffle=True,\n",
    "    validation_data=val_dataset,\n",
    "    callbacks=[early_stopping]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "\n",
    "# Predict the denoised images\n",
    "denoised_images = diffusion_model.predict(x_test_noisy)\n",
    "\n",
    "# Visualize the results\n",
    "n = 10  # Number of digits to display\n",
    "plt.figure(figsize=(20, 6))\n",
    "for i in range(n):\n",
    "    # Display original\n",
    "    ax = plt.subplot(3, n, i + 1)\n",
    "    plt.imshow(x_test[i].reshape(28, 28), cmap='gray')\n",
    "    ax.get_xaxis().set_visible(False)\n",
    "    ax.get_yaxis().set_visible(False)\n",
    "\n",
    "    # Display noisy\n",
    "    ax = plt.subplot(3, n, i + 1 + n)\n",
    "    plt.imshow(x_test_noisy[i].reshape(28, 28), cmap='gray')\n",
    "    ax.get_xaxis().set_visible(False)\n",
    "    ax.get_yaxis().set_visible(False)\n",
    "\n",
    "    # Display denoised\n",
    "    ax = plt.subplot(3, n, i + 1 + 2*n)\n",
    "    plt.imshow(denoised_images[i].reshape(28, 28), cmap='gray')\n",
    "    ax.get_xaxis().set_visible(False)\n",
    "    ax.get_yaxis().set_visible(False)\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Unfreeze the top layers of the model\n",
    "for layer in diffusion_model.layers[-4:]:\n",
    "    layer.trainable = True\n",
    "\n",
    "# Compile the model again\n",
    "diffusion_model.compile(optimizer='adam', loss='binary_crossentropy')\n",
    "\n",
    "# Train the model again\n",
    "diffusion_model.fit(x_train_noisy, x_train,\n",
    "                    epochs=10,\n",
    "                    batch_size=64,\n",
    "                    shuffle=True,\n",
    "                    validation_data=(x_test_noisy, x_test))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Develop GANs Using Keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load the MNIST dataset\n",
    "(x_train, _), (_, _) = mnist.load_data()\n",
    "\n",
    "# Normalize the pixel values to the range [-1, 1]\n",
    "x_train = x_train.astype('float32') / 127.5 - 1.\n",
    "x_train = np.expand_dims(x_train, axis=-1)\n",
    "\n",
    "# Print the shape of the data\n",
    "print(x_train.shape)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tensorflow.keras.models import Sequential \n",
    "from tensorflow.keras.layers import Dense, LeakyReLU, BatchNormalization, Reshape \n",
    "\n",
    "# Define the generator model \n",
    "def build_generator(): \n",
    "    model = Sequential() \n",
    "    model.add(Dense(256, input_dim=100)) \n",
    "    model.add(LeakyReLU(alpha=0.2)) \n",
    "    model.add(BatchNormalization(momentum=0.8)) \n",
    "    model.add(Dense(512)) \n",
    "    model.add(LeakyReLU(alpha=0.2)) \n",
    "    model.add(BatchNormalization(momentum=0.8)) \n",
    "    model.add(Dense(1024)) \n",
    "    model.add(LeakyReLU(alpha=0.2)) \n",
    "    model.add(BatchNormalization(momentum=0.8)) \n",
    "    model.add(Dense(28 * 28 * 1, activation='tanh')) \n",
    "    model.add(Reshape((28, 28, 1))) \n",
    "    return model \n",
    "\n",
    "# Build the generator \n",
    "generator = build_generator() \n",
    "generator.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tensorflow.keras.layers import Flatten \n",
    "from tensorflow.keras.models import Sequential \n",
    "from tensorflow.keras.layers import Dense, LeakyReLU\n",
    "\n",
    "# Define the discriminator model \n",
    "def build_discriminator(): \n",
    "    model = Sequential() \n",
    "    model.add(Flatten(input_shape=(28, 28, 1))) \n",
    "    model.add(Dense(512)) \n",
    "    model.add(LeakyReLU(alpha=0.2)) \n",
    "    model.add(Dense(256)) \n",
    "    model.add(LeakyReLU(alpha=0.2)) \n",
    "    model.add(Dense(1, activation='sigmoid')) \n",
    "    return model \n",
    "\n",
    "# Build and compile the discriminator \n",
    "discriminator = build_discriminator() \n",
    "discriminator.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy']) \n",
    "discriminator.summary()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This step involves combining the generator and discriminator models to create the GAN. The GAN takes a noise vector as an input, generates a synthetic image using the generator, and classifies the image using the discriminator. The discriminator is set to non-trainable when compiling the GAN to ensure that only the generator is updated during the adversarial training."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from tensorflow.keras.layers import Input \n",
    "from tensorflow.keras.models import Model \n",
    "\n",
    "# Create the GAN by stacking the generator and the discriminator \n",
    "def build_gan(generator, discriminator): \n",
    "    discriminator.trainable = False \n",
    "    gan_input = Input(shape=(100,)) \n",
    "    generated_image = generator(gan_input) \n",
    "    gan_output = discriminator(generated_image) \n",
    "    gan = Model(gan_input, gan_output) \n",
    "    gan.compile(loss='binary_crossentropy', optimizer='adam') \n",
    "    return gan \n",
    "\n",
    "# Build the GAN \n",
    "gan = build_gan(generator, discriminator) \n",
    "gan.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Training parameters \n",
    "\n",
    "batch_size = 64 \n",
    "epochs = 50\n",
    "sample_interval = 10\n",
    "\n",
    "# Adversarial ground truths \n",
    "real = np.ones((batch_size, 1)) \n",
    "fake = np.zeros((batch_size, 1)) \n",
    "\n",
    "# Training loop \n",
    "for epoch in range(epochs): \n",
    "    # Train the discriminator \n",
    "    idx = np.random.randint(0, x_train.shape[0], batch_size) \n",
    "    real_images = x_train[idx] \n",
    "    noise = np.random.normal(0, 1, (batch_size, 100)) \n",
    "    generated_images = generator.predict(noise) \n",
    "    d_loss_real = discriminator.train_on_batch(real_images, real) \n",
    "    d_loss_fake = discriminator.train_on_batch(generated_images, fake) \n",
    "    d_loss = 0.5 * np.add(d_loss_real, d_loss_fake) \n",
    "\n",
    "    # Train the generator \n",
    "    noise = np.random.normal(0, 1, (batch_size, 100)) \n",
    "    g_loss = gan.train_on_batch(noise, real) \n",
    "\n",
    "    # Print the progress \n",
    "    if epoch % sample_interval == 0: \n",
    "        print(f\"{epoch} [D loss: {d_loss[0]}] [D accuracy: {100 * d_loss[1]}%] [G loss: {g_loss}]\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After training the GAN, we need to assess the quality of the synthetic images generated by the generator. There are two main ways to evaluate the performance of GANs: qualitative assessment and quantitative assessment. \n",
    "\n",
    "### Qualitative Assessment: Visual Inspection\n",
    "\n",
    "Visual inspection is a straightforward method to assess the quality of images generated by a GAN. You can use the `sample_images` function provided in the lab to visualize a grid of generated images. During visual inspection, look for the following qualities:\n",
    "\n",
    "- **Clarity**: The images should be sharp and not blurry. Blurry images indicate that the generator is struggling to learn the patterns in the data.\n",
    "- **Coherence**: The generated images should have a coherent structure that resembles the original images in the dataset. For example, in the case of MNIST, the generated images should resemble handwritten digits with the correct number of strokes and shapes.\n",
    "- **Diversity**: There should be a variety of images generated by the GAN. If all images look similar, it might indicate that the generator is overfitting or has collapsed to a single mode.\n",
    "\n",
    "### 2. Quantitative Assessment: Metrics\n",
    "\n",
    "While visual inspection provides an intuitive understanding of the GAN’s performance, it can be subjective. To objectively evaluate GAN performance, you can use quantitative metrics such as:\n",
    "\n",
    "- **Inception Score (IS)**: This score measures both the quality and diversity of generated images by using a pre-trained classifier (such as Inception-v3) to predict the class of each image. A higher score indicates that the images are both high-quality and diverse. However, IS is not very effective for simple datasets like MNIST; it’s more suitable for complex datasets.\n",
    "\n",
    "- **Fréchet Inception Distance (FID)**: This metric calculates the distance between the distributions of generated images and real images. A lower FID score indicates that the generated images are more similar to real images. FID is commonly used and considered a reliable metric for evaluating GAN performance.\n",
    "\n",
    "- **Discriminator Accuracy**: During training, if the discriminator's accuracy is around 50%, it suggests that the generator is producing realistic images that are hard to distinguish from real ones. This metric is easy to implement and provides quick feedback on the training progress."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt \n",
    "\n",
    "def sample_images(generator, epoch, num_images=25): \n",
    "    noise = np.random.normal(0, 1, (num_images, 100)) \n",
    "    generated_images = generator.predict(noise) \n",
    "    generated_images = 0.5 * generated_images + 0.5  # Rescale to [0, 1] \n",
    "    fig, axs = plt.subplots(5, 5, figsize=(10, 10)) \n",
    "    count = 0 \n",
    "\n",
    "    for i in range(5): \n",
    "        for j in range(5): \n",
    "            axs[i, j].imshow(generated_images[count, :, :, 0], cmap='gray') \n",
    "            axs[i, j].axis('off') \n",
    "            count += 1 \n",
    "    plt.show() \n",
    "\n",
    "# Sample images at the end of training \n",
    "sample_images(generator, epochs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Calculate and print the discriminator accuracy on real vs. fake images\n",
    "noise = np.random.normal(0, 1, (batch_size, 100))\n",
    "generated_images = generator.predict(noise)\n",
    "\n",
    "# Evaluate the discriminator on real images\n",
    "real_images = x_train[np.random.randint(0, x_train.shape[0], batch_size)]\n",
    "d_loss_real = discriminator.evaluate(real_images, np.ones((batch_size, 1)), verbose=0)\n",
    "\n",
    "# Evaluate the discriminator on fake images\n",
    "d_loss_fake = discriminator.evaluate(generated_images, np.zeros((batch_size, 1)), verbose=0)\n",
    "\n",
    "print(f\"Discriminator Accuracy on Real Images: {d_loss_real[1] * 100:.2f}%\")\n",
    "print(f\"Discriminator Accuracy on Fake Images: {d_loss_fake[1] * 100:.2f}%\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Custom Training Loops in Keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Suppress all Python warnings\n",
    "warnings.filterwarnings('ignore')  # Stops Python from printing non-critical warning messages\n",
    "\n",
    "# Set TensorFlow log level to suppress warnings and info messages\n",
    "os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'  # Limits TensorFlow logging to only show errors (ignoring info and warnings)\n",
    "\n",
    "# Step 1: Set Up the Environment\n",
    "# Load the MNIST dataset - a popular dataset of hand-written digits\n",
    "(x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data() \n",
    "\n",
    "# Normalize the data by dividing by 255 to bring pixel values between 0 and 1 for faster training\n",
    "x_train, x_test = x_train / 255.0, x_test / 255.0 \n",
    "\n",
    "# Convert the training data into a TensorFlow dataset object, batched into groups of 32 examples\n",
    "train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train)).batch(32)\n",
    "\n",
    "# Step 2: Define the Model\n",
    "model = Sequential([  # A simple neural network model defined sequentially\n",
    "    Flatten(input_shape=(28, 28)),  # Flatten 28x28 images into a single 1D vector of 784 pixels\n",
    "    Dense(128, activation='relu'),  # A dense layer with 128 units, using ReLU activation for non-linearity\n",
    "    Dense(10)  # Output layer with 10 units (for the 10 digit classes 0-9)\n",
    "])\n",
    "\n",
    "# Step 3: Define Loss Function and Optimizer + Accuracy metric\n",
    "# Define the loss function: SparseCategoricalCrossentropy for multi-class classification tasks\n",
    "loss_fn = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True) \n",
    "\n",
    "# Define the optimizer: Adam, an efficient and adaptive optimizer\n",
    "optimizer = tf.keras.optimizers.Adam()  \n",
    "\n",
    "# Define a metric to track accuracy, which compares predicted vs actual labels\n",
    "accuracy_metric = tf.keras.metrics.SparseCategoricalAccuracy()  \n",
    "\n",
    "# Step 4: Implement the Custom Training Loop with Custom Callback\n",
    "epochs = 2  # Number of times to loop through the full dataset during training\n",
    "custom_callback = CustomCallback()  # Initialize a custom callback (usually for additional actions during training)\n",
    "\n",
    "# Start looping over the specified number of epochs\n",
    "for epoch in range(epochs):\n",
    "    print(f'Start of epoch {epoch + 1}')  # Informative print statement at the beginning of each epoch\n",
    "    \n",
    "    # Loop over each batch of training data in this epoch\n",
    "    for step, (x_batch_train, y_batch_train) in enumerate(train_dataset):\n",
    "        \n",
    "        with tf.GradientTape() as tape:  # Start recording operations for automatic differentiation\n",
    "            # Forward pass: Compute predictions (logits) by passing input data through the model\n",
    "            logits = model(x_batch_train, training=True)\n",
    "            # Compute the loss between actual labels and predicted logits\n",
    "            loss_value = loss_fn(y_batch_train, logits)\n",
    "        \n",
    "        # Compute gradients of the loss with respect to model parameters\n",
    "        grads = tape.gradient(loss_value, model.trainable_weights)\n",
    "        \n",
    "        # Apply gradients to the optimizer to update the model's weights\n",
    "        optimizer.apply_gradients(zip(grads, model.trainable_weights))\n",
    "        \n",
    "        # Update the accuracy metric with predictions from this batch\n",
    "        accuracy_metric.update_state(y_batch_train, logits)\n",
    "\n",
    "        # Print loss and accuracy every 200 steps for monitoring progress\n",
    "        if step % 200 == 0:\n",
    "            print(f'Epoch {epoch + 1} Step {step}: Loss = {loss_value.numpy()} Accuracy = {accuracy_metric.result().numpy()}')\n",
    "    \n",
    "    # At the end of each epoch, use the custom callback to log final epoch stats\n",
    "    custom_callback.on_epoch_end(epoch, logs={'loss': loss_value.numpy(), 'accuracy': accuracy_metric.result().numpy()})\n",
    "    \n",
    "    # Reset the accuracy metric for the next epoch to avoid mixing results\n",
    "    accuracy_metric.reset_state()  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Hyperparameter Tuning with Keras Tuner"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define a model-building function for hyperparameter tuning\n",
    "def build_model(hp):  # 'hp' represents hyperparameters that will be tuned\n",
    "    # Create a simple neural network model\n",
    "    model = Sequential([\n",
    "        # Flatten layer to convert 2D images to 1D vector for input\n",
    "        Flatten(input_shape=(28, 28)),\n",
    "        \n",
    "        # Dense layer with a variable number of units, determined by tuning\n",
    "        Dense(\n",
    "            units=hp.Int('units', min_value=32, max_value=512, step=32),  # Tune 'units' from 32 to 512 in steps of 32\n",
    "            activation='relu'  # ReLU activation adds non-linearity to the model\n",
    "        ),\n",
    "        \n",
    "        # Output layer with 10 units (for 10 classes) and 'softmax' to output class probabilities\n",
    "        Dense(10, activation='softmax')\n",
    "    ])\n",
    "\n",
    "    # Compile the model with a configurable learning rate\n",
    "    model.compile(\n",
    "        optimizer=Adam(learning_rate=hp.Float('learning_rate', min_value=1e-4, max_value=1e-2, sampling='LOG')),  # Tune learning rate from 0.0001 to 0.01\n",
    "        loss='sparse_categorical_crossentropy',  # Loss function for multi-class classification\n",
    "        metrics=['accuracy']  # Track accuracy during training\n",
    "    )\n",
    "\n",
    "    return model\n",
    "\n",
    "# Create a RandomSearch Tuner\n",
    "\n",
    "tuner = kt.RandomSearch(\n",
    "    build_model,  # The model-building function to tune\n",
    "    objective='val_accuracy',  # The goal is to maximize validation accuracy\n",
    "    max_trials=10,  # Run up to 10 different sets of hyperparameters\n",
    "    executions_per_trial=2,  # Train each model twice for more reliable results\n",
    "    directory='my_dir',  # Folder to save tuning logs\n",
    "    project_name='intro_to_kt'  # Naming the project for organization\n",
    ")\n",
    "\n",
    "# Display a summary of the search space\n",
    "tuner.search_space_summary()  # Shows details of all hyperparameters being tuned\n",
    "\n",
    "# Run the hyperparameter search\n",
    "tuner.search(\n",
    "    x_train, y_train,  # Use training data for tuning\n",
    "    epochs=5,  # Train each model for 5 epochs\n",
    "    validation_data=(x_val, y_val)  # Use validation data to assess performance\n",
    ")\n",
    "\n",
    "# Display a summary of the results\n",
    "tuner.results_summary()  # Shows the top-performing models and their hyperparameters\n",
    "\n",
    "# Step 1: Retrieve the best hyperparameters\n",
    "\n",
    "# Get the best set of hyperparameters found during tuning\n",
    "best_hps = tuner.get_best_hyperparameters(num_trials=1)[0]  # Retrieve the best combination from all trials\n",
    "\n",
    "# Print the optimal hyperparameters\n",
    "print(f\"\"\" \n",
    "\n",
    "The optimal number of units in the first dense layer is {best_hps.get('units')}. \n",
    "\n",
    "The optimal learning rate for the optimizer is {best_hps.get('learning_rate')}. \n",
    "\n",
    "\"\"\")\n",
    "\n",
    "# Step 2: Build and Train the Model with Best Hyperparameters\n",
    "\n",
    "# Build the model using the best hyperparameters\n",
    "model = tuner.hypermodel.build(best_hps)  \n",
    "\n",
    "# Train the model with the best hyperparameters on the training data\n",
    "model.fit(\n",
    "    x_train, y_train, \n",
    "    epochs=10,  # Train for a longer duration with the best hyperparameters\n",
    "    validation_split=0.2  # Use 20% of training data as validation set\n",
    ")\n",
    "\n",
    "# Evaluate the model on the test set\n",
    "test_loss, test_acc = model.evaluate(x_val, y_val)  # Assess final model performance on the test data\n",
    "\n",
    "# Print the test accuracy\n",
    "print(f'Test accuracy: {test_acc}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Implementing Q-Learning in Keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the environment \n",
    "env = gym.make('CartPole-v1') \n",
    "\n",
    "# Set random seed for reproducibility \n",
    "np.random.seed(42) \n",
    "env.action_space.seed(42) \n",
    "env.observation_space.seed(42)\n",
    "\n",
    "# Suppress warnings for a cleaner notebook or console experience\n",
    "import warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "\n",
    "# Override the default warning function\n",
    "def warn(*args, **kwargs):\n",
    "    pass\n",
    "warnings.warn = warn\n",
    "\n",
    "# Import necessary libraries for the Q-Learning model\n",
    "from tensorflow.keras.models import Sequential\n",
    "from tensorflow.keras.layers import Dense, Input  # Import Input layer\n",
    "from tensorflow.keras.optimizers import Adam\n",
    "import gym  # Ensure the environment library is available\n",
    "\n",
    "# Define the model building function\n",
    "def build_model(state_size, action_size): \n",
    "    model = Sequential() \n",
    "    model.add(Input(shape=(state_size,)))  # Use Input layer to specify the input shape \n",
    "    model.add(Dense(24, activation='relu')) \n",
    "    model.add(Dense(24, activation='relu')) \n",
    "    model.add(Dense(action_size, activation='linear')) \n",
    "    model.compile(loss='mse', optimizer=Adam(learning_rate=0.001)) \n",
    "    return model \n",
    "\n",
    "# Create the environment and set up the model\n",
    "env = gym.make('CartPole-v1')\n",
    "state_size = env.observation_space.shape[0] \n",
    "action_size = env.action_space.n \n",
    "model = build_model(state_size, action_size)\n",
    "\n",
    "import random\n",
    "import numpy as np\n",
    "from collections import deque\n",
    "import tensorflow as tf\n",
    "\n",
    "# Define epsilon and epsilon_decay\n",
    "epsilon = 1.0  # Starting with a high exploration rate\n",
    "epsilon_min = 0.01  # Minimum exploration rate\n",
    "epsilon_decay = 0.99  # Faster decay rate for epsilon after each episode\n",
    "\n",
    "# Replay memory\n",
    "memory = deque(maxlen=2000)\n",
    "\n",
    "def remember(state, action, reward, next_state, done):\n",
    "    \"\"\"Store experience in memory.\"\"\"\n",
    "    memory.append((state, action, reward, next_state, done))\n",
    "\n",
    "def replay(batch_size=64):  # Increased batch size\n",
    "    \"\"\"Train the model using a random sample of experiences from memory.\"\"\"\n",
    "    if len(memory) < batch_size:\n",
    "        return  # Skip replay if there's not enough experience\n",
    "\n",
    "    minibatch = random.sample(memory, batch_size)  # Sample a random batch from memory\n",
    "    \n",
    "    # Extract information for batch processing\n",
    "    states = np.vstack([x[0] for x in minibatch])\n",
    "    actions = np.array([x[1] for x in minibatch])\n",
    "    rewards = np.array([x[2] for x in minibatch])\n",
    "    next_states = np.vstack([x[3] for x in minibatch])\n",
    "    dones = np.array([x[4] for x in minibatch])\n",
    "    \n",
    "    # Predict Q-values for the next states in batch\n",
    "    q_next = model.predict(next_states)\n",
    "    # Predict Q-values for the current states in batch\n",
    "    q_target = model.predict(states)\n",
    "    \n",
    "    # Vectorized update of target values\n",
    "    for i in range(batch_size):\n",
    "        target = rewards[i]\n",
    "        if not dones[i]:\n",
    "            target += 0.95 * np.amax(q_next[i])  # Update Q value with the discounted future reward\n",
    "        q_target[i][actions[i]] = target  # Update only the taken action's Q value\n",
    "    \n",
    "    # Train the model with the updated targets in batch\n",
    "    model.fit(states, q_target, epochs=1, verbose=0)  # Train in batch mode\n",
    "\n",
    "    # Reduce exploration rate (epsilon) after each training step\n",
    "    global epsilon\n",
    "    if epsilon > epsilon_min:\n",
    "        epsilon *= epsilon_decay\n",
    "\n",
    "def act(state):\n",
    "    \"\"\"Choose an action based on the current state and exploration rate.\"\"\"\n",
    "    if np.random.rand() <= epsilon:\n",
    "        return random.randrange(action_size)  # Explore: choose a random action\n",
    "    act_values = model.predict(state)  # Exploit: predict action based on the state\n",
    "    return np.argmax(act_values[0])  # Return the action with the highest Q-value\n",
    "\n",
    "# Define the number of episodes you want to train the model for\n",
    "episodes = 10  # You can set this to any number you prefer\n",
    "train_frequency = 5  # Train the model every 5 steps\n",
    "\n",
    "for e in range(episodes):\n",
    "    state, _ = env.reset()  # Unpack the tuple returned by env.reset()\n",
    "    state = np.reshape(state, [1, state_size])\n",
    "    for time in range(200):  # Limit to 200 time steps per episode\n",
    "        action = act(state)\n",
    "        next_state, reward, terminated, truncated, _ = env.step(action)\n",
    "        done = terminated or truncated\n",
    "        reward = reward if not done else -10\n",
    "        next_state = np.reshape(next_state, [1, state_size])\n",
    "        remember(state, action, reward, next_state, done)  # Store experience\n",
    "        state = next_state\n",
    "        \n",
    "        if done:\n",
    "            print(f\"episode: {e+1}/{episodes}, score: {time}, e: {epsilon:.2}\")\n",
    "            break\n",
    "        \n",
    "        # Train the model every 'train_frequency' steps\n",
    "        if time % train_frequency == 0:\n",
    "            replay(batch_size=64)  # Call replay with larger batch size for efficiency\n",
    "\n",
    "env.close()\n",
    "\n",
    "# Evaluate the performance\n",
    "for e in range(10):  \n",
    "\n",
    "    state, _ = env.reset()  # Unpack the state from the tuple \n",
    "    state = np.reshape(state, [1, state_size])  # Reshape the state correctly \n",
    "    for time in range(500):  \n",
    "        env.render()  \n",
    "        action = np.argmax(model.predict(state)[0])  \n",
    "        next_state, reward, terminated, truncated, _ = env.step(action)  # Unpack the five return values \n",
    "        done = terminated or truncated  # Check if the episode is done \n",
    "        next_state = np.reshape(next_state, [1, state_size])  \n",
    "        state = next_state  \n",
    "        if done:  \n",
    "            print(f\"episode: {e+1}/10, score: {time}\")  \n",
    "            break  \n",
    "\n",
    "env.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Building a Deep Q-Network (DQN) with Keras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the environment  \n",
    "env = gym.make('CartPole-v1')\n",
    "\n",
    "# Set random seed for reproducibility  \n",
    "np.random.seed(42)\n",
    "env.reset(seed=42)\n",
    "\n",
    "# Suppress warnings for a cleaner notebook or console experience\n",
    "import warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "\n",
    "# Disable warnings for a cleaner notebook or console experience\n",
    "def warn(*args, **kwargs):\n",
    "    pass\n",
    "warnings.warn = warn\n",
    "\n",
    "# Import necessary libraries\n",
    "from tensorflow.keras.models import Sequential\n",
    "from tensorflow.keras.layers import Dense\n",
    "from tensorflow.keras.optimizers import Adam\n",
    "\n",
    "\n",
    "def build_model(state_size, action_size):\n",
    "    model = Sequential()\n",
    "    model.add(Dense(24, input_dim=state_size, activation='relu'))\n",
    "    model.add(Dense(24, activation='relu'))\n",
    "    model.add(Dense(action_size, activation='linear'))\n",
    "    model.compile(loss='mse', optimizer=Adam(learning_rate=0.001))\n",
    "    return model\n",
    "\n",
    "state_size = env.observation_space.shape[0]\n",
    "action_size = env.action_space.n\n",
    "model = build_model(state_size, action_size)\n",
    "\n",
    "from collections import deque\n",
    "import random\n",
    "\n",
    "memory = deque(maxlen=2000)\n",
    "def remember(state, action, reward, next_state, done):\n",
    "    memory.append((state, action, reward, next_state, done))\n",
    "\n",
    "# The epsilon-greedy policy balances exploration and exploitation by choosing random actions with probability epsilon.\n",
    "epsilon = 1.0\n",
    "epsilon_min = 0.01\n",
    "epsilon_decay = 0.995\n",
    " \n",
    "def act(state):\n",
    "    if np.random.rand() <= epsilon:\n",
    "        return random.randrange(action_size)\n",
    "    q_values = model.predict(state)\n",
    "    return np.argmax(q_values[0])\n",
    "\n",
    "# Implement the Q-learning update to train the DQN using experiences stored in the replay buffer.\n",
    "def replay(batch_size):\n",
    "    global epsilon\n",
    "    minibatch = random.sample(memory, batch_size)\n",
    "    for state, action, reward, next_state, done in minibatch:\n",
    "        target = reward\n",
    "        if not done:\n",
    "            target = reward + gamma * np.amax(model.predict(next_state)[0])\n",
    "        target_f = model.predict(state)\n",
    "        target_f[0][action] = target\n",
    "        model.fit(state, target_f, epochs=1, verbose=0)\n",
    "    if epsilon > epsilon_min:\n",
    "        epsilon *= epsilon_decay\n",
    "\n",
    "# Train the DQN agent by interacting with the environment and updating the Q-values using the replay buffer.\n",
    "for e in range(10):\n",
    "    state = env.reset()\n",
    "\n",
    "    # If state is a tuple, take the first element\n",
    "    if isinstance(state, tuple):\n",
    "        state = state[0]\n",
    "\n",
    "    state = np.reshape(state, [1, state_size])\n",
    "    \n",
    "    for time in range(100):\n",
    "        env.render()\n",
    "        action = np.argmax(model.predict(state)[0])\n",
    "        \n",
    "        # Handle environments that return more than 4 values\n",
    "        result = env.step(action)\n",
    "        if isinstance(result, tuple) and len(result) == 4:\n",
    "            next_state, reward, done, _ = result\n",
    "        else:\n",
    "            next_state, reward, done, _, _ = result  # Adjust based on the number of values returned\n",
    "        \n",
    "        # If next_state is a tuple, take the first element\n",
    "        if isinstance(next_state, tuple):\n",
    "            next_state = next_state[0]\n",
    "\n",
    "        next_state = np.reshape(next_state, [1, state_size])\n",
    "        state = next_state\n",
    "        \n",
    "        if done:\n",
    "            print(f\"episode: {e+1}/10, score: {time}\")\n",
    "            break\n",
    "\n",
    "env.close()\n",
    "\n",
    "# Evaluate the performance of the trained DQN agent.\n",
    "for e in range(10):\n",
    "    state = env.reset()\n",
    "\n",
    "    # Check if state is a tuple and extract the first element if it is\n",
    "    if isinstance(state, tuple):\n",
    "        state = state[0]\n",
    "\n",
    "    state = np.reshape(state, [1, state_size])\n",
    "\n",
    "    for time in range(100):\n",
    "        env.render()\n",
    "        action = np.argmax(model.predict(state)[0])\n",
    "\n",
    "        # Handle environments that return more than 4 values\n",
    "        result = env.step(action)\n",
    "        if len(result) == 4:\n",
    "            next_state, reward, done, _ = result\n",
    "        else:\n",
    "            next_state, reward, done, _, _ = result  # Adjust this based on the number of values returned\n",
    "\n",
    "        # Check if next_state is a tuple and extract the first element if it is\n",
    "        if isinstance(next_state, tuple):\n",
    "            next_state = next_state[0]\n",
    "\n",
    "        next_state = np.reshape(next_state, [1, state_size])\n",
    "        state = next_state\n",
    "\n",
    "        if done:\n",
    "            print(f\"episode: {e+1}/10, score: {time}\")\n",
    "            break\n",
    "\n",
    "env.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "base",
   "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.12.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}