Deep Learning with Python Chapters 3 and 4

my_list = [3,2,5]
sorted(my_list)

# How to Install packages in Colab: !pip install <package_name>

## Make sure to turn on the GPU runtime by going to Runtime > Change Runtime Type

import tensorflow as tf
"""
Tensors need to be created with some initial value.
All ones or All Zeros Tensors:
"""
x = tf.ones(shape=(2,1)) # Equivalent to np.ones(shape=(2,1))
print(x)
x = tf.zeros(shape=(2,1)) # Equivalent to np.zeros(shape=(2,1))
print(x)
"""
Random Tensors:
"""
x = tf.random.normal(shape=(3,1),mean=0.,stddev=1.) # Tensor of random values drawn from a normal distribution with mean 0 and a standard deviation of 1. Equivalent to np.random.normal(size=(3,1),loc=0.,scale=1.)
print(x)
x = tf.random.uniform(shape=(3,1),minval=0.,maxval=1.) # Tensor of Random values drawn from a uniform distribution between 0 and 1.
print(x)
"""
A significant difference between NumPy arrays and TensorFlow tensors is that TensorFlow tensors aren't assignable: theu're constant.
"""
import numpy as np
x = np.ones(shape=(2,2))
x[0,0] = 0. # This is acceptable for NumPy arrays
x = tf.ones(shape=(2,2))
try:
  x[0,0] = 2. # This is unacceptable in TensorFlow
except Exception as e:
  print(e)

"""
Creating a TensorFlow variable
"""
v = tf.Variable(initial_value=tf.random.normal(shape=(3,1)))
print(v)
# The stae of the variable can be modifed via its `assign` method
v.assign(tf.ones((3,1)))
print(v)
# It also works for a subset of the coefficients
v[0,0].assign(3.)
print(v)
# `assign_add()` and `assign_sub()` are efficient equivalents of += and -=
v.assign_add(tf.ones((3,1)))
print(v)

# Just Like Numpy, TensorFlow offers a large collection of Tensor operations to express mathematical formula:
a = tf.ones((2,2))
a = a + tf.ones((2,2))
b = tf.square(a) # Take the square
c = tf.sqrt(a)
d = b + c # Add Two Tensors, element-wise
e = tf.matmul(a,b) # Take the dot product of two tensors
e *= d # Multiply two tensors (element-wise)

"""
Here's something NumPy can't do: retrieve the Gradient of any differentiable expression with respect to any of its inputs.
Just oppern a `GradientTape` scope, apply some computation to one or several input tensors and retrieve the gradient of the result wrt to the inputs
"""
input_var = tf.Variable(initial_value=3.)
with tf.GradientTape() as tape:
  result = tf.square(input_var)
gradient = tape.gradient(result,input_var)
"""
This is most commonly used to retrieve the gradients of the loss of a model with respect to its weights: gradient = tape.gradient(loss,weights)
Only *trainable variables* are tracked by default. WIth a constant tensor, you'd have to manually mark it as being watched
"""
input_const = tf.constant(3.)
with tf.GradientTape() as tape:
  tape.watch(input_const) # Manually watch constant Tensor
  result = tf.square(input_const) 
gradient = tape.gradient(result,input_const)

time = tf.Variable(0.)
with tf.GradientTape() as outer_tape:
  with tf.GradientTape() as inner_tape:
    position = 4.9 * time ** 2
  speed = inner_tape.gradient(position,time)
acceleration = outer_tape.gradient(speed,time) # We compute the outer tape to compute the gradient from the inner tape
print("Acceleration = {:.1f}".format(acceleration))

### End-to-End Example A Linear Classifier in Pure TensorFlow 

