from typing import Any, Callable, Literal, Optional, Union
import time
import datetime as dt
import requests
import atexit
from .requestor import Requestor
from .study import Study
from .objects import Observation
[docs]
class Benchmark(Requestor):
"""
This class provides methods for benchmarking different studies against each other
that try to minimize the same objective. Example::
benchmark = client.create_benchmark(num_average=6)
benchmark.add_study(study1)
benchmark.add_study(study2)
benchmark.set_evaluator(evaluate)
benchmark.run()
data = benchmark.get_data(x_type='num_evaluations',y_type='objective',
average_type='mean')
fig = plt.figure(figsize=(8,4))
for idx,name in enumerate(data['names']):
X = data['X'][idx]
Y = np.array(data['Y'][idx])
std_error = np.array(data['sdev'][idx])/np.sqrt(6)
p = plt.plot(X,Y,linewidth=2.0, label=name)
plt.fill_between(X, Y-std_error, Y+std_error, alpha=0.2, color = p[0].get_color())
plt.legend(loc='upper right',ncol=1)
plt.grid()
plt.ylim([0.1,10])
plt.rc('font',family='serif')
plt.xlabel('number of iterations',fontsize=12)
plt.ylabel('average objective',fontsize=12)
plt.show()
The constructor should not be used directly since it does not create a
benchmark on the server side. Instead, one should use :func:`Client.create_benchmark`.
"""
def __init__(
self, host: str, benchmark_id: str, session: requests.Session, num_average: int
):
super(Benchmark, self).__init__(host=host, session=session)
self.id = benchmark_id
self.num_average = num_average
self._studies: list[Study] = []
self.deleted = False
atexit.register(self._delete_on_server)
def _post(
self,
purpose: str,
operation: str,
data: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
return super(Benchmark, self).post(
purpose, "benchmark", operation, self.id, data
)
def _get(self, purpose: str, type: str) -> dict[str, Any]:
return super(Benchmark, self).get(purpose, "benchmark", type, self.id)
def _delete_on_server(self) -> None:
self._post("delete benchmark", "delete")
self.deleted = True
def __del__(self) -> None:
if not self.deleted:
self._delete_on_server()
[docs]
def add_study(self, study: Study) -> None:
"""Adds a study to the benchmark. Example::
benchmark.add_study(study1)
Args:
study: A :class:`~jcmoptimizer.Study` object for which a benchmark
should be run.
"""
self._post("add study", "add_study", data={"study_id": study.id})
self._studies.append(study)
@property
def studies(self) -> list[Study]:
"""A list of studies to be run for the benchmark."""
studies: list[Study] = []
for s in self._studies:
for num_run in range(self.num_average):
studies.append(s)
return studies
[docs]
def add_study_results(self, study: Study) -> None:
"""Adds the results of a benchmark study at the end of an optimization run.
Example::
benchmark.add_study_results(study1)
Args:
study: A :class:`~jcmoptimizer.Study` object after the study was run.
"""
answer = self._post(
"add study results", "add_study_results", data={"study_id": study.id}
)
if answer["new_study_id"] != "":
study.id = answer["new_study_id"]
[docs]
def get_data(
self,
x_type: Literal["num_evaluations", "time"] = "num_evaluations",
y_type: Literal["objective", "distance"] = "objective",
average_type: Literal["mean", "median"] = "mean",
invert: bool = False,
log_scale: bool = False,
minimum: Optional[list[float]] = None,
scales: Optional[list[float]] = None,
norm: Union[str, int, None] = None,
num_samples: int = 100,
) -> dict[str, list[Union[str, float]]]:
"""Get benchmark data. Example::
data = benchmark.get_data( x_type='num_evaluations', y_type='objective',
average_type='mean')
plt.plot(data['X'][0],data['Y'][0])
Args:
x_type: Data on x-axis. Can be either 'num_evaluations' or 'time'.
The time data is given in units of seconds.
y_type: Data type on y-axis. Can be either 'objective', 'distance',
(i.e. accumulated minimum distance off all samples to overall minimum),
or 'min_distance' (i.e. distance of current minimum to overall
minimum).
average_type: Type of averaging over study runs. Can be either
'mean' w.r.t. x-axis data or 'median' w.r.t. y-axis data
invert: If True, the objective is multiplied by -1.
(Parameter not available for distance average types)
log_scale: If True, the output of Y and sdev are determined as
mean and standard deviations of the natural logarithm of the
considered y_type.
minimum: Vector with minimum position. (Only available for
distance average types)
scales: Vector with positive weights for scaling distance in
different directions. (Only available for distance average types)
norm: Order of distance norm as defined in
numpy.linalg.norm. (Only available for distance average types)
num_samples: Number of samples on y-axis. (Only available for
median average type or time on x-axis)
"""
answer = self._post(
"get data",
"get_data",
data={
"x_type": x_type,
"y_type": y_type,
"average_type": average_type,
"invert": invert,
"log_scale": log_scale,
"minimum": minimum,
"scales": scales,
"norm": norm,
"num_samples": num_samples,
},
)
return answer["data"]
def set_objective(self, objective: Callable) -> None:
raise AttributeError(
"The method 'set_objective()' is deprecated. "
"Please, use the method 'set_evaluator()' instead"
)
[docs]
def set_evaluator(self, evaluator: Callable[..., Observation]) -> None:
"""Set the function that maps design parameters to an :class:`Observation`.
Example::
def evaluate(study: Study, x1: float, x2: float) -> Observation:
observation = study.new_observation()
observation.add(x1**2 + x2**2)
return observation
benchmark.set_evaluator(evaluate)
.. note:: Call this function only after all studies have been added
to the benchmark.
Args:
evaluator: Function handle for a function of the
variable parameters that returns a corresponding :class:`Observation` object.
The function must accept a ``"study"`` argument as well as
an argument with the name of each design parameter and fixed environment
parameter.
"""
for study in self._studies:
study.set_evaluator(evaluator)
[docs]
def run(self) -> None:
"""Run the benchmark after the evaluator has been set
(see :func:`~Benchmark.set_evaluator`).
Example::
benchmark.run()
"""
time_zero_benchmark = time.time()
for study in self._studies:
self.inform(f"Running study '{study.id}'")
for i in range(self.num_average):
time_zero = time.time()
self.inform(f"Run {i+1}/{self.num_average} of study '{study.id}'")
try:
study.run()
except EnvironmentError as err:
self.warn(f"Study '{study.id}' stopped due to error: {err}")
self.add_study_results(study)
timedelta = dt.timedelta(seconds=int(time.time() - time_zero))
self.inform(f"Run of study '{study.id}' finished after {timedelta}.")
timedelta = dt.timedelta(seconds=int(time.time() - time_zero_benchmark))
self.inform(f"Benchmark finished after {timedelta}")