AutoEncoder 실습

In [ ]:

import torch
from torch import nn
import torch.nn.functional as F
from torch.utils.data import Dataset, DataLoader
from torchvision import datasets
import torch.optim as optim
from torchvision.transforms import ToTensor
import matplotlib.pyplot as plt
import numpy as np
import json
from tqdm.auto import tqdm

import time

In [ ]:

class Config:
    def __init__(self, path):
        self.path = path
        with open(self.path, 'r') as f:
            data = json.load(f)
            self.__dict__.update(data)


    @property
    def dict(self):
        return self.__dict__

In [ ]:

BATCH = 4
LR = 1e-4
training_data = datasets.FashionMNIST(
    root='../data',
    train=True,
    download=True,
    transform=ToTensor() # TODO
)

test_data = datasets.FashionMNIST(
    root='../data',
    train=False,
    download=True,
    transform=ToTensor()
)


train_dataloader = DataLoader(training_data, batch_size = BATCH, shuffle=True)
test_dataloader = DataLoader(test_data, batch_size = BATCH, shuffle=True)

In [ ]:

data, label = next(iter(train_dataloader))

In [ ]:

data.shape

Vanila Autoencoder¶

In [ ]:

class AE(nn.Module):
    def __init__(self, height, width, hidden_dim=256, latent_dim=32, dropout=0.1, color_channel=1):
        super(AE, self).__init__()
        self.height = height
        self.width = width
        self.hidden_dim = hidden_dim
        self.latent_dim = latent_dim
        self.dropout = dropout
        self.color_channel = color_channel
        
        self.encoder = nn.Sequential(
            nn.Linear(self.height*self.width*self.color_channel,self.hidden_dim),
            nn.ReLU(),
            nn.Dropout(self.dropout),
            nn.Linear(self.hidden_dim, self.latent_dim),
            nn.ReLU()
        )
        
        self.decoder = nn.Sequential(
            nn.Linear(self.latent_dim, self.hidden_dim),
            nn.ReLU(),
            nn.Dropout(self.dropout),
            nn.Linear(self.hidden_dim, self.height*self.width*self.color_channel),
            nn.Sigmoid()
        )
        
    def forward(self, x):
        batch_size = x.size(0)
        x = x.view(batch_size, -1)
        latent_variable = self.encoder(x)
        output = self.decoder(latent_variable)
        output = output.view(batch_size, -1, self.height, self.width)
        
        return output, latent_variable

Convolution Autoencoder¶

In [ ]:

