Select Git revision
test_data_preparation.py
model_class.py 23.75 KiB
import src.model_modules.keras_extensions
__author__ = "Lukas Leufen"
__date__ = '2019-12-12'
from abc import ABC
from typing import Any, Callable, Dict
import keras
from src.model_modules.inception_model import InceptionModelBase
from src.model_modules.flatten import flatten_tail
from src.model_modules.advanced_paddings import PadUtils, Padding2D
class AbstractModelClass(ABC):
"""
The AbstractModelClass provides a unified skeleton for any model provided to the machine learning workflow. The
model can always be accessed by calling ModelClass.model or directly by an model method without parsing the model
attribute name (e.g. ModelClass.model.compile -> ModelClass.compile). Beside the model, this class provides the
corresponding loss function.
"""
def __init__(self) -> None:
"""
Predefine internal attributes for model and loss.
"""
self.__model = None
self.__loss = None
self.model_name = self.__class__.__name__
self.__custom_objects = {}
def __getattr__(self, name: str) -> Any:
"""
Is called if __getattribute__ is not able to find requested attribute. Normally, the model class is saved into
a variable like `model = ModelClass()`. To bypass a call like `model.model` to access the _model attribute,
this method tries to search for the named attribute in the self.model namespace and returns this attribute if
available. Therefore, following expression is true: `ModelClass().compile == ModelClass().model.compile` as long
the called attribute/method is not part if the ModelClass itself.
:param name: name of the attribute or method to call
:return: attribute or method from self.model namespace
"""
return self.model.__getattribute__(name)
@property
def model(self) -> keras.Model:
"""
The model property containing a keras.Model instance.
:return: the keras model
"""
return self.__model
@model.setter
def model(self, value):
self.__model = value
@property
def loss(self) -> Callable:
"""
The loss property containing a callable loss function. The loss function can be any keras loss or a customised
function. If the loss is a customised function, it must contain the internal loss(y_true, y_pred) function:
def customised_loss(args):
def loss(y_true, y_pred):
return actual_function(y_true, y_pred, args)
return loss
:return: the loss function
"""
return self.__loss
@loss.setter
def loss(self, value) -> None:
self.__loss = value
@property
def custom_objects(self) -> Dict:
"""
The custom objects property collects all non-keras utilities that are used in the model class. To load such a
customised and already compiled model (e.g. from local disk), this information is required.
:return: the custom objects in a dictionary
"""
return self.__custom_objects
@custom_objects.setter
def custom_objects(self, value) -> None:
self.__custom_objects = value
def get_settings(self) -> Dict:
"""
Get all class attributes that are not protected in the AbstractModelClass as dictionary.
:return: all class attributes
"""
return dict((k, v) for (k, v) in self.__dict__.items() if not k.startswith("_AbstractModelClass__"))
def set_model(self):
pass
def set_loss(self):
pass
def set_custom_objects(self, **kwargs) -> None:
"""
Set custom objects that are not part of keras framework. These custom objects are needed if an already compiled
model is loaded from disk. There is a special treatment for the Padding2D class, which is a base class for
different padding types. For a correct behaviour, all supported subclasses are added as custom objects in
addition to the given ones.
:param kwargs: all custom objects, that should be saved
"""
if "Padding2D" in kwargs.keys():
kwargs.update(kwargs["Padding2D"].allowed_paddings)
self.custom_objects = kwargs
class MyLittleModel(AbstractModelClass):
"""
A customised model with a 1x1 Conv, and 4 Dense layers (64, 32, 16, window_lead_time), where the last layer is the
output layer depending on the window_lead_time parameter. Dropout is used between the Convolution and the first
Dense layer.
"""
def __init__(self, window_history_size, window_lead_time, channels):
"""
Sets model and loss depending on the given arguments.
:param activation: activation function
:param window_history_size: number of historical time steps included in the input data
:param channels: number of variables used in input data
:param regularizer: <not used here>
:param dropout_rate: dropout rate used in the model [0, 1)
:param window_lead_time: number of time steps to forecast in the output layer
"""
super().__init__()
# settings
self.window_history_size = window_history_size
self.window_lead_time = window_lead_time
self.channels = channels
self.dropout_rate = 0.1
self.regularizer = keras.regularizers.l2(0.1)
self.initial_lr = 1e-2
self.optimizer = keras.optimizers.SGD(lr=self.initial_lr, momentum=0.9)
self.lr_decay = src.model_modules.keras_extensions.LearningRateDecay(base_lr=self.initial_lr, drop=.94, epochs_drop=10)
self.epochs = 20
self.batch_size = int(256)
self.activation = keras.layers.PReLU
# apply to model
self.set_model()
self.set_loss()
self.set_custom_objects(loss=self.loss)
def set_model(self):
"""
Build the model.
:param activation: activation function
:param window_history_size: number of historical time steps included in the input data
:param channels: number of variables used in input data
:param dropout_rate: dropout rate used in the model [0, 1)
:param window_lead_time: number of time steps to forecast in the output layer
:return: built keras model
"""
# add 1 to window_size to include current time step t0
x_input = keras.layers.Input(shape=(self.window_history_size + 1, 1, self.channels))
x_in = keras.layers.Conv2D(32, (1, 1), padding='same', name='{}_Conv_1x1'.format("major"))(x_input)
x_in = self.activation(name='{}_conv_act'.format("major"))(x_in)
x_in = keras.layers.Flatten(name='{}'.format("major"))(x_in)
x_in = keras.layers.Dropout(self.dropout_rate, name='{}_Dropout_1'.format("major"))(x_in)
x_in = keras.layers.Dense(64, name='{}_Dense_64'.format("major"))(x_in)
x_in = self.activation()(x_in)
x_in = keras.layers.Dense(32, name='{}_Dense_32'.format("major"))(x_in)
x_in = self.activation()(x_in)
x_in = keras.layers.Dense(16, name='{}_Dense_16'.format("major"))(x_in)
x_in = self.activation()(x_in)
x_in = keras.layers.Dense(self.window_lead_time, name='{}_Dense'.format("major"))(x_in)
out_main = self.activation()(x_in)
self.model = keras.Model(inputs=x_input, outputs=[out_main])
def set_loss(self):
"""
Set the loss
:return: loss function
"""
self.loss = keras.losses.mean_squared_error
class MyBranchedModel(AbstractModelClass):
"""
A customised model
with a 1x1 Conv, and 4 Dense layers (64, 32, 16, window_lead_time), where the last layer is the
output layer depending on the window_lead_time parameter. Dropout is used between the Convolution and the first
Dense layer.
"""
def __init__(self, window_history_size, window_lead_time, channels):
"""
Sets model and loss depending on the given arguments.
:param activation: activation function
:param window_history_size: number of historical time steps included in the input data
:param channels: number of variables used in input data
:param regularizer: <not used here>
:param dropout_rate: dropout rate used in the model [0, 1)
:param window_lead_time: number of time steps to forecast in the output layer
"""
super().__init__()
# settings
self.window_history_size = window_history_size
self.window_lead_time = window_lead_time
self.channels = channels
self.dropout_rate = 0.1
self.regularizer = keras.regularizers.l2(0.1)
self.initial_lr = 1e-2
self.optimizer = keras.optimizers.SGD(lr=self.initial_lr, momentum=0.9)
self.lr_decay = src.model_modules.keras_extensions.LearningRateDecay(base_lr=self.initial_lr, drop=.94, epochs_drop=10)
self.epochs = 20
self.batch_size = int(256)
self.activation = keras.layers.PReLU
# apply to model
self.set_model()
self.set_loss()
self.set_custom_objects(loss=self.loss)
def set_model(self):
"""
Build the model.
:param activation: activation function
:param window_history_size: number of historical time steps included in the input data
:param channels: number of variables used in input data
:param dropout_rate: dropout rate used in the model [0, 1)
:param window_lead_time: number of time steps to forecast in the output layer
:return: built keras model
"""
# add 1 to window_size to include current time step t0
x_input = keras.layers.Input(shape=(self.window_history_size + 1, 1, self.channels))
x_in = keras.layers.Conv2D(32, (1, 1), padding='same', name='{}_Conv_1x1'.format("major"))(x_input)
x_in = self.activation(name='{}_conv_act'.format("major"))(x_in)
x_in = keras.layers.Flatten(name='{}'.format("major"))(x_in)
x_in = keras.layers.Dropout(self.dropout_rate, name='{}_Dropout_1'.format("major"))(x_in)
x_in = keras.layers.Dense(64, name='{}_Dense_64'.format("major"))(x_in)
x_in = self.activation()(x_in)
out_minor_1 = keras.layers.Dense(self.window_lead_time, name='{}_Dense'.format("minor_1"))(x_in)
out_minor_1 = self.activation(name="minor_1")(out_minor_1)
x_in = keras.layers.Dense(32, name='{}_Dense_32'.format("major"))(x_in)
x_in = self.activation()(x_in)
out_minor_2 = keras.layers.Dense(self.window_lead_time, name='{}_Dense'.format("minor_2"))(x_in)
out_minor_2 = self.activation(name="minor_2")(out_minor_2)
x_in = keras.layers.Dense(16, name='{}_Dense_16'.format("major"))(x_in)
x_in = self.activation()(x_in)
x_in = keras.layers.Dense(self.window_lead_time, name='{}_Dense'.format("major"))(x_in)
out_main = self.activation(name="main")(x_in)
self.model = keras.Model(inputs=x_input, outputs=[out_minor_1, out_minor_2, out_main])
def set_loss(self):
"""
Set the loss
:return: loss function
"""
self.loss = [keras.losses.mean_absolute_error] + [keras.losses.mean_squared_error] + \
[keras.losses.mean_squared_error]
class MyTowerModel(AbstractModelClass):
def __init__(self, window_history_size, window_lead_time, channels):
"""
Sets model and loss depending on the given arguments.
:param activation: activation function
:param window_history_size: number of historical time steps included in the input data
:param channels: number of variables used in input data
:param regularizer: <not used here>
:param dropout_rate: dropout rate used in the model [0, 1)
:param window_lead_time: number of time steps to forecast in the output layer
"""
super().__init__()
# settings
self.window_history_size = window_history_size
self.window_lead_time = window_lead_time
self.channels = channels
self.dropout_rate = 1e-2
self.regularizer = keras.regularizers.l2(0.1)
self.initial_lr = 1e-2
self.optimizer = keras.optimizers.adam(lr=self.initial_lr)
self.lr_decay = src.model_modules.keras_extensions.LearningRateDecay(base_lr=self.initial_lr, drop=.94, epochs_drop=10)
self.epochs = 20
self.batch_size = int(256*4)
self.activation = keras.layers.PReLU
# apply to model
self.set_model()
self.set_loss()
self.set_custom_objects(loss=self.loss)
def set_model(self):
"""
Build the model.
:param activation: activation function
:param window_history_size: number of historical time steps included in the input data
:param channels: number of variables used in input data
:param dropout_rate: dropout rate used in the model [0, 1)
:param window_lead_time: number of time steps to forecast in the output layer
:return: built keras model
"""
activation = self.activation
conv_settings_dict1 = {
'tower_1': {'reduction_filter': 8, 'tower_filter': 8 * 2, 'tower_kernel': (3, 1), 'activation': activation},
'tower_2': {'reduction_filter': 8, 'tower_filter': 8 * 2, 'tower_kernel': (5, 1), 'activation': activation},
'tower_3': {'reduction_filter': 8, 'tower_filter': 8 * 2, 'tower_kernel': (1, 1), 'activation': activation},
}
pool_settings_dict1 = {'pool_kernel': (3, 1), 'tower_filter': 8 * 2, 'activation': activation}
conv_settings_dict2 = {
'tower_1': {'reduction_filter': 8 * 2, 'tower_filter': 16 * 2 * 2, 'tower_kernel': (3, 1),
'activation': activation},
'tower_2': {'reduction_filter': 8 * 2, 'tower_filter': 16 * 2 * 2, 'tower_kernel': (5, 1),
'activation': activation},
'tower_3': {'reduction_filter': 8 * 2, 'tower_filter': 16 * 2 * 2, 'tower_kernel': (1, 1),
'activation': activation},
}
pool_settings_dict2 = {'pool_kernel': (3, 1), 'tower_filter': 16, 'activation': activation}
conv_settings_dict3 = {'tower_1': {'reduction_filter': 16 * 4, 'tower_filter': 32 * 2, 'tower_kernel': (3, 1),
'activation': activation},
'tower_2': {'reduction_filter': 16 * 4, 'tower_filter': 32 * 2, 'tower_kernel': (5, 1),
'activation': activation},
'tower_3': {'reduction_filter': 16 * 4, 'tower_filter': 32 * 2, 'tower_kernel': (1, 1),
'activation': activation},
}
pool_settings_dict3 = {'pool_kernel': (3, 1), 'tower_filter': 32, 'activation': activation}
##########################################
inception_model = InceptionModelBase()
X_input = keras.layers.Input(
shape=(self.window_history_size + 1, 1, self.channels)) # add 1 to window_size to include current time step t0
X_in = inception_model.inception_block(X_input, conv_settings_dict1, pool_settings_dict1,
regularizer=self.regularizer,
batch_normalisation=True)
X_in = keras.layers.Dropout(self.dropout_rate)(X_in)
X_in = inception_model.inception_block(X_in, conv_settings_dict2, pool_settings_dict2, regularizer=self.regularizer,
batch_normalisation=True)
X_in = keras.layers.Dropout(self.dropout_rate)(X_in)
X_in = inception_model.inception_block(X_in, conv_settings_dict3, pool_settings_dict3, regularizer=self.regularizer,
batch_normalisation=True)
#############################################
# out_main = flatten_tail(X_in, 'Main', activation=activation, bound_weight=True, dropout_rate=self.dropout_rate,
# reduction_filter=64, inner_neurons=64, output_neurons=self.window_lead_time)
out_main = flatten_tail(X_in, inner_neurons=64, activation=activation, output_neurons=self.window_lead_time,
output_activation='linear', reduction_filter=64,
name='Main', bound_weight=True, dropout_rate=self.dropout_rate,
kernel_regularizer=self.regularizer
)
self.model = keras.Model(inputs=X_input, outputs=[out_main])
def set_loss(self):
"""
Set the loss
:return: loss function
"""
self.loss = [keras.losses.mean_squared_error]
class MyPaperModel(AbstractModelClass):
def __init__(self, window_history_size, window_lead_time, channels):
"""
Sets model and loss depending on the given arguments.
:param activation: activation function
:param window_history_size: number of historical time steps included in the input data
:param channels: number of variables used in input data
:param regularizer: <not used here>
:param dropout_rate: dropout rate used in the model [0, 1)
:param window_lead_time: number of time steps to forecast in the output layer
"""
super().__init__()
# settings
self.window_history_size = window_history_size
self.window_lead_time = window_lead_time
self.channels = channels
self.dropout_rate = .3
self.regularizer = keras.regularizers.l2(0.001)
self.initial_lr = 1e-3
# self.optimizer = keras.optimizers.adam(lr=self.initial_lr, amsgrad=True)
self.optimizer = keras.optimizers.SGD(lr=self.initial_lr, momentum=0.9)
self.lr_decay = src.model_modules.keras_extensions.LearningRateDecay(base_lr=self.initial_lr, drop=.94, epochs_drop=10)
self.epochs = 150
self.batch_size = int(256 * 2)
self.activation = keras.layers.ELU
self.padding = "SymPad2D"
# apply to model
self.set_model()
self.set_loss()
self.set_custom_objects(loss=self.loss, Padding2D=Padding2D)
def set_model(self):
"""
Build the model.
:param activation: activation function
:param window_history_size: number of historical time steps included in the input data
:param channels: number of variables used in input data
:param dropout_rate: dropout rate used in the model [0, 1)
:param window_lead_time: number of time steps to forecast in the output layer
:return: built keras model
"""
activation = self.activation
first_kernel = (3,1)
first_filters = 16
conv_settings_dict1 = {
'tower_1': {'reduction_filter': 8, 'tower_filter': 16 * 2, 'tower_kernel': (3, 1),
'activation': activation},
'tower_2': {'reduction_filter': 8, 'tower_filter': 16 * 2, 'tower_kernel': (5, 1),
'activation': activation},
'tower_3': {'reduction_filter': 8, 'tower_filter': 16 * 2, 'tower_kernel': (1, 1),
'activation': activation},
# 'tower_4':{'reduction_filter':8, 'tower_filter':8*2, 'tower_kernel':(7,1), 'activation':activation},
}
pool_settings_dict1 = {'pool_kernel': (3, 1), 'tower_filter': 16, 'activation': activation}
conv_settings_dict2 = {
'tower_1': {'reduction_filter': 64, 'tower_filter': 32 * 2, 'tower_kernel': (3, 1),
'activation': activation},
'tower_2': {'reduction_filter': 64, 'tower_filter': 32 * 2, 'tower_kernel': (5, 1),
'activation': activation},
'tower_3': {'reduction_filter': 64, 'tower_filter': 32 * 2, 'tower_kernel': (1, 1),
'activation': activation},
# 'tower_4':{'reduction_filter':8*2, 'tower_filter':16*2, 'tower_kernel':(7,1), 'activation':activation},
}
pool_settings_dict2 = {'pool_kernel': (3, 1), 'tower_filter': 32, 'activation': activation}
conv_settings_dict3 = {
'tower_1': {'reduction_filter': 64 * 2, 'tower_filter': 32 * 4, 'tower_kernel': (3, 1),
'activation': activation},
'tower_2': {'reduction_filter': 64 * 2, 'tower_filter': 32 * 4, 'tower_kernel': (5, 1),
'activation': activation},
'tower_3': {'reduction_filter': 64 * 2, 'tower_filter': 32 * 4, 'tower_kernel': (1, 1),
'activation': activation},
# 'tower_4':{'reduction_filter':16*4, 'tower_filter':32, 'tower_kernel':(7,1), 'activation':activation},
}
pool_settings_dict3 = {'pool_kernel': (3, 1), 'tower_filter': 32, 'activation': activation}
##########################################
inception_model = InceptionModelBase()
X_input = keras.layers.Input(
shape=(self.window_history_size + 1, 1, self.channels)) # add 1 to window_size to include current time step t0
pad_size = PadUtils.get_padding_for_same(first_kernel)
# X_in = adv_pad.SymmetricPadding2D(padding=pad_size)(X_input)
# X_in = inception_model.padding_layer("SymPad2D")(padding=pad_size, name="SymPad")(X_input) # adv_pad.SymmetricPadding2D(padding=pad_size)(X_input)
X_in = Padding2D("SymPad2D")(padding=pad_size, name="SymPad")(X_input)
X_in = keras.layers.Conv2D(filters=first_filters,
kernel_size=first_kernel,
kernel_regularizer=self.regularizer,
name="First_conv_{}x{}".format(first_kernel[0], first_kernel[1]))(X_in)
X_in = self.activation(name='FirstAct')(X_in)
X_in = inception_model.inception_block(X_in, conv_settings_dict1, pool_settings_dict1,
regularizer=self.regularizer,
batch_normalisation=True,
padding=self.padding)
# out_minor1 = flatten_tail(X_in, 'minor_1', False, self.dropout_rate, self.window_lead_time,
# self.activation, 32, 64)
out_minor1 = flatten_tail(X_in, inner_neurons=64, activation=activation, output_neurons=self.window_lead_time,
output_activation='linear', reduction_filter=32,
name='minor_1', bound_weight=False, dropout_rate=self.dropout_rate,
kernel_regularizer=self.regularizer
)
X_in = keras.layers.Dropout(self.dropout_rate)(X_in)
X_in = inception_model.inception_block(X_in, conv_settings_dict2, pool_settings_dict2, regularizer=self.regularizer,
batch_normalisation=True, padding=self.padding)
# X_in = keras.layers.Dropout(self.dropout_rate)(X_in)
#
# X_in = inception_model.inception_block(X_in, conv_settings_dict3, pool_settings_dict3, regularizer=self.regularizer,
# batch_normalisation=True)
#############################################
out_main = flatten_tail(X_in, inner_neurons=64 * 2, activation=activation, output_neurons=self.window_lead_time,
output_activation='linear', reduction_filter=64 * 2,
name='Main', bound_weight=False, dropout_rate=self.dropout_rate,
kernel_regularizer=self.regularizer
)
self.model = keras.Model(inputs=X_input, outputs=[out_minor1, out_main])
def set_loss(self):
"""
Set the loss
:return: loss function
"""
self.loss = [keras.losses.mean_squared_error, keras.losses.mean_squared_error]