feat(right_arm): /detect/front con proxy fallback per YOLO

src/presa_mano/api/right_arm_routes.py

@@ -1,10 +1,12 @@
 """Endpoint per la tab Braccio DX (G1, polso destro, Nuwa eye-in-hand)."""
 from __future__ import annotations
 
-from typing import Any
+import logging
+from typing import Any, Optional
 
 from fastapi import APIRouter, HTTPException
 from fastapi.responses import Response
+from pydantic import BaseModel
 
 from ..clients import G1Unavailable
 from ..config import update_config
@@ -13,6 +15,7 @@ from ..timing import Stopwatch
 from . import _deps
 
 router = APIRouter()
+log = logging.getLogger(__name__)
 
 
 def _state_payload(sw: Stopwatch) -> dict[str, Any]:
@@ -73,3 +76,64 @@ def snapshot_wrist() -> Response:
     except Exception as e:
         raise HTTPException(status_code=502, detail=f"wrist nuwa unreachable: {e}")
     return Response(content=jpeg, media_type="image/jpeg")
+
+
+class DetectFrontReq(BaseModel):
+    class_name: Optional[str] = None
+    conf: Optional[float] = None
+    proxy_base_url: Optional[str] = None  # se YOLO non raggiunge la front camera
+
+
+def _yolo_detect_with_fallback(image_url: str, conf: float | None,
+                                proxy_image_url: str | None, sw: Stopwatch
+                                ) -> dict[str, Any]:
+    """Prova YOLO sull'URL diretto; se fallisce e proxy_image_url è settato, ritenta."""
+    yolo = _deps.yolo()
+    try:
+        with sw.step("yolo"):
+            return yolo.detect_url(image_url=image_url, conf=conf)
+    except Exception as e:
+        if not proxy_image_url:
+            raise HTTPException(status_code=502, detail=f"YOLO unreachable: {e}")
+        log.warning("YOLO failed on direct URL (%s), retrying via proxy %s", e, proxy_image_url)
+        try:
+            with sw.step("yolo_proxy"):
+                return yolo.detect_url(image_url=proxy_image_url, conf=conf)
+        except Exception as e2:
+            raise HTTPException(status_code=502, detail=f"YOLO unreachable (proxy): {e2}")
+
+
+def _top_detection(payload: dict[str, Any], class_name: str | None) -> dict[str, Any] | None:
+    dets = payload.get("detections") or payload.get("data", {}).get("detections") or []
+    if class_name:
+        dets = [d for d in dets if (d.get("class") or "").lower() == class_name.lower()]
+    if not dets:
+        return None
+    dets.sort(key=lambda d: float(d.get("confidence", 0.0)), reverse=True)
+    return dets[0]
+
+
+@router.post("/detect/front")
+def detect_front(req: DetectFrontReq) -> dict[str, Any]:
+    sw = Stopwatch()
+    cam = _deps.front_camera()
+    image_url = cam.rgb_url()
+    proxy_url = None
+    if req.proxy_base_url:
+        proxy_url = f"{req.proxy_base_url.rstrip('/')}/api/right_arm/snapshot/front"
+    res = _yolo_detect_with_fallback(image_url, req.conf, proxy_url, sw)
+    dets_raw = res.get("detections") or res.get("data", {}).get("detections") or []
+    if req.class_name:
+        dets_filtered = [d for d in dets_raw
+                         if (d.get("class") or "").lower() == req.class_name.lower()]
+    else:
+        dets_filtered = list(dets_raw)
+    if not dets_filtered:
+        raise HTTPException(status_code=404, detail=f"no '{req.class_name or 'object'}' detected")
+    dets_filtered.sort(key=lambda d: float(d.get("confidence", 0.0)), reverse=True)
+    intr = cam.intrinsics()
+    return {
+        "detections": dets_filtered,
+        "image_size": [int(intr.get("width", 0)), int(intr.get("height", 0))],
+        "timings_ms": sw.as_dict(),
+    }