💡 Rewriting Network Wrapper from scratch to make it more flexible

Author: JyotinderSingh

TinyFlow/Activations.py

@@ -26,8 +26,12 @@ def backward(self, dvalues):
         # Zero gradient where the input values were <= 0
         self.dvalues[self.inputs <= 0] = 0
 
+    # Calculate prediction for outputs
+    def predictions(self, outputs):
+        return outputs
+
     def __str__(self):
-        return "ReLU"
+        return "ReLU Activation"
 
 
 # Softmax activation
@@ -56,8 +60,12 @@ def backward(self, dvalues):
 
         self.dvalues = dvalues.copy()
 
+    # return predictions for outputs
+    def predictions(self, outputs):
+        return np.argmax(outputs, axis=1)
+
     def __str__(self):
-        return "Softmax"
+        return "Softmax Activation"
 
 
 # Sigmoid activation
@@ -73,6 +81,13 @@ def backward(self, dvalues):
         # d/dx(sigm(x)) = sigm(x) * [1 - sigm(x)]
         self.dvalues = dvalues * (1 - self.output) * self.output
 
+    # Calculate predictions for outputs
+    def predictions(self, outputs):
+        return (outputs > 0.5) * 1
+
+    def __str__(self):
+        return "Sigmoid Activation"
+
 
 # Linear activation
 class Activation_Linear:
@@ -90,3 +105,10 @@ def backward(self, dvalues):
         # downstream gradient = (local derivative ) * (upstream gradient)
         # self.dvalues = 1 * dvalues
         self.dvalues = dvalues.copy()
+
+    # Calculate predictions for outputs
+    def predictions(self, outputs):
+        return outputs
+
+    def __str__(self):
+        return "Linear Activation"

TinyFlow/Loss.py

@@ -36,6 +36,72 @@ def regularization_loss(self, layer):
 
         return regularization_loss
 
+    # Regularization loss calculation
+
+    def network_regularization_loss(self):
+        '''network_regularization_loss (self)\n
+            Internal method for network wrapper for auto calculation 
+            of regularization loss of all the trainable layers
+        '''
+
+        # 0 by default
+        regularization_loss = 0
+
+        # Calculate regularization loss - iterate over all trainable layers
+        for layer in self.trainable_layers:
+            # L1 regularization - weights
+            # Only calculate when factor greater than 0
+            if layer.weight_regularizer_l1 > 0:
+                regularization_loss += layer.weight_regularizer_l1 * \
+                    np.sum(np.abs(layer.weights))
+
+            # L2 regularization - weights
+            # Only calculate when factor greater than 0
+            if layer.weight_regularizer_l2 > 0:
+                regularization_loss += layer.weight_regularizer_l2 * \
+                    np.sum(layer.weights * layer.weights)
+
+            # L1 regularization - biases
+            # Only calculate when factor greater than 0
+            if layer.bias_regularizer_l1 > 0:
+                regularization_loss += layer.bias_regularizer_l1 * \
+                    np.sum(np.abs(layer.biases))
+
+            # L2 regularization - biases
+            # Only calculate when factor greater than 0
+            if layer.bias_regularizer_l2 > 0:
+                regularization_loss += layer.bias_regularizer_l2 * \
+                    np.sum(layer.biases * layer.biases)
+
+        return regularization_loss
+
+    # Set/remember trainable layers
+    def remember_trainable_layers(self, trainable_layers):
+        '''remember_trainable_layers (self, trainable_layers)\n
+            internal method for Network wrapper to keep track of trainable layers
+        '''
+
+        self.trainable_layers = trainable_layers
+
+    # Calculates the data and regularization losses
+    # given model output and ground truth values
+
+    def calculate(self, output, y):
+        '''calculate(self, output, ground_truth)\n
+            internal method for Network wrapper\n
+            Calculates the data and regularization losses
+            given model output and ground truth values
+        '''
+
+        # Calculate sample losses
+        sample_losses = self.forward(output, y)
+
+        # Calculate the mean loss
+        data_loss = np.mean(sample_losses)
+
+        # Return the data and regularization losses
+        return data_loss, self.network_regularization_loss()
+
 
 # Cross-entropy loss
 class Loss_CategoricalCrossEntropy(Loss):

TinyFlow/Metrics.py

@@ -1,6 +1,43 @@
 import numpy as np
 
 
