diff --git a/src/presa_mano/api/right_arm_routes.py b/src/presa_mano/api/right_arm_routes.py index 5920132..8f07430 100644 --- a/src/presa_mano/api/right_arm_routes.py +++ b/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(), + } diff --git a/tests/test_right_arm_routes.py b/tests/test_right_arm_routes.py index 9230b70..8d483aa 100644 --- a/tests/test_right_arm_routes.py +++ b/tests/test_right_arm_routes.py @@ -93,3 +93,75 @@ def test_snapshot_front_502_on_failure(): factory.return_value = cam r = client.get("/api/right_arm/snapshot/front") assert r.status_code == 502 + + +def _fake_yolo_top(class_name="pestello", conf=0.87): + return {"detections": [ + {"class": class_name, "confidence": conf, + "bbox": {"x1": 100.0, "y1": 80.0, "x2": 220.0, "y2": 180.0}}, + {"class": "altra", "confidence": 0.5, + "bbox": {"x1": 10.0, "y1": 10.0, "x2": 50.0, "y2": 50.0}}, + ]} + + +def test_detect_front_filters_class_and_returns_top(): + with patch("presa_mano.api.right_arm_routes._deps.yolo") as ying, \ + patch("presa_mano.api.right_arm_routes._deps.front_camera") as cam_f: + yolo = MagicMock() + yolo.detect_url.return_value = _fake_yolo_top() + ying.return_value = yolo + cam = MagicMock() + cam.rgb_url.return_value = "http://192.168.123.99:8878/api/rgb" + cam.intrinsics.return_value = {"width": 640, "height": 480} + cam_f.return_value = cam + + r = client.post("/api/right_arm/detect/front", + json={"class_name": "pestello", "conf": 0.4}) + assert r.status_code == 200 + body = r.json() + assert body["detections"][0]["class"] == "pestello" + assert body["image_size"] == [640, 480] + + +def test_detect_front_404_when_no_match(): + with patch("presa_mano.api.right_arm_routes._deps.yolo") as ying, \ + patch("presa_mano.api.right_arm_routes._deps.front_camera") as cam_f: + yolo = MagicMock() + yolo.detect_url.return_value = {"detections": []} + ying.return_value = yolo + cam = MagicMock() + cam.rgb_url.return_value = "http://x/api/rgb" + cam.intrinsics.return_value = {"width": 640, "height": 480} + cam_f.return_value = cam + r = client.post("/api/right_arm/detect/front", + json={"class_name": "pestello"}) + assert r.status_code == 404 + + +def test_detect_front_falls_back_to_proxy_url_on_yolo_failure(): + """Se YOLO non riesce sull'URL diretto, ripeti con `proxy_base_url`.""" + attempts = [] + + def _detect_url(image_url, conf=None): + attempts.append(image_url) + if "192.168" in image_url: + raise RuntimeError("dns timeout") + return _fake_yolo_top() + + with patch("presa_mano.api.right_arm_routes._deps.yolo") as ying, \ + patch("presa_mano.api.right_arm_routes._deps.front_camera") as cam_f: + yolo = MagicMock() + yolo.detect_url.side_effect = _detect_url + ying.return_value = yolo + cam = MagicMock() + cam.rgb_url.return_value = "http://192.168.123.99:8878/api/rgb" + cam.intrinsics.return_value = {"width": 640, "height": 480} + cam_f.return_value = cam + + r = client.post("/api/right_arm/detect/front", + json={"class_name": "pestello", + "proxy_base_url": "http://localhost:8890"}) + assert r.status_code == 200 + assert len(attempts) == 2 + assert "192.168" in attempts[0] + assert attempts[1].endswith("/api/right_arm/snapshot/front")