Coverage for biobb_ml/regression/polynomial_regression.py: 87%

1#!/usr/bin/env python3 

2 

3"""Module containing the PolynomialRegression class and the command line interface.""" 

4import argparse 

5import numpy as np 

6import pandas as pd 

7import joblib 

8from biobb_common.generic.biobb_object import BiobbObject 

9from sklearn.preprocessing import StandardScaler, PolynomialFeatures 

10from sklearn.model_selection import train_test_split 

11from sklearn.feature_selection import f_regression 

12from sklearn.metrics import mean_squared_error, r2_score 

13from sklearn import linear_model 

14from biobb_common.configuration import settings 

15from biobb_common.tools import file_utils as fu 

16from biobb_common.tools.file_utils import launchlogger 

17from biobb_ml.regression.common import check_input_path, check_output_path, getHeader, getIndependentVars, getIndependentVarsList, getTarget, getTargetValue, getWeight, adjusted_r2, plotResults 

18 

19 

20class PolynomialRegression(BiobbObject): 

21 """ 

22 | biobb_ml PolynomialRegression 

23 | Wrapper of the scikit-learn LinearRegression method with PolynomialFeatures. 

24 | Trains and tests a given dataset and saves the model and scaler. Visit the `LinearRegression documentation page <https://scikit-learn.org/stable/modules/generated/sklearn.linear_model.LinearRegression.html>`_ in the sklearn official website for further information. 

25 

26 Args: 

27 input_dataset_path (str): Path to the input dataset. File type: input. `Sample file <https://github.com/bioexcel/biobb_ml/raw/master/biobb_ml/test/data/regression/dataset_polynomial_regression.csv>`_. Accepted formats: csv (edam:format_3752). 

28 output_model_path (str): Path to the output model file. File type: output. `Sample file <https://github.com/bioexcel/biobb_ml/raw/master/biobb_ml/test/reference/regression/ref_output_model_polynomial_regression.pkl>`_. Accepted formats: pkl (edam:format_3653). 

29 output_test_table_path (str) (Optional): Path to the test table file. File type: output. `Sample file <https://github.com/bioexcel/biobb_ml/raw/master/biobb_ml/test/reference/regression/ref_output_test_polynomial_regression.csv>`_. Accepted formats: csv (edam:format_3752). 

30 output_plot_path (str) (Optional): Residual plot checks the error between actual values and predicted values. File type: output. `Sample file <https://github.com/bioexcel/biobb_ml/raw/master/biobb_ml/test/reference/regression/ref_output_plot_polynomial_regression.png>`_. Accepted formats: png (edam:format_3603). 

31 properties (dic - Python dictionary object containing the tool parameters, not input/output files): 

32 * **independent_vars** (*dict*) - ({}) Independent variables you want to train from your dataset. You can specify either a list of columns names from your input dataset, a list of columns indexes or a range of columns indexes. Formats: { "columns": ["column1", "column2"] } or { "indexes": [0, 2, 3, 10, 11, 17] } or { "range": [[0, 20], [50, 102]] }. In case of mulitple formats, the first one will be picked. 

33 * **target** (*dict*) - ({}) Dependent variable you want to predict from your dataset. You can specify either a column name or a column index. Formats: { "column": "column3" } or { "index": 21 }. In case of mulitple formats, the first one will be picked. 

34 * **weight** (*dict*) - ({}) Weight variable from your dataset. You can specify either a column name or a column index. Formats: { "column": "column3" } or { "index": 21 }. In case of mulitple formats, the first one will be picked. 

35 * **random_state_train_test** (*int*) - (5) [1~1000|1] Controls the shuffling applied to the data before applying the split. 

36 * **degree** (*int*) - (2) [1~100|1] Polynomial degree. 

37 * **test_size** (*float*) - (0.2) Represents the proportion of the dataset to include in the test split. It should be between 0.0 and 1.0. 

38 * **scale** (*bool*) - (False) Whether or not to scale the input dataset. 

39 * **remove_tmp** (*bool*) - (True) [WF property] Remove temporal files. 

40 * **restart** (*bool*) - (False) [WF property] Do not execute if output files exist. 

41 

42 Examples: 

43 This is a use example of how to use the building block from Python:: 

44 

45 from biobb_ml.regression.polynomial_regression import polynomial_regression 

46 prop = { 

47 'independent_vars': { 

48 'columns': [ 'column1', 'column2', 'column3' ] 

49 }, 

50 'target': { 

51 'column': 'target' 

52 }, 

53 'degree': 2, 

54 'test_size': 0.2 

55 } 

56 polynomial_regression(input_dataset_path='/path/to/myDataset.csv', 

57 output_model_path='/path/to/newModel.pkl', 

58 output_test_table_path='/path/to/newTable.csv', 

59 output_plot_path='/path/to/newPlot.png', 

60 properties=prop) 

61 

62 Info: 

63 * wrapped_software: 

64 * name: scikit-learn LinearRegression 

65 * version: >=0.24.2 

66 * license: BSD 3-Clause 

67 * ontology: 

68 * name: EDAM 

69 * schema: http://edamontology.org/EDAM.owl 

70 

71 """ 

