from typing import Any, Generic, Optional, Union
from .._runtime import Mode, blocking_property, operation
from ..requestor import OptimizerRequestor
[docs]
class Driver(Generic[Mode]):
"""This class provides methods for retrieving and setting driver-specific
information of the study. Depending on the chosen driver of
:func:`Client.create_study` the driver provides specific methods. More details
for each driver are available in the :ref:`driver reference <DriverReference>`.
Example::
study.run()
driver = study.driver
min_objective_values = driver.historic_parameter_values(
path="acquisition_function.min_objective")
The constructor should not be used directly since it does not create a
driver on the server side. Instead, one should use :attr:`Study.driver`.
"""
def __init__(
self, study_id: str, project_id: str, requestor: OptimizerRequestor
) -> None:
self._runtime = requestor.runtime
self._requestor = requestor
self.study_id = study_id
self.project_id = project_id
@property
def qualifier(self) -> str:
return self.project_id + "." + self.study_id
async def _post(
self,
purpose: str,
object: str,
operation: str,
data: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
return await self._requestor._async_post(
purpose, object, operation, self.qualifier, data
)
async def _get(self, purpose: str, object: str, type: str) -> dict[str, Any]:
return await self._requestor._async_get(purpose, object, type, self.qualifier)
async def _run_task(self, purpose: str, task: str, data: dict[str, Any]) -> Any:
return await self._requestor._async_run_task(
purpose, task, self.qualifier, data
)
async def _async_describe(self) -> dict[str, Any]:
"""Get description of all modules and their parameters that are used
by the driver. Example::
description = driver.describe()
print(description["members"]["surrogates"]["0"])
Returns: A nested dictionary with description of submodules consisting
of a name and a descriptive text. If the entry describes a module,
it has an additional ``"members"`` entry with dictionaries describing
submodules and parameters.
"""
answer = await self._get("get description of the driver", "driver", "describe")
return answer["description"]
describe = operation(_async_describe, interruptible=False)
async def _async_get_state(self, path: Optional[str] = None) -> dict[str, Any]:
"""Get state of the driver. Example::
best_sample = driver.get_state(path="best_sample")
Args:
path: A dot-separated path to a submodule or parameter.
If none, the full state is returned.
Returns: If path is None, a dictionary with information of driver state.
.. note:: A description of the meaning of each entry in the state
can be retrieved by :func:`describe`.
"""
answer = await self._post(
"get driver state", "driver", "get_state", data={"path": path}
)
return answer["state"]
get_state = operation(_async_get_state, interruptible=False)
async def _async_historic_parameter_values(self, path: str) -> list[Any]:
"""Get the values of an internal parameter for each iteration of the study. Example::
min_objective_values = driver.historic_parameter_values(
path="acquisition_function.min_objective")
Args:
path: A dot-separated path to the parameter.
.. note:: A description of the meaning of each parameter can be retrieved
by :func:`describe`.
"""
answer = await self._post(
"get historic parameter values",
"driver",
"historic_parameter_values",
data={"path": path},
)
return answer["values"]
historic_parameter_values = operation(
_async_historic_parameter_values, interruptible=False
)
class Minimizer(Driver[Mode]):
"""
This class provides methods for retrieving and setting information of the
minimization driver of the study.
"""
async def _async_get_best_sample(self) -> dict[str, Union[float, int, str]]:
"""
Best sample with minimal objective value found during the minimization. Example::
for key, value in driver.get_best_sample().items():
print(f"{key} = {value}")
"""
answer = await self._post(
"get best sample", "driver", "get_state", data={"path": "best_sample"}
)
return answer["state"]
get_best_sample = operation(_async_get_best_sample, interruptible=False)
@property
def best_sample(self) -> dict[str, Union[float, int, str]]:
"""Parameters of the best sample found so far.
With ``Client(asynchronous=True)``, use
``await driver.get_best_sample()`` instead of this property.
"""
return blocking_property(self, "get_best_sample")
async def _async_get_min_objective(self) -> float:
"""
Minimal objective value found during the minimization. Example::
min_objective = driver.get_min_objective()
"""
answer = await self._post(
"get minimal objective value",
"driver",
"get_state",
data={"path": "min_objective"},
)
return answer["state"]
get_min_objective = operation(_async_get_min_objective, interruptible=False)
@property
def min_objective(self) -> float:
"""The smallest objective value found so far.
With ``Client(asynchronous=True)``, use
``await driver.get_min_objective()`` instead of this property.
"""
return blocking_property(self, "get_min_objective")
class LeastSquaresDriver(Minimizer[Mode]):
"""
This class provides methods for retrieving information of the
result of the least squares minimization using ``scipy.optimize.least_squares``.
"""
async def _async_get_best_sample(self) -> dict[str, Union[float, int, str]]:
"""
Best sample with minimal chi-squared value found during the minimization. Example::
best_sample = driver.get_best_sample()
for key, value in best_sample.items():
print(f"{key} = {value}")
"""
return await super()._async_get_best_sample()
get_best_sample = operation(_async_get_best_sample, interruptible=False)
@property
def best_sample(self) -> dict[str, Union[float, int, str]]:
"""Parameters of the best sample found so far.
With ``Client(asynchronous=True)``, use
``await driver.get_best_sample()`` instead of this property.
"""
return blocking_property(self, "get_best_sample")
async def _async_get_min_objective(self) -> float:
"""
Minimal chi-squared value found during the minimization. Example::
min_chi_sq = study.driver.get_min_objective()
"""
return await super()._async_get_min_objective()
get_min_objective = operation(_async_get_min_objective, interruptible=False)
@property
def min_objective(self) -> float:
"""The smallest chi-squared value found so far.
With ``Client(asynchronous=True)``, use
``await driver.get_min_objective()`` instead of this property.
"""
return blocking_property(self, "get_min_objective")
async def _async_get_uncertainties(self) -> dict[str, float]:
"""
Uncertainties of the continuous parameters of the :py:attr:`best_sample`. Example::
uncertainties = driver.get_uncertainties()
for key, value in uncertainties.items():
print(f"uncertainty of {key}: {value}")
"""
answer = await self._post(
"get uncertainties",
"driver",
"get_state",
data={"path": "parameter_uncertainties"},
)
return answer["state"]
get_uncertainties = operation(_async_get_uncertainties, interruptible=False)
@property
def uncertainties(self) -> dict[str, float]:
"""Uncertainties of the continuous parameters of the best sample.
With ``Client(asynchronous=True)``, use
``await driver.get_uncertainties()`` instead of this property.
"""
return blocking_property(self, "get_uncertainties")
async def _async_get_covariance(self) -> list[list[float]]:
"""Covariance matrix of the reconstructed continuous parameters."""
answer = await self._post(
"get parameter covariance",
"driver",
"get_state",
data={"path": "parameter_covariance"},
)
return answer["state"]
get_covariance = operation(_async_get_covariance, interruptible=False)
@property
def covariance(self) -> list[list[float]]:
"""Covariance matrix of the reconstructed continuous parameters.
With ``Client(asynchronous=True)``, use
``await driver.get_covariance()`` instead of this property.
"""
return blocking_property(self, "get_covariance")