services/face_intelligence_service.py

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