72 

73 def __init__(self, input_dataset_path, output_model_path, 

74 output_test_table_path=None, output_plot_path=None, properties=None, **kwargs) -> None: 

75 properties = properties or {} 

76 

77 # Call parent class constructor 

78 super().__init__(properties) 

79 self.locals_var_dict = locals().copy() 

80 

81 # Input/Output files 

82 self.io_dict = { 

83 "in": {"input_dataset_path": input_dataset_path}, 

84 "out": {"output_model_path": output_model_path, "output_test_table_path": output_test_table_path, "output_plot_path": output_plot_path} 

85 } 

86 

87 # Properties specific for BB 

88 self.independent_vars = properties.get('independent_vars', {}) 

89 self.target = properties.get('target', {}) 

90 self.weight = properties.get('weight', {}) 

91 self.random_state_train_test = properties.get('random_state_train_test', 5) 

92 self.degree = properties.get('degree', 2) 

93 self.test_size = properties.get('test_size', 0.2) 

94 self.scale = properties.get('scale', False) 

95 self.properties = properties 

96 

97 # Check the properties 

98 self.check_properties(properties) 

99 self.check_arguments() 

100 

101 def check_data_params(self, out_log, err_log): 

102 """ Checks all the input/output paths and parameters """ 

103 self.io_dict["in"]["input_dataset_path"] = check_input_path(self.io_dict["in"]["input_dataset_path"], "input_dataset_path", out_log, self.__class__.__name__) 

104 self.io_dict["out"]["output_model_path"] = check_output_path(self.io_dict["out"]["output_model_path"], "output_model_path", False, out_log, self.__class__.__name__) 

105 if self.io_dict["out"]["output_test_table_path"]: 

106 self.io_dict["out"]["output_test_table_path"] = check_output_path(self.io_dict["out"]["output_test_table_path"], "output_test_table_path", True, out_log, self.__class__.__name__) 

107 if self.io_dict["out"]["output_plot_path"]: 

108 self.io_dict["out"]["output_plot_path"] = check_output_path(self.io_dict["out"]["output_plot_path"], "output_plot_path", True, out_log, self.__class__.__name__) 

109 

110 @launchlogger 

111 def launch(self) -> int: 

112 """Execute the :class:`PolynomialRegression <regression.polynomial_regression.PolynomialRegression>` regression.polynomial_regression.PolynomialRegression object.""" 

113 

114 # check input/output paths and parameters 

115 self.check_data_params(self.out_log, self.err_log) 

116 

117 # Setup Biobb 

118 if self.check_restart(): 

119 return 0 

120 self.stage_files() 

121 

122 # load dataset 

123 fu.log('Getting dataset from %s' % self.io_dict["in"]["input_dataset_path"], self.out_log, self.global_log) 

124 if 'columns' in self.independent_vars: 

125 labels = getHeader(self.io_dict["in"]["input_dataset_path"]) 

126 skiprows = 1 

127 else: 

128 labels = None 

129 skiprows = None 

130 data = pd.read_csv(self.io_dict["in"]["input_dataset_path"], header=None, sep="\\s+|;|:|,|\t", engine="python", skiprows=skiprows, names=labels) 

131 

132 # declare inputs, targets and weights 

133 # the inputs are all the independent variables 

134 X = getIndependentVars(self.independent_vars, data, self.out_log, self.__class__.__name__) 

135 fu.log('Independent variables: [%s]' % (getIndependentVarsList(self.independent_vars)), self.out_log, self.global_log) 

