File size: 1,798 Bytes
aac350d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 | """
Post-processing — applies final cleanup to provider results before
they enter normalization.
Examples:
- de-duplicate scraped image URLs
- clamp bounding boxes to image bounds
- strip PII from raw responses (placeholder for future policy hooks)
"""
from __future__ import annotations
from typing import Any, List
from loguru import logger
class ResultPostprocessor:
"""Cleans provider outputs before normalization."""
@staticmethod
def dedupe_images(images: List[dict]) -> List[dict]:
"""Remove duplicate image URLs."""
seen: set[str] = set()
out: List[dict] = []
for img in images:
url = img.get("url") or img.get("image_url")
if not url or url in seen:
continue
seen.add(url)
out.append(img)
return out
@staticmethod
def clamp_boxes(boxes: List[dict], width: int, height: int) -> List[dict]:
"""Clamp bounding boxes to image bounds."""
out: List[dict] = []
for b in boxes:
x = max(0, min(b["x"], width - 1))
y = max(0, min(b["y"], height - 1))
x2 = max(0, min(b["x"] + b["w"], width))
y2 = max(0, min(b["y"] + b["h"], height))
out.append({"x": x, "y": y, "w": max(0, x2 - x), "h": max(0, y2 - y)})
return out
@staticmethod
def filter_low_confidence(boxes: List[dict], confs: List[float],
threshold: float = 0.5) -> tuple[List[dict], List[float]]:
"""Drop detections below confidence threshold."""
out_boxes, out_confs = [], []
for b, c in zip(boxes, confs):
if c >= threshold:
out_boxes.append(b)
out_confs.append(c)
return out_boxes, out_confs
|