what are these

pipeline/ops/__init__.py

@@ -18,6 +18,11 @@ from pipeline.ops.geometry import (
     point_in_polygon,
 )
 from pipeline.ops.nms import nms, nms_by_class, wbf, wbf_by_class
+from pipeline.ops.selection import (
+    merge_overlapping,
+    select_for_verification,
+    select_top_k,
+)
 
 __all__ = [
     # NMS / Box Fusion
@@ -25,6 +30,10 @@ __all__ = [
     "nms_by_class",
     "wbf",
     "wbf_by_class",
+    # Selection / Budget
+    "select_top_k",
+    "select_for_verification",
+    "merge_overlapping",
     # Geometry
     "expand_box",
     "point_in_box",

pipeline/ops/selection.py

@@ -0,0 +1,174 @@
+"""Budget-aware box selection for capping expensive inference calls."""
+
+from typing import List, Literal
+
+from pipeline.types import BoundingBox, Detection
+
+type SelectPriority = Literal["area", "confidence", "area_conf"]
+
+
+def select_top_k(
+    detections: List[Detection],
+    k: int,
+    diversity_iou: float = 0.5,
+    priority: SelectPriority = "area",
+) -> List[Detection]:
+    """
+    Select top K detections with spatial diversity.
+
+    Picks the K "best" boxes while ensuring they don't overlap too much.
+    This caps expensive downstream inference (e.g., max 3 person crops → cig model).
+
+    Args:
+        detections: Candidate detections
+        k: Maximum number to select
+        diversity_iou: Reject candidates overlapping selected boxes above this IoU
+        priority: How to rank candidates:
+            - "area": Larger boxes first (closer to camera, more detail)
+            - "confidence": Higher confidence first
+            - "area_conf": area * confidence (balanced)
+
+    Returns:
+        Up to K diverse, high-priority detections
+    """
+    if not detections or k <= 0:
+        return []
+
+    # Score each detection
+    def score(det: Detection) -> float:
+        box = det.box
+        # area = (box.x2 - box.x1) * (box.y2 - box.y1)
+        area = box.area
+        if priority == "area":
+            return area
+        elif priority == "confidence":
+            return det.score
+        else:  # area_conf
+            return area * det.score
+
+    # Sort by priority (highest first)
+    ranked = sorted(detections, key=score, reverse=True)
+
+    selected = []
+    for det in ranked:
+        if len(selected) >= k:
+            break
+
+        # Check diversity: don't select if overlaps too much with already selected
+        dominated = False
+        for sel in selected:
+            if det.box.iou(sel.box) > diversity_iou:
+                dominated = True
+                break
+
+        if not dominated:
+            selected.append(det)
+
+    return selected
+
+
+def merge_overlapping(
+    detections: List[Detection],
+    iou_threshold: float = 0.3,
+) -> List[Detection]:
+    """
+    Merge overlapping boxes into larger regions.
+
+    Instead of processing each overlapping person separately, merge them
+    into one bigger crop. Good when people are clustered together.
+
+    Args:
+        detections: List of detections (e.g., people)
+        iou_threshold: Merge boxes with IoU above this
+
+    Returns:
+        Merged detections (fewer, larger boxes)
+    """
+    if not detections:
+        return []
+
+    # Sort by confidence
+    dets = sorted(detections, key=lambda d: d.score, reverse=True)
+
+    clusters: List[List[Detection]] = []
+
+    for det in dets:
+        merged = False
+        for cluster in clusters:
+            # Check if this det overlaps with any box in cluster
+            for member in cluster:
+                if det.box.iou(member.box) >= iou_threshold:
+                    cluster.append(det)
+                    merged = True
+                    break
+            if merged:
+                break
+
+        if not merged:
+            clusters.append([det])
+
+    # Create merged boxes from clusters
+    result = []
+    for cluster in clusters:
+        merged_det = _merge_cluster(cluster)
+        result.append(merged_det)
+
+    return result
+
+
+def _merge_cluster(cluster: List[Detection]) -> Detection:
+    """Merge a cluster of detections into one encompassing box."""
+    if len(cluster) == 1:
+        return cluster[0]
+
+    # Bounding box that contains all
+    x1 = min(d.box.x1 for d in cluster)
+    y1 = min(d.box.y1 for d in cluster)
+    x2 = max(d.box.x2 for d in cluster)
+    y2 = max(d.box.y2 for d in cluster)
+
+    # Confidence: max of cluster (could also use average)
+    conf = max(d.score for d in cluster)
+
+    # Use label from highest confidence detection
+    best = max(cluster, key=lambda d: d.score)
+
+    return Detection(
+        box=BoundingBox(x1=x1, y1=y1, x2=x2, y2=y2, conf=conf),
+        score=conf,
+        label=best.label,
+        metadata={
+            "merged_count": len(cluster),
+            "original_boxes": [
+                (d.box.x1, d.box.y1, d.box.x2, d.box.y2) for d in cluster
+            ],
+        },
+    )
+
+
+def select_for_verification(
+    detections: List[Detection],
+    k: int,
+    min_confidence: float = 0.0,
+) -> List[Detection]:
+    """
+    Select top K cigarette candidates for expensive verification.
+
+    Simple confidence-based selection for the verifier stage.
+    Higher confidence cigs are more likely real → verify those first.
+
+    Args:
+        detections: Cigarette candidates
+        k: Max verifications to run
+        min_confidence: Skip candidates below this
+
+    Returns:
+        Top K candidates by confidence
+    """
+    if not detections or k <= 0:
+        return []
+
+    filtered = [d for d in detections if d.score >= min_confidence]
+    ranked = sorted(filtered, key=lambda d: d.score, reverse=True)
+
+    return ranked[:k]