136 # target 

137 y = getTarget(self.target, data, self.out_log, self.__class__.__name__) 

138 fu.log('Target: %s' % (getTargetValue(self.target)), self.out_log, self.global_log) 

139 # weights 

140 if self.weight: 

141 w = getWeight(self.weight, data, self.out_log, self.__class__.__name__) 

142 fu.log('Weight column provided', self.out_log, self.global_log) 

143 

144 # train / test split 

145 fu.log('Creating train and test sets', self.out_log, self.global_log) 

146 arrays_sets = (X, y) 

147 # if user provide weights 

148 if self.weight: 

149 arrays_sets = arrays_sets + (w,) 

150 X_train, X_test, y_train, y_test, w_train, w_test = train_test_split(*arrays_sets, test_size=self.test_size, random_state=self.random_state_train_test) 

151 else: 

152 X_train, X_test, y_train, y_test = train_test_split(*arrays_sets, test_size=self.test_size, random_state=self.random_state_train_test) 

153 

154 # scale dataset 

155 if self.scale: 

156 fu.log('Scaling dataset', self.out_log, self.global_log) 

157 scaler = StandardScaler() 

158 X_train = scaler.fit_transform(X_train) 

159 

160 # regression 

161 fu.log('Training dataset applying polynomial regression', self.out_log, self.global_log) 

162 poly_features = PolynomialFeatures(degree=self.degree) 

163 X_train_poly = poly_features.fit_transform(X_train) 

164 model = linear_model.LinearRegression() 

165 model.fit(X_train_poly, y_train) 

166 

167 # scores and coefficients train 

168 y_hat_train = model.predict(X_train_poly) 

169 rmse = (np.sqrt(mean_squared_error(y_train, y_hat_train))) 

170 rss = ((y_train - y_hat_train) ** 2).sum() 

171 score = r2_score(y_hat_train, y_train) 

172 # bias = model.intercept_ 

173 coef = model.coef_ 

174 coef = ['%.3f' % item for item in coef] 

175 adj_r2 = adjusted_r2(X_train_poly, y_train, score) 

176 p_values = f_regression(X_train_poly, y_train)[1] 

177 p_values = ['%.3f' % item for item in p_values] 

178 

179 # r-squared 

180 r2_table = pd.DataFrame() 

181 r2_table["feature"] = ['R2', 'Adj. R2', 'RMSE', 'RSS'] 

182 r2_table['coefficient'] = [score, adj_r2, rmse, rss] 

183 

184 fu.log('Calculating scores and coefficients for training dataset\n\nR2, ADJUSTED R2 & RMSE\n\n%s\n' % r2_table, self.out_log, self.global_log) 

185 

186 if self.scale: 

187 X_test = scaler.transform(X_test) 

188 X_test_poly = poly_features.fit_transform(X_test) 

189 y_hat_test = model.predict(X_test_poly) 

190 test_table = pd.DataFrame(y_hat_test, columns=['prediction']) 

191 # reset y_test (problem with old indexes column) 

192 y_test = y_test.reset_index(drop=True) 

193 # add real values to predicted ones in test_table table 

194 test_table['target'] = y_test 

195 # calculate difference between target and prediction (absolute and %) 

196 test_table['residual'] = test_table['target'] - test_table['prediction'] 

197 test_table['difference %'] = np.absolute(test_table['residual']/test_table['target']*100) 

198 pd.set_option('display.float_format', lambda x: '%.2f' % x) 

199 # sort by difference in % 

200 test_table = test_table.sort_values(by=['difference %']) 

201 test_table = test_table.reset_index(drop=True) 

202 fu.log('Testing\n\nTEST DATA\n\n%s\n' % test_table, self.out_log, self.global_log) 

203 

204 # scores and coefficients test 

205 r2_test = r2_score(y_hat_test, y_test) 

206 adj_r2_test = adjusted_r2(X_test_poly, y_test, r2_test) 

207 rmse_test = np.sqrt(mean_squared_error(y_test, y_hat_test)) 

208 rss_test = ((y_test - y_hat_test) ** 2).sum() 

209 

210 # r-squared 

211 pd.set_option('display.float_format', lambda x: '%.6f' % x) 

212 r2_table_test = pd.DataFrame() 

