|
| 1 | +import logging |
| 2 | +import warnings |
| 3 | +from pathlib import Path |
| 4 | + |
| 5 | +import cv2 |
| 6 | +import numpy as np |
| 7 | + |
| 8 | +warnings.filterwarnings("ignore", message=".*estimate.*deprecated.*", category=FutureWarning) |
| 9 | +from insightface.app import FaceAnalysis |
| 10 | + |
| 11 | + |
| 12 | +PHOTO_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.heic', '.heif'} |
| 13 | + |
| 14 | + |
| 15 | +class FaceRecognizer: |
| 16 | + def __init__(self, reference_dir, similarity_threshold=0.4, det_size=(640, 640), name_map=None): |
| 17 | + self.similarity_threshold = similarity_threshold |
| 18 | + self.name_map = name_map or {} |
| 19 | + |
| 20 | + self.app = FaceAnalysis(name='buffalo_l', providers=['CPUExecutionProvider']) |
| 21 | + self.app.prepare(ctx_id=-1, det_size=det_size) |
| 22 | + |
| 23 | + self.centroids = self._load_references(Path(reference_dir)) |
| 24 | + if self.centroids: |
| 25 | + logging.info(f"Face recognition: loaded {len(self.centroids)} persons: {', '.join(self.centroids.keys())}") |
| 26 | + else: |
| 27 | + logging.warning("Face recognition: no reference embeddings loaded") |
| 28 | + |
| 29 | + def _load_references(self, reference_dir): |
| 30 | + centroids = {} |
| 31 | + for person_dir in sorted(reference_dir.iterdir()): |
| 32 | + if not person_dir.is_dir(): |
| 33 | + continue |
| 34 | + embeddings = [] |
| 35 | + for img_file in sorted(person_dir.iterdir()): |
| 36 | + if img_file.suffix.lower() not in PHOTO_EXTENSIONS: |
| 37 | + continue |
| 38 | + img = cv2.imread(str(img_file)) |
| 39 | + if img is None: |
| 40 | + logging.warning(f"Face ref: cannot read {img_file}") |
| 41 | + continue |
| 42 | + faces = self.app.get(img) |
| 43 | + if len(faces) == 0: |
| 44 | + logging.warning(f"Face ref: no face in {img_file}") |
| 45 | + elif len(faces) > 1: |
| 46 | + logging.warning(f"Face ref: multiple faces in {img_file}, skipping") |
| 47 | + else: |
| 48 | + embeddings.append(faces[0].embedding) |
| 49 | + if embeddings: |
| 50 | + centroids[person_dir.name] = np.mean(embeddings, axis=0) |
| 51 | + else: |
| 52 | + logging.warning(f"Face ref: no usable faces for '{person_dir.name}'") |
| 53 | + return centroids |
| 54 | + |
| 55 | + def _resolve_name(self, folder_name): |
| 56 | + if folder_name in self.name_map: |
| 57 | + return self.name_map[folder_name] |
| 58 | + return folder_name[0].upper() + folder_name[1:] |
| 59 | + |
| 60 | + def identify_faces(self, image_path): |
| 61 | + try: |
| 62 | + img = cv2.imread(str(image_path)) |
| 63 | + if img is None: |
| 64 | + return [] |
| 65 | + faces = self.app.get(img) |
| 66 | + if not faces: |
| 67 | + return [] |
| 68 | + except Exception as e: |
| 69 | + logging.warning(f"Face detection failed for {image_path}: {e}") |
| 70 | + return [] |
| 71 | + |
| 72 | + identified = [] |
| 73 | + for face in faces: |
| 74 | + best_name = None |
| 75 | + best_score = -1 |
| 76 | + for name, centroid in self.centroids.items(): |
| 77 | + score = np.dot(face.embedding, centroid) / ( |
| 78 | + np.linalg.norm(face.embedding) * np.linalg.norm(centroid) |
| 79 | + ) |
| 80 | + if score > best_score: |
| 81 | + best_score = score |
| 82 | + best_name = name |
| 83 | + if best_name and best_score >= self.similarity_threshold: |
| 84 | + display_name = self._resolve_name(best_name) |
| 85 | + if display_name not in identified: |
| 86 | + identified.append(display_name) |
| 87 | + |
| 88 | + return identified |
0 commit comments