Coverage for biobb_ml/neural_networks/neural_network_predict.py: 63%
103 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-10-03 14:57 +0000
1#!/usr/bin/env python3
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
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>`_
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 * **sandbox_path** (*str*) - ("./") [WF property] Parent path to the sandbox directory.
35 Examples:
36 This is a use example of how to use the building block from Python::
38 from biobb_ml.neural_networks.neural_network_predict import neural_network_predict
39 prop = {
40 'predictions': [
41 {
42 'var1': 1.0,
43 'var2': 2.0
44 },
45 {
46 'var1': 4.0,
47 'var2': 2.7
48 }
49 ]
50 }
51 neural_network_predict(input_model_path='/path/to/myModel.h5',
52 input_dataset_path='/path/to/myDataset.csv',
53 output_results_path='/path/to/newPredictedResults.csv',
54 properties=prop)
56 Info:
57 * wrapped_software:
58 * name: TensorFlow
59 * version: >2.1.0
60 * license: MIT
61 * ontology:
62 * name: EDAM
63 * schema: http://edamontology.org/EDAM.owl
65 """
67 def __init__(self, input_model_path, output_results_path,
68 input_dataset_path=None, properties=None, **kwargs) -> None:
69 properties = properties or {}
71 # Call parent class constructor
72 super().__init__(properties)
73 self.locals_var_dict = locals().copy()
75 # Input/Output files
76 self.io_dict = {
77 "in": {"input_model_path": input_model_path, "input_dataset_path": input_dataset_path},
78 "out": {"output_results_path": output_results_path}
79 }
81 # Properties specific for BB
82 self.predictions = properties.get('predictions', [])
83 self.properties = properties
85 # Check the properties
86 self.check_properties(properties)
87 self.check_arguments()
89 def check_data_params(self, out_log, err_log):
90 """ Checks all the input/output paths and parameters """
91 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__)
92 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__)
93 if self.io_dict["in"]["input_dataset_path"]:
94 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__)
96 @launchlogger
97 def launch(self) -> int:
98 """Execute the :class:`PredictNeuralNetwork <neural_networks.neural_network_predict.PredictNeuralNetwork>` neural_networks.neural_network_predict.PredictNeuralNetwork object."""
100 # check input/output paths and parameters
101 self.check_data_params(self.out_log, self.err_log)
103 # Setup Biobb
104 if self.check_restart():
105 return 0
106 self.stage_files()
108 fu.log('Getting model from %s' % self.io_dict["in"]["input_model_path"], self.out_log, self.global_log)
109 with h5py.File(self.io_dict["in"]["input_model_path"], mode='r') as f:
110 variables = f.attrs['variables']
111 new_model = hdf5_format.load_model_from_hdf5(f)
113 # get dictionary with variables
114 vars_obj = json.loads(variables)
116 stringlist = []
117 new_model.summary(print_fn=lambda x: stringlist.append(x))
118 model_summary = "\n".join(stringlist)
119 fu.log('Model summary:\n\n%s\n' % model_summary, self.out_log, self.global_log)
121 if self.io_dict["in"]["input_dataset_path"]:
122 # load dataset from input_dataset_path file
123 fu.log('Getting dataset from %s' % self.io_dict["in"]["input_dataset_path"], self.out_log, self.global_log)
124 if 'features' not in vars_obj:
125 # recurrent
126 labels = None
127 skiprows = None
128 with open(self.io_dict["in"]["input_dataset_path"]) as csvfile:
129 reader = csv.reader(csvfile, quoting=csv.QUOTE_NONNUMERIC) # change contents to floats
130 for row in reader: # each row is a list
131 self.predictions.append(row)
132 else:
133 # classification or regression
134 if 'columns' in vars_obj['features']:
135 labels = getHeader(self.io_dict["in"]["input_dataset_path"])
136 skiprows = 1
137 else:
138 labels = None
139 skiprows = None
140 new_data_table = pd.read_csv(self.io_dict["in"]["input_dataset_path"], header=None, sep="\\s+|;|:|,|\t", engine="python", skiprows=skiprows, names=labels)
141 else:
142 if vars_obj['type'] != 'recurrent':
143 new_data_table = pd.DataFrame(data=get_list_of_predictors(self.predictions), columns=get_keys_of_predictors(self.predictions))
144 else:
145 new_data_table = pd.DataFrame(data=self.predictions, columns=get_num_cols(vars_obj['window_size']))
147 # prediction
148 if vars_obj['type'] != 'recurrent':
149 # classification or regression
151 # new_data_table = pd.DataFrame(data=get_list_of_predictors(self.predictions),columns=get_keys_of_predictors(self.predictions))
152 new_data = new_data_table
153 if vars_obj['scale']:
154 new_data = scale(new_data)
156 predictions = new_model.predict(new_data)
157 predictions = np.around(predictions, decimals=2)
159 clss = ''
160 # if predictions.shape[1] > 1:
161 if vars_obj['type'] == 'classification':
162 # classification
163 pr = tuple(map(tuple, predictions))
164 clss = ' (' + ', '.join(str(x) for x in vars_obj['vs']) + ')'
165 else:
166 # regression
167 pr = np.squeeze(np.asarray(predictions))
169 new_data_table[getTargetValue(vars_obj['target']) + clss] = pr
171 else:
172 # recurrent
174 # new_data_table = pd.DataFrame(data=self.predictions, columns=get_num_cols(vars_obj['window_size']))
175 predictions = []
177 for r in self.predictions:
178 row = np.asarray(r).reshape((1, vars_obj['window_size'], 1))
180 pred = new_model.predict(row)
181 pred = np.around(pred, decimals=2)
183 predictions.append(pred[0][0])
185 # pd.set_option('display.float_format', lambda x: '%.2f' % x)
186 new_data_table["predictions"] = predictions
188 fu.log('Predicting results\n\nPREDICTION RESULTS\n\n%s\n' % new_data_table, self.out_log, self.global_log)
189 fu.log('Saving results to %s' % self.io_dict["out"]["output_results_path"], self.out_log, self.global_log)
190 new_data_table.to_csv(self.io_dict["out"]["output_results_path"], index=False, header=True, float_format='%.3f')
192 # Copy files to host
193 self.copy_to_host()
195 self.tmp_files.extend([
196 self.stage_io_dict.get("unique_dir")
197 ])
198 self.remove_tmp_files()
200 self.check_arguments(output_files_created=True, raise_exception=False)
202 return 0
205def neural_network_predict(input_model_path: str, output_results_path: str, input_dataset_path: str = None, properties: dict = None, **kwargs) -> int:
206 """Execute the :class:`PredictNeuralNetwork <neural_networks.neural_network_predict.PredictNeuralNetwork>` class and
207 execute the :meth:`launch() <neural_networks.neural_network_predict.PredictNeuralNetwork.launch>` method."""
209 return PredictNeuralNetwork(input_model_path=input_model_path,
210 output_results_path=output_results_path,
211 input_dataset_path=input_dataset_path,
212 properties=properties, **kwargs).launch()
215def main():
216 """Command line execution of this building block. Please check the command line documentation."""
217 parser = argparse.ArgumentParser(description="Makes predictions from an input dataset and a given classification model.", formatter_class=lambda prog: argparse.RawTextHelpFormatter(prog, width=99999))
218 parser.add_argument('--config', required=False, help='Configuration file')
220 # Specific args of each building block
221 required_args = parser.add_argument_group('required arguments')
222 required_args.add_argument('--input_model_path', required=True, help='Path to the input model. Accepted formats: h5.')
223 required_args.add_argument('--output_results_path', required=True, help='Path to the output results file. Accepted formats: csv.')
224 parser.add_argument('--input_dataset_path', required=False, help='Path to the dataset to predict. Accepted formats: csv.')
226 args = parser.parse_args()
227 args.config = args.config or "{}"
228 properties = settings.ConfReader(config=args.config).get_prop_dic()
230 # Specific call of each building block
231 neural_network_predict(input_model_path=args.input_model_path,
232 output_results_path=args.output_results_path,
233 input_dataset_path=args.input_dataset_path,
234 properties=properties)
237if __name__ == '__main__':
238 main()