213 r2_table_test["feature"] = ['R2', 'Adj. R2', 'RMSE', 'RSS'] 

214 r2_table_test['coefficient'] = [r2_test, adj_r2_test, rmse_test, rss_test] 

215 

216 fu.log('Calculating scores and coefficients for testing dataset\n\nR2, ADJUSTED R2 & RMSE\n\n%s\n' % r2_table_test, self.out_log, self.global_log) 

217 

218 if (self.io_dict["out"]["output_test_table_path"]): 

219 fu.log('Saving testing data to %s' % self.io_dict["out"]["output_test_table_path"], self.out_log, self.global_log) 

220 test_table.to_csv(self.io_dict["out"]["output_test_table_path"], index=False, header=True) 

221 

222 # create test plot 

223 if (self.io_dict["out"]["output_plot_path"]): 

224 fu.log('Saving residual plot to %s' % self.io_dict["out"]["output_plot_path"], self.out_log, self.global_log) 

225 y_hat_test = y_hat_test.flatten() 

226 y_hat_train = y_hat_train.flatten() 

227 plot = plotResults(y_train, y_hat_train, y_test, y_hat_test) 

228 plot.savefig(self.io_dict["out"]["output_plot_path"], dpi=150) 

229 

230 # save model, scaler and parameters 

231 variables = { 

232 'target': self.target, 

233 'independent_vars': self.independent_vars, 

234 'scale': self.scale 

235 } 

236 fu.log('Saving model to %s' % self.io_dict["out"]["output_model_path"], self.out_log, self.global_log) 

237 with open(self.io_dict["out"]["output_model_path"], "wb") as f: 

238 joblib.dump(model, f) 

239 if self.scale: 

240 joblib.dump(scaler, f) 

241 joblib.dump(poly_features, f) 

242 joblib.dump(variables, f) 

243 

244 # Copy files to host 

245 self.copy_to_host() 

246 

247 self.tmp_files.extend([ 

248 self.stage_io_dict.get("unique_dir") 

249 ]) 

250 self.remove_tmp_files() 

251 

252 self.check_arguments(output_files_created=True, raise_exception=False) 

253 

254 return 0 

255 

256 

257def polynomial_regression(input_dataset_path: str, output_model_path: str, output_test_table_path: str = None, output_plot_path: str = None, properties: dict = None, **kwargs) -> int: 

258 """Execute the :class:`PolynomialRegression <regression.polynomial_regression.PolynomialRegression>` class and 

259 execute the :meth:`launch() <regression.polynomial_regression.PolynomialRegression.launch>` method.""" 

260 

261 return PolynomialRegression(input_dataset_path=input_dataset_path, 

262 output_model_path=output_model_path, 

263 output_test_table_path=output_test_table_path, 

264 output_plot_path=output_plot_path, 

265 properties=properties, **kwargs).launch() 

266 

267 

268def main(): 

269 """Command line execution of this building block. Please check the command line documentation.""" 

270 parser = argparse.ArgumentParser(description="Wrapper of the scikit-learn LinearRegression method with PolynomialFeatures.", formatter_class=lambda prog: argparse.RawTextHelpFormatter(prog, width=99999)) 

271 parser.add_argument('--config', required=False, help='Configuration file') 

272 

273 # Specific args of each building block 

274 required_args = parser.add_argument_group('required arguments') 

275 required_args.add_argument('--input_dataset_path', required=True, help='Path to the input dataset. Accepted formats: csv.') 

276 required_args.add_argument('--output_model_path', required=True, help='Path to the output model file. Accepted formats: pkl.') 

277 parser.add_argument('--output_test_table_path', required=False, help='Path to the test table file. Accepted formats: csv.') 

278 parser.add_argument('--output_plot_path', required=False, help='Residual plot checks the error between actual values and predicted values. Accepted formats: png.') 

279 

280 args = parser.parse_args() 

281 args.config = args.config or "{}" 

282 properties = settings.ConfReader(config=args.config).get_prop_dic() 

283 

284 # Specific call of each building block 

285 polynomial_regression(input_dataset_path=args.input_dataset_path, 

286 output_model_path=args.output_model_path, 

287 output_test_table_path=args.output_test_table_path, 

288 output_plot_path=args.output_plot_path, 

289 properties=properties) 

290 

291 

292if __name__ == '__main__': 

293 main()