09 - Face Detection and Recognition: MediaPipe and Beyond

import mediapipe as mp
import cv2
import numpy as np
from dataclasses import dataclass
from typing import Optional

@dataclass
class FaceDetection:
    """Detection result for a single face."""
    bbox: tuple[int, int, int, int]  # x1, y1, x2, y2
    confidence: float
    keypoints: dict[str, tuple[int, int]]  # name -> (x, y) in pixels

class MediaPipeFaceDetector:
    """
    Face detector based on MediaPipe BlazeFace.
    Very fast on CPU: 200+ FPS on 640x480 images.
    Great for real-time, not ideal for high face density images.
    """

    KEYPOINT_NAMES = [
        'right_eye', 'left_eye', 'nose_tip',
        'mouth_center', 'right_ear_tragion', 'left_ear_tragion'
    ]

    def __init__(self, min_confidence: float = 0.5,
                 model_selection: int = 0):
        """
        model_selection:
          0 = short range (within 2m, faster)
          1 = full range (up to 5m, more accurate)
        """
        self.mp_face = mp.solutions.face_detection
        self.detector = self.mp_face.FaceDetection(
            model_selection=model_selection,
            min_detection_confidence=min_confidence
        )
        self.mp_draw = mp.solutions.drawing_utils

    def detect(self, img_bgr: np.ndarray) -> list[FaceDetection]:
        """Detect faces in a BGR image."""
        h, w = img_bgr.shape[:2]
        img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
        results = self.detector.process(img_rgb)

        faces = []
        if not results.detections:
            return faces

        for detection in results.detections:
            score = detection.score[0]
            bbox_rel = detection.location_data.relative_bounding_box

            # Relative coordinates -> pixels
            x1 = max(0, int(bbox_rel.xmin * w))
            y1 = max(0, int(bbox_rel.ymin * h))
            x2 = min(w, int((bbox_rel.xmin + bbox_rel.width) * w))
            y2 = min(h, int((bbox_rel.ymin + bbox_rel.height) * h))

            # Keypoints (eyes, nose, mouth, ears)
            keypoints = {}
            for idx, name in enumerate(self.KEYPOINT_NAMES):
                kp = detection.location_data.relative_keypoints[idx]
                keypoints[name] = (int(kp.x * w), int(kp.y * h))

            faces.append(FaceDetection(
                bbox=(x1, y1, x2, y2),
                confidence=float(score),
                keypoints=keypoints
            ))

        return faces

    def draw(self, img_bgr: np.ndarray,
             faces: list[FaceDetection]) -> np.ndarray:
        """Annotate image with detection results."""
        annotated = img_bgr.copy()
        for face in faces:
            x1, y1, x2, y2 = face.bbox
            cv2.rectangle(annotated, (x1, y1), (x2, y2), (0, 255, 0), 2)
            cv2.putText(annotated, f"{face.confidence:.2f}",
                       (x1, y1 - 8), cv2.FONT_HERSHEY_SIMPLEX,
                       0.6, (0, 255, 0), 2)

            # Draw keypoints
            for name, (kx, ky) in face.keypoints.items():
                color = (0, 0, 255) if 'eye' in name else (255, 0, 0)
                cv2.circle(annotated, (kx, ky), 4, color, -1)

        return annotated


