"""Hardened adapter over the MIT-0 guooscar 1.2.0 device/upload methods. No UI, scheduler, renderer, environment lookup, or constructor cloud calls. Retry only read-only requests or writes proven unsent/rejected; never replay ambiguous writes. """ import threading import time from collections import deque from urllib.parse import quote import requests from urllib3.exceptions import MaxRetryError, NewConnectionError from pigway_frame_flow.errors import ErrorCode as E from pigway_frame_flow.errors import FlowError from vendor import guooscar_protocol BASE_URL = "https://api.kokonna.art/openapi" class SafeTransport: MAX_ATTEMPTS = 3 MAX_RETRY_DELAY = 60 def __init__(self, key, session=None): self.key = key self.session = session or requests.Session() self.calls: deque[float] = deque() self.lock = threading.Lock() self.cooldown_until = 0.0 def request(self, endpoint, payload=None, write=False, binary=False): for attempt in range(1, self.MAX_ATTEMPTS + 1): try: return self._request_once(endpoint, payload, write, binary) except FlowError as exc: exc.details["attempts"] = attempt delay = max( 2 ** (attempt - 1), exc.details.get("retry_after_seconds", 0) ) if ( not exc.details.get("retryable") or attempt == self.MAX_ATTEMPTS or delay > self.MAX_RETRY_DELAY ): if exc.details.get("retryable"): exc.details["retry_exhausted"] = attempt == self.MAX_ATTEMPTS exc.details["retry_deferred"] = delay > self.MAX_RETRY_DELAY raise time.sleep(delay) raise AssertionError("Retry loop exhausted") def _reserve(self): with self.lock: current = time.monotonic() while self.calls and current - self.calls[0] >= 60: self.calls.popleft() delay = max(0.0, self.cooldown_until - current) if len(self.calls) >= 18: delay = max(delay, 60 - (current - self.calls[0])) if delay: raise FlowError( E.RATE_LIMITED, "Device request budget exhausted", 429, details={ "retryable": True, "retry_after_seconds": delay, "write_outcome": "not_sent", "failure_kind": "local_rate_limit", }, ) self.calls.append(current) @staticmethod def _retry_after(value): from datetime import UTC, datetime from email.utils import parsedate_to_datetime try: seconds = float(value) except (ValueError, TypeError): try: seconds = ( parsedate_to_datetime(value) - datetime.now(UTC) ).total_seconds() except (ValueError, TypeError, OverflowError): return 60.0 import math return max(0.0, seconds) if math.isfinite(seconds) else 60.0 def _request_once(self, endpoint, payload, write, binary): if not self.key: raise FlowError(E.CONFIG_REQUIRED, "KoKonna configuration required", 503) self._reserve() try: response = self.session.request( "GET" if binary else "POST", BASE_URL + endpoint, headers={} if binary else { "Authorization": "Bearer " + self.key, "Content-Type": "application/json", }, json=None if binary else payload or {}, timeout=(5, 120 if endpoint == "/upload" else 30), allow_redirects=False, ) except requests.RequestException as exc: # Never retain exception text or a URL: binary URLs contain the key. category = next( ( label for kind, label in ( (requests.ConnectTimeout, "connect_timeout"), (requests.ReadTimeout, "read_timeout"), (requests.exceptions.SSLError, "tls_error"), (requests.ConnectionError, "connection_error"), (requests.Timeout, "timeout"), ) if isinstance(exc, kind) ), "request_error", ) # A socket/DNS establishment failure precedes the HTTP write. A # dropped established connection still has an uncertain outcome. not_sent = isinstance(exc, requests.ConnectTimeout) or any( isinstance(cause, MaxRetryError) and isinstance(cause.reason, NewConnectionError) for cause in exc.args ) if not_sent and category == "connection_error": category = "connect_error" uncertain = write and not not_sent raise FlowError( E.WRITE_RESULT_UNKNOWN if uncertain else E.CLOUD_UNAVAILABLE, "Cloud write outcome is unknown" if uncertain else "Device cloud is unavailable", 502, details={ "transport_error": category, "retryable": (not write and category != "tls_error") or not_sent, "write_outcome": "unknown" if uncertain else ("not_sent" if write else "read_only"), }, ) from None try: value = None if not binary or response.status_code != 200: try: value = response.json() except ValueError: pass message = value.get("message") if isinstance(value, dict) else None status = response.status_code # Only exact documented business failures establish rejection. Generic # 5xx responses may arrive after a write; those remain UNKNOWN. business = None if status in (404, 500) and isinstance(message, str): if message == "image not found": business = ( E.NOT_FOUND, "Cloud image not found", 404, "image_not_found", ) elif status == 500 and message == "imageId is empty": business = ( E.INVALID_REQUEST, "Cloud image ID is invalid", 400, "invalid_image_id", ) elif status == 500 and message.startswith("can not find robot "): business = ( E.CONFIG_REQUIRED, "Device key does not identify a frame", 503, "device_key_invalid", ) elif status == 404 and message.startswith("can not find method "): business = ( E.UNSUPPORTED_CAPABILITY, "Cloud method not found", 422, "method_not_found", ) if business: code, safe_message, local_status, kind = business raise FlowError( code, safe_message, local_status, details={ "http_status": status, "failure_kind": kind, "retryable": False, "write_outcome": "rejected" if write else "read_only", }, ) if status != 200: limited = status == 429 rejected = 400 <= status < 500 uncertain = write and not rejected details = { "http_status": status, "retryable": limited or (not write and status >= 500), "write_outcome": "unknown" if uncertain else ("rejected" if write else "read_only"), } if limited: delay = self._retry_after(response.headers.get("Retry-After")) details["retry_after_seconds"] = delay with self.lock: self.cooldown_until = max( self.cooldown_until, time.monotonic() + delay ) raise FlowError( E.RATE_LIMITED if limited else (E.WRITE_RESULT_UNKNOWN if uncertain else E.CLOUD_UNAVAILABLE), "Device cloud request failed", 429 if limited else 502, details=details, ) if binary: if len(response.content) > 20 * 1024 * 1024: raise FlowError(E.INVALID_REQUEST, "Cloud image exceeds size limit") return response.content if ( not isinstance(value, dict) or value.get("success") is False or ("message" in value and "id" not in value) ): raise FlowError( E.WRITE_RESULT_UNKNOWN if write else E.CLOUD_UNAVAILABLE, "Invalid device cloud response", 502, details={ "failure_kind": "invalid_response", "retryable": not write, "write_outcome": "unknown" if write else "read_only", }, ) return value finally: response.close() class KokonnaClient: def __init__(self, key, write_enabled=False, session=None): self.transport = SafeTransport(key, session) self.write_enabled = write_enabled def guard(self): if not self.write_enabled: raise FlowError( E.WRITE_DISABLED, "Real device writes require explicit authorization", 403, ) def get_device(self): return guooscar_protocol.device(self.transport) def list_images(self): return self.transport.request("/listImages", {}) def upload(self, path, name): self.guard() value = guooscar_protocol.upload(self.transport, path.read_bytes(), name) if type(value.get("id")) is not int or value["id"] <= 0: raise FlowError(E.WRITE_RESULT_UNKNOWN, "Upload ID not confirmed", 502) return value["id"] def display(self, image_id): self.guard() return self.transport.request( "/displayImageById", {"imageId": image_id}, write=True ) def delete(self, image_id): self.guard() return self.transport.request("/deleteImage", {"imageId": image_id}, write=True) def download(self, image_id): return self.transport.request( f"/image/{quote(self.transport.key, safe='')}/{image_id}", binary=True ) def display_by_name(self, name): self.guard() return self.transport.request( "/displayImageByName", {"imageName": name}, write=True )