Presa-Mano/src/presa_mano/clients/open3dmodel.py

"""Client for the Open3DModel pose-estimation service.

Open3DModel API (see /opt/docker/robotserver/Open3DModel/README.md):
- GET    /api/recipes               -> list
- GET    /api/recipes/{id}/stl      -> STL bytes (HTTPException 404 if missing)
- POST   /api/pose                  -> envelope {success, data, error}
                                       data has transform_4x4, fitness,
                                       inlier_rmse_mm, translation_mm, ...
- Header X-Api-Key richiesto (403 con detail testuale altrimenti).
"""
from __future__ import annotations

import hashlib
from typing import Any

import httpx
import numpy as np


class Open3DModelUnavailable(RuntimeError):
    """Transport-level failure (connect refused, timeout)."""


class Open3DModelError(RuntimeError):
    """Semantic failure: envelope error code, HTTP 4xx/5xx with non-envelope
    body, or client-side validation (e.g. STL oversize)."""

    def __init__(self, code: str, message: str):
        super().__init__(f"{code}: {message}")
        self.code = code
        self.message = message


class Open3DModelClient:
    def __init__(self, base_url: str, api_key: str | None,
                 timeout_s: float = 30.0,
                 max_stl_bytes: int = 20_000_000):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.timeout_s = timeout_s
        self.max_stl_bytes = max_stl_bytes

    def _headers(self) -> dict[str, str]:
        h = {}
        if self.api_key:
            h["X-Api-Key"] = self.api_key
        return h

    def list_recipes(self) -> list[dict[str, Any]]:
        url = f"{self.base_url}/api/recipes"
        try:
            with httpx.Client(timeout=self.timeout_s) as c:
                r = c.get(url, headers=self._headers())
        except httpx.RequestError as e:
            raise Open3DModelUnavailable(f"open3dmodel unreachable: {e}") from e
        if r.status_code != 200:
            raise Open3DModelError(f"HTTP_{r.status_code}", self._detail_msg(r))
        try:
            body = r.json()
        except ValueError as e:
            raise Open3DModelUnavailable(f"non-JSON response: {e}") from e
        # Open3DModel wraps list responses in the {success, data, error} envelope
        # too; older mocks may return a bare list. Accept either shape.
        if isinstance(body, dict) and "success" in body:
            if not body.get("success"):
                err = body.get("error") or {}
                raise Open3DModelError(
                    str(err.get("code") or "UNKNOWN"),
                    str(err.get("message") or "no message"),
                )
            data = body.get("data")
            if not isinstance(data, list):
                raise Open3DModelError("UNEXPECTED_PAYLOAD",
                                       f"expected list in data, got {type(data).__name__}")
            return data
        if isinstance(body, list):
            return body
        raise Open3DModelError("UNEXPECTED_PAYLOAD",
                               f"expected list or envelope, got {type(body).__name__}")

    def estimate_pose(self, recipe_id: str, points_cv_m: np.ndarray,
                      intrinsics: dict[str, Any]) -> dict[str, Any]:
        body = {
            "recipe_id": str(recipe_id),
            "point_cloud": np.asarray(points_cv_m, dtype=float).tolist(),
            "intrinsics": {
                "fx": float(intrinsics["fx"]),
                "fy": float(intrinsics["fy"]),
                "cx": float(intrinsics["cx"]),
                "cy": float(intrinsics["cy"]),
                "width": int(intrinsics["width"]),
                "height": int(intrinsics["height"]),
            },
            "unit": "m",
        }
        url = f"{self.base_url}/api/pose"
        try:
            with httpx.Client(timeout=self.timeout_s) as c:
                r = c.post(url, json=body, headers=self._headers())
        except httpx.RequestError as e:
            raise Open3DModelUnavailable(f"open3dmodel unreachable: {e}") from e
        try:
            payload = r.json()
        except ValueError as e:
            raise Open3DModelUnavailable(
                f"non-JSON response (status {r.status_code}): {e}") from e
        if isinstance(payload, dict) and "success" in payload:
            if payload.get("success") is True:
                return payload.get("data") or {}
            err = payload.get("error") or {}
            raise Open3DModelError(
                str(err.get("code") or "UNKNOWN"),
                str(err.get("message") or "no message"),
            )
        # Non-envelope response (e.g. FastAPI HTTPException 403/422)
        raise Open3DModelError(f"HTTP_{r.status_code}", self._detail_msg(r))

    def get_stl_bytes(self, recipe_id: str) -> tuple[bytes, str]:
        url = f"{self.base_url}/api/recipes/{recipe_id}/stl"
        try:
            with httpx.Client(timeout=self.timeout_s) as c:
                r = c.get(url, headers=self._headers())
        except httpx.RequestError as e:
            raise Open3DModelUnavailable(f"open3dmodel unreachable: {e}") from e
        if r.status_code == 404:
            raise Open3DModelError("RECIPE_NOT_FOUND",
                                   f"STL for {recipe_id} not found")
        if r.status_code != 200:
            raise Open3DModelError(f"HTTP_{r.status_code}", self._detail_msg(r))
        body = r.content
        if len(body) > self.max_stl_bytes:
            raise Open3DModelError("STL_TOO_LARGE",
                                   f"{len(body)} bytes > {self.max_stl_bytes}")
        etag = r.headers.get("ETag") or r.headers.get("Last-Modified") \
               or hashlib.sha1(body).hexdigest()
        return body, etag

    @staticmethod
    def _detail_msg(r: httpx.Response) -> str:
        try:
            body = r.json()
            if isinstance(body, dict) and "detail" in body:
                return str(body["detail"])
        except ValueError:
            pass
        return r.text[:200] or f"http {r.status_code}"