Coverage for addmo/s3_model_tuning/models/keras_models.py: 29%

1import keras 

2import tensorflow as tf 

3import onnx 

4import numpy as np 

5from abc import ABC 

6from packaging import version 

7 

8from keras.src.models.sequential import Sequential 

9from keras.src.layers import Input, Dense, Normalization, Activation 

10from keras.src.losses import MeanSquaredError 

11from keras.src.callbacks import EarlyStopping 

12from scikeras.wrappers import KerasRegressor 

13from addmo.s3_model_tuning.models.abstract_model import AbstractMLModel 

14from addmo.s3_model_tuning.models.abstract_model import ModelMetadata 

15from addmo.util.load_save_utils import create_path_or_ask_to_override 

16 

17 

18class BaseKerasModel(AbstractMLModel, ABC): 

19 """ 

20 Base class for Keras models. 

21 """ 

22 

23 def _define_metadata(self): 

24 """ 

25 Define metadata. 

26 """ 

27 self.metadata = ModelMetadata( 

28 addmo_class=type(self).__name__, 

29 addmo_commit_id=ModelMetadata.get_commit_id(), 

30 library=keras.__name__, 

31 library_model_type=type(self.regressor.model).__name__, 

32 library_version=keras.__version__, 

33 target_name=self.y_fit.name, 

34 features_ordered=list(self.x_fit.columns), 

35 preprocessing=['Scaling as layer of the ANN.']) 

36 

37 

38class SciKerasSequential(BaseKerasModel): 

39 """"" SciKeras Sequential model. """ 

40 

41 def __init__(self): 

42 self.regressor = KerasRegressor() # SciKeras Regressor 

43 self.hyperparameters = self.default_hyperparameter() 

44 

45 def fit(self, x, y): 

46 """"" 

47 Build, compile and fit the sequential model. 

48 """ 

49 self.x_fit = x 

50 self.y_fit = y 

51 self.regressor = self.to_scikit_learn(x) 

52 self.regressor.fit(x.values.astype(np.float32), y.values.astype(np.float32)) 

53 

54 def predict(self, x): 

55 """ 

56 Make prediction. 

57 """ 

58 return self.regressor.predict(x.values.astype(np.float32)) 

59 

60 def get_params(self, deep=True): 

61 """ 

62 Get the hyperparameters of the model. 

63 """ 

64 # get scikeras params 

65 params = self.regressor.get_params(deep=deep) 

66 

67 # just info params 

68 model = self.regressor.model_ 

69 total_connections = model.count_params() 

70 params['model_complexity'] = total_connections 

71 

72 # additional params not covered by scikeras (update only if not present) 

73 for key, value in self.hyperparameters.items(): 

74 if key not in params: 

75 params[key] = value 

76 

77 return params 

78 

79 def set_params(self, hyperparameters): 

80 """"" 

81 Update the hyperparameters in internal storage, which is accessed while building the 

82 regressor. Not done here, because compilation requires the input_shape to be available. 

83 """ 

84 for key, value in hyperparameters.items(): 

85 self.hyperparameters[key] = value 

86 

87 def save_regressor(self, directory, regressor_filename, file_type='keras'): 

88 """ 

89 Save trained model as a .keras or .onnx including scaler to a file. 

90 """ 

91 

92 full_filename = f"{regressor_filename}.{file_type}" 

93 path = create_path_or_ask_to_override(full_filename, directory) 

94 

95 if file_type == 'keras': 

96 self.regressor.model_.save(path, overwrite=True) 

97 

98 elif file_type == 'onnx': 

99 # catch exceptions 

100 if version.parse(keras.__version__).major != 2: # Checking version compatibility 

101 raise ImportError("ONNX is only supported with Keras version 2") 

102 try: 

103 import tf2onnx 

104 except ImportError: 

105 raise ImportError("tf2onnx is required to save the model in ONNX format") 

106 

107 # actually save onnx 

108 spec = (tf.TensorSpec((None,) + self.regressor.model_.input_shape[1:], tf.float32, name="input"),) 

109 onnx_model, _ = tf2onnx.convert.from_keras(self.regressor.model_, input_signature=spec, opset=13) 

110 onnx.save(onnx_model, path) 

111 

112 else: 

113 raise ValueError(f'The supported file types for saving the model are: .keras and .onnx') 

114 

115 # Saving metadata 

116 self._define_metadata() 

117 self._save_metadata(directory, regressor_filename) 

118 

119 print(f"Model saved to {path}") 

120 

121 return file_type 

122 

