AutoEncoder.py

import pandas as pd
import numpy as np
from torch import nn
import torch.nn.functional as F
from torch.autograd import Variable
import torch
from sklearn.preprocessing import StandardScaler
from torch.utils.data import Dataset, DataLoader
from torch.optim import Adam

import os

os.chdir(os.path.abspath(os.path.dirname(__file__)))

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


class Autoencoder(nn.Module):
    def __init__(self, D_in, H=50, H2=12, latent_dim=5):

        # Encoder
        super(Autoencoder, self).__init__()
        self.linear1 = nn.Linear(D_in, H)
        self.lin_bn1 = nn.BatchNorm1d(num_features=H)  # Batch normalization
        self.linear2 = nn.Linear(H, H2)
        self.lin_bn2 = nn.BatchNorm1d(num_features=H2)  # Batch normalization
        self.linear3 = nn.Linear(H2, H2)
        self.lin_bn3 = nn.BatchNorm1d(num_features=H2)  # Batch normalization
        # Latent vectors mu and sigma
        self.fc1 = nn.Linear(H2, latent_dim)
        self.bn1 = nn.BatchNorm1d(num_features=latent_dim)
        self.fc21 = nn.Linear(latent_dim, latent_dim)
        self.fc22 = nn.Linear(latent_dim, latent_dim)
        # Sampling vector
        self.fc3 = nn.Linear(latent_dim, latent_dim)
        self.fc_bn3 = nn.BatchNorm1d(latent_dim)
        self.fc4 = nn.Linear(latent_dim, H2)
        self.fc_bn4 = nn.BatchNorm1d(H2)  # Batch normalization
        # Decoder
        self.linear4 = nn.Linear(H2, H2)
        self.lin_bn4 = nn.BatchNorm1d(num_features=H2)
        self.linear5 = nn.Linear(H2, H)
        self.lin_bn5 = nn.BatchNorm1d(num_features=H)
        self.linear6 = nn.Linear(H, D_in)
        self.lin_bn6 = nn.BatchNorm1d(num_features=D_in)
        self.relu = nn.ReLU()  # Activation Function

    def encode(self, x):
        lin1 = self.relu(self.lin_bn1(self.linear1(x)))
        lin2 = self.relu(self.lin_bn2(self.linear2(lin1)))
        lin3 = self.relu(self.lin_bn3(self.linear3(lin2)))
        fc1 = F.relu(self.bn1(self.fc1(lin3)))
        r1 = self.fc21(fc1)
        r2 = self.fc22(fc1)
        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):
        fc3 = self.relu(self.fc_bn3(self.fc3(z)))
        fc4 = self.relu(self.fc_bn4(self.fc4(fc3)))
        lin4 = self.relu(self.lin_bn4(self.linear4(fc4)))
        lin5 = self.relu(self.lin_bn5(self.linear5(lin4)))
        return self.lin_bn6(self.linear6(lin5))

    def forward(self, x):
        mu, logvar = self.encode(x)
        z = self.reparameterize(mu, logvar)
        return self.decode(z), mu, logvar


class customLoss(nn.Module):
    def __init__(self):
        super(customLoss, self).__init__()
        self.mse_loss = nn.MSELoss(reduction="sum")

    # x_recon ist der im forward im Model erstellte recon_batch, x ist der originale x Batch, mu ist mu und logvar ist logvar
    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


# takes in a module and applies the specified weight initialization
def weights_init_uniform_rule(m):
    classname = m.__class__.__name__
    # for every Linear layer in a model..
    if classname.find("Linear") != -1:
        # get the number of the inputs
        n = m.in_features
        y = 1.0 / np.sqrt(n)
        m.weight.data.uniform_(-y, y)
        m.bias.data.fill_(0)

# %% Data dependent method

# def load_data(path):
#     df = pd.read_csv(
#         path,
#         sep=",",
#         header=None,
#         names=[
#             "Wine",
#             "Mg",
#             "Phenols",
#             "Flavanoids",
#             "Nonflavanoid.phenols",
#             "Proanth",
#             "Color.int",
#             "Hue",
#             "OD",
#             "Proline",
#         ],
#     )
#     df = df.fillna(-99)  # replace nan with -99
#     df_base = df.iloc[:, 1:]
#     df_wine = df.iloc[:, 0].values  # get wine Label
#     x = df_base.values.reshape(-1, df_base.shape[1]).astype("float32")
#     standardizer = StandardScaler()  # stadardize values
#     x = standardizer.fit_transform(x)
#     return x, standardizer, df_wine  # 3- Dimensions


# def numpyToTensor(x):
#     x_train = torch.from_numpy(x).to(device)
#     return x_train


# class DataBuilder(Dataset):
#     def __init__(self, path):
#         self.x, self.standardizer, self.wine = load_data(path)
#         self.x = numpyToTensor(self.x)
#         self.len=self.x.shape[0]
#     def __getitem__(self,index):
#         return self.x[index]
#     def __len__(self):
#         return self.len

# %% run auto-encoder
# DATA_PATH = './test_AE/Wine.csv'
# data_set=DataBuilder(DATA_PATH)
# trainloader=DataLoader(dataset=data_set, batch_size=1024)


# D_in = data_set.x.shape[1]
# H = 50
# H2 = 12
# model = Autoencoder(D_in, H, H2).to(device)
# model.apply(weights_init_uniform_rule)
# optimizer = Adam(model.parameters(), lr=1e-3)
# loss_mse = customLoss()#LOSS

# # Training the Model
# train_losses = []
# def train(epoch):
#     model.train()
#     train_loss = 0
#     for batch_idx, data in enumerate(trainloader):
#         data = data.to(device)
#         optimizer.zero_grad()
#         recon_batch, mu, logvar = model(data)
#         loss = loss_mse(recon_batch, data, mu, logvar)
#         loss.backward()
#         train_loss += loss.item()
#         optimizer.step()

#     if epoch % 200 == 0:
#         print('====> Epoch: {} Average loss: {:.4f}'.format(
#             epoch, train_loss / len(trainloader.dataset)))
#         train_losses.append(train_loss / len(trainloader.dataset))

# epochs = 1400
# for epoch in range(1, epochs + 1):
#     train(epoch)