class CAE(nn.Module):
    def __init__(self, height, width, color_channel=3):
        super(CAE, self).__init__()
        self.height = height
        self.width = width
        self.color_channel = color_channel

        self.encoder = nn.Sequential(
            nn.Conv2d(in_channels=self.color_channel, out_channels=32, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(32),
            nn.ReLU(),

            nn.Conv2d(in_channels=32, out_channels=64, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(),

            nn.Conv2d(in_channels=64, out_channels=128, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=1, stride=2),

            nn.Conv2d(in_channels=128, out_channels=256, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(256),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=1, stride=2)
        )

        self.decoder = nn.Sequential(
            nn.ConvTranspose2d(in_channels=256, out_channels=128, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(),

            nn.ConvTranspose2d(in_channels=128, out_channels=64, kernel_size=2, stride=2, padding=0),
            nn.BatchNorm2d(64),
            nn.ReLU(),

            nn.ConvTranspose2d(in_channels=64, out_channels=32, kernel_size=2, stride=2, padding=0),
            nn.BatchNorm2d(32),
            nn.ReLU(),

            nn.ConvTranspose2d(in_channels=32, out_channels=self.color_channel, kernel_size=3, stride=1, padding=1),
            nn.Sigmoid()
        )


    def forward(self, x):
        latent_variable = self.encoder(x)
        output = self.decoder(latent_variable)

        return output, latent_variable

Train func¶

In [ ]:

dataloaders_dict = {
    'train' : train_dataloader,
    'valid' : test_dataloader
}

In [ ]:

def vis_data(image, label):
    fig = plt.figure()
    for i in range(4):
        ax = fig.add_subplot(1,4,i+1)
        ax.imshow(image[i].to('cpu').permute(1,2,0), cmap='gray')
        ax.set_title(label[i].item())
        
    plt.tight_layout()
    plt.show()

In [ ]:

def train(model, dataloaders, criterion, optimizer, num_epochs=25):
    since = time.time()
    model = model.to(device)
    val_acc_history = []

    for epoch in range(num_epochs):
        print('Epoch {}/{}'.format(epoch, num_epochs - 1))
        print('-' * 10)

        # Each epoch has a training and validation phase
        for phase in ['train', 'valid']:
            if phase == 'train':
                model.train()  # Set model to training mode
            else:
                model.eval()   # Set model to evaluate mode

            running_loss = 0.0
            running_corrects = 0

            # Iterate over data.
            for inputs, labels in tqdm(dataloaders[phase]):
                inputs = inputs.to(device)
                labels = labels.to(device)

                # zero the parameter gradients
                optimizer.zero_grad()

                # forward
                # track history if only in train
                with torch.set_grad_enabled(phase == 'train'):
                    outputs, latent_variable = model(inputs)
                    loss = criterion(outputs, inputs)
                    
                    #_, preds = torch.max(outputs, 1)

                    # backward + optimize only if in training phase
                    if phase == 'train':
                        loss.backward()
                        optimizer.step()
            running_loss += loss.item() * inputs.size(0)
        
        epoch_loss = running_loss / len(dataloaders[phase].dataset)
        print('{} Loss: {:.4f}'.format(phase, epoch_loss))
        print()

    time_elapsed = time.time() - since
    print('Training complete in {:.0f}m {:.0f}s'.format(time_elapsed // 60, time_elapsed % 60))
    
    return model

In [ ]:

model = AE(height=28, width=28)
optimizer = optim.Adam(model.parameters(), lr=LR)
criterion = nn.MSELoss()
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

In [ ]:

model = train(model, dataloaders_dict, criterion, optimizer, 100)

In [ ]:

data, label = next(iter(test_dataloader))
data = data.to(device)
output, latent = model(data)

In [ ]:

output = output.to('cpu')

In [ ]:

latent.shape

In [ ]:

print(latent[1, :])

In [ ]:

cp_latent = latent.to('cpu').detach().numpy().copy()

np.concatenate((latent.to('cpu').detach().numpy(), cp_latent), axis=0)

In [ ]:

np.concatenate((np.array(), latent.to('cpu').detach().numpy()), axis=0).shape

In [ ]:

vis_data(output.detach(), label)

In [ ]:

total_latent_variable = None
total_y = None
for input, label in tqdm(test_dataloader) :
    output, latent = model(data)
    latent = latent.to('cpu').detach().numpy()
    if total_latent_variable is None :
        total_latent_variable = latent
        total_y = label
    else :
        total_latent_variable = np.concatenate((total_latent_variable, latent), axis=0)
        total_y = np.concatenate((total_y, label), axis=0)

In [ ]:

total_latent_variable.shape

In [ ]:

total_y.shape

In [ ]:

from sklearn.manifold import TSNE

np.random.seed(42)
tsne=TSNE()
x_test_2D = tsne.fit_transform(total_latent_variable)
x_test_2D = (x_test_2D - x_test_2D.min())/(x_test_2D.max() - x_test_2D.min())

plt.scatter(x_test_2D[:,0], x_test_2D[:,1], s=10, cmap='tab10', c=total_y)

In [ ]:

# load library
from sklearn.datasets import load_digits
import matplotlib
import matplotlib.pyplot as plt

# matplotlib 설정
plt.rcParams['axes.unicode_minus'] = False # 축 - 설정
# data load
digits = load_digits()


# subplot 객체 생성
fig, axes = plt.subplots(2, 5, #  subplot객체(2x5)를 axes에 할당
subplot_kw={'xticks':(), 'yticks':()}) # subplot 축 눈금 해제

for ax, img in zip(axes.ravel(), digits.images): # axes.ravel()과 digits.images를 하나씩 할당
    ax.imshow(img, cmap='gray')

plt.show() # 그래프 출력

In [ ]:

# library import

from sklearn.manifold import TSNE

colors = ['#476A2A', '#7851B8', '#BD3430', '#4A2D4E', '#875525',
           '#A83683', '#4E655E', '#853541', '#3A3120', '#535D8E']

# t-SNE 모델 생성 및 학습
tsne = TSNE(random_state=0)
digits_tsne = tsne.fit_transform(digits.data)

In [ ]:

# 시각화
plt.scatter(digits_tsne[:, 0], digits_tsne[:, 1], c=digits.target)

plt.xlim(digits_tsne[:, 0].min(), digits_tsne[:, 0].max()) # 최소, 최대
plt.ylim(digits_tsne[:, 1].min(), digits_tsne[:, 1].max()) # 최소, 최대
plt.xlabel('t-SNE 0') # x축 이름
plt.ylabel('t-SNE 1') # y축 이름
plt.show() # 그래프 출력

In [ ]:

digits.data

In [ ]:

digits.target

NYU Autoencoder¶

In [1]:

import torch
import torchvision
from torch import nn
from torch.utils.data import DataLoader
from torchvision import transforms
from torchvision.datasets import MNIST
from matplotlib import pyplot as plt
from sklearn.manifold import TSNE

# Convert vector to image

def to_img(x):
    x = 0.5 * (x + 1)
    x = x.view(x.size(0), 28, 28)
    return x

# Displaying routine

def display_images(in_, out, n=1):
    for N in range(n):
        if in_ is not None:
            in_pic = to_img(in_.cpu().data)
            plt.figure(figsize=(18, 6))
            for i in range(4):
                plt.subplot(1,4,i+1)
                plt.imshow(in_pic[i+4*N])
                plt.axis('off')
        out_pic = to_img(out.cpu().data)
        plt.figure(figsize=(18, 6))
        for i in range(4):
            plt.subplot(1,4,i+1)
            plt.imshow(out_pic[i+4*N])
            plt.axis('off')

In [2]:

# Define data loading step

batch_size = 256

img_transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5,), (0.5,))
])

dataset = MNIST('../data', transform=img_transform, download=True)
dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)

