feat: Update documentation and API for face identification and people management
This commit enhances the README with detailed instructions on the automatic database initialization and schema compatibility between the web and desktop versions. It also introduces new API endpoints for managing unidentified faces and people, including listing, creating, and identifying faces. The schemas for these operations have been updated to reflect the new data structures. Additionally, tests have been added to ensure the functionality of the new API features, improving overall coverage and reliability.
This commit is contained in:
tanyar09
+241
-13
@@ -2,11 +2,27 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter, HTTPException, status
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||||
from fastapi.responses import FileResponse, Response
|
||||
from rq import Queue
|
||||
from redis import Redis
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from src.web.schemas.faces import ProcessFacesRequest, ProcessFacesResponse
|
||||
from src.web.db.session import get_db
|
||||
from src.web.schemas.faces import (
|
||||
ProcessFacesRequest,
|
||||
ProcessFacesResponse,
|
||||
UnidentifiedFacesQuery,
|
||||
UnidentifiedFacesResponse,
|
||||
FaceItem,
|
||||
SimilarFacesResponse,
|
||||
SimilarFaceItem,
|
||||
IdentifyFaceRequest,
|
||||
IdentifyFaceResponse,
|
||||
)
|
||||
from src.web.schemas.people import PersonCreateRequest, PersonResponse
|
||||
from src.web.db.models import Face, Person, PersonEncoding
|
||||
from src.web.services.face_service import list_unidentified_faces, find_similar_faces
|
||||
# Note: Function passed as string path to avoid RQ serialization issues
|
||||
|
||||
router = APIRouter(prefix="/faces", tags=["faces"])
|
||||
@@ -63,19 +79,231 @@ def process_faces(request: ProcessFacesRequest) -> ProcessFacesResponse:
|
||||
)
|
||||
|
||||
|
||||
@router.get("/unidentified")
|
||||
def get_unidentified_faces() -> dict:
|
||||
"""Get unidentified faces - placeholder for Phase 2."""
|
||||
return {"message": "Unidentified faces endpoint - to be implemented in Phase 2"}
|
||||
@router.get("/unidentified", response_model=UnidentifiedFacesResponse)
|
||||
def get_unidentified_faces(
|
||||
page: int = Query(1, ge=1),
|
||||
page_size: int = Query(50, ge=1, le=200),
|
||||
min_quality: float = Query(0.0, ge=0.0, le=1.0),
|
||||
date_from: str | None = Query(None),
|
||||
date_to: str | None = Query(None),
|
||||
sort_by: str = Query("quality"),
|
||||
sort_dir: str = Query("desc"),
|
||||
db: Session = Depends(get_db),
|
||||
) -> UnidentifiedFacesResponse:
|
||||
"""Get unidentified faces with filters and pagination."""
|
||||
from datetime import date as _date
|
||||
|
||||
df = _date.fromisoformat(date_from) if date_from else None
|
||||
dt = _date.fromisoformat(date_to) if date_to else None
|
||||
|
||||
faces, total = list_unidentified_faces(
|
||||
db,
|
||||
page=page,
|
||||
page_size=page_size,
|
||||
min_quality=min_quality,
|
||||
date_from=df,
|
||||
date_to=dt,
|
||||
sort_by=sort_by,
|
||||
sort_dir=sort_dir,
|
||||
)
|
||||
|
||||
items = [
|
||||
FaceItem(
|
||||
id=f.id,
|
||||
photo_id=f.photo_id,
|
||||
quality_score=float(f.quality_score),
|
||||
face_confidence=float(getattr(f, "face_confidence", 0.0)),
|
||||
location=f.location,
|
||||
)
|
||||
for f in faces
|
||||
]
|
||||
return UnidentifiedFacesResponse(items=items, page=page, page_size=page_size, total=total)
|
||||
|
||||
|
||||
@router.post("/{face_id}/identify")
|
||||
def identify_face(face_id: int) -> dict:
|
||||
"""Identify face - placeholder for Phase 2."""
|
||||
return {
|
||||
"message": f"Identify face {face_id} - to be implemented in Phase 2",
|
||||
"id": face_id,
|
||||
}
|
||||
@router.get("/{face_id}/similar", response_model=SimilarFacesResponse)
|
||||
def get_similar_faces(face_id: int, db: Session = Depends(get_db)) -> SimilarFacesResponse:
|
||||
"""Return similar unidentified faces for a given face."""
|
||||
# Validate face exists
|
||||
base = db.query(Face).filter(Face.id == face_id).first()
|
||||
if not base:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Face {face_id} not found")
|
||||
|
||||
results = find_similar_faces(db, face_id)
|
||||
items = [
|
||||
SimilarFaceItem(
|
||||
id=f.id,
|
||||
photo_id=f.photo_id,
|
||||
similarity=sim,
|
||||
location=f.location,
|
||||
quality_score=float(f.quality_score),
|
||||
)
|
||||
for f, sim in results
|
||||
]
|
||||
return SimilarFacesResponse(base_face_id=face_id, items=items)
|
||||
|
||||
|
||||
@router.post("/{face_id}/identify", response_model=IdentifyFaceResponse)
|
||||
def identify_face(
|
||||
face_id: int,
|
||||
request: IdentifyFaceRequest,
|
||||
db: Session = Depends(get_db),
|
||||
) -> IdentifyFaceResponse:
|
||||
"""Assign a face (and optional batch) to a person, creating if needed.
|
||||
|
||||
Also inserts into person_encodings for each identified face as desktop does.
|
||||
"""
|
||||
# Validate target face
|
||||
face = db.query(Face).filter(Face.id == face_id).first()
|
||||
if not face:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Face {face_id} not found")
|
||||
|
||||
target_face_ids = [face_id]
|
||||
if request.additional_face_ids:
|
||||
target_face_ids.extend([fid for fid in request.additional_face_ids if fid != face_id])
|
||||
|
||||
# Get or create person
|
||||
created_person = False
|
||||
person: Person | None = None
|
||||
if request.person_id:
|
||||
person = db.query(Person).filter(Person.id == request.person_id).first()
|
||||
if not person:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="person_id not found")
|
||||
else:
|
||||
# Validate required fields for creation
|
||||
if not (request.first_name and request.last_name and request.date_of_birth):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="first_name, last_name and date_of_birth are required to create a person",
|
||||
)
|
||||
person = Person(
|
||||
first_name=request.first_name,
|
||||
last_name=request.last_name,
|
||||
middle_name=request.middle_name,
|
||||
maiden_name=request.maiden_name,
|
||||
date_of_birth=request.date_of_birth,
|
||||
)
|
||||
db.add(person)
|
||||
db.flush() # get person.id
|
||||
created_person = True
|
||||
|
||||
# Link faces and insert person_encodings
|
||||
identified_ids: list[int] = []
|
||||
for fid in target_face_ids:
|
||||
f = db.query(Face).filter(Face.id == fid).first()
|
||||
if not f:
|
||||
continue
|
||||
f.person_id = person.id
|
||||
db.add(f)
|
||||
# Insert person_encoding
|
||||
pe = PersonEncoding(
|
||||
person_id=person.id,
|
||||
face_id=f.id,
|
||||
encoding=f.encoding,
|
||||
quality_score=f.quality_score,
|
||||
detector_backend=f.detector_backend,
|
||||
model_name=f.model_name,
|
||||
)
|
||||
db.add(pe)
|
||||
identified_ids.append(f.id)
|
||||
|
||||
db.commit()
|
||||
return IdentifyFaceResponse(identified_face_ids=identified_ids, person_id=person.id, created_person=created_person)
|
||||
|
||||
|
||||
@router.get("/{face_id}/crop")
|
||||
def get_face_crop(face_id: int, db: Session = Depends(get_db)) -> Response:
|
||||
"""Serve face crop image extracted from photo using face location."""
|
||||
import os
|
||||
import json
|
||||
import ast
|
||||
import tempfile
|
||||
from PIL import Image
|
||||
from src.web.db.models import Face, Photo
|
||||
from src.utils.exif_utils import EXIFOrientationHandler
|
||||
|
||||
face = db.query(Face).filter(Face.id == face_id).first()
|
||||
if not face:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Face {face_id} not found")
|
||||
|
||||
photo = db.query(Photo).filter(Photo.id == face.photo_id).first()
|
||||
if not photo or not os.path.exists(photo.path):
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Photo file not found")
|
||||
|
||||
try:
|
||||
# Parse location (stored as text); support JSON or Python-literal formats
|
||||
if isinstance(face.location, str):
|
||||
try:
|
||||
location = json.loads(face.location)
|
||||
except Exception:
|
||||
location = ast.literal_eval(face.location)
|
||||
else:
|
||||
location = face.location
|
||||
|
||||
# DeepFace format: {x, y, w, h}
|
||||
x = int(location.get('x', 0) or 0)
|
||||
y = int(location.get('y', 0) or 0)
|
||||
w = int(location.get('w', 0) or 0)
|
||||
h = int(location.get('h', 0) or 0)
|
||||
|
||||
# If invalid dimensions, return client error similar to desktop behavior
|
||||
if w <= 0 or h <= 0:
|
||||
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid face box")
|
||||
|
||||
# Load image with EXIF correction (same as desktop)
|
||||
# Desktop logic: use corrected image only if it's not None AND orientation != 1
|
||||
corrected_image, original_orientation = EXIFOrientationHandler.correct_image_orientation_from_path(photo.path)
|
||||
if corrected_image is not None and original_orientation and original_orientation != 1:
|
||||
# Copy the image to ensure it's not tied to closed file handle
|
||||
image = corrected_image.copy()
|
||||
else:
|
||||
# Use original image if no correction needed or correction fails
|
||||
image = Image.open(photo.path)
|
||||
|
||||
# Calculate crop bounds with padding (20% like desktop)
|
||||
padding_x = max(0, int(w * 0.2))
|
||||
padding_y = max(0, int(h * 0.2))
|
||||
crop_left = max(0, int(x - padding_x))
|
||||
crop_top = max(0, int(y - padding_y))
|
||||
crop_right = min(int(image.width), int(x + w + padding_x))
|
||||
crop_bottom = min(int(image.height), int(y + h + padding_y))
|
||||
|
||||
# Ensure bounds make a valid box
|
||||
if crop_right <= crop_left or crop_bottom <= crop_top:
|
||||
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid crop bounds")
|
||||
|
||||
face_crop = image.crop((crop_left, crop_top, crop_right, crop_bottom))
|
||||
|
||||
# Resize if too small (minimum 200px width, like desktop)
|
||||
if face_crop.width > 0 and face_crop.width < 200:
|
||||
ratio = 200 / face_crop.width
|
||||
new_width = 200
|
||||
new_height = int(face_crop.height * ratio)
|
||||
face_crop = face_crop.resize((new_width, new_height), Image.Resampling.LANCZOS)
|
||||
|
||||
# Save to bytes instead of temp file to avoid Content-Length issues
|
||||
from io import BytesIO
|
||||
output = BytesIO()
|
||||
face_crop.save(output, format="JPEG", quality=95)
|
||||
output.seek(0)
|
||||
image_bytes = output.read()
|
||||
output.close()
|
||||
|
||||
return Response(
|
||||
content=image_bytes,
|
||||
media_type="image/jpeg",
|
||||
headers={
|
||||
"Content-Disposition": "inline",
|
||||
"Cache-Control": "public, max-age=3600",
|
||||
},
|
||||
)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[Faces API] get_face_crop error for face {face_id}: {e}")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Failed to extract face crop: {str(e)}",
|
||||
)
|
||||
|
||||
|
||||
@router.post("/auto-match")
|
||||
|
||||
+41
-16
@@ -2,28 +2,53 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from src.web.db.session import get_db
|
||||
from src.web.db.models import Person
|
||||
from src.web.schemas.people import (
|
||||
PeopleListResponse,
|
||||
PersonCreateRequest,
|
||||
PersonResponse,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/people", tags=["people"])
|
||||
|
||||
|
||||
@router.get("")
|
||||
def list_people() -> dict:
|
||||
"""List people - placeholder for Phase 2."""
|
||||
return {"message": "People endpoint - to be implemented in Phase 2"}
|
||||
@router.get("", response_model=PeopleListResponse)
|
||||
def list_people(db: Session = Depends(get_db)) -> PeopleListResponse:
|
||||
"""List all people sorted by last_name, first_name."""
|
||||
people = db.query(Person).order_by(Person.last_name.asc(), Person.first_name.asc()).all()
|
||||
items = [PersonResponse.model_validate(p) for p in people]
|
||||
return PeopleListResponse(items=items, total=len(items))
|
||||
|
||||
|
||||
@router.post("")
|
||||
def create_person() -> dict:
|
||||
"""Create person - placeholder for Phase 2."""
|
||||
return {"message": "Create person endpoint - to be implemented in Phase 2"}
|
||||
@router.post("", response_model=PersonResponse, status_code=status.HTTP_201_CREATED)
|
||||
def create_person(request: PersonCreateRequest, db: Session = Depends(get_db)) -> PersonResponse:
|
||||
"""Create a new person."""
|
||||
person = Person(
|
||||
first_name=request.first_name,
|
||||
last_name=request.last_name,
|
||||
middle_name=request.middle_name,
|
||||
maiden_name=request.maiden_name,
|
||||
date_of_birth=request.date_of_birth,
|
||||
)
|
||||
db.add(person)
|
||||
try:
|
||||
db.commit()
|
||||
except Exception as e:
|
||||
db.rollback()
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
||||
db.refresh(person)
|
||||
return PersonResponse.model_validate(person)
|
||||
|
||||
|
||||
@router.get("/{person_id}")
|
||||
def get_person(person_id: int) -> dict:
|
||||
"""Get person by ID - placeholder for Phase 2."""
|
||||
return {
|
||||
"message": f"Get person {person_id} - to be implemented in Phase 2",
|
||||
"id": person_id,
|
||||
}
|
||||
@router.get("/{person_id}", response_model=PersonResponse)
|
||||
def get_person(person_id: int, db: Session = Depends(get_db)) -> PersonResponse:
|
||||
"""Get person by ID."""
|
||||
person = db.query(Person).filter(Person.id == person_id).first()
|
||||
if not person:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Person {person_id} not found")
|
||||
return PersonResponse.model_validate(person)
|
||||
|
||||
|
||||
+37
-1
@@ -3,7 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
|
||||
from fastapi.responses import JSONResponse
|
||||
from fastapi.responses import JSONResponse, FileResponse
|
||||
from rq import Queue
|
||||
from redis import Redis
|
||||
from sqlalchemy.orm import Session
|
||||
@@ -149,3 +149,39 @@ def get_photo(photo_id: int, db: Session = Depends(get_db)) -> PhotoResponse:
|
||||
|
||||
return PhotoResponse.model_validate(photo)
|
||||
|
||||
|
||||
@router.get("/{photo_id}/image")
|
||||
def get_photo_image(photo_id: int, db: Session = Depends(get_db)) -> FileResponse:
|
||||
"""Serve photo image file for display (not download)."""
|
||||
import os
|
||||
import mimetypes
|
||||
from src.web.db.models import Photo
|
||||
|
||||
photo = db.query(Photo).filter(Photo.id == photo_id).first()
|
||||
if not photo:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Photo {photo_id} not found",
|
||||
)
|
||||
|
||||
if not os.path.exists(photo.path):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Photo file not found: {photo.path}",
|
||||
)
|
||||
|
||||
# Determine media type from file extension
|
||||
media_type, _ = mimetypes.guess_type(photo.path)
|
||||
if not media_type or not media_type.startswith('image/'):
|
||||
media_type = "image/jpeg"
|
||||
|
||||
# Use FileResponse but set headers to display inline (not download)
|
||||
response = FileResponse(
|
||||
photo.path,
|
||||
media_type=media_type,
|
||||
)
|
||||
# Set Content-Disposition to inline so browser displays instead of downloads
|
||||
response.headers["Content-Disposition"] = "inline"
|
||||
response.headers["Cache-Control"] = "public, max-age=3600"
|
||||
return response
|
||||
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
"""Face processing schemas."""
|
||||
"""Face processing and identify workflow schemas."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel, Field, ConfigDict
|
||||
@@ -38,3 +39,84 @@ class ProcessFacesResponse(BaseModel):
|
||||
detector_backend: str
|
||||
model_name: str
|
||||
|
||||
|
||||
class FaceItem(BaseModel):
|
||||
"""Minimal face item for list views."""
|
||||
|
||||
model_config = ConfigDict(from_attributes=True, protected_namespaces=())
|
||||
|
||||
id: int
|
||||
photo_id: int
|
||||
quality_score: float
|
||||
face_confidence: float
|
||||
location: str
|
||||
|
||||
|
||||
class UnidentifiedFacesQuery(BaseModel):
|
||||
"""Query params for listing unidentified faces."""
|
||||
|
||||
model_config = ConfigDict(protected_namespaces=())
|
||||
|
||||
page: int = 1
|
||||
page_size: int = 50
|
||||
min_quality: float = 0.0
|
||||
date_from: Optional[date] = None
|
||||
date_to: Optional[date] = None
|
||||
sort_by: str = Field("quality", description="quality|date_taken|date_added")
|
||||
sort_dir: str = Field("desc", description="asc|desc")
|
||||
|
||||
|
||||
class UnidentifiedFacesResponse(BaseModel):
|
||||
"""Paginated unidentified faces list."""
|
||||
|
||||
model_config = ConfigDict(protected_namespaces=())
|
||||
|
||||
items: list[FaceItem]
|
||||
page: int
|
||||
page_size: int
|
||||
total: int
|
||||
|
||||
|
||||
class SimilarFaceItem(BaseModel):
|
||||
"""Similar face with similarity score (0-1)."""
|
||||
|
||||
id: int
|
||||
photo_id: int
|
||||
similarity: float
|
||||
location: str
|
||||
quality_score: float
|
||||
|
||||
|
||||
class SimilarFacesResponse(BaseModel):
|
||||
"""Response containing similar faces for a given face."""
|
||||
|
||||
model_config = ConfigDict(protected_namespaces=())
|
||||
|
||||
base_face_id: int
|
||||
items: list[SimilarFaceItem]
|
||||
|
||||
|
||||
class IdentifyFaceRequest(BaseModel):
|
||||
"""Identify a face by selecting existing or creating new person."""
|
||||
|
||||
model_config = ConfigDict(protected_namespaces=())
|
||||
|
||||
# Either provide person_id or the fields to create new person
|
||||
person_id: Optional[int] = None
|
||||
first_name: Optional[str] = None
|
||||
last_name: Optional[str] = None
|
||||
middle_name: Optional[str] = None
|
||||
maiden_name: Optional[str] = None
|
||||
date_of_birth: Optional[date] = None
|
||||
# Optionally identify a batch of face IDs along with this one
|
||||
additional_face_ids: Optional[list[int]] = None
|
||||
|
||||
|
||||
class IdentifyFaceResponse(BaseModel):
|
||||
"""Result of identify operation."""
|
||||
|
||||
model_config = ConfigDict(protected_namespaces=())
|
||||
|
||||
identified_face_ids: list[int]
|
||||
person_id: int
|
||||
created_person: bool
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
"""People schemas for web API (Phase 3)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
|
||||
class PersonResponse(BaseModel):
|
||||
"""Person DTO returned from API."""
|
||||
|
||||
model_config = ConfigDict(from_attributes=True, protected_namespaces=())
|
||||
|
||||
id: int
|
||||
first_name: str
|
||||
last_name: str
|
||||
middle_name: Optional[str] = None
|
||||
maiden_name: Optional[str] = None
|
||||
date_of_birth: Optional[date] = None
|
||||
|
||||
|
||||
class PersonCreateRequest(BaseModel):
|
||||
"""Request payload to create a new person."""
|
||||
|
||||
model_config = ConfigDict(protected_namespaces=())
|
||||
|
||||
first_name: str = Field(..., min_length=1)
|
||||
last_name: str = Field(..., min_length=1)
|
||||
middle_name: Optional[str] = None
|
||||
maiden_name: Optional[str] = None
|
||||
date_of_birth: date
|
||||
|
||||
|
||||
class PeopleListResponse(BaseModel):
|
||||
"""List of people for selection dropdowns."""
|
||||
|
||||
model_config = ConfigDict(protected_namespaces=())
|
||||
|
||||
items: list[PersonResponse]
|
||||
total: int
|
||||
|
||||
|
||||
@@ -6,12 +6,13 @@ import json
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
from typing import Callable, Optional, Tuple
|
||||
from typing import Callable, Optional, Tuple, List
|
||||
from datetime import date
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import and_
|
||||
from sqlalchemy import and_, func
|
||||
|
||||
try:
|
||||
from deepface import DeepFace
|
||||
@@ -644,3 +645,99 @@ def process_unprocessed_photos(
|
||||
|
||||
return photos_processed, total_faces_detected, total_faces_stored
|
||||
|
||||
|
||||
def list_unidentified_faces(
|
||||
db: Session,
|
||||
page: int = 1,
|
||||
page_size: int = 50,
|
||||
min_quality: float = 0.0,
|
||||
date_from: Optional[date] = None,
|
||||
date_to: Optional[date] = None,
|
||||
sort_by: str = "quality",
|
||||
sort_dir: str = "desc",
|
||||
) -> Tuple[List[Face], int]:
|
||||
"""Return paginated unidentified faces with filters.
|
||||
|
||||
Matches desktop behavior as closely as possible: filter by min quality and date_taken.
|
||||
"""
|
||||
# Base query: faces with no person
|
||||
query = db.query(Face).join(Photo, Face.photo_id == Photo.id).filter(Face.person_id.is_(None))
|
||||
|
||||
# Min quality (stored 0.0-1.0)
|
||||
if min_quality is not None:
|
||||
query = query.filter(Face.quality_score >= min_quality)
|
||||
|
||||
# Date range on photo.date_taken when available, else on date_added as fallback
|
||||
if date_from is not None:
|
||||
query = query.filter(
|
||||
(Photo.date_taken.is_not(None) & (Photo.date_taken >= date_from))
|
||||
| (Photo.date_taken.is_(None) & (func.date(Photo.date_added) >= date_from))
|
||||
)
|
||||
if date_to is not None:
|
||||
query = query.filter(
|
||||
(Photo.date_taken.is_not(None) & (Photo.date_taken <= date_to))
|
||||
| (Photo.date_taken.is_(None) & (func.date(Photo.date_added) <= date_to))
|
||||
)
|
||||
|
||||
# Sorting
|
||||
if sort_by == "quality":
|
||||
sort_col = Face.quality_score
|
||||
elif sort_by == "date_taken":
|
||||
sort_col = Photo.date_taken
|
||||
else:
|
||||
sort_col = Photo.date_added
|
||||
|
||||
if sort_dir == "asc":
|
||||
query = query.order_by(sort_col.asc().nullslast())
|
||||
else:
|
||||
query = query.order_by(sort_col.desc().nullslast())
|
||||
|
||||
# Total count for pagination
|
||||
total = query.count()
|
||||
|
||||
# Pagination
|
||||
items = query.offset((page - 1) * page_size).limit(page_size).all()
|
||||
return items, total
|
||||
|
||||
|
||||
def compute_cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
|
||||
"""Compute cosine similarity for two float vectors in range [0,1]."""
|
||||
denom = (np.linalg.norm(a) * np.linalg.norm(b))
|
||||
if denom == 0:
|
||||
return 0.0
|
||||
return float(np.dot(a, b) / denom)
|
||||
|
||||
|
||||
def find_similar_faces(
|
||||
db: Session,
|
||||
face_id: int,
|
||||
limit: int = 20,
|
||||
min_similarity: float = 0.4,
|
||||
) -> List[Tuple[Face, float]]:
|
||||
"""Find similar unidentified faces to the given face using cosine similarity.
|
||||
|
||||
Returns list of (face, similarity) sorted by similarity desc.
|
||||
"""
|
||||
base: Face = db.query(Face).filter(Face.id == face_id).first()
|
||||
if not base:
|
||||
return []
|
||||
|
||||
base_enc = np.frombuffer(base.encoding, dtype=np.float32)
|
||||
|
||||
# Compare against unidentified faces except itself
|
||||
candidates: List[Face] = (
|
||||
db.query(Face)
|
||||
.filter(Face.person_id.is_(None), Face.id != face_id)
|
||||
.all()
|
||||
)
|
||||
|
||||
scored: List[Tuple[Face, float]] = []
|
||||
for f in candidates:
|
||||
enc = np.frombuffer(f.encoding, dtype=np.float32)
|
||||
sim = compute_cosine_similarity(base_enc, enc)
|
||||
if sim >= min_similarity:
|
||||
scored.append((f, sim))
|
||||
|
||||
scored.sort(key=lambda x: x[1], reverse=True)
|
||||
return scored[:limit]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user