Coverage for mlair/model_modules/fully_connected_networks.py: 74%
103 statements
« prev ^ index » next coverage.py v6.4.2, created at 2023-06-01 13:03 +0000
1__author__ = "Lukas Leufen"
2__date__ = '2021-02-18'
4from functools import reduce, partial
6from mlair.model_modules import AbstractModelClass
7from mlair.helpers import select_from_dict
8from mlair.model_modules.loss import var_loss, custom_loss, l_p_loss
10import tensorflow.keras as keras
13class FCN(AbstractModelClass):
14 """
15 A customisable fully connected network (64, 32, 16, window_lead_time), where the last layer is the output layer depending
16 on the window_lead_time parameter.
17 """
19 _activation = {"relu": keras.layers.ReLU, "tanh": partial(keras.layers.Activation, "tanh"),
20 "sigmoid": partial(keras.layers.Activation, "sigmoid"),
21 "linear": partial(keras.layers.Activation, "linear"),
22 "selu": partial(keras.layers.Activation, "selu"),
23 "prelu": partial(keras.layers.PReLU, alpha_initializer=keras.initializers.constant(value=0.25)),
24 "leakyrelu": partial(keras.layers.LeakyReLU)}
25 _initializer = {"tanh": "glorot_uniform", "sigmoid": "glorot_uniform", "linear": "glorot_uniform",
26 "relu": keras.initializers.he_normal(), "selu": keras.initializers.lecun_normal(),
27 "prelu": keras.initializers.he_normal()}
28 _optimizer = {"adam": keras.optimizers.Adam, "sgd": keras.optimizers.SGD}
29 _regularizer = {"l1": keras.regularizers.l1, "l2": keras.regularizers.l2, "l1_l2": keras.regularizers.l1_l2}
30 _requirements = ["lr", "beta_1", "beta_2", "epsilon", "decay", "amsgrad", "momentum", "nesterov", "l1", "l2"]
31 _dropout = {"selu": keras.layers.AlphaDropout}
33 def __init__(self, input_shape: list, output_shape: list, activation="relu", activation_output="linear",
34 optimizer="adam", n_layer=1, n_hidden=10, regularizer=None, dropout=None, layer_configuration=None,
35 batch_normalization=False, **kwargs):
36 """
37 Sets model and loss depending on the given arguments.
39 :param input_shape: list of input shapes (expect len=1 with shape=(window_hist, station, variables))
40 :param output_shape: list of output shapes (expect len=1 with shape=(window_forecast))
42 Customize this FCN model via the following parameters:
44 :param activation: set your desired activation function. Chose from relu, tanh, sigmoid, linear, selu, prelu,
45 leakyrelu. (Default relu)
46 :param activation_output: same as activation parameter but exclusively applied on output layer only. (Default
47 linear)
48 :param optimizer: set optimizer method. Can be either adam or sgd. (Default adam)
49 :param n_layer: define number of hidden layers in the network. Given number of hidden neurons are used in each
50 layer. (Default 1)
51 :param n_hidden: define number of hidden units per layer. This number is used in each hidden layer. (Default 10)
52 :param layer_configuration: alternative formulation of the network's architecture. This will overwrite the
53 settings from n_layer and n_hidden. Provide a list where each element represent the number of units in the
54 hidden layer. The number of hidden layers is equal to the total length of this list.
55 :param dropout: use dropout with given rate. If no value is provided, dropout layers are not added to the
56 network at all. (Default None)
57 :param batch_normalization: use batch normalization layer in the network if enabled. These layers are inserted
58 between the linear part of a layer (the nn part) and the non-linear part (activation function). No BN layer
59 is added if set to false. (Default false)
60 """
62 assert len(input_shape) == 1
63 assert len(output_shape) == 1
64 super().__init__(input_shape[0], output_shape[0])
66 # settings
67 self.activation = self._set_activation(activation)
68 self.activation_name = activation
69 self.activation_output = self._set_activation(activation_output)
70 self.activation_output_name = activation_output
71 self.optimizer = self._set_optimizer(optimizer, **kwargs)
72 self.bn = batch_normalization
73 self.layer_configuration = (n_layer, n_hidden) if layer_configuration is None else layer_configuration
74 self._update_model_name()
75 self.kernel_initializer = self._initializer.get(activation, "glorot_uniform")
76 self.kernel_regularizer = self._set_regularizer(regularizer, **kwargs)
77 self.dropout, self.dropout_rate = self._set_dropout(activation, dropout)
79 # apply to model
80 self.set_model()
81 self.set_compile_options()
82 self.set_custom_objects(loss=self.compile_options["loss"][0], var_loss=var_loss, l_p_loss=l_p_loss(.5))
84 def _set_activation(self, activation):
85 try:
86 return self._activation.get(activation.lower())
87 except KeyError:
88 raise AttributeError(f"Given activation {activation} is not supported in this model class.")
90 def _set_optimizer(self, optimizer, **kwargs):
91 try:
92 opt_name = optimizer.lower()
93 opt = self._optimizer.get(opt_name)
94 opt_kwargs = {}
95 if opt_name == "adam": 95 ↛ 97line 95 didn't jump to line 97, because the condition on line 95 was never false
96 opt_kwargs = select_from_dict(kwargs, ["lr", "beta_1", "beta_2", "epsilon", "decay", "amsgrad"])
97 elif opt_name == "sgd":
98 opt_kwargs = select_from_dict(kwargs, ["lr", "momentum", "decay", "nesterov"])
99 return opt(**opt_kwargs)
100 except KeyError:
101 raise AttributeError(f"Given optimizer {optimizer} is not supported in this model class.")
103 def _set_regularizer(self, regularizer, **kwargs):
104 if regularizer is None or (isinstance(regularizer, str) and regularizer.lower() == "none"): 104 ↛ 106line 104 didn't jump to line 106, because the condition on line 104 was never false
105 return None
106 try:
107 reg_name = regularizer.lower()
108 reg = self._regularizer.get(reg_name)
109 reg_kwargs = {}
110 if reg_name in ["l1", "l2"]:
111 reg_kwargs = select_from_dict(kwargs, reg_name, remove_none=True)
112 if reg_name in reg_kwargs:
113 reg_kwargs["l"] = reg_kwargs.pop(reg_name)
114 elif reg_name == "l1_l2":
115 reg_kwargs = select_from_dict(kwargs, ["l1", "l2"], remove_none=True)
116 return reg(**reg_kwargs)
117 except KeyError:
118 raise AttributeError(f"Given regularizer {regularizer} is not supported in this model class.")
120 def _set_dropout(self, activation, dropout_rate):
121 if dropout_rate is None: 121 ↛ 123line 121 didn't jump to line 123, because the condition on line 121 was never false
122 return None, None
123 assert 0 <= dropout_rate < 1
124 return self._dropout.get(activation, keras.layers.Dropout), dropout_rate
126 def _update_model_name(self):
127 n_input = str(reduce(lambda x, y: x * y, self._input_shape))
128 n_output = str(self._output_shape)
129 if isinstance(self.layer_configuration, tuple) and len(self.layer_configuration) == 2:
130 n_layer, n_hidden = self.layer_configuration
131 self.model_name += "_".join(["", n_input, *[f"{n_hidden}" for _ in range(n_layer)], n_output])
132 else:
133 self.model_name += "_".join(["", n_input, *[f"{n}" for n in self.layer_configuration], n_output])
135 def set_model(self):
136 """
137 Build the model.
138 """
139 if isinstance(self.layer_configuration, tuple) is True:
140 n_layer, n_hidden = self.layer_configuration
141 conf = [n_hidden for _ in range(n_layer)]
142 else:
143 assert isinstance(self.layer_configuration, list) is True
144 conf = self.layer_configuration
146 x_input = keras.layers.Input(shape=self._input_shape)
147 x_in = keras.layers.Flatten()(x_input)
149 for layer, n_hidden in enumerate(conf):
150 x_in = keras.layers.Dense(n_hidden, kernel_initializer=self.kernel_initializer,
151 kernel_regularizer=self.kernel_regularizer)(x_in)
152 if self.bn is True: 152 ↛ 153line 152 didn't jump to line 153, because the condition on line 152 was never true
153 x_in = keras.layers.BatchNormalization()(x_in)
154 x_in = self.activation(name=f"{self.activation_name}_{layer + 1}")(x_in)
155 if self.dropout is not None: 155 ↛ 156line 155 didn't jump to line 156, because the condition on line 155 was never true
156 x_in = self.dropout(self.dropout_rate)(x_in)
158 x_in = keras.layers.Dense(self._output_shape)(x_in)
159 out = self.activation_output(name=f"{self.activation_output_name}_output")(x_in)
160 self.model = keras.Model(inputs=x_input, outputs=[out])
161 print(self.model.summary())
163 def set_compile_options(self):
164 self.compile_options = {"loss": [custom_loss([keras.losses.mean_squared_error, var_loss])],
165 "metrics": ["mse", "mae", var_loss]}
168class FCN_64_32_16(FCN):
169 """
170 A customised model 4 Dense layers (64, 32, 16, window_lead_time), where the last layer is the output layer depending
171 on the window_lead_time parameter.
172 """
174 _requirements = ["lr", "beta_1", "beta_2", "epsilon", "decay", "amsgrad"]
176 def __init__(self, input_shape: list, output_shape: list, **kwargs):
177 """
178 Sets model and loss depending on the given arguments.
180 :param input_shape: list of input shapes (expect len=1 with shape=(window_hist, station, variables))
181 :param output_shape: list of output shapes (expect len=1 with shape=(window_forecast))
182 """
183 lr = kwargs.pop("lr", 1e-2)
184 super().__init__(input_shape, output_shape, activation="prelu", activation_output="linear",
185 layer_configuration=[64, 32, 16], optimizer="adam", lr=lr, **kwargs)
187 def set_compile_options(self):
188 self.compile_options = {"loss": [keras.losses.mean_squared_error], "metrics": ["mse", "mae"]}
190 def _update_model_name(self):
191 self.model_name = "FCN"
192 super()._update_model_name()