^j)dZddlmZddlmZddlZddlmZGddeeZ de e e eeffd e e e eefffd Z d Z d ZGd dZy)zVOKS keypoint mAP metric backed by :class:`~rfdetr.evaluation.coco_eval.CocoEvaluator`.)Enum)AnyN) CocoEvaluatorc eZdZdZdZdZdZdZy)OKSKeyufKeys returned by :meth:`MetricKeypointOKS.compute`. Subclasses :class:`str` so enum members compare equal to their string values and can be used interchangeably as dict keys — ``stats[OKSKey.MAP]`` and ``stats["map"]`` both work. Examples: >>> OKSKey.MAP == "map" True >>> OKSKey.MAP_50.value 'map_50' mapmap_50map_75marN)__name__ __module__ __qualname____doc__MAPMAP_50MAP_75MARi/var/www/ramen.bs-engineer-server.com/venv/lib/python3.12/site-packages/rfdetr/evaluation/keypoint_oks.pyrrs  C F F Crr predictionsreturnc"|jDcic]b\}}||jDcic]@\}}|t|tjr|j j n|Bc}}dc}}}}Scc}}wcc}}}}w)aReturn a copy of *predictions* with all tensors detached and moved to CPU. Prevents callers from inadvertently retaining CUDA memory or autograd graphs between :meth:`MetricKeypointOKS.update` calls. Non-tensor values are kept as-is. Args: predictions: Per-image prediction dict mapping ``image_id`` to a dict of tensor-valued fields (``boxes``, ``scores``, ``labels``, ``keypoints``). Returns: New dict with the same structure; every :class:`torch.Tensor` value is replaced by its ``.detach().cpu()`` copy. Examples: >>> import torch >>> preds = {1: {"scores": torch.tensor([0.9], device="cpu"), "label": 2}} >>> sanitized = _sanitize_preds(preds) >>> sanitized[1]["label"] 2 )items isinstancetorchTensordetachcpu)rimage_idpredskeyvalues r_sanitize_predsr$$s2 +002    He fkfqfqfs XbX[]bCE5<<)H##%e S     sB AB7B B i) c eZdZdZdefdedeedzdeddfdZ e de fdZ d d Z d eeeeeffddfd Zdeeeffd Zy)MetricKeypointOKSuOKS keypoint mAP metric backed by CocoEvaluator. Plain Python facade over :class:`~rfdetr.evaluation.coco_eval.CocoEvaluator` with a :meth:`reset` / :meth:`update` / :meth:`compute` interface that mirrors the torchmetrics API shape without subclassing it. DDP synchronisation is handled inside :meth:`compute` via :meth:`~rfdetr.evaluation.coco_eval.CocoEvaluator.synchronize_between_processes`, which uses the repo's pickle-based ``all_gather`` — avoiding the torchmetrics deadlock bugs #931 / #449 that affect variable-shape state tensors. Supports arbitrary keypoint counts and per-category OKS sigmas through the underlying :class:`~rfdetr.evaluation.coco_eval._GroupedKeypointCOCOeval`. When TorchMetrics ships production-quality arbitrary-keypoint support (tracked in upstream PR #3348), the internals of :meth:`compute` can delegate to ``MeanAveragePrecision(iou_type="keypoints", keypoint_format="xyv")`` without any change to callers. Note: when migrating, ``"mar"`` will need remapping to ``"mar_"`` as TorchMetrics uses a suffixed key name. Args: coco_gt: Ground-truth COCO object. Accepted types: :class:`faster_coco_eval.COCO` or any object with a ``.dataset`` dict and optional ``.label2cat`` mapping (the duck-typed surface required by :class:`~rfdetr.evaluation.coco_eval.CocoEvaluator`). keypoint_oks_sigmas: Per-keypoint OKS sigmas. When ``None``, falls back to COCO person sigmas for 17-keypoint datasets or a uniform 0.05 sigma for other counts. max_dets: Maximum detections per image passed to the underlying :class:`~rfdetr.evaluation.coco_eval.CocoEvaluator`. Defaults to 500. Note: For keypoint evaluation the underlying COCO evaluator overrides ``maxDets`` to ``[20]`` regardless of this value — this parameter is forwarded but has no effect on keypoint evaluation. Examples: >>> from unittest.mock import MagicMock >>> metric = MetricKeypointOKS(MagicMock(), max_dets=100) >>> metric.has_updates False >>> metric.reset() # idempotent on empty state Ncoco_gtkeypoint_oks_sigmasmax_detsrc<||_||_||_g|_y)N)_coco_gt_keypoint_oks_sigmas _max_dets_batches)selfr(r)r*s r__init__zMetricKeypointOKS.__init__zs#   $7!!:< rc,t|jS)aReturn whether any predictions have been accumulated since last reset. Returns: ``True`` if :meth:`update` has been called at least once since the last :meth:`reset`. Examples: >>> from unittest.mock import MagicMock >>> metric = MetricKeypointOKS(MagicMock()) >>> metric.has_updates False >>> metric.update({1: {}}) >>> metric.has_updates True )boolr/r0s r has_updateszMetricKeypointOKS.has_updatess"DMM""rc8|jjy)a!Clear accumulated predictions. Examples: >>> from unittest.mock import MagicMock >>> metric = MetricKeypointOKS(MagicMock()) >>> metric.update({1: {}}) >>> metric.reset() >>> metric.has_updates False N)r/clearr4s rresetzMetricKeypointOKS.resets rrcL|jjt|y)u]Accumulate per-batch predictions. Each call appends one batch; predictions are replayed in order inside :meth:`compute`. Predictions for the same ``image_id`` across different calls are preserved as separate entries — no overwrite. Args: predictions: Mapping from ``image_id`` to a prediction dict with keys ``boxes`` (``[N, 4]`` xyxy pixel coords), ``scores`` (``[N]``), ``labels`` (``[N]`` int), and ``keypoints`` (``[N, K, 3]`` x/y/confidence in pixel coords). Pass an empty dict for images with no predictions. Examples: >>> from unittest.mock import MagicMock >>> metric = MetricKeypointOKS(MagicMock()) >>> metric.update({1: {}, 2: {}}) >>> metric.has_updates True N)r/appendr$)r0rs rupdatezMetricKeypointOKS.updates* _[9:rc Xt|jdg|j|jd}|jD]}|j ||j |j|jdj}|jtk(sJdtd|jdtjt|dtjt|dtj t|d tj"t|d iS) a!Run OKS keypoint evaluation and return metric dict. Constructs a fresh :class:`~rfdetr.evaluation.coco_eval.CocoEvaluator`, replays all accumulated per-batch predictions in order (matching the original per-batch ``CocoEvaluator.update()`` call pattern), synchronises across DDP ranks via :meth:`~rfdetr.evaluation.coco_eval.CocoEvaluator.synchronize_between_processes`, and accumulates COCO keypoint statistics. Returns: Dict with float values for keys :data:`METRIC_KEY_MAP` (mAP@50:95), :data:`METRIC_KEY_MAP_50` (AP@50), :data:`METRIC_KEY_MAP_75` (AP@75), and :data:`METRIC_KEY_MAR` (AR@50:95). A value of ``-1.0`` indicates the statistic was not available (e.g. no predictions matched any ground-truth annotation). Callers should filter ``value < 0`` before logging. Examples: >>> from unittest.mock import MagicMock, patch >>> import numpy as np >>> metric = MetricKeypointOKS(MagicMock(), max_dets=500) >>> fake_eval = MagicMock() >>> fake_eval.coco_eval = { ... "keypoints": MagicMock(stats=np.array([0.5, 0.7, 0.4, -1, -1, 0.6, -1, -1, -1, -1])) ... } >>> with patch("rfdetr.evaluation.keypoint_oks.CocoEvaluator", return_value=fake_eval): ... metric.update({1: {}}) ... result = metric.compute() >>> result["map"] 0.5 >>> result["map_50"] 0.7 keypointsF)r*r) log_summaryz#Expected coco keypoint stats shape z, got uR; pycocotools _summarizeKps() contract violated — check faster_coco_eval versionr)rr,r.r-r/r;synchronize_between_processes accumulate coco_evalstatsshape_KPS_STATS_SHAPErrfloatrrr)r0 evaluatorbatchrEs rcomputezMetricKeypointOKS.computes B" MM M^^ $ 9 9  ]] $E   U # $//1##K066{{.. 12B1C6%++W_ _ . JJeAh MM5q? MM5q? JJeAh   r)rN)r r rrDEFAULT_KEYPOINT_MAX_DETSrlistrHintr1propertyr3r5r8dictstrr;rKrrrr'r'Ns)\371 <<"%[4/< <  <(#T##$ ;$sDcN':";;;.6 c5j)6 rr')renumrtypingrrrfdetr.evaluation.coco_evalrrQrrPrNr$rLrGr'rrrrUsz ] 5S$(c4S>&9!:tCcSVhDW?XD  m m r