"""Face embedding endpoints.""" import logging from typing import List import numpy as np from fastapi import APIRouter, HTTPException from app.face import ( fallback_avatar_embedding, get_faces_async, load_face_app, to_pixel_bbox, validate_embedding, ) from app.image import download_image from app.resources import http_client, inference_executor from app.models import ( BBox, EmbedAvatarResponse, EmbedImageResponse, EmbedRequest, FaceEmbedding, ) logger = logging.getLogger("face_service") router = APIRouter() @router.post("/embed-avatar", response_model=EmbedAvatarResponse) async def embed_avatar(req: EmbedRequest): """ Extract face embedding from an avatar image. Returns the largest detected face. If no face is detected, falls back to center crop embedding with score=0.0. """ logger.info("embed_avatar: image_url=%s", req.image_url) fa = load_face_app() img = await download_image(str(req.image_url), http_client, inference_executor) h, w = img.shape[:2] faces = await get_faces_async(fa, img, inference_executor) if len(faces) == 0: logger.warning( "embed_avatar: no faces detected image_url=%s size=%dx%d, using fallback", req.image_url, w, h, ) fallback = fallback_avatar_embedding(fa, img, w, h) if fallback is None: raise HTTPException( status_code=422, detail="No face detected in avatar image", ) emb, bbox, score = fallback logger.info( "embed_avatar: using fallback bbox=%s score=%.4f embedding_len=%d", bbox, score, len(emb), ) return EmbedAvatarResponse(embedding=emb, bbox=bbox, score=score) # Sort by face area (largest first) faces.sort( key=lambda f: (f.bbox[2] - f.bbox[0]) * (f.bbox[3] - f.bbox[1]), reverse=True, ) face = faces[0] emb = face.normed_embedding.astype(np.float32) # Validate embedding if not validate_embedding(emb): logger.error("embed_avatar: embedding contains NaN/Inf values") raise HTTPException( status_code=422, detail="Failed to generate valid face embedding", ) emb_list = emb.tolist() bbox = to_pixel_bbox(face.bbox, w, h) score = float(getattr(face, "det_score", 1.0)) logger.info( "embed_avatar: using face bbox=%s score=%.4f embedding_len=%d", face.bbox, score, len(emb_list), ) return EmbedAvatarResponse(embedding=emb_list, bbox=bbox, score=score) @router.post("/embed-image", response_model=EmbedImageResponse) async def embed_image(req: EmbedRequest): """ Extract face embeddings from all faces in an image. Returns all detected faces sorted by detection score (highest first). Returns empty list if no faces detected. """ fa = load_face_app() img = await download_image(str(req.image_url), http_client, inference_executor) h, w = img.shape[:2] faces = await get_faces_async(fa, img, inference_executor) if len(faces) == 0: logger.warning( "embed_image: no faces detected image_url=%s size=%dx%d", req.image_url, w, h, ) return EmbedImageResponse(faces=[]) logger.info( "embed_image: detected %d faces image_url=%s size=%dx%d", len(faces), req.image_url, w, h, ) # Sort by detection score (highest first) faces.sort( key=lambda f: float(getattr(f, "det_score", 1.0)), reverse=True, ) result: List[FaceEmbedding] = [] for f in faces: emb = f.normed_embedding.astype(np.float32) # Skip faces with invalid embeddings if not validate_embedding(emb): logger.warning("embed_image: skipping face with NaN/Inf embedding") continue emb_list = emb.tolist() bbox = to_pixel_bbox(f.bbox, w, h) score = float(getattr(f, "det_score", 1.0)) result.append(FaceEmbedding(bbox=bbox, score=score, embedding=emb_list)) return EmbedImageResponse(faces=result)