"""
Coming up with some nicely linearly separable synthetic dat to work with: two classes of points in a 2D plane. 
"""
num_samples_per_class = 1000
negative_samples = np.random.multivariate_normal(mean=[0,3],cov=[[1,0.5],[0.5,1]],size=num_samples_per_class) # Generate a clss of points, 1000 random 2D points. cov corresponds to an oval-like point oreiented from bottom left to top right
positive_samples = np.random.multivariate_normal(mean=[3,0],cov=[[1,0.5],[0.5,1]],size=num_samples_per_class) # Generate the other class with a different mean and the same covariance matrix
import matplotlib.pyplot as plt 
def plot_data():
    fig, ax = plt.subplots(1,1,layout="constrained")
    ax.scatter(negative_samples[:,0],negative_samples[:,1],c='y',label="Negative Samples")
    ax.scatter(positive_samples[:,0],positive_samples[:,1],c='b',label="Positive Samples")
    ax.legend()
    ax.set_title("Visualizing Data")
    return ax
plot_data()
plt.show()
inputs = np.vstack((negative_samples,positive_samples)).astype("float32")
targets = np.vstack((np.zeros((num_samples_per_class,1),dtype=np.float32), np.ones((num_samples_per_class,1),dtype=np.float32)))
"""
A linear classifier is an affine transformation `(prediction = W * input + b) trained to minimize the sequare of the difference between predictions and the targets
"""
input_dim = 2 # The input will be 2D points
output_dim = 1 # The output prediction will be a single score per sample (0 or 1)
W = tf.Variable(initial_value=tf.random.uniform(shape=(input_dim,output_dim)))
b = tf.Variable(initial_value=tf.zeros(shape=(output_dim,)))

def model(inputs):
    """
    This is our foward pass function
    """
    return tf.matmul(inputs,W)+b

def square_loss(targets,predictions):
    # per_sample_losses will be a tensor with the same shape as the targets and predictions, containing per-sample loss score
    per_sample_losses = tf.square(targets-predictions)
    # We need to average these per-sample loss scores into a single scalar loss value: this is what reduce_mean does
    return tf.reduce_mean(per_sample_losses)

learning_rate = 0.1 
def training_step(inputs,targets):
    """
    The training steps receives some training data and updates the weights W and b so as to minimize the loss on the data
    """
    with tf.GradientTape() as tape:
        """
        Forwaed pass inside a gradient tape scope
        """
        predictions = model(inputs)
        loss = square_loss(predictions,targets)
    grad_loss_wrt_W, grad_loss_wrt_b = tape.gradient(loss,[W,b])
    """
    Update the weights
    """
    W.assign_sub(grad_loss_wrt_W * learning_rate)
    b.assign_sub(grad_loss_wrt_b * learning_rate)
    return loss 

"""
Below, for simplicity, we will do batch training instead of mini-batch training. On the one hand, this means each training step will take longer to run,
on the other, it means that each gradient update will be much more effective at reducing the loss of the training data . As a result, we will need fewer steps of training, and we 
should use a larger training rate than we would typically use for mini-batch 
"""
for step in range(40):
    loss = training_step(inputs,targets)
    print(f"Loss at step {step}: {loss:.4f}")

ax = plot_data()
x = np.linspace(-1,4,100)
"""
> Recall that the prediction value for a given point [x, y] is simply prediction == [[w1], [w2]] • [x, y] + b == w1 * x + w2 * y + b. Thus, class 0 is defined as w1 * x + w2 * y + b < 0.5, and class 1 is defined as w1 * x + w2 * y + b > 0.5. 
"""
y = -W[0] / W[1] * x + (0.5-b) / W[1] # Line Equation 
ax.plot(x,y,"-r")
plt.show()
"""
This is really what a linear classifier is all about: finding the parameters of a line (orm in higher-dimensional spaces, a hyperplane) neatly separating two classes of data
"""

from tensorflow import keras 

