text_plain_output_1.png
from PIL import Image

from sklearn.model_selection import train_test_split

from torch.autograd import Variable

import matplotlib.pyplot as plt

import numpy as np

import pandas as pd

import random

import torch

import torch.nn as nn

import torch.optim as optim

train = pd.read_csv('../input/train.csv')
test = pd.read_csv('../input/test.csv')

x_train = train.iloc[:, 1:].values / 255.0
y_train = train.iloc[:, 0].values
x_test = test.values / 255

x_train = np.reshape(x_train, (-1, 1, 28, 28))
x_test = np.reshape(x_test, (-1, 1, 28, 28))
(x_train.shape, x_test.shape)

random_seed = 234
x_train, x_val, y_train, y_val = train_test_split(x_train, y_train, test_size=0.1, random_state=random_seed)
(x_train.shape, x_val.shape, y_train.shape, y_val.shape)

def display(rows, columns, images, values=[], predictions=[], save_image=False, filename='image.png', show=False):
    fig = plt.figure(figsize=(15, 15))
    
#     plt.ioff()

    ax = []

    for i in range( columns*rows ):
        img = images[i]
        ax.append(fig.add_subplot(rows, columns, i+1))
        
        title = ""
        
        if(len(values) == 0 and len(predictions) != 0):
            title = "Pred:" + str(predictions[i])
        elif(len(values) != 0 and len(predictions) == 0):
            title = "Value:" + str(values[i])
        elif(len(values) != 0 and len(predictions) != 0):
            title = "\nValue:" + str(values[i]) + "\nPred:" + str(predictions[i])
        
        ax[-1].set_title(title)  # set title
        plt.imshow(img)
        plt.axis('off')
        
    if save_image:
        plt.savefig(filename)
    
    if not show:
        plt.close(fig)
        
#     plt.show()
    
idx = np.random.randint(1, 1000, size=9)

images = x_train[idx,:]
images = images[:,0]

values = y_train[idx]
print(images.shape)
display(rows=3, columns=3, images=images, values=values, predictions=[], save_image=True, filename='Autoencoder-validation.png', show=True)

device = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu')

class customLoss(nn.Module):
    def __init__(self):
        super(customLoss, self).__init__()
        self.mse_loss = nn.MSELoss(reduction='sum')
    def forward(self, x_recon, x, mu, logvar):
        loss_MSE = self.mse_loss(x_recon, x)
        loss_KLD = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp())
        return loss_MSE + loss_KLD

