cnu/worker-gesture-perception Private

1234567891011121314151617

from __future__ import annotationsCOMMANDS = {"STOP": "SAFETY_STOP", "FOLLOW": "FOLLOW_WORKER", "CALL": "APPROACH_WORKER"}def execute(payload: dict) -> dict:    threshold = float(payload.get("confidenceThreshold", 0.8))    candidates = [track for track in payload.get("tracks", []) if float(track.get("confidence", 0)) >= threshold]    candidates.sort(key=lambda track: (-float(track["confidence"]), track["id"]))    selected = candidates[0] if candidates else None    command = "WAIT" if not selected else COMMANDS.get(selected.get("gesture"), "WAIT")    worker = None if not selected else {"id": selected["id"], "confidence": float(selected["confidence"]), "distanceM": float(selected.get("distanceM", 0)), "gesture": selected.get("gesture", "UNKNOWN")}    return {"component": "cnu/worker-gesture-perception", "accepted": selected is not None,            "status": "WORKER_READY" if selected else "NO_WORKER", "target": "vpu",            "selectedWorker": worker, "command": command, "stopRequired": command == "SAFETY_STOP",            "consideredTracks": len(payload.get("tracks", []))}