{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import math\n",
    "from random import randint\n",
    "import h5py\n",
    "\n",
    "\n",
    "class ConvolutionalNeuralNetwork:\n",
    "    def __init__(self, in_dim,classes=10, filter_size=3, num_channels=1):\n",
    "       \n",
    "        self.d = in_dim\n",
    "        self.k = classes\n",
    "        self.filter_size = filter_size\n",
    "        \n",
    "        self.c = num_channels\n",
    "        \n",
    "        self.out_dim = self.d - self.filter_size + 1\n",
    "\n",
    "\n",
    "        np.random.seed(72)\n",
    "       \n",
    "        self.filter = np.random.randn(self.filter_size, self.filter_size, self.c) * \\\n",
    "                     math.sqrt(2.0 / (in_dim * in_dim))\n",
    "        self.w = np.random.rand(self.k, self.out_dim, self.out_dim, self.c) * \\\n",
    "                     math.sqrt(2.0 / (in_dim * in_dim))\n",
    "        self.b = np.zeros(self.k)\n",
    "    \n",
    "\n",
    "    def train(self, X,Y, learning_rate=0.01, epochs=50):\n",
    " \n",
    "        \n",
    "\n",
    "      \n",
    "        for epoch in range(1, epochs + 1):\n",
    "            \n",
    "        \n",
    "\n",
    "            correct_sum = 0\n",
    "            for _ in range(X.shape[0]):\n",
    "                index = randint(0, X.shape[0] - 1)\n",
    "                x = X[index]\n",
    "                y = Y[index]\n",
    "\n",
    "                \n",
    "                z, h, u, f = self.forward(x)\n",
    "                g_b, g_w, g_f = self.backward(x, y, z, h, u, f)\n",
    "                self.weight_update(learning_rate, g_b, g_w, g_f)\n",
    "\n",
    "                if self.predict(f) == y:\n",
    "                    correct_sum += 1\n",
    "\n",
    "            acc = correct_sum / np.float(X.shape[0])\n",
    "            print(\"epoch {}, training accuracy = {}\".format(epoch, acc))\n",
    "            \n",
    "            \n",
    "\n",
    "    def forward(self, x):\n",
    "        \"\"\"Calculate output f and intermediary network values.\"\"\"\n",
    "        x = self._reshape_x_to_matrix(x)\n",
    "\n",
    "        convolution = Convolve_Operation(x, self.filter)\n",
    "        z = convolution.convolve()\n",
    "\n",
    "\n",
    "        h = relu(z)\n",
    "        u = np.zeros(self.k)\n",
    "        for depth_slice in range(self.c):\n",
    "            u += np.sum(\n",
    "                    np.multiply(self.w[:, :, :, depth_slice], h[:, :, depth_slice]), \n",
    "                    axis=(1,2)) \\\n",
    "                 + self.b\n",
    "\n",
    "        f = softmax(u)\n",
    "        return z, h, u, f\n",
    "\n",
    "    def backward(self, x, y, z, h, u, f):\n",
    "        \"\"\"Calculate the gradient w.r.t parameters.\"\"\"\n",
    "        x = self._reshape_x_to_matrix(x)\n",
    "\n",
    "        e_y = np.zeros(self.k)\n",
    "        e_y[y] = 1\n",
    "        de_u = - (e_y - f)\n",
    "\n",
    "        de_b = de_u\n",
    "        \n",
    "        de_w = np.zeros((self.k, self.out_dim, self.out_dim, self.c))\n",
    "\n",
    "        \n",
    "        for i in range(self.k):\n",
    "            de_w[i] = de_u[i] * h\n",
    "            \n",
    "        de_u_times_w = np.multiply(np.reshape(de_u, (self.k, 1, 1, 1)), self.w)\n",
    "        delta = de_u_times_w.sum(axis=0)\n",
    "\n",
    "        relu_prime = gradient_for_relu(z)\n",
    "\n",
    "        convolution = Convolve_Operation(x, np.multiply(relu_prime, delta))\n",
    "        gradient_filter = convolution.convolve()\n",
    "        return de_b, de_w,gradient_filter\n",
    "\n",
    "    def weight_update(self, learning_rate, g_b, g_w, g_f):\n",
    "        self.b -= learning_rate * g_b\n",
    "        self.filter -= learning_rate * g_f\n",
    "        self.w -= learning_rate * g_w\n",
    "\n",
    "    def predict(self, f):\n",
    "        return np.argmax(f)\n",
    "\n",
    "    def test(self, X,Y):\n",
    "        \"\"\"Test the trained model.\n",
    "    \n",
    "        \"\"\"\n",
    "        \n",
    "        correct_sum = 0\n",
    "\n",
    "        for i in range(X.shape[0]):\n",
    "            x = X[i]\n",
    "            y = Y[i]\n",
    "            _, _, _, f = self.forward(x)\n",
    "            if self.predict(f) == y:\n",
    "                correct_sum += 1\n",
    "\n",
    "        acc = correct_sum / np.float(X.shape[0])\n",
    "        print(\"testing accuracy = {}\".format(acc))\n",
    "    def _reshape_x_to_matrix(self, x):\n",
    "        \"\"\"Reshape the input vector x into square matrix.\"\"\"\n",
    "        return np.reshape(x, (self.d, self.d))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "\n",
    "def relu(z):\n",
    "    return np.maximum(z, 0)\n",
    "\n",
    "def gradient_for_relu(z):\n",
    "    z[z >= 0] = 1\n",
    "    z[z < 0] = 0\n",
    "    return z\n",
    "\n",
    "def softmax(z):\n",
    "    return np.exp(z) / np.sum(np.exp(z))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def load_data(path):\n",
    "    \"\"\"\n",
    "    Read data from HDF5 files.\n",
    "    \n",
    "    \"\"\"\n",
    "    data_hf = h5py.File(path, 'r')\n",
    "    x_train = np.array(data_hf['x_train'])\n",
    "    y_train = np.array(data_hf['y_train'][:, 0])\n",
    "    x_test = np.array(data_hf['x_test'])\n",
    "    y_test = np.array(data_hf['y_test'][:, 0])\n",
    "\n",
    "    data_hf.close()\n",
    "    return x_train, y_train, x_test, y_test\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Convolve_Operation:\n",
    "    def __init__(self, x, filter_):\n",
    "       \n",
    "        self.in_dim = x.shape[0]\n",
    "        self.filter_size = filter_.shape[0]\n",
    "        self.c = filter_.shape[2]\n",
    "        self.out_dim = self.in_dim - self.filter_size + 1\n",
    "\n",
    "        self.feature_map = np.zeros((self.out_dim, self.out_dim, self.c))\n",
    "\n",
    "        self.x = x\n",
    "        self.filter = filter_\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "    def convolve(self):\n",
    "       \n",
    "\n",
    "        x_col = self.image_col_converter()\n",
    "        filter_row = self.filter_row_converter()\n",
    "\n",
    "        \n",
    "        self.feature_map = np.dot(filter_row, x_col)\n",
    "\n",
    "        \n",
    "        self.feature_map = self.feature_map.T.reshape((self.out_dim, self.out_dim, self.c))\n",
    "\n",
    "        return self.feature_map\n",
    "    \n",
    "\n",
    "    def image_col_converter(self):\n",
    "        \"\"\"Stretch the local regions in the input image into columns\n",
    "        \"\"\"\n",
    "\n",
    "        x_col = np.zeros((self.filter_size * self.filter_size, \n",
    "                          self.out_dim * self.out_dim))\n",
    "        index = 0\n",
    "        for i in range(self.out_dim):\n",
    "            for j in range(self.out_dim):\n",
    "                x_col[:, index] = self.x[i:i+self.filter_size, j:j+self.filter_size].flatten()\n",
    "                index += 1\n",
    "\n",
    "        return x_col\n",
    "    \n",
    "\n",
    "    def filter_row_converter(self):\n",
    "        \"\"\"Stretch the weights in the filter into rows.\n",
    "        \"\"\"\n",
    "\n",
    "        filter_row = np.zeros((self.c, self.filter_size * self.filter_size))\n",
    "        for i in range(self.c):\n",
    "            filter_row[i] = self.filter[:, :, i].T.flatten()\n",
    "        \n",
    "        return filter_row"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "import math"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "loading dataset...\n",
      "training model...\n",
      "hyperparameters:\n",
      "             learning rate = 0.005\n",
      "             epochs = 10\n",
      "             filter size = 6\n",
      "             num of channels = 16\n",
      "epoch 1, training accuracy = 0.9245333333333333\n",
      "epoch 2, training accuracy = 0.9502166666666667\n",
      "epoch 3, training accuracy = 0.9573166666666667\n",
      "epoch 4, training accuracy = 0.9617333333333333\n",
      "epoch 5, training accuracy = 0.96625\n",
      "epoch 6, training accuracy = 0.964\n",
      "epoch 7, training accuracy = 0.9636666666666667\n",
      "epoch 8, training accuracy = 0.9641\n",
      "epoch 9, training accuracy = 0.96015\n",
      "epoch 10, training accuracy = 0.9624166666666667\n",
      "testing model...\n",
      "testing accuracy = 0.9515\n"
     ]
    }
   ],
   "source": [
    "\n",
    "\n",
    "\n",
    "FILTER_SIZE = 6\n",
    "NUM_CHANNELS = 16\n",
    "LR = 0.005\n",
    "EPOCHS = 10\n",
    "\n",
    "def main():\n",
    "    \n",
    "\n",
    "    \n",
    "    print(\"loading dataset...\")\n",
    "    x_train, y_train, x_test, y_test =  load_data(\"MNISTdata.hdf5\")\n",
    "    \n",
    "\n",
    "\n",
    "    \n",
    "\n",
    "    input_dim = int(math.sqrt(x_train.shape[1]))  \n",
    "  \n",
    "    cnn = ConvolutionalNeuralNetwork(input_dim, filter_size=FILTER_SIZE, \n",
    "                                     num_channels=NUM_CHANNELS)\n",
    "\n",
    "    # Train.\n",
    "    print(\"training model...\")\n",
    "    print(\"\"\"hyperparameters:\n",
    "             learning rate = {}\n",
    "             epochs = {}\n",
    "             filter size = {}\n",
    "             num of channels = {}\"\"\"\n",
    "            .format(LR, EPOCHS, FILTER_SIZE, NUM_CHANNELS))\n",
    "    \n",
    "   \n",
    "    cnn.train(x_train, y_train, learning_rate=LR, epochs=EPOCHS)\n",
    "\n",
    "    \n",
    "    \n",
    "\n",
    "    # Test\n",
    "    print(\"testing model...\")\n",
    "    cnn.test(x_test, y_test)\n",
    "\n",
    "\n",
    "main()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