def run_face_detection_webcam() -> None:
    """Real-time detection on webcam."""
    detector = MediaPipeFaceDetector(min_confidence=0.5)
    cap = cv2.VideoCapture(0)

    while True:
        ret, frame = cap.read()
        if not ret:
            break

        faces = detector.detect(frame)
        annotated = detector.draw(frame, faces)

        cv2.putText(annotated, f"Faces: {len(faces)}",
                   (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
        cv2.imshow('MediaPipe Face Detection', annotated)

        if cv2.waitKey(1) & 0xFF == ord('q'):
            break

    cap.release()
    cv2.destroyAllWindows()

import mediapipe as mp
import cv2
import numpy as np

class FaceMeshAnalyzer:
    """
    MediaPipe Face Mesh: 468 3D landmarks in real-time.
    Includes utilities: eye aspect ratio (drowsiness), head pose, etc.
    """

    # MediaPipe landmark indices for eyes
    LEFT_EYE_IDX = [362, 385, 387, 263, 373, 380]
    RIGHT_EYE_IDX = [33, 160, 158, 133, 153, 144]

    def __init__(self, max_faces: int = 1,
                 refine_landmarks: bool = True):
        """
        refine_landmarks=True: adds landmarks around eyes and irises
        (468 -> 478 total points)
        """
        self.mp_mesh = mp.solutions.face_mesh
        self.face_mesh = self.mp_mesh.FaceMesh(
            max_num_faces=max_faces,
            refine_landmarks=refine_landmarks,
            min_detection_confidence=0.5,
            min_tracking_confidence=0.5
        )
        self.mp_styles = mp.solutions.drawing_styles

    def process(self, img_bgr: np.ndarray) -> Optional[list]:
        """Process image and return landmark list per face."""
        h, w = img_bgr.shape[:2]
        img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
        results = self.face_mesh.process(img_rgb)

        if not results.multi_face_landmarks:
            return None

        all_faces_lm = []
        for face_landmarks in results.multi_face_landmarks:
            lm_pixels = []
            for lm in face_landmarks.landmark:
                lm_pixels.append((int(lm.x * w), int(lm.y * h), lm.z))
            all_faces_lm.append(lm_pixels)

        return all_faces_lm

    def eye_aspect_ratio(self, landmarks: list,
                          eye_indices: list) -> float:
        """
        Eye Aspect Ratio (EAR) - drowsiness indicator.
        EAR < 0.2 for 20+ consecutive frames = closed eye.
        Formula: EAR = (|p2-p6| + |p3-p5|) / (2 * |p1-p4|)
        """
        pts = [np.array(landmarks[i][:2]) for i in eye_indices]

        # Vertical distances
        A = np.linalg.norm(pts[1] - pts[5])
        B = np.linalg.norm(pts[2] - pts[4])
        # Horizontal distance
        C = np.linalg.norm(pts[0] - pts[3])

        return (A + B) / (2.0 * C) if C > 0 else 0.0


def drowsiness_detector(threshold: float = 0.22,
                          consec_frames: int = 20) -> None:
    """Drowsiness alert system based on EAR."""
    analyzer = FaceMeshAnalyzer(max_faces=1)
    cap = cv2.VideoCapture(0)
    ear_counter = 0

    while True:
        ret, frame = cap.read()
        if not ret:
            break

        landmarks_list = analyzer.process(frame)

        if landmarks_list:
            lms = landmarks_list[0]  # first face

            ear_l = analyzer.eye_aspect_ratio(lms, analyzer.LEFT_EYE_IDX)
            ear_r = analyzer.eye_aspect_ratio(lms, analyzer.RIGHT_EYE_IDX)
            avg_ear = (ear_l + ear_r) / 2.0

            if avg_ear < threshold:
                ear_counter += 1
                if ear_counter >= consec_frames:
                    cv2.putText(frame, "ALERT: DROWSINESS!",
                               (50, 200), cv2.FONT_HERSHEY_SIMPLEX,
                               1.5, (0, 0, 255), 3)
            else:
                ear_counter = 0

            cv2.putText(frame, f"EAR: {avg_ear:.3f}",
                       (10, 30), cv2.FONT_HERSHEY_SIMPLEX,
                       0.8, (0, 255, 0), 2)

        cv2.imshow('Drowsiness Detector', frame)
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break

    cap.release()
    cv2.destroyAllWindows()

from mtcnn import MTCNN
import cv2
import numpy as np

class MTCNNFaceProcessor:
    """
    MTCNN for precise detection + face alignment.
    Produces normalized 112x112 images, optimal for FaceNet/ArcFace.
    """

    def __init__(self, min_face_size: int = 40,
                 thresholds: list = None,
                 scale_factor: float = 0.709):
        self.detector = MTCNN(
            min_face_size=min_face_size,
            thresholds=thresholds or [0.6, 0.7, 0.7],
            scale_factor=scale_factor
        )

    def detect_and_align(self, img_bgr: np.ndarray,
                          output_size: int = 112) -> list[np.ndarray]:
        """
        Detect faces and return them aligned (112x112 by default).
        Alignment uses an affine transform on 5 landmarks to bring
        eyes into canonical position.

        Returns: list of aligned face images (BGR, float32 [0,1])
        """
        img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
        detections = self.detector.detect_faces(img_rgb)

        aligned_faces = []
        for det in detections:
            if det['confidence'] < 0.90:
                continue

            keypoints = det['keypoints']
            src_pts = np.array([
                keypoints['left_eye'],
                keypoints['right_eye'],
                keypoints['nose'],
                keypoints['mouth_left'],
                keypoints['mouth_right']
            ], dtype=np.float32)

            # Canonical destination points for 112x112
            dst_pts = np.array([
                [38.2946, 51.6963],
                [73.5318, 51.6963],
                [56.0252, 71.7366],
                [41.5493, 92.3655],
                [70.7299, 92.3655]
            ], dtype=np.float32)

            # Scale for output sizes other than 112
            scale = output_size / 112.0
            dst_pts *= scale

            # Affine transform -> aligned image
            M = cv2.estimateAffinePartial2D(src_pts, dst_pts)[0]
            aligned = cv2.warpAffine(img_bgr, M,
                                     (output_size, output_size))
            aligned_faces.append(aligned.astype(np.float32) / 255.0)

        return aligned_faces

import insightface
from insightface.app import FaceAnalysis
import numpy as np
import cv2
import pickle
from pathlib import Path
from sklearn.preprocessing import normalize
from sklearn.neighbors import KNeighborsClassifier
from typing import Optional

class FaceRecognitionSystem:
    """
    Complete face recognition system based on InsightFace (ArcFace).
    Supports registration of new identities and real-time recognition.

    Install: pip install insightface onnxruntime scikit-learn
    """

    def __init__(self, db_path: str = 'face_db.pkl',
                 recognition_threshold: float = 0.5):
        """
        recognition_threshold: cosine threshold for a valid match
          (0.5 is a good default for 512D ArcFace embeddings)
        """
        self.app = FaceAnalysis(
            providers=['CUDAExecutionProvider', 'CPUExecutionProvider']
        )
        self.app.prepare(ctx_id=0, det_size=(640, 640))

        self.db_path = Path(db_path)
        self.threshold = recognition_threshold
        self.database: dict[str, list[np.ndarray]] = {}
        self.knn: Optional[KNeighborsClassifier] = None

        if self.db_path.exists():
            self._load_database()

    def register_person(self, name: str,
                         images: list[np.ndarray],
                         max_faces_per_image: int = 1) -> int:
        """
        Register a new person in the database.

        name: person identifier
        images: list of BGR images (at least 5 for robustness)
        Returns: number of embeddings successfully registered
        """
        embeddings = []
        for img in images:
            faces = self.app.get(img)
            if not faces:
                continue

            # Take the largest face (for single-person images)
            face = max(faces,
                       key=lambda f: (f.bbox[2]-f.bbox[0]) *
                                     (f.bbox[3]-f.bbox[1]))
            emb = normalize(face.embedding.reshape(1, -1))[0]
            embeddings.append(emb)

        if not embeddings:
            print(f"[WARN] No face detected for {name}")
            return 0

        if name not in self.database:
            self.database[name] = []
        self.database[name].extend(embeddings)

        self._rebuild_knn()
        self._save_database()
        print(f"Registered {name}: {len(embeddings)} embeddings")
        return len(embeddings)

    def recognize(self, img_bgr: np.ndarray) -> list[dict]:
        """
        Recognize all faces in an image.
        Returns: list of dicts with bbox, identity, confidence per face
        """
        faces = self.app.get(img_bgr)
        results = []

        for face in faces:
            emb = normalize(face.embedding.reshape(1, -1))[0]
            identity, confidence = self._match_embedding(emb)

            x1, y1, x2, y2 = face.bbox.astype(int)
            results.append({
                'bbox': (x1, y1, x2, y2),
                'identity': identity,
                'confidence': confidence,
                'is_known': confidence >= self.threshold
            })

        return results

    def _match_embedding(self, emb: np.ndarray) -> tuple[str, float]:
        """Find the best match in the database."""
        if not self.database or self.knn is None:
            return ('unknown', 0.0)

        dist, idx = self.knn.kneighbors([emb], n_neighbors=1)
        labels = [name for name, embs in self.database.items()
                  for _ in embs]
        best_name = labels[idx[0][0]]
        similarity = 1.0 - dist[0][0]

        return (best_name, float(similarity))

    def _rebuild_knn(self) -> None:
        """Rebuild KNN classifier after database updates."""
        all_embs = []
        all_labels = []
        for name, embs in self.database.items():
            all_embs.extend(embs)
            all_labels.extend([name] * len(embs))

        if len(all_embs) < 2:
            return

        self.knn = KNeighborsClassifier(
            n_neighbors=min(3, len(all_embs)),
            metric='cosine',
            algorithm='brute'
        )
        self.knn.fit(np.array(all_embs), all_labels)

    def _save_database(self) -> None:
        with open(self.db_path, 'wb') as f:
            pickle.dump(self.database, f)

    def _load_database(self) -> None:
        with open(self.db_path, 'rb') as f:
            self.database = pickle.load(f)
        self._rebuild_knn()
        print(f"Database loaded: {len(self.database)} identities")

    def annotate(self, img_bgr: np.ndarray,
                  results: list[dict]) -> np.ndarray:
        """Annotate image with recognition results."""
        annotated = img_bgr.copy()
        for r in results:
            x1, y1, x2, y2 = r['bbox']
            color = (0, 255, 0) if r['is_known'] else (0, 0, 255)
            cv2.rectangle(annotated, (x1, y1), (x2, y2), color, 2)
            label = (f"{r['identity']} ({r['confidence']:.2f})"
                     if r['is_known'] else "Unknown")
            cv2.putText(annotated, label, (x1, y1 - 8),
                       cv2.FONT_HERSHEY_SIMPLEX, 0.6, color, 2)
        return annotated

import numpy as np
from sklearn.metrics import roc_curve, auc

def cosine_similarity(emb1: np.ndarray, emb2: np.ndarray) -> float:
    """Cosine similarity between two normalized embeddings."""
    emb1_n = emb1 / (np.linalg.norm(emb1) + 1e-10)
    emb2_n = emb2 / (np.linalg.norm(emb2) + 1e-10)
    return float(np.dot(emb1_n, emb2_n))

def find_optimal_threshold(same_person_pairs: list[tuple],
                             diff_person_pairs: list[tuple]) -> dict:
    """
    Find optimal threshold by analyzing the ROC curve.

    same_person_pairs: list of (emb1, emb2) pairs - same person
    diff_person_pairs: list of (emb1, emb2) pairs - different people

    Returns: {threshold, eer, auc, far, frr}
    """
    scores = []
    labels = []

    for emb1, emb2 in same_person_pairs:
        scores.append(cosine_similarity(emb1, emb2))
        labels.append(1)  # same person

    for emb1, emb2 in diff_person_pairs:
        scores.append(cosine_similarity(emb1, emb2))
        labels.append(0)  # different people

    scores_arr = np.array(scores)
    labels_arr = np.array(labels)

    # ROC curve
    fpr, tpr, thresholds = roc_curve(labels_arr, scores_arr)
    roc_auc = auc(fpr, tpr)

    # Equal Error Rate (EER): point where FAR = FRR
    fnr = 1 - tpr
    eer_idx = np.argmin(np.abs(fpr - fnr))
    eer = (fpr[eer_idx] + fnr[eer_idx]) / 2.0
    optimal_threshold = thresholds[eer_idx]

    # Metrics at optimal threshold
    predictions = (scores_arr >= optimal_threshold).astype(int)
    tp = np.sum((predictions == 1) & (labels_arr == 1))
    fp = np.sum((predictions == 1) & (labels_arr == 0))
    fn = np.sum((predictions == 0) & (labels_arr == 1))
    tn = np.sum((predictions == 0) & (labels_arr == 0))

    far = fp / (fp + tn) if (fp + tn) > 0 else 0  # False Accept Rate
    frr = fn / (fn + tp) if (fn + tp) > 0 else 0  # False Reject Rate

    print(f"=== Face Verification Metrics ===")
    print(f"AUC-ROC: {roc_auc:.4f}")
    print(f"EER: {eer*100:.2f}%")
    print(f"Optimal threshold: {optimal_threshold:.4f}")
    print(f"FAR @ EER: {far*100:.2f}%")
    print(f"FRR @ EER: {frr*100:.2f}%")

    return {
        'threshold': float(optimal_threshold),
        'eer': float(eer),
        'auc': float(roc_auc),
        'far': float(far),
        'frr': float(frr)
    }

import cv2
import numpy as np
import torch
import torch.nn as nn
import torchvision.models as models
import torchvision.transforms as transforms
from typing import Optional

class LivenessDetector:
    """
    Liveness detection based on two complementary signals:
    1. Texture analysis (CNN) - detects print attacks
    2. Micro-motion analysis - detects replay attacks (static videos lack natural micro-movements)

    For serious deployment, consider: SilentFace, FAS-SGTD, CentralDiff-CNN
    Datasets: CelebA-Spoof, OULU-NPU, MSU-MFSD
    """

    def __init__(self, model_path: Optional[str] = None,
                 device: str = 'auto'):
        self.device = torch.device(
            'cuda' if torch.cuda.is_available() and device == 'auto'
            else 'cpu'
        )
        self.model = self._build_model(model_path)
        self.model.eval()

        self.transform = transforms.Compose([
            transforms.ToPILImage(),
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                                  std=[0.229, 0.224, 0.225])
        ])

        self.frame_buffer: list[np.ndarray] = []
        self.buffer_size = 10  # 10 frames ~= 333ms @ 30FPS

    def _build_model(self, model_path: Optional[str]) -> nn.Module:
        """
        MobileNetV2 fine-tuned for binary classification: real vs spoof.
        Lightweight (3.4M params) - suitable for edge deployment.
        """
        model = models.mobilenet_v2(pretrained=False)
        model.classifier = nn.Sequential(
            nn.Dropout(0.2),
            nn.Linear(model.last_channel, 2)  # [spoof, real]
        )
        if model_path:
            state_dict = torch.load(model_path, map_location=self.device)
            model.load_state_dict(state_dict)
        return model.to(self.device)

    def is_live_texture(self, face_roi: np.ndarray,
                         threshold: float = 0.7) -> tuple[bool, float]:
        """
        CNN texture analysis: classify face as real or spoof.

        face_roi: BGR face crop [H, W, 3]
        Returns: (is_live, confidence_score)
        """
        img_rgb = cv2.cvtColor(face_roi, cv2.COLOR_BGR2RGB)
        tensor = self.transform(img_rgb).unsqueeze(0).to(self.device)

        with torch.no_grad():
            logits = self.model(tensor)
            probs = torch.softmax(logits, dim=1)
            live_prob = probs[0, 1].item()  # index 1 = "real"

        return live_prob >= threshold, live_prob

    def analyze_micro_motion(self, frame_bgr: np.ndarray) -> tuple[bool, float]:
        """
        Micro-motion analysis: detects natural face movements (micro-expressions,
        breathing, eye blinks) absent in photos/static videos.

        Returns: (has_micro_motion, motion_score)
        A replay video typically has motion_score < 0.5
        """
        gray = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2GRAY)
        self.frame_buffer.append(gray)

        if len(self.frame_buffer) > self.buffer_size:
            self.frame_buffer.pop(0)

        if len(self.frame_buffer) < 3:
            return True, 1.0  # Not enough frames, assume live

        if len(self.frame_buffer) >= self.buffer_size:
            all_flows = []
            for i in range(len(self.frame_buffer) - 1):
                flow = cv2.calcOpticalFlowFarneback(
                    self.frame_buffer[i], self.frame_buffer[i+1],
                    None, 0.5, 3, 15, 3, 5, 1.2, 0
                )
                magnitude, _ = cv2.cartToPolar(flow[..., 0], flow[..., 1])
                all_flows.append(np.mean(magnitude))

            # High variance = irregular natural movements = live
            motion_variance = np.var(all_flows)
            motion_score = min(1.0, motion_variance * 100)
        else:
            motion_score = 0.5

        return motion_score > 0.3, float(motion_score)

    def predict(self, face_roi: np.ndarray,
                frame_bgr: np.ndarray) -> dict:
        """
        Combined prediction: texture CNN + micro-motion.
        Conservative AND fusion rule for security.
        """
        is_live_tex, tex_score = self.is_live_texture(face_roi)
        has_motion, motion_score = self.analyze_micro_motion(frame_bgr)

        combined_score = 0.6 * tex_score + 0.4 * motion_score
        is_live = is_live_tex and (motion_score > 0.2)

        return {
            'is_live': is_live,
            'combined_score': combined_score,
            'texture_score': tex_score,
            'motion_score': motion_score,
            'verdict': 'LIVE' if is_live else 'SPOOF'
        }

