{"cells":[{"metadata":{},"cell_type":"markdown","source":"<pre>\n                       !\n                      /^\\\n                    /     \\\n |               | (       ) |               |\n/^\\  |          /^\\ \\     / /^\\          |  /^\\\n|O| /^\\        (   )|-----|(   )        /^\\ |O|\n|_| |-|    |^-^|---||-----||---|^-^|    |-| |_|\n|O| |O|    |/^\\|/^\\||  |  ||/^\\|/^\\|    |O| |O|\n|-| |-|    ||_|||_||| /^\\ |||_|||_||    |-| |-|\n|O| |O|    |/^\\|/^\\||(   )||/^\\|/^\\|    |O| |O|\n|-| |-|    ||_|||_||||   ||||_|||_||    |-| |-|\n|O| |_|----|___|___|||___|||___|_|_|    |O| |O|\n|_|                                         |_|\n   /_______________________________________\\\n__|_______________________________________|___|\n\n<b>Kannada - Unsupervised Learning Autoencoders</b>\nby Alin Cijov\n</pre>"},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"_kg_hide-input":true},"cell_type":"code","source":"import numpy as np\nimport pandas as pd\nfrom typing import Optional, Tuple\n\nimport torch\nimport torch.nn as nn\nfrom torch import Tensor\nimport torch.optim as optim\nfrom torch.optim import lr_scheduler\nfrom torch.optim import Optimizer\nfrom torch.nn.modules.loss import _Loss\nfrom torch.utils.data import DataLoader\nfrom torch.nn.modules.loss import _Loss\n\nfrom sklearn.model_selection import train_test_split\nfrom sklearn.manifold import TSNE\n\nimport matplotlib.pyplot as plt\nimport seaborn as sns\n%matplotlib inline","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Prepare Data"},{"metadata":{"trusted":true},"cell_type":"code","source":"path = '../input/Kannada-MNIST/'\ndf_train = pd.read_csv(path + 'train.csv')\ndf_test = pd.read_csv(path + 'test.csv')\ndf_train.head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"labels = df_train['label'].values\nfeatures = df_train[list(df_train.columns)[1:]].values\nfeatures = features.reshape(-1, 28, 28, 1)\nfeatures = torch.tensor(features).type(torch.FloatTensor)\nfeatures = features.permute(0, 3, 1, 2)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"X_train, X_test, y_train, y_test = train_test_split(\n                                    features, labels, test_size=0.3, random_state=42)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"X_train_auto = (X_train - X_train.min()) / (X_train.max() - X_train.min()) * 2 - 1\nX_test_auto = (X_test - X_train.min()) / (X_train.max() - X_train.min()) * 2 - 1","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Implementation"},{"metadata":{"trusted":true},"cell_type":"code","source":"class Layer(nn.Module):\n\n    def __init__(self) -> None:\n        super().__init__()\n\n    def forward(self, x: Tensor,\n                inference: bool = False) -> Tensor:\n        raise NotImplementedError()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class DeconvLayer(Layer):\n    def __init__(self,\n                 in_channels: int,\n                 out_channels: int,\n                 filter_size: int,\n                 activation: nn.Module = None,\n                 dropout: float = 1.0,\n                 flatten: bool = False) -> None:\n        super().__init__()\n        self.deconv = nn.ConvTranspose2d(in_channels, out_channels, filter_size, \n                                       padding=filter_size // 2)\n        self.activation = activation\n        self.flatten = flatten\n        if dropout < 1.0:\n            self.dropout = nn.Dropout(1 - dropout)\n\n    def forward(self, x: Tensor) -> Tensor:\n\n        x = self.deconv(x)\n        if self.activation:\n            x = self.activation(x)\n        if self.flatten:\n            x = x.view(x.shape[0], x.shape[1] * x.shape[2] * x.shape[3])\n        if hasattr(self, \"dropout\"):\n            x = self.dropout(x)            \n            \n        return x","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class Model(nn.Module):\n\n    def __init__(self) -> None:\n        super().__init__()\n\n    def forward(self, x: Tensor) -> Tuple[Tensor]:\n        raise NotImplementedError()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class ConvLayer(Layer):\n    def __init__(self,\n                 in_channels: int,\n                 out_channels: int,\n                 filter_size: int,\n                 activation: nn.Module = None,\n                 dropout: float = 1.0,\n                 flatten: bool = False) -> None:\n        super().__init__()\n        self.conv = nn.Conv2d(in_channels, out_channels, filter_size, \n                              padding=filter_size // 2)\n        self.activation = activation\n        self.flatten = flatten\n        if dropout < 1.0:\n            self.dropout = nn.Dropout(1 - dropout)\n\n    def forward(self, x: Tensor) -> Tensor:\n\n        x = self.conv(x)\n        if self.activation:\n            x = self.activation(x)\n        if self.flatten:\n            x = x.view(x.shape[0], x.shape[1] * x.shape[2] * x.shape[3])\n        if hasattr(self, \"dropout\"):\n            x = self.dropout(x)            \n            \n        return x","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class DenseLayer(Layer):\n    def __init__(self,\n                 input_size: int,\n                 neurons: int,\n                 dropout: float = 1.0,\n                 activation: nn.Module = None) -> None:\n\n        super().__init__()\n        self.linear = nn.Linear(input_size, neurons)\n        self.activation = activation\n        if dropout < 1.0:\n            self.dropout = nn.Dropout(1 - dropout)\n\n    def forward(self, x: Tensor,\n                inference: bool = False) -> Tensor:\n        if inference:\n            self.apply(inference_mode)\n\n        x = self.linear(x)\n        if self.activation:\n            x = self.activation(x)\n        if hasattr(self, \"dropout\"):\n            x = self.dropout(x)\n\n        return x","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class Autoencoder(Model):\n    def __init__(self,\n                 hidden_dim: int = 28):\n        super(Autoencoder, self).__init__()\n        self.conv1 = ConvLayer(1, 14, 5, activation=nn.Tanh())\n        self.conv2 = ConvLayer(14, 7, 5, activation=nn.Tanh(), flatten=True)\n        \n        self.dense1 = DenseLayer(7 * 28 * 28, hidden_dim, activation=nn.Tanh())\n        self.dense2 = DenseLayer(hidden_dim, 7 * 28 * 28, activation=nn.Tanh())\n        \n        self.conv3 = ConvLayer(7, 14, 5, activation=nn.Tanh()) \n        self.conv4 = ConvLayer(14, 1, 5, activation=nn.Tanh())         \n\n    def forward(self, x: Tensor) -> Tensor:\n        assert_dim(x, 4)\n            \n        x = self.conv1(x)\n        x = self.conv2(x)\n\n        encoding = self.dense1(x)\n        \n        x = self.dense2(encoding)\n        \n        x = x.view(-1, 7, 28, 28)\n        \n        x = self.conv3(x)\n        x = self.conv4(x)\n\n        return x, encoding","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def permute_data(X: Tensor, y: Tensor, seed=1) -> Tuple[Tensor]:\n    perm = torch.randperm(X.shape[0])\n    return X[perm], y[perm]\n\ndef assert_dim(t: Tensor,\n               dim: Tensor):\n    assert len(t.shape) == dim, \\\n        '''\n        Tensor expected to have dimension {0}, instead has dimension {1}\n        '''.format(dim, len(t.shape))\n    return None","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Training"},{"metadata":{"trusted":true},"cell_type":"code","source":"class Trainer(object):\n    def __init__(self,\n                 model: Model,\n                 optim: Optimizer,\n                 criterion: _Loss):\n        self.model = model\n        self.optim = optim\n        self.loss = criterion\n        self._check_optim_net_aligned()\n\n    def _check_optim_net_aligned(self):\n        assert self.optim.param_groups[0]['params']\\\n            == list(self.model.parameters())\n\n    def _generate_batches(self,\n                          X: Tensor,\n                          y: Tensor,\n                          size: int = 32) -> Tuple[Tensor]:\n\n        N = X.shape[0]\n\n        for ii in range(0, N, size):\n            X_batch, y_batch = X[ii:ii+size], y[ii:ii+size]\n\n            yield X_batch, y_batch\n\n    def fit(self, X_train: Tensor = None,\n            y_train: Tensor = None,\n            X_test: Tensor = None,\n            y_test: Tensor = None,\n            train_dataloader: DataLoader = None,\n            test_dataloader: DataLoader = None,\n            epochs: int=100,\n            eval_every: int=10,\n            batch_size: int=32,\n            final_lr_exp: int = None):\n\n        init_lr = self.optim.param_groups[0]['lr']\n        if final_lr_exp:\n            decay = (final_lr_exp / init_lr) ** (1.0 / (epochs + 1))\n            scheduler = lr_scheduler.ExponentialLR(self.optim, gamma=decay)\n        for e in range(epochs):\n\n            if final_lr_exp:\n                scheduler.step()\n\n            if not train_dataloader:\n                X_train, y_train = permute_data(X_train, y_train)\n\n                batch_generator = self._generate_batches(X_train, y_train,\n                                                         batch_size)\n\n                self.model.train()\n\n                for ii, (X_batch, y_batch) in enumerate(batch_generator):\n\n                    self.optim.zero_grad()\n\n                    output = self.model(X_batch)[0]\n\n                    loss = self.loss(output, y_batch)\n                    loss.backward()\n                    self.optim.step()\n\n                if e % eval_every == 0:\n                    with torch.no_grad():\n                        self.model.eval()\n                        output = self.model(X_test)[0]\n                        loss = self.loss(output, y_test)\n                        print(\"Epoch:\", e+1, \", Loss:\", loss.item())\n\n            else:\n                for X_batch, y_batch in train_dataloader:\n\n                    self.optim.zero_grad()\n\n                    output = self.model(X_batch)[0]\n\n                    loss = self.loss(output, y_batch)\n                    loss.backward()\n                    self.optim.step()\n\n                if e % eval_every == 0:\n                    with torch.no_grad():\n                        self.model.eval()\n                        losses = []\n                        for X_batch, y_batch in test_dataloader:\n                            output = self.model(X_batch)[0]\n                            loss = self.loss(output, y_batch)\n                            losses.append(loss.item())\n                        print(\"Epoch:\", e, \", Loss:\",\n                              round(torch.Tensor(losses).mean().item(), 4))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"model = Autoencoder(hidden_dim=28)\ncriterion = nn.MSELoss()\noptimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.9)\n\ntrainer = Trainer(model, optimizer, criterion)\n\ntrainer.fit(X_train_auto, X_train_auto,\n            X_test_auto, X_test_auto,\n            epochs=10,\n            batch_size=60)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Reconstruct image from trained model"},{"metadata":{"trusted":true},"cell_type":"code","source":"reconstructed_images, image_representations = model(X_test_auto)","execution_count":null,"outputs":[]},{"metadata":{"_kg_hide-input":true,"trusted":true},"cell_type":"code","source":"def display_image(ax,\n    t: Tensor):\n    n = t.detach().numpy()\n    ax.imshow(n.reshape(28, 28))\n\nf, axarr = plt.subplots(5,8, figsize=(14,9))\n\nk = 0\nfor i in range(5):\n    \n    for j in range(7):\n        display_image(axarr[i][j + (j % 2)], X_test[k])\n        display_image(axarr[i][j + (j % 2 + 1)], reconstructed_images[k])\n        k += 1\n\n    for j in range(7):\n        axarr[i][j + (j % 2)].set_title(\"Original\")\n        axarr[i][j + (j % 2 + 1)].set_title(\"Reconstructed\")\n\n    for j in range(7):\n        axarr[i][j + (j % 2)].axis('off')\n        axarr[i][j + (j % 2 + 1)].axis('off')","execution_count":null,"outputs":[]}],"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat":4,"nbformat_minor":4}