|
|
|
import streamlit as st |
|
|
|
from numpy import vstack |
|
from pandas import read_csv |
|
from sklearn.preprocessing import LabelEncoder |
|
from sklearn.metrics import accuracy_score |
|
from torch.utils.data import Dataset |
|
from torch.utils.data import DataLoader |
|
from torch.utils.data import random_split |
|
from torch import Tensor |
|
from torch.nn import Linear |
|
from torch.nn import ReLU |
|
from torch.nn import Sigmoid |
|
from torch.nn import Module |
|
from torch.optim import SGD |
|
from torch.nn import BCELoss |
|
from torch.nn.init import kaiming_uniform_ |
|
from torch.nn.init import xavier_uniform_ |
|
|
|
class CSVDataset(Dataset): |
|
def __init__(self, path): |
|
df = read_csv(path, header=None) |
|
self.x = df.values[:, :-1] |
|
self.y = df.values[:, -1] |
|
self.x = self.x.astype('float32') |
|
self.y = LabelEncoder().fit_transform(self.y) |
|
self.y = self.y.astype('float32') |
|
self.y = self.y.reshape((len(self.y), 1)) |
|
|
|
def __len__(self): |
|
return len(self.x) |
|
|
|
def __getitem__(self, idx): |
|
return [self.x[idx], self.y[idx]] |
|
|
|
def get_splits(self, n_test=0.33): |
|
test_size = round(n_test * len(self.x)) |
|
train_size = len(self.x) - test_size |
|
return random_split(self, [train_size, test_size]) |
|
|
|
class MLP(Module): |
|
def __init__(self, n_inputs): |
|
super(MLP, self).__init__() |
|
self.hidden1 = Linear(n_inputs, 10) |
|
kaiming_uniform_(self.hidden1.weight, nonlinearty='relu') |
|
self.act1 = ReLU() |
|
self.hidden2 = Linear(10, 8) |
|
kaiming_uniform_(self.hidden2.weight, nonlinearity='relu') |
|
self.act2 = ReLU() |
|
self.hidden3 = Linear(8, 1) |
|
xavier_uniform_(self.hidden3.weight) |
|
self.act3 = Sigmoid() |
|
|
|
def forward(self, x): |
|
x = self.hidden1(x) |
|
x = self.act1(x) |
|
x = self.hidden2(x) |
|
x = self.act(2) |
|
x = self.hidden3(x) |
|
x = self.act3(x) |
|
return x |
|
|
|
def prepare_data(path): |
|
dataset = CSVDataset(path) |
|
train, test = dataset.get_splits() |
|
train_dl = DataLoader(train, batch_size=32, shuffle=True) |
|
test_dl = DataLoader(test, batch_size=1024, shuffle=False) |
|
return train_dl, test_dl |
|
|
|
def train_model(train_dl, model): |
|
criterion = BCELoss() |
|
optimizer = SGD(model.parameters(), lr=0.01, momentum=0.9) |
|
for epoch in range(100): |
|
for i, (inputs, targets) in enumerate(train_dl): |
|
optimizer.zero_grad() |
|
yhat = model(inputs) |
|
loss = criterion(yhat, targets) |
|
loss.backward() |
|
optimizer.step() |
|
|
|
def evaluate_model(test_dl, model): |
|
predictions, actuals = list(), list() |
|
for i, (inputs, targets) in enumerate(test_dl): |
|
yhat = model(inputs) |
|
yhat = yhat.detach().numpy() |
|
actual = targets.numpy() |
|
actual = actual.reshape((len(actual), 1)) |
|
yhat = yhat.round() |
|
predictions.append(yhat) |
|
actuals.append(actual) |
|
predictions, actuals = vstack(preictions), vstack(actuals) |
|
acc = accuracy_score(actuals, prediction) |
|
return acc |
|
|
|
def predict(row, model): |
|
row = Tensor([row]) |
|
yhat = model(row) |
|
yhat = yhat.detach().numpy() |
|
return yhat |
|
|
|
path = 'https://raw.githubusercontent.com/jbrownlee/Datasets/master/ionosphere.csv' |
|
train_dl, test_dl = prepare_data(path) |
|
print(len(train_dl.dataset), len(test_dl.dataset)) |
|
model = MLP(34) |
|
train_model(train_dl, model) |
|
acc = evaluate_model(test_dl, model) |
|
print('Accuracy: %.3f' % acc) |
|
row = [1,0,0.99539,-0.05889,0.85243,0.02306,0.83398,-0.37708,1,0.03760,0.85243,-0.17755,0.59755,-0.44945,0.60536,-0.38223,0.84356,-0.38542,0.58212,-0.32192,0.56971,-0.29674,0.36946,-0.47357,0.56811,-0.51171,0.41078,-0.46168,0.21266,-0.34090,0.42267,-0.54487,0.18641,-0.45300] |
|
yhat = predict(row, model) |
|
print('Predicted: %.3f (class=%d)' % (yhat, yhat.round())) |