class SimpleDense(keras.layers.Layer):
    """
    All Keras layers inherit from the base Layer class
    """
    def __init__(self, units, activation=None):
        super().__init__()
        self.units = units 
        self.activation = activation 
    
    def build(self,input_shape):
        """
        Weight creation takes place in the build() method
        """
        input_dim = input_shape[-1]
        """
        add_weight() method is a sortcut for creating weights. It is also possible to create standalone variables and assign them to layer attributes, like self.W = tf.variable(tf.random.uniform(w_shape))
        """
        self.W = self.add_weight(shape=(input_dim, self.units), initializer="random_normal")
        self.b = self.add_weight(shape=(self.units,), initializer="zeros")
    
    def call(self,inputs):
        """
        The forward pass computation is defined in the call() method
        """
        y = tf.matmul(inputs,self.W) + self.b 
        if self.activation is not None:
            y = self.activation(y)
        return y 
my_dense = SimpleDense(units=32, activation=tf.nn.relu) # Instantiate our layer, defined previously 
input_tensor = tf.ones(shape=(2,784)) # Create some test inputs 
output_tensor = my_dense(input_tensor) # Call the layer on the inputs, kust like a function
print(output_tensor.shape)

from tensorflow.keras.datasets import imdb
"""
- IMDB reviews are sequences of words that have been turned into sequences of integers, where each integer stands for a specific word in a dictionary
- num_words=10000 means to only keep the top 10,000 most frequently occuring words in the training data
- labels are either 0 (negative) or 1 (positive)
"""
(train_data, train_labels), (test_data, test_labels) = imdb.load_data(num_words=10000)
print("Train Data[0]:",train_data[0])
print("Train Label[0]:",train_labels[0])

"""
Decoding Reviews back to text
"""
word_index = imdb.get_word_index() # word_index is a dictionary mapping words to an interger index
reverse_word_index = dict([(value, key) for (key, value) in word_index.items()]) # Reverses it, mapping integer indices to words
decoded_review = " ".join([reverse_word_index.get(i - 3, "?") for i in train_data[0]]) # Decodes the review. Note that the indices are offset by 3, because 0, 1m and 2 are reserved indices for "padding", "start of sequence", and "unknown"

"""
Mult-hot encoding to prepare data
"""
import numpy as np
def vectorize_sequences(sequences, dimension=10_000):
    results = np.zeros((len(sequences), dimension)) # Create an all-zero matrix of shape (len(sequences), dimension)
    for i, sequence in enumerate(sequences):
        for j in sequence:
            results[i, j] = 1. # Set Specific inices of results[i] to 1s
    return results

x_train = vectorize_sequences(train_data)
x_test = vectorize_sequences(test_data)

print("x_train[0]:",x_train[0])

y_train = np.asarray(train_labels).astype("float32")
y_test = np.asarray(test_labels).astype("float32")

"""
There are two key architecture decisions to be made about stacks of Dense layers:
1. How many layers to use
2. How many units to use for each layer
"""

from tensorflow import keras
from tensorflow.keras import layers

model = keras.Sequential([
    # The first agument is the nmber of units in the layer
    # Having 16 units means the weight matrix W wll have shape (input_dimension, 16): the dot
    # product with W will project the input data onto a 16-dimensional representation space.
    # You can intuitively understand the dimensionality of your representation space as "how
    # much freedom you're allowing the model to have when learning internal representations."
    # Having more units allows the model to learn more-complex representations, but it makes
    # the model more computationally expensive and may lead to learning unwanted pattern (overfitting data)
    layers.Dense(16, activation="relu"), # Each dense layer computes: output = relu(dot(input, W) + b)
    layers.Dense(16, activation="relu"),
    # The sigmoid activation function outputs a provabability (score between 0 and 1) indicating
    # how likely the sample is to have a target "1". A relu (rectified linear unit) is a function meant to zero out
    # negative values, whereas a sigmoid "squashes" arbitrary values into the [0,1] interval
    layers.Dense(1, activation="sigmoid")
])

