Coverage for mlair/model_modules/fully_connected_networks.py: 74%

103 statements  

« prev     ^ index     » next       coverage.py v6.4.2, created at 2023-06-30 10:40 +0000

1__author__ = "Lukas Leufen" 

2__date__ = '2021-02-18' 

3 

4from functools import reduce, partial 

5 

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 

9 

10import tensorflow.keras as keras 

11 

12 

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 """ 

18 

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} 

32 

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. 

38 

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)) 

41 

42 Customize this FCN model via the following parameters: 

43 

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 """ 

61 

62 assert len(input_shape) == 1 

63 assert len(output_shape) == 1 

64 super().__init__(input_shape[0], output_shape[0]) 

65 

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) 

78 

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)) 

83 

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.") 

89 

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.") 

102 

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.") 

119 

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 

125 

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]) 

134 

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 

145 

146 x_input = keras.layers.Input(shape=self._input_shape) 

147 x_in = keras.layers.Flatten()(x_input) 

148 

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) 

157 

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()) 

162 

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]} 

166 

167 

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 """ 

173 

174 _requirements = ["lr", "beta_1", "beta_2", "epsilon", "decay", "amsgrad"] 

175 

176 def __init__(self, input_shape: list, output_shape: list, **kwargs): 

177 """ 

178 Sets model and loss depending on the given arguments. 

179 

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) 

186 

187 def set_compile_options(self): 

188 self.compile_options = {"loss": [keras.losses.mean_squared_error], "metrics": ["mse", "mae"]} 

189 

190 def _update_model_name(self): 

191 self.model_name = "FCN" 

192 super()._update_model_name()