Coverage for biobb_ml/neural_networks/neural_network_predict.py: 63%

1#!/usr/bin/env python3 

2 

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

4import argparse 

5import h5py 

6import json 

7import csv 

8import numpy as np 

9import pandas as pd 

10from biobb_common.generic.biobb_object import BiobbObject 

11from tensorflow.python.keras.saving import hdf5_format 

12from sklearn.preprocessing import scale 

13from biobb_common.configuration import settings 

14from biobb_common.tools import file_utils as fu 

15from biobb_common.tools.file_utils import launchlogger 

16from biobb_ml.neural_networks.common import check_input_path, check_output_path, getHeader, getTargetValue, get_list_of_predictors, get_keys_of_predictors, get_num_cols 

17 

18 

19class PredictNeuralNetwork(BiobbObject): 

20 """ 

21 | biobb_ml PredictNeuralNetwork 

22 | Makes predictions from an input dataset and a given model. 

23 | Makes predictions from an input dataset (provided either as a file or as a dictionary property) and a given model trained with `TensorFlow Keras Sequential <https://www.tensorflow.org/api_docs/python/tf/keras/Sequential>`_ and `TensorFlow Keras LSTM <https://www.tensorflow.org/api_docs/python/tf/keras/layers/LSTM>`_ 

24 

25 Args: 

26 input_model_path (str): Path to the input model. File type: input. `Sample file <https://github.com/bioexcel/biobb_ml/raw/master/biobb_ml/test/data/neural_networks/input_model_predict.h5>`_. Accepted formats: h5 (edam:format_3590). 

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

28 output_results_path (str): Path to the output results file. File type: output. `Sample file <https://github.com/bioexcel/biobb_ml/raw/master/biobb_ml/test/reference/neural_networks/ref_output_predict.csv>`_. Accepted formats: csv (edam:format_3752). 

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

30 * **predictions** (*list*) - (None) List of dictionaries with all values you want to predict targets. It will be taken into account only in case **input_dataset_path** is not provided. Format: [{ 'var1': 1.0, 'var2': 2.0 }, { 'var1': 4.0, 'var2': 2.7 }] for datasets with headers and [[ 1.0, 2.0 ], [ 4.0, 2.7 ]] for datasets without headers. 

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

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

33 

34 Examples: 

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

36 

37 from biobb_ml.neural_networks.neural_network_predict import neural_network_predict 

38 prop = { 

39 'predictions': [ 

40 { 

41 'var1': 1.0, 

42 'var2': 2.0 

43 }, 

44 { 

45 'var1': 4.0, 

46 'var2': 2.7 

47 } 

48 ] 

49 } 

50 neural_network_predict(input_model_path='/path/to/myModel.h5', 

51 input_dataset_path='/path/to/myDataset.csv', 

52 output_results_path='/path/to/newPredictedResults.csv', 

53 properties=prop) 

54 

55 Info: 

56 * wrapped_software: 

57 * name: TensorFlow 

58 * version: >2.1.0 

59 * license: MIT 

60 * ontology: 

61 * name: EDAM 

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

63 

64 """ 

65 

66 def __init__(self, input_model_path, output_results_path, 

67 input_dataset_path=None, properties=None, **kwargs) -> None: 

68 properties = properties or {} 

69 

70 # Call parent class constructor 

71 super().__init__(properties) 

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

73 

74 # Input/Output files 

75 self.io_dict = { 

76 "in": {"input_model_path": input_model_path, "input_dataset_path": input_dataset_path}, 

77 "out": {"output_results_path": output_results_path} 

78 } 

79 

80 # Properties specific for BB 

81 self.predictions = properties.get('predictions', []) 

82 self.properties = properties 

83 

84 # Check the properties 

85 self.check_properties(properties) 

86 self.check_arguments() 

87 

88 def check_data_params(self, out_log, err_log): 

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

90 self.io_dict["in"]["input_model_path"] = check_input_path(self.io_dict["in"]["input_model_path"], "input_model_path", False, out_log, self.__class__.__name__) 

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

92 if self.io_dict["in"]["input_dataset_path"]: 

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

94 

95 @launchlogger 

96 def launch(self) -> int: 

97 """Execute the :class:`PredictNeuralNetwork <neural_networks.neural_network_predict.PredictNeuralNetwork>` neural_networks.neural_network_predict.PredictNeuralNetwork object.""" 

98 

99 # check input/output paths and parameters 

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

101 

102 # Setup Biobb 

103 if self.check_restart(): 

104 return 0 

105 self.stage_files() 

106 

107 fu.log('Getting model from %s' % self.io_dict["in"]["input_model_path"], self.out_log, self.global_log) 

108 with h5py.File(self.io_dict["in"]["input_model_path"], mode='r') as f: 

109 variables = f.attrs['variables'] 

110 new_model = hdf5_format.load_model_from_hdf5(f) 

111 

112 # get dictionary with variables 

113 vars_obj = json.loads(variables) 

114 

115 stringlist = [] 