Downloading http://yann.lecun.com/exdb/mnist/train-images-idx3-ubyte.gz
Downloading http://yann.lecun.com/exdb/mnist/train-images-idx3-ubyte.gz to ../data/MNIST/raw/train-images-idx3-ubyte.gz

  0%|          | 0/9912422 [00:00<?, ?it/s]

Extracting ../data/MNIST/raw/train-images-idx3-ubyte.gz to ../data/MNIST/raw

Downloading http://yann.lecun.com/exdb/mnist/train-labels-idx1-ubyte.gz
Downloading http://yann.lecun.com/exdb/mnist/train-labels-idx1-ubyte.gz to ../data/MNIST/raw/train-labels-idx1-ubyte.gz

  0%|          | 0/28881 [00:00<?, ?it/s]

Extracting ../data/MNIST/raw/train-labels-idx1-ubyte.gz to ../data/MNIST/raw

Downloading http://yann.lecun.com/exdb/mnist/t10k-images-idx3-ubyte.gz
Downloading http://yann.lecun.com/exdb/mnist/t10k-images-idx3-ubyte.gz to ../data/MNIST/raw/t10k-images-idx3-ubyte.gz

  0%|          | 0/1648877 [00:00<?, ?it/s]

Extracting ../data/MNIST/raw/t10k-images-idx3-ubyte.gz to ../data/MNIST/raw

Downloading http://yann.lecun.com/exdb/mnist/t10k-labels-idx1-ubyte.gz
Downloading http://yann.lecun.com/exdb/mnist/t10k-labels-idx1-ubyte.gz to ../data/MNIST/raw/t10k-labels-idx1-ubyte.gz

  0%|          | 0/4542 [00:00<?, ?it/s]

Extracting ../data/MNIST/raw/t10k-labels-idx1-ubyte.gz to ../data/MNIST/raw

In [3]:

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

In [8]:

# Define model architecture and reconstruction loss

# n = 28 x 28 = 784
d = 30  # for standard AE (under-complete hidden layer)
# d = 500  # for denoising AE (over-complete hidden layer)

class Autoencoder(nn.Module):
    def __init__(self):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(28 * 28, d),
            nn.Tanh(),
        )
        self.decoder = nn.Sequential(
            nn.Linear(d, 28 * 28),
            nn.Tanh(),
        )

    def forward(self, x):
        latent_vec = self.encoder(x)
        x = self.decoder(latent_vec)
        return x, latent_vec
    
model = Autoencoder().to(device)
criterion = nn.MSELoss()

In [17]:

