diff --git a/backend/app/services/vision/faces.py b/backend/app/services/vision/faces.py index 42fbbf0..63c3448 100644 --- a/backend/app/services/vision/faces.py +++ b/backend/app/services/vision/faces.py @@ -68,7 +68,7 @@ class YuNetSFaceProcessor(FaceProcessor): def process(self, image: np.ndarray) -> list[FaceDetection]: orig_h, orig_w = image.shape[:2] - # Scale image for YuNet (expects fixed input size) + # 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) @@ -79,12 +79,17 @@ class YuNetSFaceProcessor(FaceProcessor): dtype=np.uint8, ) - # YuNet expects BGR, uint8, NHWC - bgr = resized[:, :, ::-1].copy() + # 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 + + # YuNet expects BGR + bgr = canvas[:, :, ::-1].copy() # Run detection det_input = self._detector.get_inputs()[0] - # YuNet uses dynamic input — reshape 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: @@ -109,11 +114,11 @@ class YuNetSFaceProcessor(FaceProcessor): if face_size < self._min_face_size: continue - # Rescale to original image coords - x1 = x / scale - y1 = y / scale - x2 = (x + w) / scale - y2 = (y + h) / scale + # 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 bbox = [ max(0, x1 / orig_w), @@ -124,7 +129,9 @@ class YuNetSFaceProcessor(FaceProcessor): # Extract landmarks (5 points) for alignment if len(det) >= 15: - landmarks = det[5:15].reshape(5, 2) / scale + 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 diff --git a/backend/app/tasks/vision.py b/backend/app/tasks/vision.py index 0f9d4d8..2677b15 100644 --- a/backend/app/tasks/vision.py +++ b/backend/app/tasks/vision.py @@ -5,24 +5,32 @@ face recognition. All tasks run on the dedicated `vision` queue with limited concurrency (memory-bound CPU inference). They read thumbnails generated by generate_thumbnails, so they MUST run after thumbs complete. + +DB access uses sync psycopg2 sessions (not asyncpg) because Celery +forks workers and asyncpg connections can't be shared across forks. """ -import asyncio import logging from pathlib import Path import numpy as np from celery import shared_task -from sqlalchemy import select, delete, text +from sqlalchemy import create_engine, text as sa_text, select, delete +from sqlalchemy.orm import Session, sessionmaker from PIL import Image -from app.database import AsyncSessionLocal -from app.models import Photo from app.models.embeddings import Embedding from app.config import settings logger = logging.getLogger(__name__) +def _get_sync_session() -> Session: + """Create a sync DB session for use in Celery workers.""" + sync_url = settings.database_url.replace("+asyncpg", "+psycopg2").replace("+aiosqlite", "") + engine = create_engine(sync_url, pool_pre_ping=True) + return sessionmaker(bind=engine)() + + def _load_thumb(photo_id: str, size: str = "medium") -> np.ndarray | None: """Load a thumbnail as an RGB numpy array.""" thumb_path = Path(f"/data/thumbs/{photo_id}/{size}.webp") @@ -38,10 +46,7 @@ def embed_photo(photo_id: str): """Generate CLIP embedding for a photo and store in pgvector.""" if not settings.vision.enabled: return {'status': 'skipped', 'reason': 'vision disabled'} - return asyncio.run(_embed_photo_async(photo_id)) - -async def _embed_photo_async(photo_id: str): image = _load_thumb(photo_id, "medium") # 640px if image is None: return {'status': 'error', 'message': 'thumbnail not found'} @@ -52,9 +57,9 @@ async def _embed_photo_async(photo_id: str): model_name = settings.vision.embedder.name - async with AsyncSessionLocal() as session: - # Upsert: delete existing then insert - await session.execute( + session = _get_sync_session() + try: + session.execute( delete(Embedding).where( Embedding.photo_id == photo_id, Embedding.model == model_name, @@ -66,7 +71,9 @@ async def _embed_photo_async(photo_id: str): vector=vector.tolist(), ) session.add(emb) - await session.commit() + session.commit() + finally: + session.close() logger.info("Embedded photo %s with %s", photo_id, model_name) return {'status': 'success', 'photo_id': photo_id} @@ -95,10 +102,7 @@ def ocr_photo(photo_id: str): """Run OCR on a photo and store text regions.""" if not settings.vision.enabled or not settings.vision.ocr.enabled: return {'status': 'skipped', 'reason': 'OCR disabled'} - return asyncio.run(_ocr_photo_async(photo_id)) - -async def _ocr_photo_async(photo_id: str): image = _load_thumb(photo_id, "large") # 1280px for better OCR accuracy if image is None: return {'status': 'error', 'message': 'thumbnail not found'} @@ -113,11 +117,9 @@ async def _ocr_photo_async(photo_id: str): from app.models.ocr_text import OCRText - async with AsyncSessionLocal() as session: - # Delete existing OCR results for this photo (re-run safe) - await session.execute( - delete(OCRText).where(OCRText.photo_id == photo_id) - ) + session = _get_sync_session() + try: + session.execute(delete(OCRText).where(OCRText.photo_id == photo_id)) for r in results: session.add(OCRText( photo_id=photo_id, @@ -126,7 +128,9 @@ async def _ocr_photo_async(photo_id: str): confidence=r.confidence, bbox=r.bbox, )) - await session.commit() + session.commit() + finally: + session.close() logger.info("OCR: %d text regions for photo %s", len(results), photo_id) return {'status': 'success', 'photo_id': photo_id, 'regions': len(results)} @@ -138,10 +142,7 @@ def detect_objects(photo_id: str): link via photo_tags with confidence/bbox/source.""" if not settings.vision.enabled or not settings.vision.detector.enabled: return {'status': 'skipped', 'reason': 'detection disabled'} - return asyncio.run(_detect_objects_async(photo_id)) - -async def _detect_objects_async(photo_id: str): image = _load_thumb(photo_id, "medium") # 640px if image is None: return {'status': 'error', 'message': 'thumbnail not found'} @@ -158,38 +159,47 @@ async def _detect_objects_async(photo_id: str): source_name = "vision:yolov8n" - async with AsyncSessionLocal() as session: + session = _get_sync_session() + try: # Wipe previous detection results for this photo from this model - await session.execute( + session.execute( delete(photo_tags).where( photo_tags.c.photo_id == photo_id, photo_tags.c.source == source_name, ) ) + # Group detections by label, keep highest confidence per label + best_per_label: dict[str, tuple[float, list]] = {} for det in detections: + if det.label not in best_per_label or det.confidence > best_per_label[det.label][0]: + best_per_label[det.label] = (det.confidence, det.bbox) + + for label, (confidence, bbox) in best_per_label.items(): # Find or create the object tag - result = await session.execute( - select(Tag).where(Tag.name == det.label, Tag.kind == 'object') - ) - tag = result.scalar_one_or_none() + tag = session.execute( + select(Tag).where(Tag.name == label, Tag.kind == 'object') + ).scalar_one_or_none() + if not tag: - tag = Tag(name=det.label, kind='object', source=source_name) + tag = Tag(name=label, kind='object', source=source_name) session.add(tag) - await session.flush() # get tag.id + session.flush() # get tag.id # Insert photo_tags association with ML metadata - await session.execute( + session.execute( photo_tags.insert().values( photo_id=photo_id, tag_id=tag.id, - confidence=det.confidence, - bbox=det.bbox, + confidence=confidence, + bbox=bbox, source=source_name, ) ) - await session.commit() + session.commit() + finally: + session.close() labels = [d.label for d in detections] logger.info("Detected %d objects in photo %s: %s", len(detections), photo_id, labels) @@ -202,10 +212,7 @@ def extract_faces(photo_id: str): handled separately by recluster_faces (periodic task).""" if not settings.vision.enabled or not settings.vision.faces.enabled: return {'status': 'skipped', 'reason': 'faces disabled'} - return asyncio.run(_extract_faces_async(photo_id)) - -async def _extract_faces_async(photo_id: str): image = _load_thumb(photo_id, "large") # 1280px for better face detection if image is None: return {'status': 'error', 'message': 'thumbnail not found'} @@ -220,20 +227,20 @@ async def _extract_faces_async(photo_id: str): from app.models.face_embedding import FaceEmbedding - async with AsyncSessionLocal() as session: - # Wipe previous face results for this photo (re-run safe) - await session.execute( - delete(FaceEmbedding).where(FaceEmbedding.photo_id == photo_id) - ) + session = _get_sync_session() + try: + session.execute(delete(FaceEmbedding).where(FaceEmbedding.photo_id == photo_id)) for face in faces: session.add(FaceEmbedding( photo_id=photo_id, bbox=face.bbox, vector=face.embedding.tolist(), quality=face.quality, - cluster_id=None, # assigned by recluster_faces + cluster_id=None, )) - await session.commit() + session.commit() + finally: + session.close() logger.info("Extracted %d faces from photo %s", len(faces), photo_id) return {'status': 'success', 'photo_id': photo_id, 'faces': len(faces)} @@ -242,36 +249,27 @@ async def _extract_faces_async(photo_id: str): @shared_task(name='recluster_faces', queue='vision') def recluster_faces(): """Run DBSCAN clustering over all face embeddings and assign/create - Tag(kind=face_cluster) entries. Should be called periodically or - manually after a batch of new faces is extracted.""" + Tag(kind=face_cluster) entries.""" if not settings.vision.enabled or not settings.vision.faces.enabled: return {'status': 'skipped', 'reason': 'faces disabled'} - return asyncio.run(_recluster_faces_async()) - -async def _recluster_faces_async(): from app.models.face_embedding import FaceEmbedding from app.models.tags import Tag from app.services.vision.clustering import cluster_faces - async with AsyncSessionLocal() as session: - # Load all face embeddings - result = await session.execute( + session = _get_sync_session() + try: + face_rows = session.execute( select(FaceEmbedding).order_by(FaceEmbedding.created_at) - ) - face_rows = result.scalars().all() + ).scalars().all() if len(face_rows) < 2: logger.info("Not enough faces for clustering (%d)", len(face_rows)) return {'status': 'success', 'clusters': 0} embeddings = np.array([f.vector for f in face_rows], dtype=np.float32) - labels = cluster_faces( - embeddings, - eps=settings.vision.faces.cluster_eps, - ) + labels = cluster_faces(embeddings, eps=settings.vision.faces.cluster_eps) - # Map cluster labels → Tag(kind=face_cluster) cluster_tag_map: dict[int, str] = {} source_name = "vision:sface" @@ -281,16 +279,15 @@ async def _recluster_faces_async(): continue if label not in cluster_tag_map: - # Check if a cluster tag already exists for this cluster cluster_name = f"Person {label + 1}" - tag_result = await session.execute( + tag = session.execute( select(Tag).where( Tag.kind == 'face_cluster', Tag.source == source_name, Tag.name == cluster_name, ) - ) - tag = tag_result.scalar_one_or_none() + ).scalar_one_or_none() + if not tag: tag = Tag( name=cluster_name, @@ -299,12 +296,14 @@ async def _recluster_faces_async(): representative_photo_id=face_rows[i].photo_id, ) session.add(tag) - await session.flush() + session.flush() cluster_tag_map[label] = tag.id face_rows[i].cluster_id = cluster_tag_map[label] - await session.commit() + session.commit() + finally: + session.close() n_clusters = len(cluster_tag_map) logger.info("Face clustering: %d clusters from %d faces", n_clusters, len(face_rows)) @@ -313,27 +312,25 @@ async def _recluster_faces_async(): @shared_task(name='backfill_vision') def backfill_vision(task: str | None = None, limit: int | None = None): - """Queue vision tasks for photos that haven't been processed yet.""" - return asyncio.run(_backfill_vision_async(task, limit)) - - -async def _backfill_vision_async(task: str | None, limit: int | None): + """Queue vision tasks for photos that haven't been processed yet. + Uses a sync DB connection to avoid asyncpg conflicts in Celery.""" model_name = settings.vision.embedder.name + sql = """ + SELECT p.id FROM photos p + LEFT JOIN embeddings e ON e.photo_id = p.id AND e.model = :model + WHERE e.photo_id IS NULL + AND p.processing_status = 'completed' + ORDER BY p.added_at DESC + """ + if limit: + sql += f" LIMIT {limit}" - async with AsyncSessionLocal() as session: - # Find photos without embeddings - stmt = text(""" - SELECT p.id FROM photos p - LEFT JOIN embeddings e ON e.photo_id = p.id AND e.model = :model - WHERE e.photo_id IS NULL - AND p.processing_status = 'completed' - ORDER BY p.created_at DESC - """) - if limit: - stmt = text(str(stmt) + f" LIMIT {limit}") - - result = await session.execute(stmt, {"model": model_name}) + session = _get_sync_session() + try: + result = session.execute(sa_text(sql), {"model": model_name}) photo_ids = [row[0] for row in result.fetchall()] + finally: + session.close() count = 0 for pid in photo_ids: diff --git a/docker-compose.yml b/docker-compose.yml index d1e1e57..0f243ad 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -62,7 +62,7 @@ services: context: ./backend dockerfile: Dockerfile container_name: mulita-worker - command: sh -c "python -m app.services.vision.bootstrap_models && celery -A app.tasks.celery worker --loglevel=${LOG_LEVEL:-info} --concurrency=${CELERYD_CONCURRENCY:-4}" + command: sh -c "python -m app.services.vision.bootstrap_models && celery -A app.tasks.celery worker --loglevel=${LOG_LEVEL:-info} --concurrency=${CELERYD_CONCURRENCY:-4} -Q default,high,low,vision" volumes: - ./mulita.yml:/app/config/mulita.yml:ro - ${PHOTO_DIRS:-./photos}:/photos:rw