""" Face analysis helpers — quality, blur, pose estimation, size/orientation, best-face selection, clustering, duplicate elimination. Pure OpenCV + NumPy — no external deps. Used by the FaceIntelligenceService. """ from __future__ import annotations from typing import Dict, List, Optional, Tuple import cv2 import numpy as np from cores.vision.geometry import BBox, crop_region, boxes_iou from cores.vision.quality import sharpness from cores.face.helpers import cosine_similarity def blur_score(face_img: np.ndarray) -> float: """Estimate face blur via variance of Laplacian. Higher score = sharper (less blurry). <50 = very blurry, >200 = sharp. """ if face_img.size == 0: return 0.0 gray = cv2.cvtColor(face_img, cv2.COLOR_BGR2GRAY) if face_img.ndim == 3 else face_img return float(cv2.Laplacian(gray, cv2.CV_64F).var()) def is_blurry(face_img: np.ndarray, threshold: float = 50.0) -> bool: """True if the face is blurry (Laplacian variance < threshold).""" return blur_score(face_img) < threshold def face_size(bbox: BBox) -> int: """Face size in pixels (width * height of bbox).""" return bbox.area def face_size_label(bbox: BBox) -> str: """Classify face size: 'small' (<2500px), 'medium' (<10000px), 'large' (>=10000px).""" s = face_size(bbox) if s < 2500: return "small" elif s < 10000: return "medium" else: return "large" def estimate_pose_landmark(landmarks: Optional[dict]) -> Tuple[float, float, float, str]: """Estimate pose (yaw, pitch, roll, label) from 5-point landmarks. Uses eye + nose positions to estimate yaw. If landmarks are not available, returns (0, 0, 0, 'unknown'). Landmarks dict should have: left_eye, right_eye, nose, mouth_left, mouth_right Each as (x, y) tuples. """ if not landmarks: return 0.0, 0.0, 0.0, "unknown" try: le = landmarks.get("left_eye") re = landmarks.get("right_eye") nose = landmarks.get("nose") if not le or not re or not nose: return 0.0, 0.0, 0.0, "unknown" le_x, le_y = le re_x, re_y = re nose_x, nose_y = nose # Eye midpoint eye_mid_x = (le_x + re_x) / 2.0 eye_mid_y = (le_y + re_y) / 2.0 # Yaw: horizontal offset of nose from eye midpoint eye_dist = abs(re_x - le_x) if eye_dist < 1: return 0.0, 0.0, 0.0, "unknown" yaw = (nose_x - eye_mid_x) / eye_dist * 45.0 # scale to degrees # Pitch: vertical offset of nose from eye midpoint pitch = (nose_y - eye_mid_y) / eye_dist * 30.0 # Clamp pitch = max(-45.0, min(45.0, pitch)) yaw = max(-90.0, min(90.0, yaw)) # Roll: angle of eye line import math roll = math.degrees(math.atan2(re_y - le_y, re_x - le_x)) roll = max(-45.0, min(45.0, roll)) # Label abs_yaw = abs(yaw) if abs_yaw < 15: label = "frontal" elif abs_yaw < 45: label = "profile" else: label = "extreme" return round(yaw, 2), round(pitch, 2), round(roll, 2), label except Exception: return 0.0, 0.0, 0.0, "unknown" def estimate_pose_bbox(bbox: BBox, img_shape: Tuple[int, int]) -> Tuple[float, float, float, str]: """Estimate pose from bbox position alone (when no landmarks). Less accurate than landmark-based estimation. Returns (0, 0, 0, 'unknown') since bbox alone can't determine pose reliably. """ return 0.0, 0.0, 0.0, "unknown" def face_orientation(roll: float) -> str: """Classify face orientation based on roll angle.""" abs_roll = abs(roll) if abs_roll < 10: return "upright" elif abs_roll < 25: return "tilted" else: return "rotated" def face_quality_score( face_img: np.ndarray, bbox: BBox, blur: Optional[float] = None, pose_label: str = "frontal", ) -> float: """Composite 0-1 quality score for a face. Factors: - Sharpness (Laplacian variance) - Face size - Pose (frontal = best) - Blur threshold """ if face_img.size == 0: return 0.0 # Blur component if blur is None: blur = blur_score(face_img) blur_component = min(1.0, blur / 200.0) # Size component size = face_size(bbox) size_component = min(1.0, size / 10000.0) # Pose component pose_weights = { "frontal": 1.0, "profile": 0.6, "extreme": 0.3, "unknown": 0.8, } pose_component = pose_weights.get(pose_label, 0.5) # Weighted average return round(0.4 * blur_component + 0.3 * size_component + 0.3 * pose_component, 4) def select_best_face( quality_scores: List[float], face_sizes: List[int], pose_labels: List[str], ) -> int: """Select the index of the best face for recognition. Prefers: frontal pose + large size + high quality. """ if not quality_scores: return -1 best_idx = 0 best_score = -1.0 for i, qs in enumerate(quality_scores): # Pose weight pose_w = {"frontal": 1.0, "unknown": 0.8, "profile": 0.5, "extreme": 0.2}.get( pose_labels[i] if i < len(pose_labels) else "unknown", 0.5 ) # Size weight (log scale) size_w = min(1.0, np.log1p(face_sizes[i] if i < len(face_sizes) else 0) / np.log1p(10000)) combined = qs * 0.5 + pose_w * 0.3 + size_w * 0.2 if combined > best_score: best_score = combined best_idx = i return best_idx def cluster_faces( embeddings: List[np.ndarray], threshold: float = 0.6, ) -> List[dict]: """Cluster faces by embedding similarity. Uses greedy agglomerative clustering with cosine similarity. Returns list of clusters: {cluster_id, face_indices, representative_index, num_faces, avg_similarity} """ if not embeddings: return [] n = len(embeddings) assigned: List[int] = [-1] * n # -1 = unassigned cluster_id = 0 clusters: List[dict] = [] for i in range(n): if assigned[i] != -1: continue # Start a new cluster assigned[i] = cluster_id members = [i] sims = [] for j in range(i + 1, n): if assigned[j] != -1: continue sim = cosine_similarity(embeddings[i], embeddings[j]) if sim >= threshold: assigned[j] = cluster_id members.append(j) sims.append(sim) avg_sim = sum(sims) / len(sims) if sims else 1.0 # Representative = the member with highest average similarity to others if len(members) == 1: rep = members[0] else: # Compute avg similarity of each member to the rest best_rep = members[0] best_avg = -1.0 for m in members: other_sims = [ cosine_similarity(embeddings[m], embeddings[o]) for o in members if o != m ] m_avg = sum(other_sims) / len(other_sims) if other_sims else 0.0 if m_avg > best_avg: best_avg = m_avg best_rep = m rep = best_rep clusters.append({ "cluster_id": cluster_id, "face_indices": members, "representative_index": rep, "num_faces": len(members), "avg_similarity": round(avg_sim, 4), }) cluster_id += 1 return clusters def find_duplicate_faces( boxes: List[dict], iou_threshold: float = 0.7, ) -> List[int]: """Find duplicate face indices by IoU overlap. Returns indices of faces that are duplicates (lower-priority copies). Keeps the first (highest confidence) face in each overlap group. """ if len(boxes) <= 1: return [] duplicates: list[int] = [] bboxes = [BBox(b["x"], b["y"], b["w"], b["h"]) for b in boxes] for i in range(1, len(bboxes)): for j in range(i): if j in duplicates: continue if boxes_iou(bboxes[i], bboxes[j]) >= iou_threshold: duplicates.append(i) break return duplicates