cores/face/analysis.py

"""
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