import faiss
import numpy as np
import pickle
from pathlib import Path
from collections import Counter

class FAISSFaceDatabase:
    """
    Scalable face embedding database using FAISS.
    Approximate search (HNSW) for 1M+ embeddings in < 1ms.

    Install: pip install faiss-cpu  (or faiss-gpu for GPU)
    """

    def __init__(self, embedding_dim: int = 512,
                 db_path: str = 'faiss_face_db',
                 index_type: str = 'hnsw'):
        """
        index_type:
          'flat'  - Exact search, O(n), for < 100K embeddings
          'hnsw'  - Approximate HNSW, for 100K - 10M embeddings
          'ivf'   - Inverted File Index, for 10M+ embeddings
        """
        self.embedding_dim = embedding_dim
        self.db_path = Path(db_path)
        self.db_path.mkdir(exist_ok=True)
        self.index_type = index_type

        self.index = self._build_index()
        self.id_to_name: dict[int, str] = {}
        self.next_id = 0

        if (self.db_path / 'index.faiss').exists():
            self._load()

    def _build_index(self) -> faiss.Index:
        """Build appropriate FAISS index."""
        if self.index_type == 'flat':
            return faiss.IndexFlatIP(self.embedding_dim)

        elif self.index_type == 'hnsw':
            # M=32: connections per node (higher = more accurate but more RAM)
            # efConstruction=200: index quality during build
            index = faiss.IndexHNSWFlat(self.embedding_dim, 32)
            index.hnsw.efConstruction = 200
            index.hnsw.efSearch = 64  # accuracy/speed tradeoff at query time
            return index

        elif self.index_type == 'ivf':
            n_lists = 100  # number of clusters
            quantizer = faiss.IndexFlatIP(self.embedding_dim)
            return faiss.IndexIVFFlat(quantizer, self.embedding_dim,
                                       n_lists, faiss.METRIC_INNER_PRODUCT)

        raise ValueError(f"Unknown index type: {self.index_type}")

    def add_person(self, name: str,
                    embeddings: list[np.ndarray]) -> int:
        """Add multiple embeddings for the same person."""
        for emb in embeddings:
            emb_norm = emb / (np.linalg.norm(emb) + 1e-10)
            self.index.add(emb_norm.astype(np.float32).reshape(1, -1))
            self.id_to_name[self.next_id] = name
            self.next_id += 1
        return len(embeddings)

    def identify(self, query_embedding: np.ndarray,
                  threshold: float = 0.5) -> tuple[str, float]:
        """Identify person with highest similarity (top-3 majority vote)."""
        if self.next_id == 0:
            return 'unknown', 0.0

        emb_norm = query_embedding / (np.linalg.norm(query_embedding) + 1e-10)
        emb_norm = emb_norm.astype(np.float32).reshape(1, -1)

        k = min(3, self.next_id)
        similarities, indices = self.index.search(emb_norm, k)

        candidates = []
        for sim, idx in zip(similarities[0], indices[0]):
            if idx != -1 and sim >= threshold:
                candidates.append((self.id_to_name[int(idx)], float(sim)))

        if not candidates:
            return 'unknown', 0.0

        names = [c[0] for c in candidates]
        best_name = Counter(names).most_common(1)[0][0]
        best_sim = max(c[1] for c in candidates if c[0] == best_name)

        return best_name, best_sim

    def save(self) -> None:
        """Save FAISS index and ID->name mapping to disk."""
        faiss.write_index(self.index,
                          str(self.db_path / 'index.faiss'))
        with open(self.db_path / 'id_map.pkl', 'wb') as f:
            pickle.dump({'id_to_name': self.id_to_name,
                         'next_id': self.next_id}, f)

    def _load(self) -> None:
        """Load FAISS index and mapping from disk."""
        self.index = faiss.read_index(str(self.db_path / 'index.faiss'))
        with open(self.db_path / 'id_map.pkl', 'rb') as f:
            data = pickle.load(f)
            self.id_to_name = data['id_to_name']
            self.next_id = data['next_id']
        print(f"Database loaded: {self.next_id} embeddings, "
              f"{len(set(self.id_to_name.values()))} identities")

    def stats(self) -> dict:
        names = list(self.id_to_name.values())
        name_counts = Counter(names)
        return {
            'total_embeddings': self.next_id,
            'total_identities': len(name_counts),
            'avg_per_person': np.mean(list(name_counts.values()))
                              if name_counts else 0,
            'index_type': self.index_type
        }


