302 lines
12 KiB
Python
302 lines
12 KiB
Python
"""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
|
|
)
|