class CAE(nn.Module):
    def __init__(self, height=28, width=28, color_channel=1):
        super(CAE, self).__init__()
        self.height = height
        self.width = width
        self.color_channel = color_channel

        self.encoder = nn.Sequential(
            nn.Conv2d(in_channels=self.color_channel, out_channels=32, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(32),
            nn.ReLU(),

            nn.Conv2d(in_channels=32, out_channels=64, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(),

            nn.Conv2d(in_channels=64, out_channels=128, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=1, stride=2),

            nn.Conv2d(in_channels=128, out_channels=256, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(256),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=1, stride=2)
        )

        self.decoder = nn.Sequential(
            nn.ConvTranspose2d(in_channels=256, out_channels=128, kernel_size=3, stride=1, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(),

            nn.ConvTranspose2d(in_channels=128, out_channels=64, kernel_size=2, stride=2, padding=0),
            nn.BatchNorm2d(64),
            nn.ReLU(),

            nn.ConvTranspose2d(in_channels=64, out_channels=32, kernel_size=2, stride=2, padding=0),
            nn.BatchNorm2d(32),
            nn.ReLU(),

            nn.ConvTranspose2d(in_channels=32, out_channels=self.color_channel, kernel_size=3, stride=1, padding=1),
            nn.Sigmoid()
        )


    def forward(self, x):
        latent_variable = self.encoder(x)
        output = self.decoder(latent_variable)

        return output, latent_variable
    
model = CAE().to(device)
criterion = nn.MSELoss()

In [18]:

# Configure the optimiser

learning_rate = 1e-3

optimizer = torch.optim.Adam(
    model.parameters(),
    lr=learning_rate,
)

In [44]:

# Train standard or denoising autoencoder (AE)

num_epochs = 20
# do = nn.Dropout()  # comment out for standard AE
for epoch in range(num_epochs):
    for data in dataloader:
        img, _ = data
        img = img.to(device)
        # img = img.view(img.size(0), -1)
        noise = torch.ones(img.shape).to(device)
        img_bad = (img * noise).to(device)  # comment out for standard AE
        # ===================forward=====================
        output, _ = model(img_bad)  # feed <img> (for std AE) or <img_bad> (for denoising AE)
        loss = criterion(output, img.data)
        # ===================backward====================
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    # ===================log========================
    print(f'epoch [{epoch + 1}/{num_epochs}], loss:{loss.item():.4f}')
    display_images(None, output)  # pass (None, output) for std AE, (img_bad, output) for denoising AE

epoch [1/20], loss:0.8368
epoch [2/20], loss:0.8364
epoch [3/20], loss:0.8349
epoch [4/20], loss:0.8392
epoch [5/20], loss:0.8333
epoch [6/20], loss:0.8310
epoch [7/20], loss:0.8279
epoch [8/20], loss:0.8369
epoch [9/20], loss:0.8339
epoch [10/20], loss:0.8339
epoch [11/20], loss:0.8261
epoch [12/20], loss:0.8298
epoch [13/20], loss:0.8386
epoch [14/20], loss:0.8391
epoch [15/20], loss:0.8371
epoch [16/20], loss:0.8324
epoch [17/20], loss:0.8421
epoch [18/20], loss:0.8261
epoch [19/20], loss:0.8324
epoch [20/20], loss:0.8379

In [38]:

test_loader = DataLoader(dataset, batch_size=1024, shuffle=True)

In [39]:

data, label = next(iter(test_loader))
data = data.to(device)
# data = data.view(data.size(0), -1)
output, latent_vec = model(data)
latent_vec = latent_vec.to('cpu').detach().numpy()
latent_vec = latent_vec.reshape(data.size(0), -1)

In [41]:

# t-SNE 모델 생성 및 학습
tsne = TSNE(random_state=0)

digits_tsne = tsne.fit_transform(latent_vec)

/usr/local/lib/python3.8/dist-packages/sklearn/manifold/t_sne.py:333: DeprecationWarning: `np.float` is a deprecated alias for the builtin `float`. To silence this warning, use `float` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.float64` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  error = np.finfo(np.float).max
/usr/local/lib/python3.8/dist-packages/sklearn/manifold/t_sne.py:334: DeprecationWarning: `np.float` is a deprecated alias for the builtin `float`. To silence this warning, use `float` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.float64` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  best_error = np.finfo(np.float).max
/usr/local/lib/python3.8/dist-packages/sklearn/manifold/t_sne.py:333: DeprecationWarning: `np.float` is a deprecated alias for the builtin `float`. To silence this warning, use `float` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.float64` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  error = np.finfo(np.float).max
/usr/local/lib/python3.8/dist-packages/sklearn/manifold/t_sne.py:334: DeprecationWarning: `np.float` is a deprecated alias for the builtin `float`. To silence this warning, use `float` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.float64` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  best_error = np.finfo(np.float).max

In [42]:

# 시각화
plt.scatter(digits_tsne[:, 0], digits_tsne[:, 1], c=label)

plt.xlim(digits_tsne[:, 0].min(), digits_tsne[:, 0].max()) # 최소, 최대
plt.ylim(digits_tsne[:, 1].min(), digits_tsne[:, 1].max()) # 최소, 최대
plt.xlabel('t-SNE 0') # x축 이름
plt.ylabel('t-SNE 1') # y축 이름
plt.show() # 그래프 출력

In [ ]:

import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import datasets, transforms
from sklearn.manifold import TSNE
import matplotlib.pyplot as plt

# Set device
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# Define transforms
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.1307,), (0.3081,))
])

# Load dataset
train_dataset = datasets.MNIST(root='./data', train=True, transform=transform, download=True)
test_dataset = datasets.MNIST(root='./data', train=False, transform=transform, download=True)

# Define data loader
train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=128, shuffle=True)
test_loader = torch.utils.data.DataLoader(test_dataset, batch_size=128, shuffle=False)

# Define autoencoder model
class Autoencoder(nn.Module):
    def __init__(self):
        super(Autoencoder, self).__init__()
        self.encoder = nn.Sequential(
            nn.Linear(28*28, 256),