# Benchmark: sklearn KNN vs FAISS
def benchmark_backends(n_identities: int = 10000,
                        embs_per_person: int = 5) -> None:
    """Compare search times: sklearn KNN vs FAISS HNSW."""
    import time
    from sklearn.neighbors import KNeighborsClassifier
    from sklearn.preprocessing import normalize

    n_total = n_identities * embs_per_person
    dim = 512

    embeddings = np.random.randn(n_total, dim).astype(np.float32)
    embeddings = normalize(embeddings)
    labels = np.repeat(np.arange(n_identities), embs_per_person)
    query = normalize(np.random.randn(1, dim).astype(np.float32))

    # sklearn KNN
    knn = KNeighborsClassifier(n_neighbors=3, metric='cosine',
                                algorithm='brute')
    knn.fit(embeddings, labels)
    t0 = time.perf_counter()
    for _ in range(100):
        knn.predict(query)
    knn_ms = (time.perf_counter() - t0) / 100 * 1000

    # FAISS HNSW
    index = faiss.IndexHNSWFlat(dim, 32)
    index.add(embeddings)
    t0 = time.perf_counter()
    for _ in range(100):
        index.search(query, 3)
    faiss_ms = (time.perf_counter() - t0) / 100 * 1000

    print(f"Search benchmark ({n_total:,} embeddings, dim={dim}):")
    print(f"  sklearn KNN: {knn_ms:.2f} ms/query")
    print(f"  FAISS HNSW:  {faiss_ms:.3f} ms/query")
    print(f"  Speedup:     {knn_ms/faiss_ms:.0f}x")