116 new_model.summary(print_fn=lambda x: stringlist.append(x)) 

117 model_summary = "\n".join(stringlist) 

118 fu.log('Model summary:\n\n%s\n' % model_summary, self.out_log, self.global_log) 

119 

120 if self.io_dict["in"]["input_dataset_path"]: 

121 # load dataset from input_dataset_path file 

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

123 if 'features' not in vars_obj: 

124 # recurrent 

125 labels = None 

126 skiprows = None 

127 with open(self.io_dict["in"]["input_dataset_path"]) as csvfile: 

128 reader = csv.reader(csvfile, quoting=csv.QUOTE_NONNUMERIC) # change contents to floats 

129 for row in reader: # each row is a list 

130 self.predictions.append(row) 

131 else: 

132 # classification or regression 

133 if 'columns' in vars_obj['features']: 

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

135 skiprows = 1 

136 else: 

137 labels = None 

138 skiprows = None 

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

140 else: 

141 if vars_obj['type'] != 'recurrent': 

142 new_data_table = pd.DataFrame(data=get_list_of_predictors(self.predictions), columns=get_keys_of_predictors(self.predictions)) 

143 else: 

144 new_data_table = pd.DataFrame(data=self.predictions, columns=get_num_cols(vars_obj['window_size'])) 

145 

146 # prediction 

147 if vars_obj['type'] != 'recurrent': 

148 # classification or regression 

149 

150 # new_data_table = pd.DataFrame(data=get_list_of_predictors(self.predictions),columns=get_keys_of_predictors(self.predictions)) 

151 new_data = new_data_table 

152 if vars_obj['scale']: 

153 new_data = scale(new_data) 

154 

155 predictions = new_model.predict(new_data) 

156 predictions = np.around(predictions, decimals=2) 

157 

158 clss = '' 

159 # if predictions.shape[1] > 1: 

160 if vars_obj['type'] == 'classification': 

161 # classification 

162 pr = tuple(map(tuple, predictions)) 

163 clss = ' (' + ', '.join(str(x) for x in vars_obj['vs']) + ')' 

164 else: 

165 # regression 

166 pr = np.squeeze(np.asarray(predictions)) 

167 

168 new_data_table[getTargetValue(vars_obj['target']) + clss] = pr 

169 

170 else: 

171 # recurrent 

172 

173 # new_data_table = pd.DataFrame(data=self.predictions, columns=get_num_cols(vars_obj['window_size'])) 

174 predictions = [] 

175 

176 for r in self.predictions: 

177 row = np.asarray(r).reshape((1, vars_obj['window_size'], 1)) 

178 

179 pred = new_model.predict(row) 

180 pred = np.around(pred, decimals=2) 

181 

182 predictions.append(pred[0][0]) 

183 

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

185 new_data_table["predictions"] = predictions 

186 

187 fu.log('Predicting results\n\nPREDICTION RESULTS\n\n%s\n' % new_data_table, self.out_log, self.global_log) 

188 fu.log('Saving results to %s' % self.io_dict["out"]["output_results_path"], self.out_log, self.global_log) 

189 new_data_table.to_csv(self.io_dict["out"]["output_results_path"], index=False, header=True, float_format='%.3f') 

190 

191 # Copy files to host 

192 self.copy_to_host() 

193 

194 self.tmp_files.extend([ 

195 self.stage_io_dict.get("unique_dir") 

196 ]) 

197 self.remove_tmp_files() 

198 

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

200 

201 return 0 

202 

203 

204def neural_network_predict(input_model_path: str, output_results_path: str, input_dataset_path: str = None, properties: dict = None, **kwargs) -> int: 

205 """Execute the :class:`PredictNeuralNetwork <neural_networks.neural_network_predict.PredictNeuralNetwork>` class and 

206 execute the :meth:`launch() <neural_networks.neural_network_predict.PredictNeuralNetwork.launch>` method.""" 

207 

208 return PredictNeuralNetwork(input_model_path=input_model_path, 

209 output_results_path=output_results_path, 

210 input_dataset_path=input_dataset_path, 

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

212 

213 

214def main(): 

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

216 parser = argparse.ArgumentParser(description="Makes predictions from an input dataset and a given classification model.", formatter_class=lambda prog: argparse.RawTextHelpFormatter(prog, width=99999)) 

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

218 

219 # Specific args of each building block 

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

221 required_args.add_argument('--input_model_path', required=True, help='Path to the input model. Accepted formats: h5.') 

222 required_args.add_argument('--output_results_path', required=True, help='Path to the output results file. Accepted formats: csv.') 

223 parser.add_argument('--input_dataset_path', required=False, help='Path to the dataset to predict. Accepted formats: csv.') 

224 

225 args = parser.parse_args() 

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

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

228 

229 # Specific call of each building block 

230 neural_network_predict(input_model_path=args.input_model_path, 

231 output_results_path=args.output_results_path, 

232 input_dataset_path=args.input_dataset_path, 

233 properties=properties) 

234 

235 

236if __name__ == '__main__': 

237 main()