"""A lab response surface with measurement noise against the hosted TauOpt service with no package installed: maximize a quadratic bowl with an interaction term under Gaussian measurement error, reporting the instrument's known repeatability through ``value_stddev`` on every observation, with ``num_solutions: 3`` so best_assignments returns three diverse good settings. The package-free twin of ``experimental_design.py``: the same hidden surface and noise, the ask/observe loop over the REST routes. The reported stddev enters the GP as fixed per-point noise variance. The seed rides every ask as an explicit ``config`` with ``timebox_mode: "caps"`` (deterministic work ceilings, so the same seed replays the same sequence on the service; the service refuses ``"off"`` for a non-exempt user). Needs ``TAUOPT_BASE_URL`` and ``TAUOPT_TOKEN`` (the API https://api.tauopt.com; a token minted on the web app's tokens page); without them it prints one line and exits 0. python experimental_design_rest.py --seed 7 """ from __future__ import annotations import argparse import json import os import sys import random import time import urllib.error import urllib.request # --- rest helper (identical in every *_rest.py example; tests/test_rest_examples.py pins it) --- class ApiError(RuntimeError): """A non-2xx answer: the service's one error shape, {"error": {"type", "message", "details"}}.""" def __init__(self, status: int, body: dict, headers) -> None: err = body.get("error") or {} self.status = status self.error_type = err.get("type", "http_error") self.details = err.get("details") ra = headers.get("Retry-After") self.retry_after = float(ra) if ra and ra.replace(".", "", 1).isdigit() else None super().__init__(f"[{status}] {self.error_type}: {err.get('message', '')}") class _NoRedirect(urllib.request.HTTPRedirectHandler): """Never follow a redirect: a 301 from http:// to https:// would turn a POST into a GET.""" def redirect_request(self, *args, **kwargs): return None class Api: """The hosted REST API over the standard library: a bearer token, JSON in and out.""" def __init__( self, base_url: str, token: str, timeout_s: float = 120.0, wait_s: float = 600.0 ) -> None: self._base = base_url.rstrip("/") self._token = token self._timeout = timeout_s self._wait = wait_s # the budget per call for the waits a 429 or a 503 names self._opener = urllib.request.build_opener(_NoRedirect) self.last_exhaustion: dict | None = None # the 409 space_exhausted details ask() last saw def call( self, method: str, path: str, body: dict | None = None, wait_s: float | None = None ) -> dict | None: """One request. A non-2xx answer raises ApiError (a 429 carries retry_after); 204 returns None; a body that is not JSON raises ApiError whatever the status (the service answers JSON or 204; anything else is a proxy's page). A 429 that names a wait in Retry-After (``bandwidth_full``, a quota, the rate limit) and a 503 ``experiment_busy`` are waited out, the named interval (at least 1 s) in sleeps of at most 30 s, and re-sent, within a budget per call: the constructor's ``wait_s`` unless the call passes its own (a script with a deadline passes what is left of it). A wait that would end past the budget raises the answer instead, with no further request. ``experiment_quota_exceeded`` carries no Retry-After and raises at once. A failure before the answer is retried twice (see ``_send``); a timeout after the request was sent is raised as is. """ budget = self._wait if wait_s is None else wait_s started = time.monotonic() while True: try: return self._send(method, path, body) except ApiError as e: busy = e.status == 503 and e.error_type == "experiment_busy" if e.retry_after is None or not (e.status == 429 or busy): raise left = max(e.retry_after, 1.0) # a Retry-After of 0 is never a tight loop if time.monotonic() - started + left > budget: raise while left > 0: time.sleep(min(left, 30.0)) left -= 30.0 def _send(self, method: str, path: str, body: dict | None) -> dict | None: req = urllib.request.Request( self._base + path, data=None if body is None else json.dumps(body).encode(), method=method, headers={ "Authorization": f"Bearer {self._token}", "Content-Type": "application/json", "Accept": "application/json", }, ) for attempt in range(3): try: with self._opener.open(req, timeout=self._timeout) as r: status, raw, headers = r.status, r.read(), r.headers break except urllib.error.HTTPError as e: # every non-2xx answer, the 3xx included status, raw, headers = e.code, e.read(), e.headers break except urllib.error.URLError: # a failure before the answer: DNS, connect, the TLS handshake, or the send # itself (urllib wraps all of them in URLError; whether any byte reached the # service is unknowable here, so a resend of a landed POST is possible in # principle) -- retried twice; a timeout while WAITING for the answer is a # bare TimeoutError, not caught here, because the request landed: list the # open suggestions first if attempt == 2: raise time.sleep(1.0 + attempt) if status == 204: return None try: data = json.loads(raw) except ValueError: data = {"error": {"type": "http_error", "message": raw[:200].decode(errors="replace")}} raise ApiError(status, data, headers) from None # a 2xx that is not JSON too if status >= 300: raise ApiError(status, data, headers) return data def ask( api: Api, exp_id, body: dict | None = None, wait_s: float | None = None ) -> dict | None: """One suggestion, or None on proven exhaustion (a completion, not an error; the 409's details stay on ``api.last_exhaustion``). ``body`` is the optional SuggestionIn (a ``config``, ``max_wait_s``). A 429 that names a wait and a 503 ``experiment_busy`` are waited out by ``Api.call`` (``wait_s`` is this call's budget); ``experiment_quota_exceeded`` carries no wait: archive an experiment first. """ try: return api.call( "POST", f"/v1/experiments/{exp_id}/suggestions", body or {}, wait_s=wait_s ) except ApiError as e: if e.status == 409 and e.error_type == "space_exhausted": api.last_exhaustion = e.details return None raise def archive(api: Api, exp_id, wait_s: float | None = None) -> None: """DELETE the experiment; a 404 means it is already archived (the call is idempotent). ``wait_s`` is this call's budget for a wait the answer names.""" try: api.call("DELETE", f"/v1/experiments/{exp_id}", wait_s=wait_s) except ApiError as e: if e.error_type != "not_found": raise # --- end rest helper --- NOISE_STD = 0.5 # one measurement's repeatability, known from instrument calibration OPTIMUM = {"temp": 0.4, "conc": -0.2} # coded units; the true response there is 85.0 def response(temp: float, conc: float) -> float: """The hidden true surface in coded factor units: a bowl with a cross term.""" u, v = temp - OPTIMUM["temp"], conc - OPTIMUM["conc"] return 85.0 - 7.0 * u**2 - 5.0 * v**2 + 3.0 * u * v def main() -> int: ap = argparse.ArgumentParser(description=__doc__.strip().split("\n\n")[0]) ap.add_argument( "--seed", type=int, default=7, help="the seed every ask's config carries, so a rerun replays the same points" " (default %(default)s)", ) ap.add_argument( "--budget", type=int, default=25, help="ask/observe rounds before the result is read (default %(default)s)", ) ap.add_argument("--keep", action="store_true", help="leave the experiment active at the end") args = ap.parse_args() base_url, token = os.environ.get("TAUOPT_BASE_URL"), os.environ.get("TAUOPT_TOKEN") if not base_url or not token: pair = (("TAUOPT_BASE_URL", base_url), ("TAUOPT_TOKEN", token)) missing = "/".join(name for name, value in pair if not value) print(f"{missing} not set; skipping" " (export TAUOPT_BASE_URL=https://api.tauopt.com and a TAUOPT_TOKEN minted at app.tauopt.com/settings/tokens)") return 0 if args.budget < 3: ap.error("--budget must be at least 3 (best returns three solutions)") # the measurement noise is seeded too, so one seed replays one run exactly rng = random.Random(args.seed) api = Api(base_url, token) exp = api.call( "POST", "/v1/experiments", { "name": f"response-surface-rest-seed{args.seed}", "parameters": [ {"name": "temp", "type": "double", "bounds": [-1.0, 1.0]}, {"name": "conc", "type": "double", "bounds": [-1.0, 1.0]}, ], "metric": {"name": "response", "objective": "maximize"}, # three diverse top settings from best_assignments: single-objective only "num_solutions": 3, "observation_budget": args.budget, "metadata": {"seed": args.seed, "noise_std": NOISE_STD}, }, ) exp_id = exp["id"] try: # caps: deterministic work ceilings, so same-seed runs match on the service config = {"strategy": "auto", "seed": args.seed, "timebox_mode": "caps"} for i in range(args.budget): sug = ask(api, exp_id, body={"config": config}) if sug is None: break a = sug["assignments"] measured = response(a["temp"], a["conc"]) + rng.gauss(0.0, NOISE_STD) # the stddev rides with the value: the GP treats it as known noise # variance at this point instead of inferring one noise level for all api.call( "POST", f"/v1/experiments/{exp_id}/observations", {"suggestion_id": sug["id"], "value": measured, "value_stddev": NOISE_STD}, ) print(json.dumps({"i": i, "assignments": {k: round(v, 4) for k, v in a.items()}, "measured": round(measured, 3), "value_stddev": NOISE_STD, "strategy": sug["provenance"]["strategy"]}, sort_keys=True)) best = api.call("GET", f"/v1/experiments/{exp_id}/best_assignments") assert best["count"] == 3, f"expected three solutions, got {best['count']}" print(json.dumps({"solutions": best["count"], "true_optimum": OPTIMUM, "true_optimum_response": 85.0}, sort_keys=True)) for rank, entry in enumerate(best["data"], start=1): a = entry["assignments"] true_response = response(a["temp"], a["conc"]) print(f"rank {rank}: temp={a['temp']:.4f} conc={a['conc']:.4f}" f" measured={entry['value']:.3f} true_response={true_response:.3f}") finally: if not args.keep: failing = sys.exc_info()[1] # the error above, if any, stays the one reported try: archive(api, exp_id) print(f"experiment {exp_id} archived") except Exception as e: # noqa: BLE001 -- never hide the error above if failing is None: raise print(f"experiment {exp_id} not archived: {e}", file=sys.stderr) return 0 if __name__ == "__main__": raise SystemExit(main())