+# Common Accuracy class
+class Accuracy:
+
+    # Calculates accuracy, given predictions and ground truth values
+    def calculate(self, predictions, y):
+
+        # Get comparison results
+        comparisons = self.compare(predictions, y)
+
+        # Calculate an accuracy
+        accuracy = np.mean(comparisons)
+
+        # Return the accuracy
+        return accuracy
+
+
+class Accuracy_Regression(Accuracy):
+
+    def __init__(self):
+        # Create a precision property
+        self.precision = None
+
+    # Calculates precision value based on passed in ground truth values
+    def init(self, y, reinit=False):
+        '''init (self, y, reinit=False)\n
+        reinit - Forces reinitilization of precision\n
+        Calculates precision value based on passed in ground truth values
+        '''
+
+        if self.precision is None or reinit:
+            self.precision = np.std(y) / 500
+
+    # Compare predictions to the ground truth values
+    def compare(self, predictions, y):
+        return np.absolute(predictions - y) < self.precision
+
+
 def model_accuracy_softmax(outputs, labels):
     '''Returns the accuracy of the model on the current batch'''
 

TinyFlow/Network.py

@@ -13,9 +13,10 @@ def __init__(self):
     def add(self, layer):
         self.layers.append(layer)
 
-    def set(self, *, loss, optimizer):
+    def set(self, *, loss, optimizer, accuracy):
         self.loss = loss
         self.optimizer = optimizer
+        self.accuracy = accuracy
 
     def compile_model(self):
 
@@ -25,6 +26,9 @@ def compile_model(self):
         # Count all the objects
         layer_count = len(self.layers)
 
+        # Initialize a list containing trainable layers
+        self.trainable_layers = []
+
         # Iterate through all the objects
         for i in range(layer_count):
 
@@ -40,11 +44,23 @@ def compile_model(self):
                 self.layers[i].next = self.layers[i + 1]
 
             # The last layer will have the next object as the loss
+            # We also save aside the reference to the last object
+            # whose output is the model's output
             else:
                 self.layers[i].prev = self.layers[i - 1]
                 self.layers[i].next = self.loss
+                self.output_layer_activation = self.layers[i]
+
+            # If the layer contains an attribute called 'weights', it's a
+            # trainable layer, add it to the list of trainable layers
+            # We dont't need to check for biases, checking for weights is enough
+            if hasattr(self.layers[i], 'weights'):
+                self.trainable_layers.append(self.layers[i])
+
+            # Update the loss object with trainable layers
+            self.loss.remember_trainable_layers(self.trainable_layers)
 
-    def forward(self, X):
+    def forward(self, X, training):
 
         # Call the forward method on the input layer
         # This will set the output property that the first layer
@@ -59,19 +75,50 @@ def forward(self, X):
         # 'layer' is now the last object from the list, return its output
         return layer.output
 
-    def train(self, X, y, *, epochs=1, print_every=1):
+    def backward(self, output, y):
+
+        # First call the backward method on loss
+        # This will set the dvalues property that the last
+        # layer will try to access shortly
+        self.loss.backward(output, y)
+
+        # Call backward method going through all the objects in
+        # reversed order, passing down the dvalues as a parameter
+        for layer in reversed(self.layers):
+            layer.backward(layer.next.dvalues)
+
+    def train(self, X, y, *, epochs=1, print_every=1, validation_data=None):
+
+        # Initialize accuracy object
+        self.accuracy.init(y)
 
         # Main training loop
         for epoch in range(1, epochs+1):
 
             # Perform the forward pass
-            output = self.forward(X)
+            output = self.forward(X, training=True)
+
+            # Calculate loss
+            data_loss, regularization_loss = self.loss.calculate(output, y)
+
+            loss = data_loss + regularization_loss
+
+            # Get predictions and calculate accuracy
+            predictions = self.output_layer_activation.predictions(output)
+            accuracy = self.accuracy.calculate(predictions, y)
 
-            # Temporary
-            print(output)
-            exit()
+            # Backward pass
+            self.backward(output, y)
 
+            # Optimize (update parameters)
+            self.optimizer.pre_update_params()
+            for layer in self.trainable_layers:
+                self.optimizer.update_params(layer)
+            self.optimizer.post_update_params()
 
+            # Print a summary
+            if not epoch % print_every:
+                print(f'epoch: {epoch}, acc: {accuracy:.3f}, loss: {loss:.3f} (data_loss: {data_loss:.3f}, reg_loss: {regularization_loss:.3f}), lr: {self.optimizer.current_learning_rate}')
 
 
 class Layer_Input:

demo_Network.py

@@ -9,7 +9,7 @@
 from TinyFlow.Activations import Activation_Linear, Activation_ReLU
 from TinyFlow.Optimizers import Optimizer_Adam
 from TinyFlow.Loss import Loss_MeanSquaredError
-
+from TinyFlow.Metrics import Accuracy_Regression
 # Create a dataset
 X, y = sine_data()
 
@@ -25,7 +25,7 @@
 model.add(Activation_Linear())
 
 # Set loss and optimizer objects
-model.set(loss=Loss_MeanSquaredError(), optimizer=Optimizer_Adam(decay=1e-8))
+model.set(loss=Loss_MeanSquaredError(), optimizer=Optimizer_Adam(decay=1e-8), accuracy=Accuracy_Regression())
 
 # Compile the model
 model.compile_model()