File size: 2,999 Bytes
aac350d 892fa81 aac350d 892fa81 aac350d 892fa81 aac350d 892fa81 aac350d 23d337e aac350d 892fa81 aac350d 892fa81 aac350d 892fa81 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 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """
Feature extraction — uses cores.vision for cropping + cores.face for
box conversions. No duplicated crop logic.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import List, Optional
import numpy as np
from loguru import logger
from cores.vision import BBox, crop_region
from providers.base import Provider, ProviderResult
@dataclass
class FaceCrop:
image: np.ndarray
box: dict
confidence: float = 1.0
detector: str = ""
@dataclass
class PipelineOutput:
"""The normalized payload the orchestrator consumes."""
image: np.ndarray
image_hash: str
width: int
height: int
source: str
original_bytes: Optional[bytes] = None
original_format: Optional[str] = None
face_crops: List[FaceCrop] = field(default_factory=list)
primary_detector: str = ""
gallery: Optional[dict] = None
scrape_url: Optional[str] = None
@property
def num_faces(self) -> int:
return len(self.face_crops)
class FeatureExtractor:
"""Uses a detection provider to pre-extract face crops."""
def __init__(self, detector: Optional[Provider] = None) -> None:
self._detector = detector
def set_detector(self, provider: Provider) -> None:
self._detector = provider
def extract(
self,
image: np.ndarray,
image_hash: str,
width: int,
height: int,
source: str,
original_bytes: Optional[bytes] = None,
original_format: Optional[str] = None,
) -> PipelineOutput:
crops: List[FaceCrop] = []
detector_name = ""
if self._detector is not None and self._detector.is_available():
try:
result: ProviderResult = self._detector.execute(image)
if result.success and result.normalized.get("boxes"):
detector_name = result.provider
boxes = result.normalized["boxes"]
confs = result.normalized.get("confidences", [1.0] * len(boxes))
for box_dict, conf in zip(boxes, confs):
bbox = BBox(box_dict["x"], box_dict["y"],
box_dict["w"], box_dict["h"])
crop = crop_region(image, bbox, margin=0.2)
crops.append(FaceCrop(
image=crop, box=box_dict,
confidence=float(conf), detector=detector_name,
))
except Exception as e:
logger.warning(f"Feature extraction failed: {e}")
else:
logger.debug("No detector available; pipeline output will have 0 face crops.")
return PipelineOutput(
image=image, image_hash=image_hash,
width=width, height=height, source=source,
original_bytes=original_bytes, original_format=original_format,
face_crops=crops, primary_detector=detector_name,
)
|