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
|
"""Contains classes and functions for logging."""
from __future__ import annotations
import json
from contextlib import suppress
from pathlib import Path
from typing import TYPE_CHECKING, Any
import numpy as np
from colorama import Fore, just_fix_windows_console
from bayes_opt.event import Events
from bayes_opt.observer import _Tracker
if TYPE_CHECKING:
from os import PathLike
from bayes_opt.bayesian_optimization import BayesianOptimization
just_fix_windows_console()
def _get_default_logger(verbose: int, is_constrained: bool) -> ScreenLogger:
"""
Return the default logger.
Parameters
----------
verbose : int
Verbosity level of the logger.
is_constrained : bool
Whether the underlying optimizer uses constraints (this requires
an additional column in the output).
Returns
-------
ScreenLogger
The default logger.
"""
return ScreenLogger(verbose=verbose, is_constrained=is_constrained)
class ScreenLogger(_Tracker):
"""Logger that outputs text, e.g. to log to a terminal.
Parameters
----------
verbose : int
Verbosity level of the logger.
is_constrained : bool
Whether the logger is associated with a constrained optimization
instance.
"""
_default_cell_size = 9
_default_precision = 4
_colour_new_max = Fore.MAGENTA
_colour_regular_message = Fore.RESET
_colour_reset = Fore.RESET
def __init__(self, verbose: int = 2, is_constrained: bool = False) -> None:
self._verbose = verbose
self._is_constrained = is_constrained
self._header_length = None
super().__init__()
@property
def verbose(self) -> int:
"""Return the verbosity level."""
return self._verbose
@verbose.setter
def verbose(self, v: int) -> None:
"""Set the verbosity level.
Parameters
----------
v : int
New verbosity level of the logger.
"""
self._verbose = v
@property
def is_constrained(self) -> bool:
"""Return whether the logger is constrained."""
return self._is_constrained
def _format_number(self, x: float) -> str:
"""Format a number.
Parameters
----------
x : number
Value to format.
Returns
-------
A stringified, formatted version of `x`.
"""
if isinstance(x, int):
s = f"{x:<{self._default_cell_size}}"
else:
s = f"{x:<{self._default_cell_size}.{self._default_precision}}"
if len(s) > self._default_cell_size:
if "." in s:
return s[: self._default_cell_size]
return s[: self._default_cell_size - 3] + "..."
return s
def _format_bool(self, x: bool) -> str:
"""Format a boolean.
Parameters
----------
x : boolean
Value to format.
Returns
-------
A stringified, formatted version of `x`.
"""
x_ = ("T" if x else "F") if self._default_cell_size < 5 else str(x)
return f"{x_:<{self._default_cell_size}}"
def _format_key(self, key: str) -> str:
"""Format a key.
Parameters
----------
key : string
Value to format.
Returns
-------
A stringified, formatted version of `x`.
"""
s = f"{key:^{self._default_cell_size}}"
if len(s) > self._default_cell_size:
return s[: self._default_cell_size - 3] + "..."
return s
def _step(self, instance: BayesianOptimization, colour: str = _colour_regular_message) -> str:
"""Log a step.
Parameters
----------
instance : bayesian_optimization.BayesianOptimization
The instance associated with the event.
colour :
(Default value = _colour_regular_message, equivalent to Fore.RESET)
Returns
-------
A stringified, formatted version of the most recent optimization step.
"""
res: dict[str, Any] = instance.res[-1]
keys: list[str] = instance.space.keys
# iter, target, allowed [, *params]
cells: list[str | None] = [None] * (3 + len(keys))
cells[:2] = self._format_number(self._iterations + 1), self._format_number(res["target"])
if self._is_constrained:
cells[2] = self._format_bool(res["allowed"])
params = res.get("params", {})
cells[3:] = [self._format_number(params.get(key, float("nan"))) for key in keys]
return "| " + " | ".join(colour + x + self._colour_reset for x in cells if x is not None) + " |"
def _header(self, instance: BayesianOptimization) -> str:
"""Print the header of the log.
Parameters
----------
instance : bayesian_optimization.BayesianOptimization
The instance associated with the header.
Returns
-------
A stringified, formatted version of the most header.
"""
keys: list[str] = instance.space.keys
# iter, target, allowed [, *params]
cells: list[str | None] = [None] * (3 + len(keys))
cells[:2] = self._format_key("iter"), self._format_key("target")
if self._is_constrained:
cells[2] = self._format_key("allowed")
cells[3:] = [self._format_key(key) for key in keys]
line = "| " + " | ".join(x for x in cells if x is not None) + " |"
self._header_length = len(line)
return line + "\n" + ("-" * self._header_length)
def _is_new_max(self, instance: BayesianOptimization) -> bool:
"""Check if the step to log produced a new maximum.
Parameters
----------
instance : bayesian_optimization.BayesianOptimization
The instance associated with the step.
Returns
-------
boolean
"""
if instance.max is None:
# During constrained optimization, there might not be a maximum
# value since the optimizer might've not encountered any points
# that fulfill the constraints.
return False
if self._previous_max is None:
self._previous_max = instance.max["target"]
return instance.max["target"] > self._previous_max
def update(self, event: str, instance: BayesianOptimization) -> None:
"""Handle incoming events.
Parameters
----------
event : str
One of the values associated with `Events.OPTIMIZATION_START`,
`Events.OPTIMIZATION_STEP` or `Events.OPTIMIZATION_END`.
instance : bayesian_optimization.BayesianOptimization
The instance associated with the step.
"""
line = ""
if event == Events.OPTIMIZATION_START:
line = self._header(instance) + "\n"
elif event == Events.OPTIMIZATION_STEP:
is_new_max = self._is_new_max(instance)
if self._verbose != 1 or is_new_max:
colour = self._colour_new_max if is_new_max else self._colour_regular_message
line = self._step(instance, colour=colour) + "\n"
elif event == Events.OPTIMIZATION_END:
line = "=" * self._header_length + "\n"
if self._verbose:
print(line, end="")
self._update_tracker(event, instance)
class JSONLogger(_Tracker):
"""
Logger that outputs steps in JSON format.
The resulting file can be used to restart the optimization from an earlier state.
Parameters
----------
path : str or os.PathLike
Path to the file to write to.
reset : bool
Whether to overwrite the file if it already exists.
"""
def __init__(self, path: str | PathLike[str], reset: bool = True):
self._path = Path(path)
if reset:
with suppress(OSError):
self._path.unlink(missing_ok=True)
super().__init__()
def update(self, event: str, instance: BayesianOptimization) -> None:
"""
Handle incoming events.
Parameters
----------
event : str
One of the values associated with `Events.OPTIMIZATION_START`,
`Events.OPTIMIZATION_STEP` or `Events.OPTIMIZATION_END`.
instance : bayesian_optimization.BayesianOptimization
The instance associated with the step.
"""
if event == Events.OPTIMIZATION_STEP:
data = dict(instance.res[-1])
now, time_elapsed, time_delta = self._time_metrics()
data["datetime"] = {"datetime": now, "elapsed": time_elapsed, "delta": time_delta}
if "allowed" in data: # fix: github.com/fmfn/BayesianOptimization/issues/361
data["allowed"] = bool(data["allowed"])
if "constraint" in data and isinstance(data["constraint"], np.ndarray):
data["constraint"] = data["constraint"].tolist()
with self._path.open("a") as f:
f.write(json.dumps(data) + "\n")
self._update_tracker(event, instance)
|