class VAE_CNN(nn.Module):
    def __init__(self):
        super(VAE_CNN, self).__init__()
        self.e_conv1 = nn.Conv2d(1, 16, kernel_size=3, stride=1, padding=1, bias=False)
        self.e_bn1 = nn.BatchNorm2d(16)
        self.e_conv2 = nn.Conv2d(16, 32, kernel_size=3, stride=2, padding=1, bias=False)
        self.e_bn2 = nn.BatchNorm2d(32)
        self.e_conv3 = nn.Conv2d(32, 64, kernel_size=3, stride=1, padding=1, bias=False)
        self.e_bn3 = nn.BatchNorm2d(64)
        self.e_conv4 = nn.Conv2d(64, 16, kernel_size=3, stride=2, padding=1, bias=False)
        self.e_bn4 = nn.BatchNorm2d(16)
        self.e_fc1 = nn.Linear(7 * 7 * 16, 512)
        self.e_fc_bn1 = nn.BatchNorm1d(512)
        self.e_fc2 = nn.Linear(512, 256)
        self.e_fc_bn2 = nn.BatchNorm1d(256)
        self.lv_fc1 = nn.Linear(256, 256)
        self.lv_fc2 = nn.Linear(256, 256)
        self.d_fc1 = nn.Linear(256, 512)
        self.d_fc_bn1 = nn.BatchNorm1d(512)
        self.d_fc2 = nn.Linear(512, 7 * 7 * 16)
        self.d_fc_bn2 = nn.BatchNorm1d(7 * 7 * 16)
        self.d_conv1 = nn.ConvTranspose2d(16, 64, kernel_size=3, stride=2, padding=1, output_padding=1, bias=False)
        self.d_bn1 = nn.BatchNorm2d(64)
        self.d_conv2 = nn.ConvTranspose2d(64, 32, kernel_size=3, stride=1, padding=1, bias=False)
        self.d_bn2 = nn.BatchNorm2d(32)
        self.d_conv3 = nn.ConvTranspose2d(32, 16, kernel_size=3, stride=2, padding=1, output_padding=1, bias=False)
        self.d_bn3 = nn.BatchNorm2d(16)
        self.d_conv4 = nn.ConvTranspose2d(16, 1, kernel_size=3, stride=1, padding=1, bias=False)
        self.relu = nn.ReLU()
    def encode(self, x, test=False):
        x = self.relu(self.e_bn1(self.e_conv1(x)))
        x = self.relu(self.e_bn2(self.e_conv2(x)))
        x = self.relu(self.e_bn3(self.e_conv3(x)))
        x = self.relu(self.e_bn4(self.e_conv4(x)))
        x = x.view(-1, 16 * 7 * 7)
        x = self.relu(self.e_fc_bn1(self.e_fc1(x)))
        x = self.relu(self.e_fc_bn2(self.e_fc2(x)))
        r1 = self.lv_fc1(x)
        r2 = self.lv_fc1(x)
        return (r1, r2)
    def reparameterize(self, mu, logvar):
        if self.training:
            std = logvar.mul(0.5).exp_()
            eps = Variable(std.data.new(std.size()).normal_())
            return eps.mul(std).add_(mu)
        else:
            return mu
    def decode(self, z, test=False):
        z = self.relu(self.d_fc_bn1(self.d_fc1(z)))
        z = self.relu(self.d_fc_bn2(self.d_fc2(z)))
        z = z.view(-1, 16, 7, 7)
        z = self.relu(self.d_bn1(self.d_conv1(z)))
        z = self.relu(self.d_bn2(self.d_conv2(z)))
        z = self.relu(self.d_bn3(self.d_conv3(z)))
        z = self.d_conv4(z)
        return z
    def forward(self, x):
        mu, logvar = self.encode(x)
        z = self.reparameterize(mu, logvar)
        return (self.decode(z), mu, logvar)
    def print(self, t, test=False):
        pass
    def setup(self, train_data_loader, val_data_loader, epochs=10, log_interval=10):
        self.optimizer = optim.Adam(self.parameters(), lr=0.001)
        self.loss_mse = customLoss()
        self.epochs = epochs
        self.epoch = 0
        self.train_data_loader = train_data_loader
        self.val_data_loader = val_data_loader
        self.log_interval = log_interval
        self.is_cuda = True
    def validate_epoch(self):
        it = iter(self.val_data_loader)
        ii = random.randint(1, 30)
        for i in range(ii):
            data, _ = it.next()
        output, _, _ = model(data.cuda())
        data = data.numpy()
        output = output.cpu().data.numpy()
        data = data[:, 0]
        output = output[:, 0]
        images = []
        for i in range(18):
            images.append(data[i])
            images.append(output[i])
    def train_epoch(self):
        self.train()
        train_loss = 0.0
        for batch_idx, (data, _) in enumerate(self.train_data_loader):
            if self.is_cuda:
                data = data.cuda()
            self.optimizer.zero_grad()
            recon_batch, mu, logvar = self(data)
            loss = self.loss_mse(recon_batch, data, mu, logvar)
            loss.backward()
            train_loss += loss.item()
            self.optimizer.step()
    def train_model(self):
        self.train()
        self.validate_epoch()
        for i in range(self.epochs):
            self.train_epoch()
            self.epoch += 1
            self.validate_epoch()
model = VAE_CNN().cuda()

import PIL
images = []
values = []
for i in range(10):
    image = Image.open('Autoencoder-validation-' + str(i) + '.png')
    images.append(np.asarray(image))
    values.append('Epoch ' + str(i))
display(rows=3, columns=3, images=images, values=values, predictions=[], show=True)