Mercurial > repos > bgruening > keras_batch_models

comparison keras_train_and_eval.py @ 5:79efb5472c2e draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload for repository https://github.com/bgruening/galaxytools/tree/master/tools/sklearn commit 5b2ac730ec6d3b762faa9034eddd19ad1b347476"
author bgruening
date Mon, 16 Dec 2019 09:47:30 +0000
parents
children 8edfc7381344
comparison
equal deleted inserted replaced
4:ca0719aa1de3 5:79efb5472c2e
1 import argparse
2 import joblib
3 import json
4 import numpy as np
5 import os
6 import pandas as pd
7 import pickle
8 import warnings
9 from itertools import chain
10 from scipy.io import mmread
11 from sklearn.pipeline import Pipeline
12 from sklearn.metrics.scorer import _check_multimetric_scoring
13 from sklearn import model_selection
14 from sklearn.model_selection._validation import _score
15 from sklearn.model_selection import _search, _validation
16 from sklearn.utils import indexable, safe_indexing
17
18 from galaxy_ml.externals.selene_sdk.utils import compute_score
19 from galaxy_ml.model_validations import train_test_split
20 from galaxy_ml.keras_galaxy_models import _predict_generator
21 from galaxy_ml.utils import (SafeEval, get_scoring, load_model,
22 read_columns, try_get_attr, get_module,
23 clean_params, get_main_estimator)
24
25
26 _fit_and_score = try_get_attr('galaxy_ml.model_validations', '_fit_and_score')
27 setattr(_search, '_fit_and_score', _fit_and_score)
28 setattr(_validation, '_fit_and_score', _fit_and_score)
29
30 N_JOBS = int(os.environ.get('GALAXY_SLOTS', 1))
31 CACHE_DIR = os.path.join(os.getcwd(), 'cached')
32 del os
33 NON_SEARCHABLE = ('n_jobs', 'pre_dispatch', 'memory', '_path',
34 'nthread', 'callbacks')
35 ALLOWED_CALLBACKS = ('EarlyStopping', 'TerminateOnNaN', 'ReduceLROnPlateau',
36 'CSVLogger', 'None')
37
38
39 def _eval_swap_params(params_builder):
40 swap_params = {}
41
42 for p in params_builder['param_set']:
43 swap_value = p['sp_value'].strip()
44 if swap_value == '':
45 continue
46
47 param_name = p['sp_name']
48 if param_name.lower().endswith(NON_SEARCHABLE):
49 warnings.warn("Warning: `%s` is not eligible for search and was "
50 "omitted!" % param_name)
51 continue
52
53 if not swap_value.startswith(':'):
54 safe_eval = SafeEval(load_scipy=True, load_numpy=True)
55 ev = safe_eval(swap_value)
56 else:
57 # Have `:` before search list, asks for estimator evaluatio
58 safe_eval_es = SafeEval(load_estimators=True)
59 swap_value = swap_value[1:].strip()
60 # TODO maybe add regular express check
61 ev = safe_eval_es(swap_value)
62
63 swap_params[param_name] = ev
64
65 return swap_params
66
67
68 def train_test_split_none(*arrays, **kwargs):
69 """extend train_test_split to take None arrays
70 and support split by group names.
71 """
72 nones = []
73 new_arrays = []
74 for idx, arr in enumerate(arrays):
75 if arr is None:
76 nones.append(idx)
77 else:
78 new_arrays.append(arr)
79
80 if kwargs['shuffle'] == 'None':
81 kwargs['shuffle'] = None
82
83 group_names = kwargs.pop('group_names', None)
84
85 if group_names is not None and group_names.strip():
86 group_names = [name.strip() for name in
87 group_names.split(',')]
88 new_arrays = indexable(*new_arrays)
89 groups = kwargs['labels']
90 n_samples = new_arrays[0].shape[0]
91 index_arr = np.arange(n_samples)
92 test = index_arr[np.isin(groups, group_names)]
93 train = index_arr[~np.isin(groups, group_names)]
94 rval = list(chain.from_iterable(
95 (safe_indexing(a, train),
96 safe_indexing(a, test)) for a in new_arrays))
97 else:
98 rval = train_test_split(*new_arrays, **kwargs)
99
100 for pos in nones:
101 rval[pos * 2: 2] = [None, None]
102
103 return rval
104
105
106 def _evaluate(y_true, pred_probas, scorer, is_multimetric=True):
107 """ output scores based on input scorer
108
109 Parameters
110 ----------
111 y_true : array
112 True label or target values
113 pred_probas : array
114 Prediction values, probability for classification problem
115 scorer : dict
116 dict of `sklearn.metrics.scorer.SCORER`
117 is_multimetric : bool, default is True
118 """
119 if y_true.ndim == 1 or y_true.shape[-1] == 1:
120 pred_probas = pred_probas.ravel()
121 pred_labels = (pred_probas > 0.5).astype('int32')
122 targets = y_true.ravel().astype('int32')
123 if not is_multimetric:
124 preds = pred_labels if scorer.__class__.__name__ == \
125 '_PredictScorer' else pred_probas
126 score = scorer._score_func(targets, preds, **scorer._kwargs)
127
128 return score
129 else:
130 scores = {}
131 for name, one_scorer in scorer.items():
132 preds = pred_labels if one_scorer.__class__.__name__\
133 == '_PredictScorer' else pred_probas
134 score = one_scorer._score_func(targets, preds,
135 **one_scorer._kwargs)
136 scores[name] = score
137
138 # TODO: multi-class metrics
139 # multi-label
140 else:
141 pred_labels = (pred_probas > 0.5).astype('int32')
142 targets = y_true.astype('int32')
143 if not is_multimetric:
144 preds = pred_labels if scorer.__class__.__name__ == \
145 '_PredictScorer' else pred_probas
146 score, _ = compute_score(preds, targets,
147 scorer._score_func)
148 return score
149 else:
150 scores = {}
151 for name, one_scorer in scorer.items():
152 preds = pred_labels if one_scorer.__class__.__name__\
153 == '_PredictScorer' else pred_probas
154 score, _ = compute_score(preds, targets,
155 one_scorer._score_func)
156 scores[name] = score
157
158 return scores
159
160
161 def main(inputs, infile_estimator, infile1, infile2,
162 outfile_result, outfile_object=None,
163 outfile_weights=None, outfile_y_true=None,
164 outfile_y_preds=None, groups=None,
165 ref_seq=None, intervals=None, targets=None,
166 fasta_path=None):
167 """
168 Parameter
169 ---------
170 inputs : str
171 File path to galaxy tool parameter
172
173 infile_estimator : str
174 File path to estimator
175
176 infile1 : str
177 File path to dataset containing features
178
179 infile2 : str
180 File path to dataset containing target values
181
182 outfile_result : str
183 File path to save the results, either cv_results or test result
184
185 outfile_object : str, optional
186 File path to save searchCV object
187
188 outfile_weights : str, optional
189 File path to save deep learning model weights
190
191 outfile_y_true : str, optional
192 File path to target values for prediction
193
194 outfile_y_preds : str, optional
195 File path to save deep learning model weights
196
197 groups : str
198 File path to dataset containing groups labels
199
200 ref_seq : str
201 File path to dataset containing genome sequence file
202
203 intervals : str
204 File path to dataset containing interval file
205
206 targets : str
207 File path to dataset compressed target bed file
208
209 fasta_path : str
210 File path to dataset containing fasta file
211 """
212 warnings.simplefilter('ignore')
213
214 with open(inputs, 'r') as param_handler:
215 params = json.load(param_handler)
216
217 # load estimator
218 with open(infile_estimator, 'rb') as estimator_handler:
219 estimator = load_model(estimator_handler)
220
221 estimator = clean_params(estimator)
222
223 # swap hyperparameter
224 swapping = params['experiment_schemes']['hyperparams_swapping']
225 swap_params = _eval_swap_params(swapping)
226 estimator.set_params(**swap_params)
227
228 estimator_params = estimator.get_params()
229
230 # store read dataframe object
231 loaded_df = {}
232
233 input_type = params['input_options']['selected_input']
234 # tabular input
235 if input_type == 'tabular':
236 header = 'infer' if params['input_options']['header1'] else None
237 column_option = (params['input_options']['column_selector_options_1']
238 ['selected_column_selector_option'])
239 if column_option in ['by_index_number', 'all_but_by_index_number',
240 'by_header_name', 'all_but_by_header_name']:
241 c = params['input_options']['column_selector_options_1']['col1']
242 else:
243 c = None
244
245 df_key = infile1 + repr(header)
246 df = pd.read_csv(infile1, sep='\t', header=header,
247 parse_dates=True)
248 loaded_df[df_key] = df
249
250 X = read_columns(df, c=c, c_option=column_option).astype(float)
251 # sparse input
252 elif input_type == 'sparse':
253 X = mmread(open(infile1, 'r'))
254
255 # fasta_file input
256 elif input_type == 'seq_fasta':
257 pyfaidx = get_module('pyfaidx')
258 sequences = pyfaidx.Fasta(fasta_path)
259 n_seqs = len(sequences.keys())
260 X = np.arange(n_seqs)[:, np.newaxis]
261 for param in estimator_params.keys():
262 if param.endswith('fasta_path'):
263 estimator.set_params(
264 **{param: fasta_path})
265 break
266 else:
267 raise ValueError(
268 "The selected estimator doesn't support "
269 "fasta file input! Please consider using "
270 "KerasGBatchClassifier with "
271 "FastaDNABatchGenerator/FastaProteinBatchGenerator "
272 "or having GenomeOneHotEncoder/ProteinOneHotEncoder "
273 "in pipeline!")
274
275 elif input_type == 'refseq_and_interval':
276 path_params = {
277 'data_batch_generator__ref_genome_path': ref_seq,
278 'data_batch_generator__intervals_path': intervals,
279 'data_batch_generator__target_path': targets
280 }
281 estimator.set_params(**path_params)
282 n_intervals = sum(1 for line in open(intervals))
283 X = np.arange(n_intervals)[:, np.newaxis]
284
285 # Get target y
286 header = 'infer' if params['input_options']['header2'] else None
287 column_option = (params['input_options']['column_selector_options_2']
288 ['selected_column_selector_option2'])
289 if column_option in ['by_index_number', 'all_but_by_index_number',
290 'by_header_name', 'all_but_by_header_name']:
291 c = params['input_options']['column_selector_options_2']['col2']
292 else:
293 c = None
294
295 df_key = infile2 + repr(header)
296 if df_key in loaded_df:
297 infile2 = loaded_df[df_key]
298 else:
299 infile2 = pd.read_csv(infile2, sep='\t',
300 header=header, parse_dates=True)
301 loaded_df[df_key] = infile2
302
303 y = read_columns(
304 infile2,
305 c=c,
306 c_option=column_option,
307 sep='\t',
308 header=header,
309 parse_dates=True)
310 if len(y.shape) == 2 and y.shape[1] == 1:
311 y = y.ravel()
312 if input_type == 'refseq_and_interval':
313 estimator.set_params(
314 data_batch_generator__features=y.ravel().tolist())
315 y = None
316 # end y
317
318 # load groups
319 if groups:
320 groups_selector = (params['experiment_schemes']['test_split']
321 ['split_algos']).pop('groups_selector')
322
323 header = 'infer' if groups_selector['header_g'] else None
324 column_option = \
325 (groups_selector['column_selector_options_g']
326 ['selected_column_selector_option_g'])
327 if column_option in ['by_index_number', 'all_but_by_index_number',
328 'by_header_name', 'all_but_by_header_name']:
329 c = groups_selector['column_selector_options_g']['col_g']
330 else:
331 c = None
332
333 df_key = groups + repr(header)
334 if df_key in loaded_df:
335 groups = loaded_df[df_key]
336
337 groups = read_columns(
338 groups,
339 c=c,
340 c_option=column_option,
341 sep='\t',
342 header=header,
343 parse_dates=True)
344 groups = groups.ravel()
345
346 # del loaded_df
347 del loaded_df
348
349 # cache iraps_core fits could increase search speed significantly
350 memory = joblib.Memory(location=CACHE_DIR, verbose=0)
351 main_est = get_main_estimator(estimator)
352 if main_est.__class__.__name__ == 'IRAPSClassifier':
353 main_est.set_params(memory=memory)
354
355 # handle scorer, convert to scorer dict
356 scoring = params['experiment_schemes']['metrics']['scoring']
357 scorer = get_scoring(scoring)
358 scorer, _ = _check_multimetric_scoring(estimator, scoring=scorer)
359
360 # handle test (first) split
361 test_split_options = (params['experiment_schemes']
362 ['test_split']['split_algos'])
363
364 if test_split_options['shuffle'] == 'group':
365 test_split_options['labels'] = groups
366 if test_split_options['shuffle'] == 'stratified':
367 if y is not None:
368 test_split_options['labels'] = y
369 else:
370 raise ValueError("Stratified shuffle split is not "
371 "applicable on empty target values!")
372
373 X_train, X_test, y_train, y_test, groups_train, groups_test = \
374 train_test_split_none(X, y, groups, **test_split_options)
375
376 exp_scheme = params['experiment_schemes']['selected_exp_scheme']
377
378 # handle validation (second) split
379 if exp_scheme == 'train_val_test':
380 val_split_options = (params['experiment_schemes']
381 ['val_split']['split_algos'])
382
383 if val_split_options['shuffle'] == 'group':
384 val_split_options['labels'] = groups_train
385 if val_split_options['shuffle'] == 'stratified':
386 if y_train is not None:
387 val_split_options['labels'] = y_train
388 else:
389 raise ValueError("Stratified shuffle split is not "
390 "applicable on empty target values!")
391
392 X_train, X_val, y_train, y_val, groups_train, groups_val = \
393 train_test_split_none(X_train, y_train, groups_train,
394 **val_split_options)
395
396 # train and eval
397 if hasattr(estimator, 'validation_data'):
398 if exp_scheme == 'train_val_test':
399 estimator.fit(X_train, y_train,
400 validation_data=(X_val, y_val))
401 else:
402 estimator.fit(X_train, y_train,
403 validation_data=(X_test, y_test))
404 else:
405 estimator.fit(X_train, y_train)
406
407 if hasattr(estimator, 'evaluate'):
408 steps = estimator.prediction_steps
409 batch_size = estimator.batch_size
410 generator = estimator.data_generator_.flow(X_test, y=y_test,
411 batch_size=batch_size)
412 predictions, y_true = _predict_generator(estimator.model_, generator,
413 steps=steps)
414 scores = _evaluate(y_true, predictions, scorer, is_multimetric=True)
415
416 else:
417 if hasattr(estimator, 'predict_proba'):
418 predictions = estimator.predict_proba(X_test)
419 else:
420 predictions = estimator.predict(X_test)
421
422 y_true = y_test
423 scores = _score(estimator, X_test, y_test, scorer,
424 is_multimetric=True)
425 if outfile_y_true:
426 try:
427 pd.DataFrame(y_true).to_csv(outfile_y_true, sep='\t',
428 index=False)
429 pd.DataFrame(predictions).astype(np.float32).to_csv(
430 outfile_y_preds, sep='\t', index=False,
431 float_format='%g', chunksize=10000)
432 except Exception as e:
433 print("Error in saving predictions: %s" % e)
434
435 # handle output
436 for name, score in scores.items():
437 scores[name] = [score]
438 df = pd.DataFrame(scores)
439 df = df[sorted(df.columns)]
440 df.to_csv(path_or_buf=outfile_result, sep='\t',
441 header=True, index=False)
442
443 memory.clear(warn=False)
444
445 if outfile_object:
446 main_est = estimator
447 if isinstance(estimator, Pipeline):
448 main_est = estimator.steps[-1][-1]
449
450 if hasattr(main_est, 'model_') \
451 and hasattr(main_est, 'save_weights'):
452 if outfile_weights:
453 main_est.save_weights(outfile_weights)
454 del main_est.model_
455 del main_est.fit_params
456 del main_est.model_class_
457 del main_est.validation_data
458 if getattr(main_est, 'data_generator_', None):
459 del main_est.data_generator_
460
461 with open(outfile_object, 'wb') as output_handler:
462 pickle.dump(estimator, output_handler,
463 pickle.HIGHEST_PROTOCOL)
464
465
466 if __name__ == '__main__':
467 aparser = argparse.ArgumentParser()
468 aparser.add_argument("-i", "--inputs", dest="inputs", required=True)
469 aparser.add_argument("-e", "--estimator", dest="infile_estimator")
470 aparser.add_argument("-X", "--infile1", dest="infile1")
471 aparser.add_argument("-y", "--infile2", dest="infile2")
472 aparser.add_argument("-O", "--outfile_result", dest="outfile_result")
473 aparser.add_argument("-o", "--outfile_object", dest="outfile_object")
474 aparser.add_argument("-w", "--outfile_weights", dest="outfile_weights")
475 aparser.add_argument("-l", "--outfile_y_true", dest="outfile_y_true")
476 aparser.add_argument("-p", "--outfile_y_preds", dest="outfile_y_preds")
477 aparser.add_argument("-g", "--groups", dest="groups")
478 aparser.add_argument("-r", "--ref_seq", dest="ref_seq")
479 aparser.add_argument("-b", "--intervals", dest="intervals")
480 aparser.add_argument("-t", "--targets", dest="targets")
481 aparser.add_argument("-f", "--fasta_path", dest="fasta_path")
482 args = aparser.parse_args()
483
484 main(args.inputs, args.infile_estimator, args.infile1, args.infile2,
485 args.outfile_result, outfile_object=args.outfile_object,
486 outfile_weights=args.outfile_weights,
487 outfile_y_true=args.outfile_y_true,
488 outfile_y_preds=args.outfile_y_preds,
489 groups=args.groups,
490 ref_seq=args.ref_seq, intervals=args.intervals,
491 targets=args.targets, fasta_path=args.fasta_path)