Skip to content
Snippets Groups Projects
Select Git revision
  • 9ec5d18db36105ac380fc08b507fed2ed91a7f5f
  • 2023 default protected
2 results

mpi4py-3.0.3-gpsmpi-2020-Python-3.8.5.eb

Blame
  • 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]