"""
Coosing Loss Function and Optimizer
----------------------------------------------------

Because you're facing a binary classification problem and the output of your modek si a probability
(you end your model with asingle-unit layer ith a sigmoid activation), it's best to use `binary_crossentropy`
loss. Crossentropy is the best choice when you're doing models that output probabilities. Crossentropy is a quantity
from the field of information theory that measures the distance between probability distributions, or, in this case,
between the ground-truth distribution and your predictions.

rmsprop optimizer is a good default choice for virtually any problem.
"""
model.compile(optimizer="rmsprop", loss="binary_crossentropy", metrics=["accuracy"])
x_val = x_train[:10_000]
partial_x_train = x_train[10_000:]
partial_y_train = y_train[10_000:]
y_val = y_train[:10_000]
partial_y_train = y_train[10_000:]
history = model.fit(
    partial_x_train,
    partial_y_train,
    epochs=20,
    batch_size=512,
    validation_data=(x_val, y_val)
)

history_dict = history.history

"""
The dictionary contains four entries: one per metric that was being monitored during training and validation. Plotting Training and Validation Loss side by side:
"""
import matplotlib.pyplot as plt
loss_values = history_dict["loss"]
val_loss_values = history_dict["val_loss"]
epochs = range(1, len(loss_values) + 1)
plt.plot(epochs, loss_values, "bo", label="Training loss")
plt.plot(epochs, val_loss_values, "b", label="Validation loss")
plt.title("Training and validation loss")
plt.xlabel("Epochs")
plt.ylabel("Loss")
plt.legend()
plt.show()

plt.clf() # Clears the figure
acc = history_dict["accuracy"]
val_acc = history_dict["val_accuracy"]
plt.plot(epochs, acc, "bo", label="Training acc")
plt.plot(epochs, val_acc, "b", label="Validation acc")
plt.title("Training and validation accuracy")
plt.xlabel("Epochs")
plt.ylabel("Accuracy")
plt.legend()
plt.show()

model = keras.Sequential([
    layers.Dense(16,activation="relu"),
    layers.Dense(16,activation="relu"),
    layers.Dense(1, activation="sigmoid")
])
model.compile(optimizer="rmsprop", loss="binary_crossentropy", metrics=["accuracy"])
model.fit(x_train, y_train, epochs=4, batch_size=512)
results = model.evaluate(x_test, y_test)
print("Results [test loss, test accuracy]=",results)
predictions = model.predict(x_test)

!pip install keras-tuner -q

import keras_tuner

def build_model(hp):
  model = keras.models.Sequential()
  model.add(layers.Dense(hp.Choice('units_layer_1',[16,32,64,128]),activation=hp.Choice('activation_1',['relu','tanh'])))
  if hp.Boolean('two_layers'):
    model.add(layers.Dense(hp.Choice('units_layer_2',[16,32,64,128]),activation=hp.Choice('activation_2',['relu','tanh'])))
  if hp.Boolean('three_layers'):
    model.add(layers.Dense(hp.Choice('units_layer_3',[16,32,64,128]),activation=hp.Choice('activation_3',['relu','tanh'])))
  model.add(layers.Dense(1,activation="sigmoid"))
  model.compile(optimizer="rmsprop",loss=hp.Choice("loss",["binary_crossentropy","mse"]),metrics=["accuracy"])
  return model
tuner = keras_tuner.Hyperband(hypermodel=build_model,objective="val_accuracy",max_epochs=10)

"""
Implementing early stopping
"""
early_stopping_cb = keras.callbacks.EarlyStopping(patience=10,restore_best_weights=True)

tuner.search(x_train,y_train,epochs=100,validation_split=0.2,callbacks=[early_stopping_cb])

print(tuner.results_summary())
best_hp = tuner.get_best_hyperparameters()[0]
model = tuner.hypermodel.build(best_hp)
model.fit(x_train,y_train)
results = model.evaluate(x_test,y_test)
print(results)

from tensorflow.keras.datasets import reuters

(train_data, train_labels), (test_data, test_labels) = reuters.load_data(num_words=10_000)

