""" Face Intelligence service — post-processes face detections into structured intelligence: quality scores, pose, blur, best-face selection, clustering, and duplicate elimination. This service consumes the output of detection + recognition providers and produces a FaceIntelligenceResult. It does NOT run providers itself — it expects the caller to have already run detection (and optionally recognition for embeddings). Pure computation via cores.face.analysis — no external calls. """ from __future__ import annotations import time from typing import List, Optional import numpy as np from cores.face import ( blur_score, is_blurry, face_size, face_size_label, estimate_pose_landmark, face_orientation, face_quality_score, select_best_face, cluster_faces, find_duplicate_faces, extract_face_crops, ) from cores.vision.geometry import BBox from models.jobs import JobRequest from models.reports import ( FaceIntelligenceResult, FaceQualityMetrics, FaceCluster, ) from pipeline import InputValidator, ImagePreprocessor, ImageHasher, FeatureExtractor from utils.logging import execution_context, new_execution_id class FaceIntelligenceService: """Post-processes face detections into structured intelligence.""" def __init__( self, validator: InputValidator, preprocessor: ImagePreprocessor, hasher: ImageHasher, feature_extractor: FeatureExtractor, ) -> None: self._validator = validator self._preprocessor = preprocessor self._hasher = hasher self._feature_extractor = feature_extractor async def analyze(self, request: JobRequest) -> dict: """Run face intelligence on an image. Uses the feature extractor (which wraps a detection provider) to detect faces, then computes quality/pose/cluster intelligence. """ eid = new_execution_id() with execution_context(execution_id=eid, provider_id="face_intelligence_service"): t0 = time.perf_counter() vr = self._validator.validate( image_url=request.image_url, image_base64=request.image_base64, ) if not vr.valid: return {"success": False, "error": vr.error, "error_type": "ValidationError"} if vr.source == "url": pre = self._preprocessor.from_url(request.image_url) else: pre = self._preprocessor.from_bytes(vr.image_bytes, vr.source) img_hash = self._hasher.hash(pre.image) pipeline_output = self._feature_extractor.extract( pre.image, img_hash, pre.width, pre.height, pre.source, original_bytes=pre.original_bytes, original_format=pre.original_format, ) face_crops = pipeline_output.face_crops boxes = [fc.box for fc in face_crops] total_faces = len(face_crops) if total_faces == 0: result = FaceIntelligenceResult( total_faces=0, best_face_index=None, elapsed_ms=0.0, ) elapsed = (time.perf_counter() - t0) * 1000.0 result.elapsed_ms = round(elapsed, 3) return {"success": True, "face_intelligence": result.model_dump(), "elapsed_ms": round(elapsed, 3)} # Compute quality metrics per face quality_metrics: List[FaceQualityMetrics] = [] blur_scores: List[float] = [] sizes: List[int] = [] pose_labels: List[str] = [] for i, fc in enumerate(face_crops): face_img = fc.image bbox = BBox(fc.box["x"], fc.box["y"], fc.box["w"], fc.box["h"]) blur = blur_score(face_img) blur_scores.append(blur) sizes.append(face_size(bbox)) # Pose estimation — use landmarks if available, else bbox landmarks = None # landmarks would come from the detector; for now use None yaw, pitch, roll, pose_label = estimate_pose_landmark(landmarks) pose_labels.append(pose_label) orientation = face_orientation(roll) qs = face_quality_score(face_img, bbox, blur=blur, pose_label=pose_label) quality_metrics.append(FaceQualityMetrics( quality_score=qs, blur_score=round(blur, 4), is_blurry=is_blurry(face_img), face_size=face_size(bbox), face_size_label=face_size_label(bbox), pose_yaw=yaw, pose_pitch=pitch, pose_roll=roll, pose_label=pose_label, orientation=orientation, is_best_face=False, )) # Best face selection best_idx = select_best_face( [m.quality_score for m in quality_metrics], sizes, pose_labels, ) if best_idx >= 0: quality_metrics[best_idx].is_best_face = True # Clustering — only if we have embeddings (from recognition provider) # For now, we don't have embeddings here; clustering is optional clusters: List[FaceCluster] = [] # If embeddings were attached to pipeline_output.gallery, use them # (This would require the caller to have run recognition first) # Duplicate face elimination by IoU duplicate_indices = find_duplicate_faces(boxes) elapsed = (time.perf_counter() - t0) * 1000.0 result = FaceIntelligenceResult( total_faces=total_faces, best_face_index=best_idx if best_idx >= 0 else None, quality_metrics=quality_metrics, clusters=clusters, duplicate_face_indices=duplicate_indices, elapsed_ms=round(elapsed, 3), ) return { "success": True, "face_intelligence": result.model_dump(), "elapsed_ms": round(elapsed, 3), } async def analyze_with_embeddings( self, image: np.ndarray, boxes: List[dict], embeddings: List[np.ndarray], cluster_threshold: float = 0.6, ) -> FaceIntelligenceResult: """Analyze faces when embeddings are already available. This is used when detection + recognition have already run, and we want to add quality + clustering intelligence. """ t0 = time.perf_counter() # Extract crops crops = extract_face_crops(image, boxes) quality_metrics: List[FaceQualityMetrics] = [] blur_scores: List[float] = [] sizes: List[int] = [] pose_labels: List[str] = [] for i, crop in enumerate(crops): bbox = BBox(boxes[i]["x"], boxes[i]["y"], boxes[i]["w"], boxes[i]["h"]) blur = blur_score(crop) blur_scores.append(blur) sizes.append(face_size(bbox)) yaw, pitch, roll, pose_label = estimate_pose_landmark(None) pose_labels.append(pose_label) quality_metrics.append(FaceQualityMetrics( quality_score=face_quality_score(crop, bbox, blur=blur, pose_label=pose_label), blur_score=round(blur, 4), is_blurry=is_blurry(crop), face_size=face_size(bbox), face_size_label=face_size_label(bbox), pose_yaw=yaw, pose_pitch=pitch, pose_roll=roll, pose_label=pose_label, orientation=face_orientation(roll), is_best_face=False, )) best_idx = select_best_face( [m.quality_score for m in quality_metrics], sizes, pose_labels ) if best_idx >= 0: quality_metrics[best_idx].is_best_face = True # Cluster by embedding similarity cluster_data = cluster_faces(embeddings, threshold=cluster_threshold) clusters = [ FaceCluster( cluster_id=c["cluster_id"], face_indices=c["face_indices"], representative_index=c["representative_index"], num_faces=c["num_faces"], avg_similarity=c["avg_similarity"], ) for c in cluster_data ] duplicate_indices = find_duplicate_faces(boxes) elapsed = (time.perf_counter() - t0) * 1000.0 return FaceIntelligenceResult( total_faces=len(crops), best_face_index=best_idx if best_idx >= 0 else None, quality_metrics=quality_metrics, clusters=clusters, duplicate_face_indices=duplicate_indices, elapsed_ms=round(elapsed, 3), )