123 def load_regressor(self, regressor, input_shape): 

124 """"" 

125 Load trained model for serialisation. 

126 """ 

127 # Create dummy system_data for initialization of loaded model 

128 x = np.zeros((1, input_shape)) 

129 y = np.zeros((1,)) 

130 self.regressor = KerasRegressor(regressor) 

131 # Initialize model to avoid re-fitting 

132 self.regressor.initialize(x, y) 

133 

134 def _build_regressor_architecture(self, input_shape): 

135 """ 

136 Builds a sequential model. 

137 """ 

138 sequential_regressor = Sequential() 

139 # normalizer = Normalization(axis=-1) # Preprocessing layer 

140 sequential_regressor.add(Input(shape=input_shape)) 

141 # sequential_regressor.add(normalizer) 

142 # Adding hidden layers based on hyperparameters 

143 for units in self.hyperparameters['hidden_layer_sizes']: 

144 sequential_regressor.add(Dense(units=units, activation=self.hyperparameters['activation'])) 

145 

146 sequential_regressor.add(Dense(1, activation='linear')) 

147 return sequential_regressor 

148 

149 def _build_regressor(self, x): 

150 """"" 

151 Returns a compiled sequential model. 

152 """ 

153 input_shape = (len(x.columns),) 

154 sequential_regressor = self._build_regressor_architecture(input_shape) 

155 

156 # Normalisation of first layer (input system_data). 

157 # sequential_regressor.layers[0].adapt(x.to_numpy()) # Normalisation initialisation works only on np arrays 

158 

159 # define optimizer explicitly to avoid user warnings that optimizer could not be loaded 

160 optimizer = tf.keras.optimizers.RMSprop() 

161 optimizer.build(sequential_regressor.trainable_variables) 

162 

163 sequential_regressor.compile(loss=self.hyperparameters['loss'], optimizer=optimizer) 

164 return sequential_regressor 

165 

166 def to_scikit_learn(self, x): 

167 """"" 

168 Convert Keras Model to Scikeras Regressor for tuning. 

169 """ 

170 

171 regressor_scikit = KerasRegressor(model=self._build_regressor(x), 

172 batch_size=self.hyperparameters['batch_size'], 

173 loss=self.hyperparameters['loss'], 

174 epochs=self.hyperparameters['epochs'], 

175 verbose=0, 

176 callbacks=self.hyperparameters['callbacks']) 

177 return regressor_scikit 

178 

179 def default_hyperparameter(self): 

180 """" 

181 Return default hyperparameters. 

182 """ 

183 regressor = KerasRegressor() 

184 hyperparameters = regressor.get_params() 

185 # Define default loss if not present 

186 if hyperparameters['loss'] is None: 

187 hyperparameters['loss'] = MeanSquaredError() 

188 hyperparameters['hidden_layer_sizes'] = [16] 

189 hyperparameters['activation'] = 'softplus' 

190 hyperparameters['batch_size'] = 50 

191 hyperparameters['epochs'] = 1000 

192 hyperparameters['callbacks'] = [EarlyStopping(monitor="loss", min_delta=0.000000001, verbose=1, 

193 patience=100)] 

194 

195 

196 return hyperparameters 

197 

198 def optuna_hyperparameter_suggest(self, trial): 

199 """ 

200 Suggest hyperparameters for optimization. 

201 """ 

202 n_layers = trial.suggest_int("n_layers", 1, 2) 

203 if n_layers == 1: 

204 hidden_layer_sizes = tuple(trial.suggest_int(f"n_units_l{i}", 1, 1000) for i in range(1, n_layers + 1, 1)) 

205 if n_layers == 2: 

206 hidden_layer_sizes = tuple(trial.suggest_int(f"n_units_l{i}", 1, 100) for i in range(1, n_layers + 1, 1)) 

207 hyperparameters = { 

208 "hidden_layer_sizes": hidden_layer_sizes, 

209 "callbacks": [EarlyStopping(monitor="loss", min_delta=0.000000001, verbose=1, 

210 patience=100)], 

211 "activation": trial.suggest_categorical("activation", ["relu", "linear", "softplus", "sigmoid"]), 

212 } 

213 return hyperparameters 

214 

215 def grid_search_hyperparameter(self): 

216 """ 

217 Suggest hyperparameters for optimization. 

218 """ 

219 hyperparameter_grid = { 

220 "hidden_layer_sizes": [(64,), (128, 64), (256, 128, 64)], 

221 "loss": [MeanSquaredError()], 

222 "epochs": [50] 

223 } 

224 return hyperparameter_grid