diff --git a/backend/app/services/vision/faces.py b/backend/app/services/vision/faces.py index 63c3448..d79309c 100644 --- a/backend/app/services/vision/faces.py +++ b/backend/app/services/vision/faces.py @@ -1,15 +1,20 @@ """ -Face detection (YuNet) + recognition (SFace) using ONNX Runtime. +Face detection (YuNet) + recognition (SFace) using OpenCV DNN. + +YuNet is loaded via cv2.FaceDetectorYN which handles the multi-scale +anchor decoding and NMS internally. SFace recognition uses raw ONNX +Runtime for the 128-d embedding. Both models are from opencv_zoo (Apache 2.0 license). Expects {models_dir}/face/: - - yunet.onnx (~75 KB) + - yunet.onnx (~233 KB) - sface.onnx (~37 MB, 128-d embeddings) """ import logging from pathlib import Path import numpy as np +import cv2 import onnxruntime as ort from app.config import VisionSettings @@ -17,131 +22,114 @@ from app.services.vision.base import FaceProcessor, FaceDetection logger = logging.getLogger(__name__) -_YUNET_INPUT_SIZE = 640 - def _align_face(image: np.ndarray, landmarks: np.ndarray) -> np.ndarray: - """Align and crop a 112x112 face patch using 5-point landmarks. - landmarks shape: (5, 2) — left_eye, right_eye, nose, left_mouth, right_mouth.""" - from PIL import Image - import math - + """Align and crop a 112x112 face patch using 5-point landmarks.""" left_eye = landmarks[0] right_eye = landmarks[1] dx = right_eye[0] - left_eye[0] dy = right_eye[1] - left_eye[1] - angle = math.degrees(math.atan2(dy, dx)) + angle = np.degrees(np.arctan2(dy, dx)) eye_center = ((left_eye[0] + right_eye[0]) / 2, (left_eye[1] + right_eye[1]) / 2) - eye_dist = math.sqrt(dx * dx + dy * dy) - scale = 64.0 / max(eye_dist, 1e-6) # target: eyes at ~64px apart in 112x112 + eye_dist = np.sqrt(dx * dx + dy * dy) - img = Image.fromarray(image) - img = img.rotate(-angle, center=eye_center, resample=Image.BICUBIC) + M = cv2.getRotationMatrix2D(eye_center, angle, 1.0) + rotated = cv2.warpAffine(image, M, (image.shape[1], image.shape[0])) + # Crop around face center + scale = 64.0 / max(eye_dist, 1e-6) cx, cy = eye_center half = 56.0 / scale - crop = img.crop((int(cx - half), int(cy - half * 0.8), int(cx + half), int(cy + half * 1.2))) - crop = crop.resize((112, 112), Image.BICUBIC) - return np.array(crop, dtype=np.float32) + x1 = max(0, int(cx - half)) + y1 = max(0, int(cy - half * 0.8)) + x2 = min(rotated.shape[1], int(cx + half)) + y2 = min(rotated.shape[0], int(cy + half * 1.2)) + crop = rotated[y1:y2, x1:x2] + + if crop.size == 0: + return np.zeros((112, 112, 3), dtype=np.float32) + + return cv2.resize(crop, (112, 112)).astype(np.float32) class YuNetSFaceProcessor(FaceProcessor): def __init__(self, settings: VisionSettings): face_dir = Path(settings.models_dir) / "face" - yunet_path = face_dir / "yunet.onnx" - sface_path = face_dir / "sface.onnx" + yunet_path = str(face_dir / "yunet.onnx") + sface_path = str(face_dir / "sface.onnx") + # YuNet via OpenCV's FaceDetectorYN — handles anchor decoding + NMS + self._detector = cv2.FaceDetectorYN.create( + yunet_path, + "", + (640, 640), + settings.faces.recognition_threshold, + 0.3, # NMS threshold + 5000, # top_k + ) + logger.info("YuNet face detector loaded via OpenCV") + + # SFace via ONNX Runtime opts = ort.SessionOptions() opts.inter_op_num_threads = 2 opts.intra_op_num_threads = 2 - - logger.info("Loading YuNet from %s", yunet_path) - self._detector = ort.InferenceSession(str(yunet_path), opts, providers=["CPUExecutionProvider"]) - - logger.info("Loading SFace from %s", sface_path) - self._recognizer = ort.InferenceSession(str(sface_path), opts, providers=["CPUExecutionProvider"]) + ort.set_default_logger_severity(3) + self._recognizer = ort.InferenceSession(sface_path, opts, providers=["CPUExecutionProvider"]) + logger.info("SFace recognizer loaded via ONNX Runtime") self._min_face_size = settings.faces.min_face_size - self._score_threshold = settings.faces.recognition_threshold def process(self, image: np.ndarray) -> list[FaceDetection]: orig_h, orig_w = image.shape[:2] - # Scale + letterbox to exactly 640x640 (YuNet fixed input) - scale = min(_YUNET_INPUT_SIZE / orig_w, _YUNET_INPUT_SIZE / orig_h) - new_w = int(orig_w * scale) - new_h = int(orig_h * scale) + # Convert RGB → BGR for OpenCV + bgr = cv2.cvtColor(image, cv2.COLOR_RGB2BGR) - from PIL import Image as PILImage - resized = np.array( - PILImage.fromarray(image).resize((new_w, new_h), PILImage.BICUBIC), - dtype=np.uint8, - ) + # Set input size to actual image dimensions + self._detector.setInputSize((orig_w, orig_h)) - # Letterbox pad to 640x640 - canvas = np.full((_YUNET_INPUT_SIZE, _YUNET_INPUT_SIZE, 3), 128, dtype=np.uint8) - pad_y = (_YUNET_INPUT_SIZE - new_h) // 2 - pad_x = (_YUNET_INPUT_SIZE - new_w) // 2 - canvas[pad_y:pad_y + new_h, pad_x:pad_x + new_w] = resized + # Detect faces + _, faces_raw = self._detector.detect(bgr) - # YuNet expects BGR - bgr = canvas[:, :, ::-1].copy() - - # Run detection - det_input = self._detector.get_inputs()[0] - blob = bgr.astype(np.float32)[np.newaxis] # (1, H, W, 3) - # Some YuNet ONNX exports expect (1, 3, H, W) - if det_input.shape and len(det_input.shape) == 4 and det_input.shape[1] == 3: - blob = blob.transpose(0, 3, 1, 2) - - detections_raw = self._detector.run(None, {det_input.name: blob}) - dets = detections_raw[0] # (N, 15): x,y,w,h,score, 5x landmark pairs - - if dets is None or len(dets) == 0: + if faces_raw is None or len(faces_raw) == 0: return [] results = [] - for det in dets: - score = float(det[4]) if len(det) > 4 else float(det[-1]) - if score < self._score_threshold: - continue - - x, y, w, h = det[0], det[1], det[2], det[3] + for face in faces_raw: + # face: [x, y, w, h, right_eye_x, right_eye_y, left_eye_x, left_eye_y, + # nose_x, nose_y, right_mouth_x, right_mouth_y, left_mouth_x, left_mouth_y, score] + x, y, w, h = int(face[0]), int(face[1]), int(face[2]), int(face[3]) + score = float(face[14]) # Filter small faces - face_size = max(w, h) / scale + face_size = max(w, h) if face_size < self._min_face_size: continue - # Remove letterbox padding and rescale to original image coords - x1 = (x - pad_x) / scale - y1 = (y - pad_y) / scale - x2 = (x + w - pad_x) / scale - y2 = (y + h - pad_y) / scale - + # Normalized bbox bbox = [ - max(0, x1 / orig_w), - max(0, y1 / orig_h), - min(1, x2 / orig_w), - min(1, y2 / orig_h), + max(0, x / orig_w), + max(0, y / orig_h), + min(1, (x + w) / orig_w), + min(1, (y + h) / orig_h), ] - # Extract landmarks (5 points) for alignment - if len(det) >= 15: - landmarks = det[5:15].reshape(5, 2) - landmarks[:, 0] = (landmarks[:, 0] - pad_x) / scale - landmarks[:, 1] = (landmarks[:, 1] - pad_y) / scale - else: - # Fallback: no landmarks, skip recognition - continue + # Extract 5-point landmarks for alignment + landmarks = np.array([ + [face[4], face[5]], # right eye + [face[6], face[7]], # left eye + [face[8], face[9]], # nose + [face[10], face[11]], # right mouth + [face[12], face[13]], # left mouth + ], dtype=np.float32) # Align face for recognition face_crop = _align_face(image, landmarks) - # SFace expects (1, 3, 112, 112) float32, BGR, normalized - face_bgr = face_crop[:, :, ::-1].copy() - face_blob = (face_bgr / 255.0).transpose(2, 0, 1)[np.newaxis].astype(np.float32) + # SFace expects (1, 3, 112, 112) float32, BGR + face_bgr = cv2.cvtColor(face_crop.astype(np.uint8), cv2.COLOR_RGB2BGR) + face_blob = (face_bgr.astype(np.float32) / 255.0).transpose(2, 0, 1)[np.newaxis] rec_input = self._recognizer.get_inputs()[0].name embedding = self._recognizer.run(None, {rec_input: face_blob})[0][0] diff --git a/backend/app/tasks/vision.py b/backend/app/tasks/vision.py index 2677b15..ac4df17 100644 --- a/backend/app/tasks/vision.py +++ b/backend/app/tasks/vision.py @@ -206,6 +206,42 @@ def detect_objects(photo_id: str): return {'status': 'success', 'photo_id': photo_id, 'objects': len(detections)} +def _load_original(photo_id: str) -> np.ndarray | None: + """Load the original photo file as an RGB numpy array, resized to + max 1280px on the longest edge for face detection.""" + from sqlalchemy import create_engine, select as sa_select, text as sa_text + from app.models import Photo + + session = _get_sync_session() + try: + photo = session.execute( + sa_select(Photo).where(Photo.id == photo_id) + ).scalar_one_or_none() + if not photo or not photo.filepath: + return None + filepath = photo.filepath + finally: + session.close() + + if not Path(filepath).exists(): + logger.warning("Original file not found: %s", filepath) + return None + + try: + img = Image.open(filepath).convert("RGB") + # Cap at 4000px on longest edge to avoid OOM, but keep as large + # as possible for face detection accuracy + max_dim = 4000 + w, h = img.size + if max(w, h) > max_dim: + scale = max_dim / max(w, h) + img = img.resize((int(w * scale), int(h * scale)), Image.BICUBIC) + return np.array(img) + except Exception as e: + logger.warning("Failed to load original %s: %s", filepath, e) + return None + + @shared_task(name='extract_faces', queue='vision') def extract_faces(photo_id: str): """Detect faces and store recognition embeddings. Clustering is @@ -213,9 +249,13 @@ def extract_faces(photo_id: str): if not settings.vision.enabled or not settings.vision.faces.enabled: return {'status': 'skipped', 'reason': 'faces disabled'} - image = _load_thumb(photo_id, "large") # 1280px for better face detection + # Use original file for face detection — thumbnails are often too + # small (240px) for reliable face detection. + image = _load_original(photo_id) if image is None: - return {'status': 'error', 'message': 'thumbnail not found'} + image = _load_thumb(photo_id, "large") + if image is None: + return {'status': 'error', 'message': 'no image available'} from app.services.vision.registry import registry face_proc = registry.get_face_processor()