# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# # Simple Dense Neural Network / Fully connected Neural Network

from torchvision import datasets
from PIL import Image
import numpy as np
import matplotlib.pyplot as plt
import random
import sys

# Set training and test datasets
train_data = datasets.MNIST("data", train=True, download=True)
test_data = datasets.MNIST("data", train=False, download=True)

# +
# Convert to numpy arrays
train_features = np.array(train_data.data) / 255.
train_labels = np.array(train_data.targets)
valid_features = np.array(test_data.data[:len(test_data.data)//2])
valid_labels = np.array(test_data.targets[:len(test_data.targets)//2])
test_features = np.array(test_data.data[len(test_data.data)//2:])
test_labels = np.array(test_data.targets[len(test_data.targets)//2:])

print(train_features.shape)
print(train_labels.shape)
print(valid_features.shape)
print(valid_labels.shape)
print(test_features.shape)
print(test_labels.shape)

# +
# show a random image from train dataset

height = 2
width = 8

f, axarr = plt.subplots(height, width, figsize=(20, 5))

for y in range(width):
    for x in range(height):
        r_idx = random.randint(0, len(train_data.data))
        axarr[x,y].set_title(int(train_data.targets[r_idx]))
        axarr[x,y].imshow(train_data.data[r_idx], cmap='gray')
        axarr[x,y].axis("off")
        
plt.show()
# -

# ## Torch implementation

# +
import torch
from torchvision import datasets 
from torch import nn
import torchvision.transforms as transforms
import torch.optim as optim
import time


batch_size = 100
num_workers = 0 
n_epochs = 1

# define tranformation pipeline
transform = transforms.Compose([ 
    transforms.ToTensor()
    #transforms.Normalize(0.5, 0.5) 
    ]) 

# Set training and test datasets
train_data = datasets.MNIST("data", train=True, download=True, transform=transform)
test_data = datasets.MNIST("data", train=False, download=True, transform=transform)

# prepare data loaders
train_loader = torch.utils.data.DataLoader(train_data, batch_size=batch_size, num_workers=num_workers) 
test_loader = torch.utils.data.DataLoader(test_data, batch_size=batch_size, num_workers=num_workers) 

torch_model = nn.Sequential(
    nn.Linear(28**2, 40, bias=False),
    #nn.ReLU(),
    nn.Linear(40, 10, bias=False)
)

optimizer = optim.SGD(torch_model.parameters(), lr=0.01)
criterion = nn.CrossEntropyLoss()

dataiter = iter(train_loader)
images, labels = dataiter.next() 
images = images.numpy()
print(images.shape)

for epoch in range(1, n_epochs+1):
    train_loss = 0.0
    valid_loss = 0.0
    start = time.time()
    # TRAINING
    torch_model.train()
    for data, target in train_loader:
        data = data.view(data.size(0), -1)
        optimizer.zero_grad()
        output = torch_model(data)
        loss = criterion(output, target)
        loss.backward()
        optimizer.step()
        train_loss += loss.item()*data.size(0)
        
    # VALIDATION
    
    train_loss = train_loss/len(train_loader.dataset)
    print("Epoch: {} / {} | Train Loss: {} | Time: {:.3f}".format(
        epoch, n_epochs, train_loss, time.time() - start))



# +
# Test accuracy
import torch.nn.functional as F 

torch_model.eval()
total = 0
correct = 0
for data, target in test_loader:
    data = data.view(data.size(0), -1)
    output = torch_model(data)
    output = F.softmax(output, dim=1)
    _, pred = torch.max(output, dim=1)
    correct += torch.sum(target == pred).item()
    total += len(target)

print("Accuracy: {:.2f} %".format((correct/total)*100))


# +
def sigmoid(x):
    return 1 / (1 + np.exp(-x))

def sigmoid_derivative(x):
    sig = sigmoid(x)
    return sig * (1 - sig)

def sigmoid_derived(x):
    return x * (1 - x)

def relu(x):
    return np.maximum(0, x, x)

def cross_entropy(x):
    x = np.exp(x)
    sums = np.sum(x, axis=1)
    return x / sums.reshape((-1, 1))

def cross_entropy_derived(x):
    return x * (1 - x)


# -

x = [[0.5,0.5,0.5,-0.5],[1,2,3,4]]
print(cross_entropy(x))
np.sum(cross_entropy(x),axis=1)

# %matplotlib inline
# %load_ext autoreload
# %autoreload 2

# +
import numpy as np
import minitorch as mt

a = mt.Input_Node(4)

# -

# ## Model 1

class NN:
    def __init__(self, input_nodes, hidden_nodes, output_nodes):
        self.input_nodes = input_nodes
        self.hidden_nodes = hidden_nodes
        self.output_nodes = output_nodes
        
        self.input_weights = np.random.normal(0, 1.0, (input_nodes, hidden_nodes))
        self.hidden_weights = np.random.normal(0, 1.0, (hidden_nodes, output_nodes))
    
    def predict(self, x):
        """
        x = a batch of vectors (n, features)
        """
        # FORWARD PROPAGATION
        n_count = x.shape[0]
        
        
        ## input to hidden
        h_score = np.matmul(x, self.input_weights) # (n, hidden_nodes)
        h_out = sigmoid(h_score) # (n, hidden_nodes)
        
        ## hidden to out
        output_score = np.matmul(h_out, self.hidden_weights) # (n, output_nodes)
        output = cross_entropy(output_score) # (n, output_nodes)
        
        return output
        
    def train(self, x, y, lr=0.01, train=True):
        """
        x = a batch of vectors (n, features)
        y = labels (one-hot-encoded) (n, outputs)
        """
        # FORWARD PROPAGATION
        n_count = x.shape[0]
        
        
        ## input to hidden
        h_score = np.matmul(x, self.input_weights) # (n, hidden_nodes)
        h_out = sigmoid(h_score) # (n, hidden_nodes)
        
        ## hidden to out
        output_score = np.matmul(h_out, self.hidden_weights) # (n, output_nodes)
        output = cross_entropy(output_score) # (n, output_nodes)

        error = -(1/n_count)*np.sum(np.log(output)*y) # (n, output_nodes)
        
        if not train:
            return error
        
        # BACKWARD PROPAGATION
        error_d = y / output
        output_d = np.zeros((n_count, self.output_nodes, self.output_nodes))
        for node_idx in range(self.output_nodes):
            for s_idx in range(self.output_nodes):
                if node_idx == s_idx:
                    output_d[:, node_idx, s_idx] = output[:,s_idx]*(1 - output[:,s_idx]) * error_d[:,s_idx]
                else:
                    output_d[:, node_idx, s_idx] = -output[:,node_idx]*output[:,s_idx] * error_d[:,s_idx]
        
        h_o_gradient = np.zeros((self.hidden_nodes, self.output_nodes)) # (n, hidden_nodes, output_nodes)
        for h_idx in range(self.hidden_nodes):
            for out_idx in range(self.output_nodes):
                h_o_gradient[h_idx, out_idx] = \
                    (np.sum(output_d[:, out_idx, :] * h_out[:, h_idx].reshape(-1,1))) / n_count
         
        # get hidden layer errors
        hidden_d = np.zeros((n_count, self.hidden_nodes, self.output_nodes, self.output_nodes))
        for h_idx in range(self.hidden_nodes):
            for out_idx in range(self.output_nodes):
                for s_idx in range(self.output_nodes):
                    hidden_d[:, h_idx, node_idx, s_idx] = \
                        output_d[:, out_idx, s_idx] * self.hidden_weights[h_idx, out_idx] * \
                        sigmoid_derived(h_out[:,h_idx])
                 
        i_h_gradient = np.zeros((self.input_nodes, self.hidden_nodes))
        for i_idx in range(self.input_nodes):
            for h_idx in range(self.hidden_nodes):
                i_h_gradient[i_idx, h_idx] = \
                    (np.sum(hidden_d[:, h_idx, -1] * x[:, i_idx].reshape(-1,1))) / n_count
        
        self.input_weights += i_h_gradient * lr
        self.hidden_weights += h_o_gradient * lr

        return error


# ## Model 2 - ReLu

class NN_relu:
    def __init__(self, input_nodes, hidden_nodes, output_nodes):
        self.input_nodes = input_nodes
        self.hidden_nodes = hidden_nodes
        self.output_nodes = output_nodes
        
        self.input_weights = np.random.normal(0, 1.0, (input_nodes, hidden_nodes))
        self.hidden_weights = np.random.normal(0, 1.0, (hidden_nodes, output_nodes))
    
    def predict(self, x):
        """
        x = a batch of vectors (n, features)
        """
        # FORWARD PROPAGATION
        n_count = x.shape[0]
        
        
        ## input to hidden
        h_score = np.matmul(x, self.input_weights) # (n, hidden_nodes)
        h_out = relu(h_score) # (n, hidden_nodes)
        
        ## hidden to out
        output_score = np.matmul(h_out, self.hidden_weights) # (n, output_nodes)
        output = cross_entropy(output_score) # (n, output_nodes)
        
        return output
        
    def train(self, x, y, lr=0.01, train=True):
        """
        x = a batch of vectors (n, features)
        y = labels (one-hot-encoded) (n, outputs)
        """
        # FORWARD PROPAGATION
        n_count = x.shape[0]
        
        
        ## input to hidden
        h_score = np.matmul(x, self.input_weights) # (n, hidden_nodes)
        h_out = relu(h_score) # (n, hidden_nodes)
        
        ## hidden to out
        output_score = np.matmul(h_out, self.hidden_weights) # (n, output_nodes)
        output = cross_entropy(output_score) # (n, output_nodes)

        error = -(1/n_count)*np.sum(np.log(output)*y) # (n, output_nodes)
        
        if not train:
            return error
        
        # BACKWARD PROPAGATION
        error_d = y / output
        output_d = np.zeros((n_count, self.output_nodes, self.output_nodes))
        for node_idx in range(self.output_nodes):
            for s_idx in range(self.output_nodes):
                if node_idx == s_idx:
                    output_d[:, node_idx, s_idx] = output[:,s_idx]*(1 - output[:,s_idx]) * error_d[:,s_idx]
                else:
                    output_d[:, node_idx, s_idx] = -output[:,node_idx]*output[:,s_idx] * error_d[:,s_idx]
        
        h_o_gradient = np.zeros((self.hidden_nodes, self.output_nodes)) # (n, hidden_nodes, output_nodes)
        for h_idx in range(self.hidden_nodes):
            for out_idx in range(self.output_nodes):
                h_o_gradient[h_idx, out_idx] = \
                    (np.sum(output_d[:, out_idx, :] * h_out[:, h_idx].reshape(-1,1))) / n_count
         
        # get hidden layer errors
        hidden_d = np.zeros((n_count, self.hidden_nodes, self.output_nodes, self.output_nodes))
        for h_idx in range(self.hidden_nodes):
            for out_idx in range(self.output_nodes):
                for s_idx in range(self.output_nodes):
                    hidden_d[:, h_idx, node_idx, s_idx] = \
                        output_d[:, out_idx, s_idx] * self.hidden_weights[h_idx, out_idx] * \
                        h_out[:,h_idx]
                 
        i_h_gradient = np.zeros((self.input_nodes, self.hidden_nodes))
        for i_idx in range(self.input_nodes):
            for h_idx in range(self.hidden_nodes):
                i_h_gradient[i_idx, h_idx] = \
                    (np.sum(hidden_d[:, h_idx, -1] * x[:, i_idx].reshape(-1,1))) / n_count
        
        self.input_weights += i_h_gradient * lr
        self.hidden_weights += h_o_gradient * lr

        return error

# ## Model 3 - Added bias

test = np.array([[1,2,3],[4,5,6]])
new = np.ones((test.shape[0], test.shape[1]+1))
new[:,:-1] = test
print(new.shape)
print(new)


class NN_bias:
    def __init__(self, input_nodes, hidden_nodes, output_nodes):
        self.input_nodes = input_nodes
        self.hidden_nodes = hidden_nodes
        self.output_nodes = output_nodes
        
        self.input_weights = np.random.normal(0, 1.0, (input_nodes+1, hidden_nodes))
        self.hidden_weights = np.random.normal(0, 1.0, (hidden_nodes+1, output_nodes))
    
    def predict(self, x):
        """
        x = a batch of vectors (n, features)
        """
        # FORWARD PROPAGATION
        n_count = x.shape[0]
        
        ## add bias to input layer
        new = np.ones((x.shape[0], x.shape[1]+1)) # (n, features+1)
        new[:,:-1] = x
        x = new
        
        ## input to hidden
        h_score = np.matmul(x, self.input_weights) # (n, features+1) * (features+1, h_n) = (n, hidden_nodes)
        h_out = relu(h_score) # (n, hidden_nodes)
        
        ## add bias to hidden layer
        new = np.ones((h_out.shape[0], h_out.shape[1]+1)) # (n, hidden_nodes+1)
        new[:,:-1] = h_out
        h_out = new
        
        ## hidden to out
        output_score = np.matmul(h_out, self.hidden_weights) # (n, output_nodes)
        output = cross_entropy(output_score) # (n, output_nodes)
        
        return output
        
    def train(self, x, y, lr=0.01, train=True):
        """
        x = a batch of vectors (n, features)
        y = labels (one-hot-encoded) (n, outputs)
        """
        # FORWARD PROPAGATION
        n_count = x.shape[0]
        
        ## add bias to input layer
        new = np.ones((x.shape[0], x.shape[1]+1)) # (n, features+1)
        new[:,:-1] = x
        x = new
        
        ## input to hidden
        h_score = np.matmul(x, self.input_weights) # (n, hidden_nodes)
        h_out = relu(h_score) # (n, hidden_nodes)
        
        ## add bias to hidden layer
        new = np.ones((h_out.shape[0], h_out.shape[1]+1)) # (n, hidden_nodes+1)
        new[:,:-1] = h_out
        h_out = new
        
        ## hidden to out
        output_score = np.matmul(h_out, self.hidden_weights) # (n, output_nodes)
        output = cross_entropy(output_score) # (n, output_nodes)

        error = -(1/n_count)*np.sum(np.log(output)*y) # (n, output_nodes)
        
        if not train:
            return error
        
        # BACKWARD PROPAGATION
        error_d = y / output
        output_d = np.zeros((n_count, self.output_nodes, self.output_nodes))
        for node_idx in range(self.output_nodes):
            for s_idx in range(self.output_nodes):
                if node_idx == s_idx:
                    output_d[:, node_idx, s_idx] = output[:,s_idx]*(1 - output[:,s_idx]) * error_d[:,s_idx]
                else:
                    output_d[:, node_idx, s_idx] = -output[:,node_idx]*output[:,s_idx] * error_d[:,s_idx]
        
        h_o_gradient = np.zeros((self.hidden_nodes+1, self.output_nodes)) # (n, hidden_nodes, output_nodes)
        for h_idx in range(self.hidden_nodes+1):
            for out_idx in range(self.output_nodes):
                h_o_gradient[h_idx, out_idx] = \
                    (np.sum(output_d[:, out_idx, :] * h_out[:, h_idx].reshape(-1,1))) / n_count
         
        # get hidden layer errors
        hidden_d = np.zeros((n_count, self.hidden_nodes, self.output_nodes, self.output_nodes))
        for h_idx in range(self.hidden_nodes):
            for out_idx in range(self.output_nodes):
                for s_idx in range(self.output_nodes):
                    hidden_d[:, h_idx, node_idx, s_idx] = \
                        output_d[:, out_idx, s_idx] * self.hidden_weights[h_idx, out_idx] * \
                        h_out[:,h_idx]
                 
        i_h_gradient = np.zeros((self.input_nodes+1, self.hidden_nodes))
        for i_idx in range(self.input_nodes+1):
            for h_idx in range(self.hidden_nodes):
                i_h_gradient[i_idx, h_idx] = \
                    (np.sum(hidden_d[:, h_idx, -1] * x[:, i_idx].reshape(-1,1))) / n_count
        
        self.input_weights += i_h_gradient * lr
        self.hidden_weights += h_o_gradient * lr

        return error


# ## Training

# +
#model = NN(28**2, 40, 10)
#model = NN_relu(28**2, 40, 10)
model = NN(28**2, 40, 10)

batchsize = 100
batchsize_valid = 100
epochs = 1

for epoch in range(1, epoch+1):
    # Training
    for idx in range(0, train_features.shape[0], batchsize):
        features = train_features[idx:idx+batchsize].reshape((-1, 784))
        labels = train_labels[idx:idx+batchsize]
        labels_enc = np.zeros((batchsize, 10))
        for l_idx in range(len(labels_enc)):
            labels_enc[l_idx, labels[l_idx]] = 1

        error = model.train(features, labels_enc)
        #print("{} / {}".format(idx, train_features.shape[0]) , error)
        if idx % (batchsize * 50) == 0:
            # Validation
            error_valid = 0
            for idx_v in range(0, len(valid_features), batchsize_valid):
                features_valid = valid_features[idx_v:idx+batchsize_valid].reshape((-1,784))
                labels_valid = valid_labels[idx_v:idx_v+batchsize_valid]
                labels_enc_valid = np.zeros((batchsize_valid, 10))
                error_valid += model.train(features, labels_enc, train=False)
            
            error_valid /= len(valid_features) / batchsize_valid
            
            print("Epoch: {} | Batch: {} / {} | Train Error: {:.8f} | Valid Error: {:.8f}".format(
                epoch, idx+(batchsize * 50), train_features.shape[0] , error, error_valid))


# +
"""
nn_relu = 19.8 %
nn_bias = 34.3 %
"""


correct = 0

for idx in range(0,len(test_features)):
    features = test_features[idx].reshape((1,784))
    result = np.argmax(model.predict(features))
    if result == test_labels[idx]:
        correct += 1
    #print("Real: {} Prediction: {}".format(test_labels[idx], result))
print("Accuracy: {:.1f} %".format((correct/len(test_features))*100))
# -

test = np.array([1,2,3,-4,-5,-6])
np.maximum(0, test, test)
test

StarcoderdataJupyterScriptsDedupFiltered