"""KineJing State PushT client. Python 3.10+, standard library only. One logical rollout keeps one body and Idempotency-Key across bounded retries. No redirects are followed. Keys are never included in client error messages. """ from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timezone from email.utils import parsedate_to_datetime import http.client import ipaddress import json import math import os import socket import ssl import time from typing import Callable, Mapping from urllib.error import HTTPError, URLError from urllib.parse import urlsplit from urllib.request import HTTPRedirectHandler, HTTPSHandler, Request, build_opener import uuid __version__ = "0.2.0" DEFAULT_ORIGIN = "https://kineworld.com" MODEL = "kinejing-state-pusht-v1" MAX_RESPONSE_BYTES = 2 * 1024 * 1024 RETRY_STATUSES = frozenset({429, 502, 503, 504}) @dataclass(frozen=True) class HTTPResponse: status: int headers: Mapping[str, str] body: bytes class KineJingError(Exception): """Bounded diagnostics. Retain idempotency_key for an explicit later retry.""" def __init__(self, message, *, status=None, request_id=None, receipt=None, code=None, idempotency_key=None, retry_after=None): super().__init__(message) self.status = status self.status_code = status self.request_id = request_id self.receipt = receipt self.code = code self.idempotency_key = idempotency_key self.retry_after = retry_after class APIError(KineJingError): """A non-success HTTP response; 409 receipts are available as .receipt.""" class TransportError(KineJingError): """Connection/timeout failure or exhausted time budget; outcome may be unknown.""" class ProtocolError(KineJingError): """An oversized, non-JSON or malformed success response; do not change key blindly.""" class _NoRedirect(HTTPRedirectHandler): def redirect_request(self, request, file_pointer, code, message, headers, new_url): return None def _origin(value): if (not isinstance(value, str) or any(c.isspace() or ord(c) < 32 or ord(c) == 127 for c in value) or "\\" in value or "?" in value or "#" in value): raise ValueError("base_url must be an HTTPS origin or loopback HTTP origin") try: parsed = urlsplit(value) host = parsed.hostname port = parsed.port except ValueError: raise ValueError("Invalid base_url origin") from None if (not host or not host.isascii() or "%" in host or parsed.netloc.endswith(":") or parsed.username is not None or parsed.password is not None or parsed.path not in ("", "/") or parsed.query or parsed.fragment or (port is not None and not 1 <= port <= 65535)): raise ValueError("base_url must contain only scheme, host and optional port") try: loopback = host.lower() == "localhost" or ipaddress.ip_address(host).is_loopback except ValueError: loopback = host.lower() == "localhost" if parsed.scheme != "https" and not (parsed.scheme == "http" and loopback): raise ValueError("HTTPS is required except for explicit loopback HTTP development") return parsed.scheme + "://" + parsed.netloc def _visible_key(value): return isinstance(value, str) and 1 <= len(value) <= 128 and value.isascii() and all(33 <= ord(c) <= 126 for c in value) def _redact(value, secret): if isinstance(value, str): return value.replace(secret, "[REDACTED]") if isinstance(value, dict): return {_redact(str(k), secret): _redact(v, secret) for k, v in value.items()} if isinstance(value, list): return [_redact(item, secret) for item in value] return value def _retry_after(headers): raw = headers.get("retry-after", "").strip() if raw.isascii() and raw.isdigit(): # Very large valid delta-seconds must not turn into an early retry. return float(raw) if len(raw) <= 10 else 10_000_000_000.0 if raw: try: when = parsedate_to_datetime(raw) if when.tzinfo is None: when = when.replace(tzinfo=timezone.utc) return max(0.0, (when - datetime.now(timezone.utc)).total_seconds()) except (ValueError, TypeError, OverflowError): pass return None def _validate_states(context_states, actions): if not isinstance(context_states, (list, tuple)) or len(context_states) != 2: raise ValueError("context_states must contain two states 0.1 seconds apart") if not isinstance(actions, (list, tuple)) or not 1 <= len(actions) <= 16: raise ValueError("actions must contain 1 to 16 absolute XY targets") for rows, width, kind in ((context_states, 5, "state"), (actions, 2, "action")): for row in rows: if not isinstance(row, (list, tuple)) or len(row) != width: raise ValueError("Invalid state or action dimensions") for index, value in enumerate(row): try: finite = type(value) in (int, float) and math.isfinite(value) except OverflowError: finite = False if not finite: raise ValueError("State and action values must be finite JSON numbers") if kind == "action" and not 0 <= value <= 512: raise ValueError("Absolute action targets must be in [0, 512] pixels") if kind == "state" and (abs(value) > 10000 or (index < 4 and not -256 <= value <= 768)): raise ValueError("State values exceed the supported simulator range") class KineJingClient: def __init__(self, api_key=None, *, base_url=DEFAULT_ORIGIN, timeout=20.0, retry_budget=45.0, max_retries=2, transport: Callable | None = None, _clock=time.monotonic, _sleep=time.sleep): key = os.environ.get("KINEJING_API_KEY", "") if api_key is None else api_key if not isinstance(key, str) or not 32 <= len(key) <= 512 or not key.isascii() or any(c.isspace() or ord(c) < 33 or ord(c) > 126 for c in key): raise ValueError("Set KINEJING_API_KEY to your valid server-side API key") if type(max_retries) is not int or not 0 <= max_retries <= 2: raise ValueError("max_retries must be an integer from zero to two") for value in (timeout, retry_budget): if type(value) not in (int, float) or not math.isfinite(value) or value <= 0: raise ValueError("timeout and retry_budget must be positive finite seconds") self.base_url = _origin(base_url) self.timeout = float(timeout) self.retry_budget = float(retry_budget) self.max_retries = max_retries self._api_key = key self._clock = _clock self._sleep = _sleep self._opener = build_opener(_NoRedirect(), HTTPSHandler(context=ssl.create_default_context())) self._transport = transport or self._http def __repr__(self): return "KineJingClient(base_url=" + repr(_redact(self.base_url, self._api_key)) + ")" def _http(self, method, url, headers, body, timeout): request = Request(url, data=body, headers=headers, method=method) try: response = self._opener.open(request, timeout=timeout) except HTTPError as error: response = error with response: data = response.read(MAX_RESPONSE_BYTES + 1) return HTTPResponse(response.status, dict(response.headers), data) def _error(self, response, idem): headers = {str(k).lower(): str(v) for k, v in response.headers.items()} try: data = json.loads(response.body) if len(response.body) <= MAX_RESPONSE_BYTES else {} data = _redact(data, self._api_key) except (ValueError, UnicodeError, TypeError, RecursionError): # A small body can still exceed the parser or redaction recursion # limit. Keep the HTTP/header diagnostics and original recovery key. data = {} if not isinstance(data, dict): data = {} nested = data.get("error") detail = data.get("detail") if not isinstance(detail, str): detail = nested.get("message") if isinstance(nested, dict) else nested if not isinstance(detail, str): detail = "KineJing returned HTTP " + str(response.status) request_id = headers.get("x-request-id") or data.get("request_id") if not isinstance(request_id, str): request_id = None return APIError(detail[:1000], status=response.status, request_id=_redact(request_id, self._api_key), receipt=data.get("receipt", data.get("usage_receipt")), code=(nested.get("code") or data.get("code") or nested.get("type")) if isinstance(nested, dict) else data.get("code"), idempotency_key=_redact(idem, self._api_key), retry_after=_retry_after(headers)) def rollout(self, context_states, actions, *, idempotency_key=None): """Return predicted_states and receipt. Angles are radians, XY are raw pixels.""" _validate_states(context_states, actions) idem = uuid.uuid4().hex if idempotency_key is None else idempotency_key if not _visible_key(idem): raise ValueError("Idempotency-Key must contain 1 to 128 visible ASCII characters") if self._api_key in idem: raise ValueError("Do not use an API key as the Idempotency-Key") payload = {"model": MODEL, "state_schema": "pusht-512xy-angle-rad-v1", "action_schema": "pusht-absolute-xy-10hz-v1", "context_states": context_states, "actions": actions} return self._prediction(payload, "/v1/rollouts", MODEL, "predicted_states", len(actions), 5, idem) def actions(self, image_png_base64, wrist_png_base64, state8, *, acknowledge_experimental=False, idempotency_key=None): """Fixed-task simulation actions. A delivered response is charged; success is unverified.""" if acknowledge_experimental is not True: raise ValueError("Set acknowledge_experimental=True for this simulation-only model") if not isinstance(state8,list) or len(state8)!=8 or any(type(x) not in (int,float) or not math.isfinite(x) or abs(x)>10000 for x in state8): raise ValueError("state8 requires eight finite simulator values") for value in (image_png_base64,wrist_png_base64): if not isinstance(value,str) or not 1<=len(value)<=349528: raise ValueError("Supply two base64 RGB256 PNG images") idem=uuid.uuid4().hex if idempotency_key is None else idempotency_key if not _visible_key(idem) or self._api_key in idem: raise ValueError("Use a separate visible Idempotency-Key, never the API key") payload={"model":"kine-a1", "task":"Pick up the black bowl between the rectangular board and the small ramekin, and place it on the plate.", "image_orientation":"robosuite_raw_unflipped", "state_schema":"robosuite-eef-pos-axis-angle-gripper-qpos-8d-v1", "images":{"image":image_png_base64,"image2":wrist_png_base64}, "state8":state8,"acknowledge_experimental":True} return self._prediction(payload,"/v1/actions",payload["model"],"actions",50,7,idem) def _prediction(self,payload,endpoint,expected_model,rows_key,row_count,row_width,idem): body = json.dumps(payload, allow_nan=False, separators=(",", ":")).encode("utf-8") headers = {"Authorization": "Bearer " + self._api_key, "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "kinejing-python/" + __version__, "Idempotency-Key": idem, "X-KineJing-Replay": "recover"} deadline = self._clock() + self.retry_budget last_error = None for attempt in range(self.max_retries + 1): remaining = deadline - self._clock() if remaining <= 0: break try: response = self._transport("POST", self.base_url + endpoint, dict(headers), body, min(self.timeout, remaining)) except ssl.SSLError: raise TransportError("TLS verification or protocol failure; request was not retried", idempotency_key=_redact(idem, self._api_key)) from None except (URLError, TimeoutError, ConnectionError, socket.timeout, http.client.IncompleteRead, http.client.RemoteDisconnected) as error: last_error = TransportError("Connection or timeout failure; retain the same Idempotency-Key for recovery", idempotency_key=_redact(idem, self._api_key)) # Certificate/protocol errors do not improve through retry. if isinstance(error, URLError) and isinstance(error.reason, ssl.SSLError): raise last_error from None delay = float(2 ** attempt) except http.client.HTTPException: # A bad status line/header may arrive after server settlement. # Keep its recovery identity, without exposing the raw response. raise ProtocolError("Invalid HTTP response; retain the same Idempotency-Key for recovery", idempotency_key=_redact(idem, self._api_key)) from None else: if not isinstance(response, HTTPResponse) or type(response.status) is not int: raise ProtocolError("Transport returned an invalid HTTP response", idempotency_key=_redact(idem, self._api_key)) if 200 <= response.status < 300: try: if len(response.body) > MAX_RESPONSE_BYTES: raise ValueError("oversized response") result = json.loads(response.body) if not isinstance(result, dict) or result.get("model") != expected_model: raise ValueError("response model does not match request") states = result.get(rows_key) if not isinstance(states, list) or len(states) != row_count: raise ValueError("predicted state count does not match actions") for row in states: if not isinstance(row, list) or len(row) != row_width: raise ValueError("invalid predicted state dimensions") for value in row: if type(value) not in (int, float) or not math.isfinite(value): raise ValueError("predicted states must be finite JSON numbers") if rows_key=="actions" and abs(value)>1: raise ValueError("action outside normalized controller range") result = _redact(result, self._api_key) except (ValueError, UnicodeError, TypeError, OverflowError, RecursionError): diagnostics = self._error(response, idem) raise ProtocolError("KineJing returned an invalid rollout response; retain the same Idempotency-Key", status=response.status, request_id=diagnostics.request_id, receipt=diagnostics.receipt, code=diagnostics.code, idempotency_key=diagnostics.idempotency_key) from None response_headers = {str(k).lower(): str(v) for k, v in response.headers.items()} attempt_id = response_headers.get("x-request-id") or result.get("request_id") if isinstance(attempt_id, str): result["_request_id"] = _redact(attempt_id, self._api_key) return result last_error = self._error(response, idem) if response.status not in RETRY_STATUSES: raise last_error delay = last_error.retry_after if last_error.retry_after is not None else float(2 ** attempt) if attempt >= self.max_retries or delay >= deadline - self._clock(): break if delay > 0: self._sleep(delay) if last_error is not None: raise last_error from None raise TransportError("Total retry time budget exhausted", idempotency_key=_redact(idem, self._api_key))