-
Notifications
You must be signed in to change notification settings - Fork 66
Expand file tree
/
Copy pathcallback.py
More file actions
477 lines (372 loc) · 16.7 KB
/
Copy pathcallback.py
File metadata and controls
477 lines (372 loc) · 16.7 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
"""The callback module contains sub-classes of the ``Callback`` class.
The ``Callback`` class is used to trigger custom actions on the start and
completion of jobs by the ``Evaluator``. Callbacks can be used with any
``Evaluator`` implementation.
"""
import abc
import csv
import logging
import os
from typing import List
import numpy as np
from deephyper.evaluator import HPOJob, Job
from deephyper.evaluator.utils import test_ipython_interpretor
from deephyper.skopt.moo import hypervolume
if test_ipython_interpretor():
from tqdm.notebook import tqdm
else:
from tqdm import tqdm
__all__ = ["Callback", "LoggerCallback", "TqdmCallback", "SearchEarlyStopping"]
logger = logging.getLogger(__name__)
class Callback(abc.ABC):
"""Callback interface."""
def on_launch(self, job: Job):
"""Called each time a ``Job`` is created by the ``Evaluator``.
Args:
job (Job): The created job.
"""
def on_done(self, job: Job):
"""Called each time a local ``Job`` has been gathered by the Evaluator.
Args:
job (Job): The completed job.
"""
def on_done_other(self, job: Job):
"""Called after local ``Job`` have been gathered for each remote ``Job`` that is done.
Args:
job (Job): The completed Job.
"""
def on_gather(self, local_jobs: List[Job], other_jobs: List[Job]):
"""Called after gathering jobs.
Args:
local_jobs (List[Job]):
gathered jobs from local evaluator instance.
other_jobs (List[Job]):
gathered jobs from other evaluators using the same storage.
"""
def on_close(self):
"""Called when the evaluator is being closed."""
class ObjectiveRecorder:
"""Records the objective values of the jobs.
:meta: private
"""
def __init__(self):
self._objectives = []
self.is_multi_objective = False
self._last_return = -float("inf")
def __call__(self, job):
"""Called when a local job has been gathered."""
# Only add the objective if it is not a string (i.e., failure...)
if not isinstance(job.objective, str):
self._objectives.append(job.objective)
# Then check if the objective is multi-objective
if np.ndim(job.objective) > 0:
self.is_multi_objective = True
# If no objectives are received but only failures then return -inf
if len(self._objectives) == 0:
return self._last_return
# If single objective then returns the maximum
if not self.is_multi_objective:
self._last_return = max(self._objectives[-1], self._last_return)
return self._last_return
else:
objectives = -np.asarray(self._objectives)
ref = np.max(objectives, axis=0) # reference point
return hypervolume(objectives, ref)
class LoggerCallback(Callback):
"""Print information when jobs are completed by the ``Evaluator``.
An example usage can be:
>>> evaluator.create(method="ray", method_kwargs={..., "callbacks": [LoggerCallback()]})
"""
def __init__(self):
self._best_objective = None
self._n_done = 0
self._objective_func = ObjectiveRecorder()
def on_done_other(self, job):
"""Called after gathering local jobs on available remote jobs that are done."""
self.on_done(job)
def on_done(self, job):
"""Called when a local job has been gathered."""
self._n_done += 1
# Test if multi objectives are received
if np.ndim(job.objective) > 0:
if np.isreal(job.objective).all():
self._best_objective = self._objective_func(job)
tmp = tuple(round(o, 5) if not isinstance(o, str) else o for o in job.objective)
print(
f"[{self._n_done:05d}] -- HVI Objective: {self._best_objective:.5f} -- "
f"Last Objective: {tmp}"
)
elif np.any(type(res) is str and "F" == res[0] for res in job.objective):
print(f"[{self._n_done:05d}] -- Last Failure: {job.objective}")
elif np.isreal(job.objective):
self._best_objective = self._objective_func(job)
print(
f"[{self._n_done:05d}] -- Maximum Objective: {self._best_objective:.5f} -- "
f"Last Objective: {job.objective:.5f}"
)
elif type(job.objective) is str and "F" == job.objective[0]:
print(f"[{self._n_done:05d}] -- Last Failure: {job.objective}")
class TqdmCallback(Callback):
"""Print information when jobs are completed by the ``Evaluator``.
Args:
description (str, optional): an optional description to add to the progressbar.
An example usage can be:
>>> evaluator.create(method="ray", method_kwargs={..., "callbacks": [TqdmCallback()]})
"""
def __init__(self, description: str = None):
self._best_objective = None
self._n_done = 0
self._n_failures = 0
self._max_evals = None
self._tqdm = None
self._objective_func = ObjectiveRecorder()
self._description = description
def set_max_evals(self, max_evals):
"""Setter for the maximum number of evaluations.
It is used to initialize the tqdm progressbar.
"""
self._max_evals = max_evals
self._tqdm = None
def on_done_other(self, job):
"""Called after gathering local jobs on available remote jobs that are done."""
self.on_done(job)
def on_done(self, job):
"""Called when a local job has been gathered."""
if self._tqdm is None:
if self._max_evals:
self._tqdm = tqdm(total=self._max_evals)
else:
self._tqdm = tqdm()
if self._description:
self._tqdm.set_description(self._description)
self._n_done += 1
self._tqdm.update(1)
if isinstance(job, HPOJob):
# Test if multi objectives are received
if np.ndim(job.objective) > 0:
if not (any(not (np.isreal(objective_i)) for objective_i in job.objective)):
self._best_objective = self._objective_func(job)
else:
self._n_failures += 1
self._tqdm.set_postfix({"failures": self._n_failures, "hvi": self._best_objective})
else:
if np.isreal(job.objective):
self._best_objective = self._objective_func(job)
else:
self._n_failures += 1
self._tqdm.set_postfix(objective=self._best_objective, failures=self._n_failures)
if self._max_evals == self._n_done:
self._tqdm.close()
class SearchEarlyStopping(Callback):
"""Stop the search gracefully when it does not improve for a given number of evaluations.
Args:
patience (int, optional):
The number of not improving evaluations to wait for before
stopping the search. Defaults to ``10``.
objective_func (callable, optional):
A function that takes a ``Job`` has input and returns the maximized scalar value
monitored by this callback. Defaults to computes the maximum for single-objective
optimization and the hypervolume for multi-objective optimization.
threshold (float, optional):
The threshold to reach before activating the patience to stop the
search. Defaults to ``None``, patience is reinitialized after
each improving observation.
verbose (bool, optional): Activation or deactivate the verbose mode. Defaults to ``True``.
"""
def __init__(
self,
patience: int = 10,
objective_func=None,
threshold: float = None,
verbose: bool = 1,
):
self._best_objective = None
self._n_lower = 0
self._patience = patience
self._objective_func = ObjectiveRecorder() if objective_func is None else objective_func
self._threshold = threshold
self._verbose = verbose
self.search_stopped = False
def on_done_other(self, job):
"""Called after gathering local jobs on available remote jobs that are done."""
self.on_done(job)
def on_done(self, job):
"""Called when a local job has been gathered."""
job_objective = self._objective_func(job)
if self._best_objective is None:
self._best_objective = job_objective
else:
if job_objective > self._best_objective:
if self._verbose:
print(
"Objective has improved from "
f"{self._best_objective:.5f} -> {job_objective:.5f}"
)
self._best_objective = job_objective
self._n_lower = 0
else:
self._n_lower += 1
if self._n_lower >= self._patience:
if self._threshold is None:
if self._verbose:
print(
"Stopping the search because it did not improve for the last "
f"{self._patience} evaluations!"
)
self.search_stopped = True
else:
if self._best_objective > self._threshold:
if self._verbose:
print(
"Stopping the search because it did not improve for the last "
f"{self._patience} evaluations!"
)
self.search_stopped = True
# TODO: Add unit tests
# This class is made to be used by people who wants to log results from the
# evaluator without using it within the Search.
class CSVLoggerCallback(Callback):
"""Dump jobs done to a CSV file.
Args:
path (str): The path where the CSV is being dumped.
"""
def __init__(self, path: str = "results.csv"):
self.path = os.path.abspath(path)
if not os.path.exists(os.path.dirname(path)):
raise ValueError(f"Directory not found {self.path}")
self.jobs_done = []
self.num_objective = None
self._start_dumping = False
self._columns_dumped = None
self._job_class = None
def on_gather(self, local_jobs: List[Job], other_jobs: List[Job]):
"""Called after gathering jobs.
Args:
local_jobs (List[Job]):
gathered jobs from local evaluator instance.
other_jobs (List[Job]):
gathered jobs from other evaluators using the same storage.
"""
self.jobs_done.extend(local_jobs)
self.jobs_done.extend(other_jobs)
self.dump_jobs_done_to_csv(self.path)
def on_close(self):
self.dump_jobs_done_to_csv(self.path, flush=True)
def dump_jobs_done_to_csv(self, path: str, flush: bool = False):
"""Dump completed jobs to a CSV file.
This will reset the ``Evaluator.jobs_done`` attribute to an empty list.
Args:
path (str):
The path of the file where the CSV is being dumped.
flush (bool):
A boolean indicating if the results should be flushed (i.e., forcing the dumping).
"""
if len(self.jobs_done) > 0:
if self._job_class is None:
self._job_class = type(self.jobs_done[0])
else:
return
logger.info("Dumping completed jobs to CSV...")
if self._job_class is HPOJob:
self._dump_jobs_done_to_csv_as_hpo_format(path, flush)
else:
self._dump_jobs_done_to_csv_as_regular_format(path)
logger.info("Dumping done")
def _dump_jobs_done_to_csv_as_regular_format(self, path: str):
"""Dump completed jobs to a CSV file for regular job format.
Args:
path (str):
The path of the file where the CSV is being dumped.
"""
records_list = []
for job in self.jobs_done:
# Start with job.id
result = {"job_id": int(job.id.split(".")[1])}
# Add job.status
result["job_status"] = job.status.name
# input arguments: add prefix for all keys found in "args"
result.update({f"p:{k}": v for k, v in job.args.items()})
# output
if isinstance(job.output, dict):
output = {f"o:{k}": v for k, v in job.output.items()}
else:
output = {"o:": job.output}
result.update(output)
# metadata
metadata = {f"m:{k}": v for k, v in job.metadata.items() if k[0] != "_"}
result.update(metadata)
records_list.append(result)
if len(records_list) != 0:
mode = "a" if self._start_dumping else "w"
with open(path, mode) as fp:
if not (self._start_dumping):
self._columns_dumped = records_list[0].keys()
if self._columns_dumped is not None:
writer = csv.DictWriter(fp, self._columns_dumped, extrasaction="ignore")
if not (self._start_dumping):
writer.writeheader()
self._start_dumping = True
writer.writerows(records_list)
self.jobs_done = []
def _dump_jobs_done_to_csv_as_hpo_format(self, path: str, flush: bool = False):
"""Dump completed jobs to a CSV file for the hyperparameter optimization format.
This will reset the ``Evaluator.jobs_done`` attribute to an empty list.
Args:
path (str):
The path of the file where the CSV is being dumped.
flush (bool):
A boolean indicating if the results should be flushed (i.e., forcing the dumping).
"""
resultsList = []
for job in self.jobs_done:
# add prefix for all keys found in "args"
result = {f"p:{k}": v for k, v in job.args.items()}
# when the returned value of the run-function is a dict we flatten it to add in csv
result["objective"] = job.objective
print(f"{job.objective=}")
# when the objective is a tuple (multi-objective) we create 1 column per tuple-element
if isinstance(result["objective"], tuple) or isinstance(result["objective"], list):
obj = result.pop("objective")
if self.num_objective is None:
self.num_objective = len(obj)
for i, objval in enumerate(obj):
result[f"objective_{i}"] = objval
else:
if self.num_objective is None:
self.num_objective = 1
if self.num_objective > 1:
obj = result.pop("objective")
for i in range(self.num_objective):
result[f"objective_{i}"] = obj
# Add job.id
result["job_id"] = int(job.id.split(".")[1])
# Add job.status
result["job_status"] = job.status.name
# Profiling and other
# methdata keys starting with "_" are not saved (considered as internal)
metadata = {f"m:{k}": v for k, v in job.metadata.items() if k[0] != "_"}
result.update(metadata)
resultsList.append(result)
if len(resultsList) != 0:
mode = "a" if self._start_dumping else "w"
with open(path, mode) as fp:
if not (self._start_dumping):
for result in resultsList:
# Waiting to start receiving non-failed jobs before dumping results
is_single_obj_and_has_success = (
"objective" in result and type(result["objective"]) is not str
)
is_multi_obj_and_has_success = (
"objective_0" in result and type(result["objective_0"]) is not str
)
print(f"{is_single_obj_and_has_success=}, {is_multi_obj_and_has_success=}")
if is_single_obj_and_has_success or is_multi_obj_and_has_success or flush:
self._columns_dumped = result.keys()
break
if self._columns_dumped is not None:
writer = csv.DictWriter(fp, self._columns_dumped, extrasaction="ignore")
if not (self._start_dumping):
writer.writeheader()
self._start_dumping = True
writer.writerows(resultsList)
self.jobs_done = []