import asyncio
import webbrowser
from datetime import datetime, timedelta
from typing import Any, Generic, Optional, Union
from . import drivers
from ._runtime import (
Mode, Runtime, operation, register_exit_cleanup, settle, unregister_exit_cleanup,
)
from .benchmark import Benchmark
from .drivers import Driver
from .requestor import (
CloudRequestor,
OptimizerRequestor,
inform,
read_yml_config_files,
warn,
)
from .study import Study
from .version import __version__
[docs]
class Client(Generic[Mode]):
"""Connect to an optimization server and create studies and benchmarks.
If the :ref:`Configuration` is set up, no connection arguments are needed.
For example::
design_space = [
{'name': 'x1', 'type': 'continuous', 'domain': (-1.5, 1.5)},
{'name': 'x2', 'type': 'continuous', 'domain': (-1.5, 1.5)},
]
with Client() as client:
study = client.create_study(
design_space=design_space, study_name='example'
)
By default, each call waits for its result. When writing an ``async``
function, use ``Client(asynchronous=True)`` and ``await`` calls that contact
the server. After defining your evaluation function, you can run a study::
async with Client(asynchronous=True) as client:
study = await client.create_study(design_space=design_space)
await study.configure(max_iter=20)
study.set_evaluator(evaluate)
await study.run()
minimum = await study.driver.get_min_objective()
Methods that prepare observations or set the evaluation function, such as
``new_observation`` and ``set_evaluator``, do not need ``await``.
Use ``with Client()`` or ``async with Client(asynchronous=True)`` to close
the connection automatically when you leave the block. If the client started
a server, it also stops that server unless ``persistent=True``. A server
that was already running is left running.
Without a ``with`` block, call ``client.close()`` when finished, or
``await client.close()`` when using ``asynchronous=True``.
Args:
host: Address of a running server, for example
``'http://localhost:4554'``. If neither ``host`` nor ``server_id`` is
given, the client uses the configuration to find or start a server.
For cloud connections, it first checks for a running server.
server_id: ID of a running cloud server to connect to when its address
is not known.
token: API access token for cloud servers. Create a token on the
`JCMoptimizer Cloud <https://optimizer.jcmwave.com/cloud/tokens/list>`_
website. If omitted, the token is read from the :ref:`Configuration`.
verbose: If true, print messages from the optimization server.
check: If true, check the server connection and version when connecting.
cloud_endpoint: Address of the cloud API. Usually leave this unchanged.
asynchronous: If true, use ``await`` for calls that contact the server.
Connection and server startup begin when entering ``async with`` or
making the first awaited call. Default: false.
persistent: If true, a server started by this client keeps running after
the client closes. Servers that were already running are always left
running. Default: false.
"""
async def _async_check_server(self) -> None:
"""Checks if the optimization server is running and if it is compatible to the client.
Example::
>>> client.check_server()
Polling server at http://localhost:4554
Optimization server is running
"""
if self.requestor.verbose:
inform(f"Polling server at {self.requestor.endpoint}")
answer = await self.requestor._async_get("check server")
optimizer_version = str(answer["version"])
if optimizer_version != __version__:
warn(
f"Version mismatch between Python client ({__version__}) and optimization server ({optimizer_version}). This can lead to undefined behavior. Consider upgrading the {('client' if __version__ < optimizer_version else 'server')}."
)
inform(answer["message"])
check_server = operation(_async_check_server, interruptible=False)
async def _async_shutdown_server(self, force: bool = False) -> None:
"""Shuts down the optimization server. Example::
client.shutdown_server()
Args:
force: If true, the optimization server is closed even if a study
is not yet finished.
"""
inform(f"Server at {self.requestor.endpoint} is shutting down.")
try:
await self.requestor._async_post(
"shutdown server", "server", "shutdown", data={"force": force}
)
self._shutdown_requested = True
except Exception as err:
raise err from err
shutdown_server = operation(_async_shutdown_server, interruptible=False)
async def _async_shutdown_server_at(self, shutdown_time: datetime) -> None:
"""Shuts down the optimization server at a certain time. Example::
import datetime as dt
client.shutdown_server_at(dt.datetime.now() + dt.timedelta(days=1))
Args:
shutdown_time: The time at which the optimization server should be
shut down.
"""
await self.requestor._async_post(
"set server shutdown time",
"server",
"update_shutdown_at",
data={"shutdown_at": shutdown_time.astimezone().isoformat()},
)
shutdown_server_at = operation(_async_shutdown_server_at, interruptible=False)
async def _async_create_study(
self,
design_space: Optional[list[dict[str, Any]]] = None,
environment: Optional[list[dict[str, Any]]] = None,
constraints: Optional[list[dict[str, Any]]] = None,
study_name: Optional[str] = None,
study_id: Optional[str] = None,
project_name: str = "Project",
project_id: str = "project",
save_dir: Optional[str] = None,
driver: Optional[str] = None,
output_precision: float = 1e-10,
dashboard: bool = True,
open_browser: Optional[bool] = None,
clear_storage: bool = False,
extend_on_request: Optional[timedelta] = None,
extend_on_suggestion: Optional[timedelta] = None,
**kwargs: Any,
) -> Study:
"""Creates a new :class:`~jcmoptimizer.Study` instance. Example::
study = client.create_study(
design_space=design_space,
study_name='example',
study_id="example_01"
)
Args:
design_space: List of domain definitions for each parameter. A design space
definition consists of a list of dictionary with the entries
:name: Name of the parameter. E.g. ``'x1'``. The name should contain
no spaces and must not be equal to function names or mathematical
constants like ``'sin'``, ``'cos'``, ``'exp'``, ``'pi'`` etc.
:type: Type of the parameter. Either ``'continuous'``, ``'discrete'``,
or ``'categorial'``.
:domain: The domain of the parameter. For continuous parameters this
is a tuple (min, max). For discrete parameters this is a list
of values, e.g. ``[1.0,2.5,3.0]``. For categorial inputs it is a list
of strings, e.g. ``['cat1','cat2','cat3']``. Note, that categorial
values are internally mapped to integer representations, which
are allowed to have a correlation. The categorial values should
therefore be ordered according to their similarity.
For fixed parameters the domain is a single parameter value.
Example::
design_space = [
{'name': 'x1', 'type': 'continuous', 'domain': (-2.0,2.0)},
{'name': 'x2', 'type': 'continuous', 'domain': (-2.0,2.0)},
{'name': 'x3', 'type': 'discrete', 'domain': [-1.0,0.0,1.0]},
{'name': 'x4', 'type': 'categorial', 'domain': ['a','b']}
]
If not specified, the design space configuration from the study history
is used. If no historic information exists, an error is raised.
environment: Environment parameters are those which influence the
behaviour of the system, but are not design parameters.
Examples are uncontrollable environment parameters (e.g. temperature, time)
or parameters that are scanned in each evaluation (e.g. wavelength, angle)
for each run of the system can be described by environment parameters.
Alternatively, scans can be described by surrogates that are trained on multiple
inputs (one for each scan value). The environment definition consists of a
list of dictionary with the entries:
:name: Name of the parameter. E.g. ``'x1'``. The name should contain
no spaces and must not be equal to function names or mathematical
constants like ``'sin'``, ``'cos'``, ``'exp'``, ``'PI'`` etc.
:type: Type of the parameter. Either ``'variable'`` or ``'fixed'``.
Numeric fixed parameters can be used in constraint functions or other
expressions. String fixed parameters are labels passed to evaluators;
they are excluded from numeric expressions. Use a new study ID
when changing a fixed string value.
:domain: The domain of the parameter. For fixed parameters, this is a
single numeric or string value; for variable parameters this can be a tuple
(min, max). If no bounds are specified
for variable parameters, the environment is considered to be
unconstrained. In this case the scale has to be set.
:scale: The scale at which environment parameter
changes affect the system. If the environment parameter
describes unknown drifts and aging effects, the length scale
is equal to the timescale at which the system behaviour
changes due to drifts or aging.
Example::
environment = [
{'name': 'wavelength', 'type': 'continuous', 'domain': (300, 600)},
{'name': 'time', 'type': 'continuous', 'scale': 0.1},
{'name': 'radius', 'type': 'fixed', 'domain': 1.5}
]
If not specified, the environment configuration from the study history
is used. If no historic information exists, the environment is
assumed to be empty (``environment = []``).
constraints: List of constraints on the design space. Each list element is a
dictionary with the entries
:name: Name of the constraint.
:constraint: A string defining an inequality constraint in the
for `a <= b` or `a >= b`.
The following operations and functions may be for example used:
+,-,*,/,^,sqrt,sin,cos,tan,exp,log,log10,abs,round,sign,trunc.
E.g. ``'x1^2 + x2^2 <= sin(x1+x2)'``.
For a more complete list of supported functions,
see the :ref:`expression variable reference
<ActiveLearning.Expression.expression>`.
Example::
constraints = [
{'name': 'circle', 'expression': 'x1^2 + x2^2 <= radius^2'},
{'name': 'triangle', 'expression': 'x1 >= x2'},
]
If not specified, the constraints configuration from the study history
is used. If no historic information exists, the list of constraints is
assumed to be empty (``constraints = []``).
study_name: The name of the study.
study_id: A unique identifier of the study. All relevant information on
the study are saved in a file named study_id+'.jcmo'
If the study already exists, the ``design_space``, ``environment``, and
``constraints`` do not need to be provided. If not set, the study_id is set to
a random unique string.
project_name: The name of the project that the study belongs to.
.. note:: This parameter is only relevant when connecting to
a cloud-based server.
project_id: A unique identifier of the project.
.. note:: This parameter is only relevant when connecting to
a cloud-based server.
save_dir: The path to a directory, where the study file (jcmo-file)
is saved. Default is a directory in the system's temporary directory.
.. note:: This parameter is only relevant when connecting to
a self-hosted server.
driver: Driver used for the study. If not specified, the driver from
the study history is used. If no historic information exists,
the 'ActiveLearning' driver is used.
For a list of drivers, see the :ref:`driver reference <DriverReference>`.
output_precision: Precision level for output of parameters.
.. note:: Rounding the output can potentially lead to a slight
breaking of constraints.
dashboard: If true, a dashboard will be served for the study.
open_browser: If true, the study dashboard will be opened in the default
web browser. If not specified, the value is retrieved from the
:ref:`Configuration` file.
clear_storage: If true, any stored data (study configuration, observations
etc.) from previous runs of the study is cleared before creating the
study.
extend_on_request: If specified, the shutdown time of a cloud-based
server is extended in each request, such that the server will always run
for at least the specified amount of time after the request.
If not specified, the value is retrieved from the
:ref:`Configuration` file. Default: 5 minutes. Example::
study = client.create_study(
study_id="my-study",
extend_on_request=timedelta(minutes=10)
)
extend_on_suggestion: If specified, the shutdown time of a cloud-based
server is extended in each suggestion requested with ``study.get_suggestion()``,
such that the server will always run for at least the specified amount of time
after the request. If not specified, the value is retrieved from the
:ref:`Configuration` file. Default: 30 minutes. Example::
study = client.create_study(
study_id="my-study",
extend_on_suggestion=timedelta(hours=1, minutes=30)
)
"""
if "domain" in kwargs:
raise ValueError(
"The 'domain' argument is deprecated. Please use the 'design_space' argument instead."
)
if "name" in kwargs:
raise ValueError(
"The 'name' argument is deprecated. Please use the 'study_name' argument instead."
)
if extend_on_request is None:
if "extend_on_request" in self.config:
extend_on_request = self.config["extend_on_request"]
if not isinstance(extend_on_request, timedelta):
raise TypeError(
"Invalid format of 'extend_on_request' in config file"
)
else:
extend_on_request = timedelta(minutes=5)
if extend_on_suggestion is None:
if "extend_on_suggestion" in self.config:
extend_on_suggestion = self.config["extend_on_suggestion"]
if not isinstance(extend_on_suggestion, timedelta):
raise TypeError(
"Invalid format of 'extend_on_suggestion' in config file"
)
else:
extend_on_suggestion = timedelta(minutes=30)
answer = await self.requestor._async_post(
"create study",
"study",
"create",
data={
"study_id": study_id,
"study_name": study_name,
"project_id": project_id,
"project_name": project_name,
"design_space": design_space,
"environment": environment,
"constraints": constraints,
"driver": driver,
"save_dir": save_dir,
"output_precision": output_precision,
"dashboard": dashboard,
"clear_storage": clear_storage,
},
)
driver_class: type[Driver] = getattr(drivers, answer["driver"])
driver_instance = driver_class(
study_id=answer["study_id"],
project_id=answer["project_id"],
requestor=self.requestor,
)
if open_browser is None:
open_browser = self.config.get("open_browser", True)
if answer["dashboard_path"]:
dashboard_path = self.requestor.endpoint + answer["dashboard_path"]
inform(f"The dashboard is accessible via {dashboard_path}")
if open_browser:
webbrowser.open(dashboard_path)
study: Study[Mode] = Study(
study_id=answer["study_id"],
project_id=answer["project_id"],
driver=driver_instance,
requestor=self.requestor,
extend_on_request=extend_on_request,
extend_on_suggestion=extend_on_suggestion,
)
self._resources.append(study)
return study
create_study = operation(_async_create_study, interruptible=False)
async def _async_create_benchmark(
self,
benchmark_id: Optional[str] = None,
num_average: int = 6,
remove_completed_studies: bool = False,
) -> Benchmark:
"""Creates a new :class:`~jcmoptimizer.Benchmark` object for benchmarking
different optimization studies against each other. Example::
benchmark = client.create_benchmark(num_average=6)
Args:
benchmark_id: A unique identifier of the benchmark.
num_average: Number of study runs to determine average
study performance
remove_completed_studies: If true, studies that are completed
(i.e. some stopping criterion was met) are removed from the server
provided no other client is still using the study.
"""
answer = await self.requestor._async_post(
"create benchmark",
"benchmark",
"create",
data={
"benchmark_id": benchmark_id,
"num_average": num_average,
"remove_completed_studies": remove_completed_studies,
},
)
benchmark: Benchmark[Mode] = Benchmark(
benchmark_id=answer["benchmark_id"],
num_average=num_average,
requestor=self.requestor,
)
self._resources.append(benchmark)
return benchmark
create_benchmark = operation(_async_create_benchmark, interruptible=False)
async def _async_get_drivers(self) -> tuple[dict[str, str], str]:
out = await self.requestor._async_post(
"get available drivers", "server", "get_drivers", data={}
)
return out["drivers"]
async def _async_get_driver_config_rsts(
self, driver_name: str, language: str
) -> tuple[str, dict[str, str]]:
out = await self.requestor._async_post(
"get driver rst info",
"server",
"get_driver_config_rsts",
data={"driver_name": driver_name, "language": language},
)
return (out["config_rst"], out["submodule_config_rst_dict"])
def __init__(
self,
host: Optional[str] = None,
server_id: Optional[str] = None,
token: Optional[str] = None,
verbose: bool = True,
check: bool = True,
cloud_endpoint: str = "https://optimizer.jcmwave.com/cloud/api/v1/",
*,
asynchronous: bool = False,
persistent: bool = False,
) -> None:
self._runtime = Runtime(asynchronous)
self.config = read_yml_config_files()
self._options: dict[str, Any] = dict(
host=host,
server_id=server_id,
token=token,
verbose=verbose,
check=check,
cloud_endpoint=cloud_endpoint,
persistent=persistent,
)
self._connected = False
self._connect_lock: Optional[asyncio.Lock] = None
self._close_task: Optional[asyncio.Task[None]] = None
self._resources: list[Union[Study[Mode], Benchmark[Mode]]] = []
self._owned_server = None
self._shutdown_requested = False
if not asynchronous:
try:
self.connect()
except BaseException:
self.close()
raise
register_exit_cleanup(self.close)
async def _async_connect(self) -> None:
"""Connect to the configured server, starting one if needed.
Usually you do not need to call this method: the client connects
automatically. You can call it to check the connection before creating
a study.
"""
if self._connected:
return
if self._connect_lock is None:
self._connect_lock = asyncio.Lock()
async with self._connect_lock:
if self._connected:
return
options = self._options
host, server_id = (options["host"], options["server_id"])
token = (
options["token"]
if options["token"] is not None
else self.config.get("token")
)
endpoint = options["cloud_endpoint"]
from .server import Server, _async_wait_for_server_startup
try:
if (
host is None
and token is not None
and (
server_id is not None
or self.config.get("server_location") != "local"
)
):
cloud = CloudRequestor(endpoint, token, runtime=self._runtime)
if server_id is not None:
response = await cloud._async_request(
"GET", "get server url", f"servers/{server_id}/"
)
host = response.json()["url"]
else:
response = await cloud._async_request(
"GET", "get server list", "servers/"
)
servers = response.json()
if servers:
server_id, host = (
servers[0]["server_id"],
servers[0]["url"],
)
if host is not None:
await _async_wait_for_server_startup(
endpoint, token, server_id, host, self._runtime,
verbose=options["verbose"],
)
elif host is None and server_id is not None:
raise ValueError("A cloud server ID requires a token.")
if host is None:
server = Server.__new__(Server)
server._runtime = self._runtime
self._owned_server = server
await server._async_initialize(
token=token,
cloud_endpoint=endpoint,
persistent=options["persistent"],
verbose=options["verbose"],
)
host = server.host
if server.server_location == "local":
token = None
self.requestor = OptimizerRequestor(
host=host,
token=token or "selfhosted",
verbose=options["verbose"],
runtime=self._runtime,
)
if options["check"]:
await self._async_check_server()
self._connected = True
except BaseException:
if self._owned_server is not None:
await self._owned_server._async_abort_startup()
self._owned_server = None
raise
connect = operation(_async_connect)
async def _async_close(self) -> None:
"""Close the connection when you have finished using the client.
Any running operations started through this client are stopped. If the
client started a server, it also stops that server unless
``persistent=True``. A server that was already running is left running.
A ``with`` or ``async with`` block calls this method automatically.
Otherwise, call ``client.close()`` or ``await client.close()`` when
using ``asynchronous=True``.
"""
if self._close_task is None:
self._close_task = asyncio.create_task(self._close_impl())
await settle(self._close_task)
async def _close_impl(self) -> None:
self._runtime.closing = True
await self._runtime.cancel_pending()
try:
if self._connected and (not self._shutdown_requested):
for resource in reversed(self._resources):
try:
await resource._delete_on_server()
except (ConnectionError, OSError):
resource.deleted = True
if self._owned_server is not None and (not self._options["persistent"]):
await self._owned_server._async_shutdown(force=True)
finally:
self._connected = False
for resource in self._resources:
resource.deleted = True
self._resources.clear()
await self._runtime.close_sessions()
if not self._runtime.asynchronous:
unregister_exit_cleanup(self.close)
close = operation(_async_close)
async def _async_cancel_result(self, name: str, result: Any) -> None:
if isinstance(result, (Study, Benchmark)):
await result._delete_on_server()
def __enter__(self) -> "Client[Mode]":
if self._runtime.asynchronous:
raise RuntimeError("Use async with for Client(asynchronous=True).")
return self
def __exit__(self, *exc: Any) -> None:
self.close()
async def __aenter__(self) -> "Client[Mode]":
if not self._runtime.asynchronous:
raise RuntimeError("Use Client(asynchronous=True) in an async context.")
try:
await self.connect()
except BaseException:
if self._close_task is None:
await self._async_close()
raise
return self
async def __aexit__(self, *exc: Any) -> None:
await self._async_close()
_get_drivers = operation(_async_get_drivers, name="_get_drivers")
_get_driver_config_rsts = operation(
_async_get_driver_config_rsts, name="_get_driver_config_rsts"
)