print("Train Data Length:",len(train_data))
print("Test Data Length:",len(test_data))
"""
As with the IMDB reviews, each sample is a list of indices
"""
print("Train Data Instance:",train_data[10])

"""
How to decode the newswires back to text
"""
word_index = reuters.get_word_index()
reverse_word_index = dict(
    [(value, key) for (key, value) in word_index.items()]
)
decoded_newswire = " ".join(
    [reverse_word_index.get(i - 3, "?") for i in train_data[0]]
) # Note that the indices are offset by 3 because 0, 1, and 2 are reserved indices for "passing", "start of sequence", and "unkown"
print("Train Label of Instance",train_labels[10])
"""
The label associated with an example is an integer between 0 and 45 - a topic index.
"""
x_train = vectorize_sequences(train_data) # Vectorize training data
x_test = vectorize_sequences(test_data) # Vectorized test data

def to_one_hot(labels,dimension=46):
  results = np.zeros((len(labels), dimension))
  for i, label in enumerate(labels):
    results[i, label] = 1
  return results
y_train = to_one_hot(train_labels) # Vectorized Train Labels
y_test = to_one_hot(test_labels) # Vectorized Test Labels
"""
There is a built in way to do this in Keras
"""
from tensorflow.keras.utils import to_categorical
y_train = to_categorical(train_labels)
y_test = to_categorical(test_labels)
model = keras.Sequential([
    layers.Dense(64, activation="relu"),
    layers.Dense(64, activation="relu"),
    layers.Dense(46, activation="softmax"),
])
"""
Each layer in a stack of Dense layers can only access information rpesent in
the output of the previous layer. If one layer drops some information relevant to the classification problem, this information can never be recobered by later
layers: each layer can potentially bacomje an information-bottleneck. IOn the previous example, we used 16-dimensional intermediate layers, but a
16-dimensional space may be too limited to learn form 46 different classes: such small layers may act as information bottlenecks, permenanelt dropping
relevant information.
"""
model = keras.Sequential([
    layers.Dense(64, activation="relu"),
    layers.Dense(64, activation="relu"),
    # 46 units because 46 classes
    # softmax activation -> the model will output a probability distribution
    # over the 46 different output classes - for every input sample, the model will produce a 46-diemnsional output vector, where `output[i]` is the probability that the sample belongs to class i, the 46 scores will sum to 1
    layers.Dense(46, activation="softmax")
])
"""
The ebest loss function to use in this case is categorical_crossentropy. It measures the distance between two probability distributions here, between the probability distribution output by the model and the true distribution of the labels. By minimizing the distance between these two distributions, you train the model to output something as close as possible to the true labels.
"""
model.compile(optimizer="rmsprop",loss="categorical_crossentropy",metrics=["accuracy"])

"""
Validating samples
"""
x_val = x_train[:1_000]
partial_x_train = x_train[1_000:]
y_val = y_train[:1_000]
partial_y_train = y_train[1_000:]
history = model.fit(partial_x_train, partial_y_train, epochs=20, batch_size=512, validation_data=(x_val, y_val))
# Plotting the Training and Validation Loss
loss = history.history["loss"]
val_loss = history.history["val_loss"]
epochs = range(1, len(loss) + 1)
plt.plot(epochs, loss, "bo", label="Training loss")
plt.plot(epochs, val_loss, "b", label="Validation loss")
plt.title("Training and validation loss")
plt.xlabel("Epochs")
plt.ylabel("Loss")
plt.legend()
plt.show()
# Plotting the Validation Accuracy
plt.clf()
acc = history.history["accuracy"]
val_acc = history.history["val_accuracy"]
plt.plot(epochs, acc, "bo", label="Training accuracy")
plt.plot(epochs, val_acc, "b", label="Validation accuracy")
plt.title("Training and validation accuracy")
plt.xlabel("Epochs")
plt.ylabel("Accuracy")
plt.legend()
plt.show()

