"""Calibrate a damped harmonic oscillator against the hosted TauOpt service with no package installed: recover the damping, the stiffness (log scale), and the initial velocity from 50 noisy position samples by minimizing the mean squared error between the simulated and the observed trajectory (scipy.integrate.solve_ivp does the simulation). The package-free twin of ``ode_calibration.py``: the same model and data (numpy and scipy installed, nothing else), the ask/observe loop over the REST routes; the engine is handed log10(MSE), the printed values are the raw MSE. 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 ode_calibration_rest.py --seed 7 """ from __future__ import annotations import argparse import json import math import os import sys 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 --- TRUE = {"damping": 0.4, "stiffness": 9.0, "v0": -1.5} T_END, N_SAMPLES, NOISE_STD = 5.0, 50, 0.05 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 try: import numpy as np from scipy.integrate import solve_ivp except ImportError as e: raise ImportError("this example needs numpy and scipy: pip install numpy scipy") from e t_eval = np.linspace(0.0, T_END, N_SAMPLES) def simulate(damping: float, stiffness: float, v0: float): # x'' + damping * x' + stiffness * x = 0 with x(0) = 1 known and # x'(0) = v0 unknown; tight tolerances keep solver error out of the MSE sol = solve_ivp( lambda t, z: [z[1], -damping * z[1] - stiffness * z[0]], (0.0, T_END), [1.0, v0], t_eval=t_eval, rtol=1e-8, atol=1e-10, ) return sol.y[0] # the dataset is fixed so --seed moves only the search, never the data data = simulate(**TRUE) + np.random.RandomState(0).normal(0.0, NOISE_STD, N_SAMPLES) noise_floor = float(np.mean((simulate(**TRUE) - data) ** 2)) def mse(a: dict) -> float: sim = simulate(float(a["damping"]), float(a["stiffness"]), float(a["v0"])) return float(np.mean((sim - data) ** 2)) api = Api(base_url, token) exp = api.call( "POST", "/v1/experiments", { "name": f"oscillator-rest-seed{args.seed}", "parameters": [ {"name": "damping", "type": "double", "bounds": [0.0, 1.0]}, # stiffness sets the frequency; a log scale treats 2 -> 6 and # 10 -> 30 as the same step {"name": "stiffness", "type": "double", "bounds": [2.0, 40.0], "log_scale": True}, {"name": "v0", "type": "double", "bounds": [-3.0, 3.0]}, ], "metric": {"name": "log10_mse", "objective": "minimize"}, "observation_budget": args.budget, "metadata": {"seed": args.seed, "true": TRUE}, }, ) 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 err = mse(sug["assignments"]) api.call( "POST", f"/v1/experiments/{exp_id}/observations", {"suggestion_id": sug["id"], "value": math.log10(err)}, ) shown = {k: round(v, 4) for k, v in sug["assignments"].items()} print(json.dumps({"i": i, "assignments": shown, "mse": round(err, 6), "strategy": sug["provenance"]["strategy"]}, sort_keys=True)) best = api.call("GET", f"/v1/experiments/{exp_id}/best_assignments") assert best["data"], "no successful observations" recovered = best["data"][0]["assignments"] print(json.dumps({"best_mse": round(10 ** best["data"][0]["value"], 6), "noise_floor_mse": round(noise_floor, 6)}, sort_keys=True)) for name in ("damping", "stiffness", "v0"): rel = abs(recovered[name] - TRUE[name]) / abs(TRUE[name]) print(f"{name:>9}: recovered {recovered[name]:8.4f} true {TRUE[name]:8.4f}" f" rel_error {rel:6.1%}") 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())