wehub-resource-sync
182 lines
5.0 KiB
Python
182 lines
5.0 KiB
Python
import logging
|
|
|
|
import numpy as np
|
|
from autograd import elementwise_grad
|
|
|
|
from mla.base import BaseEstimator
|
|
from mla.metrics.metrics import get_metric
|
|
from mla.neuralnet.layers import PhaseMixin
|
|
from mla.neuralnet.loss import get_loss
|
|
from mla.utils import batch_iterator
|
|
|
|
np.random.seed(9999)
|
|
|
|
"""
|
|
Architecture inspired from:
|
|
|
|
https://github.com/fchollet/keras
|
|
https://github.com/andersbll/deeppy
|
|
"""
|
|
|
|
|
|
class NeuralNet(BaseEstimator):
|
|
fit_required = False
|
|
|
|
def __init__(
|
|
self,
|
|
layers,
|
|
optimizer,
|
|
loss,
|
|
max_epochs=10,
|
|
batch_size=64,
|
|
metric="mse",
|
|
shuffle=False,
|
|
verbose=True,
|
|
):
|
|
self.verbose = verbose
|
|
self.shuffle = shuffle
|
|
self.optimizer = optimizer
|
|
|
|
self.loss = get_loss(loss)
|
|
|
|
# TODO: fix
|
|
if loss == "categorical_crossentropy":
|
|
self.loss_grad = lambda actual, predicted: -(actual - predicted)
|
|
else:
|
|
self.loss_grad = elementwise_grad(self.loss, 1)
|
|
self.metric = get_metric(metric)
|
|
self.layers = layers
|
|
self.batch_size = batch_size
|
|
self.max_epochs = max_epochs
|
|
self._n_layers = 0
|
|
self.log_metric = True if loss != metric else False
|
|
self.metric_name = metric
|
|
self.bprop_entry = self._find_bprop_entry()
|
|
self.training = False
|
|
self._initialized = False
|
|
|
|
def _setup_layers(self, x_shape):
|
|
"""Initialize model's layers."""
|
|
x_shape = list(x_shape)
|
|
x_shape[0] = self.batch_size
|
|
|
|
for layer in self.layers:
|
|
layer.setup(x_shape)
|
|
x_shape = layer.shape(x_shape)
|
|
|
|
self._n_layers = len(self.layers)
|
|
# Setup optimizer
|
|
self.optimizer.setup(self)
|
|
self._initialized = True
|
|
logging.info("Total parameters: %s" % self.n_params)
|
|
|
|
def _find_bprop_entry(self):
|
|
"""Find entry layer for back propagation."""
|
|
|
|
if len(self.layers) > 0 and not hasattr(self.layers[-1], "parameters"):
|
|
return -1
|
|
return len(self.layers)
|
|
|
|
def fit(self, X, y=None):
|
|
if not self._initialized:
|
|
self._setup_layers(X.shape)
|
|
|
|
if y.ndim == 1:
|
|
# Reshape vector to matrix
|
|
y = y[:, np.newaxis]
|
|
self._setup_input(X, y)
|
|
|
|
self.is_training = True
|
|
# Pass neural network instance to an optimizer
|
|
self.optimizer.optimize(self)
|
|
self.is_training = False
|
|
|
|
def update(self, X, y):
|
|
# Forward pass
|
|
y_pred = self.fprop(X)
|
|
|
|
# Backward pass
|
|
grad = self.loss_grad(y, y_pred)
|
|
for layer in reversed(self.layers[: self.bprop_entry]):
|
|
grad = layer.backward_pass(grad)
|
|
return self.loss(y, y_pred)
|
|
|
|
def fprop(self, X):
|
|
"""Forward propagation."""
|
|
for layer in self.layers:
|
|
X = layer.forward_pass(X)
|
|
return X
|
|
|
|
def _predict(self, X=None):
|
|
if not self._initialized:
|
|
self._setup_layers(X.shape)
|
|
|
|
y = []
|
|
X_batch = batch_iterator(X, self.batch_size)
|
|
for Xb in X_batch:
|
|
y.append(self.fprop(Xb))
|
|
return np.concatenate(y)
|
|
|
|
@property
|
|
def parametric_layers(self):
|
|
for layer in self.layers:
|
|
if hasattr(layer, "parameters"):
|
|
yield layer
|
|
|
|
@property
|
|
def parameters(self):
|
|
"""Returns a list of all parameters."""
|
|
params = []
|
|
for layer in self.parametric_layers:
|
|
params.append(layer.parameters)
|
|
return params
|
|
|
|
def error(self, X=None, y=None):
|
|
"""Calculate an error for given examples."""
|
|
training_phase = self.is_training
|
|
if training_phase:
|
|
# Temporally disable training.
|
|
# Some layers work differently while training (e.g. Dropout).
|
|
self.is_training = False
|
|
if X is None and y is None:
|
|
y_pred = self._predict(self.X)
|
|
score = self.metric(self.y, y_pred)
|
|
else:
|
|
y_pred = self._predict(X)
|
|
score = self.metric(y, y_pred)
|
|
if training_phase:
|
|
self.is_training = True
|
|
return score
|
|
|
|
@property
|
|
def is_training(self):
|
|
return self.training
|
|
|
|
@is_training.setter
|
|
def is_training(self, train):
|
|
self.training = train
|
|
for layer in self.layers:
|
|
if isinstance(layer, PhaseMixin):
|
|
layer.is_training = train
|
|
|
|
def shuffle_dataset(self):
|
|
"""Shuffle rows in the dataset."""
|
|
n_samples = self.X.shape[0]
|
|
indices = np.arange(n_samples)
|
|
np.random.shuffle(indices)
|
|
self.X = self.X.take(indices, axis=0)
|
|
self.y = self.y.take(indices, axis=0)
|
|
|
|
@property
|
|
def n_layers(self):
|
|
"""Returns the number of layers."""
|
|
return self._n_layers
|
|
|
|
@property
|
|
def n_params(self):
|
|
"""Return the number of trainable parameters."""
|
|
return sum([layer.parameters.n_params for layer in self.parametric_layers])
|
|
|
|
def reset(self):
|
|
self._initialized = False
|