| """ |
| 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_mid_x = (le_x + re_x) / 2.0 |
| eye_mid_y = (le_y + re_y) / 2.0 |
|
|
| |
| 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 |
|
|
| |
| pitch = (nose_y - eye_mid_y) / eye_dist * 30.0 |
| |
| pitch = max(-45.0, min(45.0, pitch)) |
| yaw = max(-90.0, min(90.0, yaw)) |
|
|
| |
| import math |
| roll = math.degrees(math.atan2(re_y - le_y, re_x - le_x)) |
| roll = max(-45.0, min(45.0, roll)) |
|
|
| |
| 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 |
|
|
| |
| if blur is None: |
| blur = blur_score(face_img) |
| blur_component = min(1.0, blur / 200.0) |
|
|
| |
| size = face_size(bbox) |
| size_component = min(1.0, size / 10000.0) |
|
|
| |
| pose_weights = { |
| "frontal": 1.0, |
| "profile": 0.6, |
| "extreme": 0.3, |
| "unknown": 0.8, |
| } |
| pose_component = pose_weights.get(pose_label, 0.5) |
|
|
| |
| 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_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_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 |
| cluster_id = 0 |
| clusters: List[dict] = [] |
|
|
| for i in range(n): |
| if assigned[i] != -1: |
| continue |
| |
| 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 |
| |
| if len(members) == 1: |
| rep = members[0] |
| else: |
| |
| 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 |
|
|