"""
The model begins to overfit after 9 epochs -> train model from scratch with 9 epochs
"""
model = keras.Sequential([
 layers.Dense(64, activation="relu"),
 layers.Dense(64, activation="relu"),
 layers.Dense(46, activation="softmax")
])
model.compile(optimizer="rmsprop",
 loss="categorical_crossentropy",
 metrics=["accuracy"])
model.fit(x_train,
 y_train,
 epochs=9,
 batch_size=512)
results = model.evaluate(x_test, y_test)
results = model.evaluate(x_test, y_test)
print("Final Results")
print(results)

from tensorflow.keras.datasets import boston_housing
(train_data, train_targets), (test_data, test_targets) = boston_housing.load_data()

print(train_data.shape)
print(test_data.shape)

print(train_targets)

mean = train_data.mean(axis=0)
train_data -= mean
std = train_data.std(axis=0)
train_data /= std
test_data -= mean
test_data /= std

def build_model():
  """
  Since we need to instantiate the same model multiple times, we use a function to construct it.
  """
  model = keras.Sequential([
      layers.Dense(64,activation="relu"),
      layers.Dense(64, activation="relu"),
      layers.Dense(1)
  ])
  model.compile(optimizer="rmsprop",loss="mse",metrics=["mae"])
  return model

k = 4
num_val_samples = len(train_data) // k
num_epochs = 100
all_scores = []
for i in range(k):
  print(f"Processing Fold ${i}")
  # Prepares the validation data: data from partition #k
  val_data = train_data[i*num_val_samples: (i+1)*num_val_samples]
  val_targets = train_targets[i * num_val_samples: (i+1) * num_val_samples]
  # Prepare the training data: data from all other partitions
  partial_train_data = np.concatenate(
      [train_data[:i * num_val_samples],
       train_data[(i+1)*num_val_samples:]],
      axis=0
  )
  partial_train_targets = np.concatenate(
      [train_targets[:i * num_val_samples],
       train_targets[(i+1)*num_val_samples:]],
      axis=0
  )
  model = build_model()
  # Trains in silent mode (verbose = 0)
  model.fit(partial_train_data, partial_train_targets, epochs=num_epochs, batch_size=16, verbose=0)
  val_mse, val_mae = model.evaluate(val_data, val_targets, verbose=0)
  all_scores.append(val_mae)

all_scores

np.mean(all_scores)

 num_epochs = 500
 all_mae_histories = []
 for i in range(k):
  print(f"Processing Fold #{i}")
  val_data = train_data[i*num_val_samples : (i+1) * num_val_samples] # Prepares the validation data: data from partition #k
  val_targets = train_targets[i * num_val_samples : (i+1) * num_val_samples]
  # Prepares the training data: data from all other partitions
  partial_train_data = np.concatenate(
      [train_data[:i * num_val_samples],
       train_data[(i+1) * num_val_samples:]],
      axis=0
  )
  partial_train_targets = np.concatenate(
  [train_targets[:i * num_val_samples],
   train_targets[(i+1) * num_val_samples:]],
   axis=0
  )
  # Builds the Keras model (already compiled)
  model = build_model()
  # Trains the model (in silent mode, verbose=0)
  history = model.fit(partial_train_data,partial_train_targets,validation_data=(val_data, val_targets), epochs=num_epochs, batch_size=16, verbose=0)
  mae_history = history.history["val_mae"]
  all_mae_histories.append(mae_history)

average_mae_history = [np.mean([x[i] for x in all_mae_histories]) for i in range(num_epochs)]
plt.plot(range(1, len(average_mae_history) + 1),average_mae_history)
plt.xlabel("Epochs")
plt.ylabel("Validation MAE")
plt.show()
plt.clf()
truncated_mae_history = average_mae_history[10:]
plt.plot(range(1, len(truncated_mae_history) + 1), truncated_mae_history)
plt.xlabel("Epochs")
plt.ylabel("Validation MAE")
plt.show()