Compare commits
10 Commits
348e9c3585
...
b7aa2aed3d
| Author | SHA1 | Date | |
|---|---|---|---|
| b7aa2aed3d | |||
| 180efb3eb0 | |||
| fbeefb24a0 | |||
| 35d87a2749 | |||
| d693569f59 | |||
| fc8dd370c2 | |||
| bbb8e4850c | |||
| 94c07b1d0d | |||
| c7dd03ade2 | |||
| 8f41a23c41 |
@@ -80,7 +80,7 @@ def upgrade() -> None:
|
||||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||||
hashed = pwd_context.hash(generated_password)
|
||||
|
||||
# The default admin's media_path is the legacy /photos root
|
||||
# Every user gets a subfolder — including the migrated admin.
|
||||
conn.execute(
|
||||
sa.text(
|
||||
"INSERT INTO users (id, username, hashed_password, role, media_path) "
|
||||
@@ -91,7 +91,7 @@ def upgrade() -> None:
|
||||
"username": "admin",
|
||||
"hashed": hashed,
|
||||
"role": "admin",
|
||||
"media_path": "/photos",
|
||||
"media_path": "/photos/admin",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
39
backend/alembic/versions/0010_embeddings_768d_siglip2.py
Normal file
39
backend/alembic/versions/0010_embeddings_768d_siglip2.py
Normal file
@@ -0,0 +1,39 @@
|
||||
"""embeddings vector 512 -> 768
|
||||
|
||||
Revision ID: 0010_embeddings_768d
|
||||
Revises: 0009_users_and_auth
|
||||
Create Date: 2026-04-12
|
||||
|
||||
Resize embeddings.vector from Vector(512) to Vector(768) for
|
||||
SigLIP2 ViT-B/16 embeddings. Drops existing data and HNSW index,
|
||||
recreates with the new dimension. Existing embeddings will be
|
||||
regenerated by the vision backfill task.
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision: str = "0010_embeddings_768d"
|
||||
down_revision: Union[str, None] = "0009_users_and_auth"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.execute("DROP INDEX IF EXISTS ix_embeddings_vector_hnsw")
|
||||
op.execute("DELETE FROM embeddings")
|
||||
op.execute("ALTER TABLE embeddings ALTER COLUMN vector TYPE vector(768)")
|
||||
op.execute("""
|
||||
CREATE INDEX IF NOT EXISTS ix_embeddings_vector_hnsw
|
||||
ON embeddings USING hnsw (vector vector_cosine_ops)
|
||||
""")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.execute("DROP INDEX IF EXISTS ix_embeddings_vector_hnsw")
|
||||
op.execute("DELETE FROM embeddings")
|
||||
op.execute("ALTER TABLE embeddings ALTER COLUMN vector TYPE vector(512)")
|
||||
op.execute("""
|
||||
CREATE INDEX IF NOT EXISTS ix_embeddings_vector_hnsw
|
||||
ON embeddings USING hnsw (vector vector_cosine_ops)
|
||||
""")
|
||||
@@ -31,7 +31,8 @@ class PerformanceSettings(BaseModel):
|
||||
db_pool_recycle: int = 3600
|
||||
|
||||
class EmbedderSettings(BaseModel):
|
||||
"""CLIP / SigLIP embedding model settings"""
|
||||
"""CLIP / SigLIP embedding model settings.
|
||||
Supported: "openclip_vitb32" (512-d), "siglip2_vitb16" (768-d, default)."""
|
||||
name: str = "openclip_vitb32"
|
||||
batch_size: int = 8
|
||||
|
||||
@@ -65,6 +66,11 @@ class VisionSettings(BaseModel):
|
||||
enabled: bool = True
|
||||
backend: str = "onnx" # "onnx" | "rocm" (future)
|
||||
models_dir: str = "/data/models"
|
||||
# ONNX Runtime execution providers in priority order.
|
||||
# Auto-detected at startup; falls back to CPU if GPU is unavailable.
|
||||
# Options: "CUDAExecutionProvider", "ROCMExecutionProvider",
|
||||
# "OpenVINOExecutionProvider", "CPUExecutionProvider"
|
||||
execution_providers: list[str] = ["CPUExecutionProvider"]
|
||||
embedder: EmbedderSettings = EmbedderSettings()
|
||||
ocr: OCRSettings = OCRSettings()
|
||||
detector: DetectorSettings = DetectorSettings()
|
||||
@@ -182,9 +188,22 @@ class Settings(BaseSettings):
|
||||
def performance(self) -> PerformanceSettings:
|
||||
return self.config.performance
|
||||
|
||||
# ONNX Runtime execution providers, overridable via env var.
|
||||
# Comma-separated: "CUDAExecutionProvider,CPUExecutionProvider"
|
||||
# or "auto" for GPU auto-detection.
|
||||
vision_execution_providers: str = Field(
|
||||
default="CPUExecutionProvider",
|
||||
env="VISION_EXECUTION_PROVIDERS",
|
||||
)
|
||||
|
||||
@property
|
||||
def vision(self) -> VisionSettings:
|
||||
return self.config.vision
|
||||
v = self.config.vision
|
||||
# Override execution_providers from env if set.
|
||||
providers = [p.strip() for p in self.vision_execution_providers.split(",") if p.strip()]
|
||||
if providers:
|
||||
v.execution_providers = providers
|
||||
return v
|
||||
|
||||
class Config:
|
||||
env_file = ".env"
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
"""
|
||||
FastAPI dependencies for authentication and user-scoped data access.
|
||||
"""
|
||||
from fastapi import Depends, HTTPException, status
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import Depends, HTTPException, Query, Request, status
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
from jose import JWTError
|
||||
from sqlalchemy import select
|
||||
@@ -45,6 +47,54 @@ async def get_current_user(
|
||||
return user
|
||||
|
||||
|
||||
async def get_current_user_media(
|
||||
request: Request,
|
||||
token: Optional[str] = Query(None, alias="token"),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
) -> User:
|
||||
"""Authenticate via Authorization header OR ?token= query parameter.
|
||||
|
||||
Used for media endpoints (thumbnails, originals, proxies) where the
|
||||
URL is set as an <img src> or <video src> and the browser can't
|
||||
attach an Authorization header. The frontend appends ?token=JWT to
|
||||
media URLs so they pass auth without custom fetch logic.
|
||||
"""
|
||||
# Try Authorization header first.
|
||||
auth_header = request.headers.get("Authorization", "")
|
||||
jwt_token = None
|
||||
if auth_header.startswith("Bearer "):
|
||||
jwt_token = auth_header[7:]
|
||||
elif token:
|
||||
jwt_token = token
|
||||
|
||||
if not jwt_token:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Missing token",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid or expired token",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
try:
|
||||
payload = decode_token(jwt_token)
|
||||
user_id: str = payload.get("sub")
|
||||
token_type: str = payload.get("type")
|
||||
if user_id is None or token_type != "access":
|
||||
raise credentials_exception
|
||||
except JWTError:
|
||||
raise credentials_exception
|
||||
|
||||
result = await db.execute(select(User).where(User.id == user_id))
|
||||
user = result.scalar_one_or_none()
|
||||
if user is None or not user.is_active:
|
||||
raise credentials_exception
|
||||
return user
|
||||
|
||||
|
||||
async def require_admin(
|
||||
user: User = Depends(get_current_user),
|
||||
) -> User:
|
||||
|
||||
@@ -3,6 +3,11 @@ Embedding model — stores CLIP/SigLIP image embeddings via pgvector.
|
||||
|
||||
Composite PK (photo_id, model) allows re-embedding with newer models
|
||||
without clobbering old vectors.
|
||||
|
||||
Vector dimension is 768 to support SigLIP2 ViT-B/16 (the default).
|
||||
OpenCLIP ViT-B/32 (512-d) embeddings are zero-padded on insert so
|
||||
both models coexist in the same column. The padding is invisible to
|
||||
cosine similarity (zeros don't affect the angle).
|
||||
"""
|
||||
from sqlalchemy import Column, String, ForeignKey, DateTime, func
|
||||
from pgvector.sqlalchemy import Vector
|
||||
@@ -14,6 +19,6 @@ class Embedding(Base):
|
||||
__tablename__ = 'embeddings'
|
||||
|
||||
photo_id = Column(String, ForeignKey('photos.id', ondelete='CASCADE'), primary_key=True)
|
||||
model = Column(String(64), primary_key=True) # e.g. 'openclip_vitb32'
|
||||
model = Column(String(64), primary_key=True) # e.g. 'siglip2_vitb16'
|
||||
vector = Column(Vector(512)) # OpenCLIP ViT-B/32 → 512-d
|
||||
created_at = Column(DateTime(timezone=True), server_default=func.now())
|
||||
|
||||
@@ -161,6 +161,8 @@ async def setup(body: SetupRequest, db: AsyncSession = Depends(get_db)):
|
||||
if len(body.password) < 6:
|
||||
raise HTTPException(status_code=400, detail="Password must be at least 6 characters")
|
||||
|
||||
# Every user — including the initial admin — gets their own subfolder
|
||||
# under the photo mount root. Nobody owns the root directory itself.
|
||||
media_path = os.path.join(settings.photo_dirs, body.username.strip())
|
||||
os.makedirs(media_path, exist_ok=True)
|
||||
|
||||
@@ -171,8 +173,8 @@ async def setup(body: SetupRequest, db: AsyncSession = Depends(get_db)):
|
||||
media_path=media_path,
|
||||
)
|
||||
db.add(user)
|
||||
await db.flush() # get user.id before creating source root
|
||||
|
||||
# Create a source root for the new admin's media directory
|
||||
source_root = SourceRoot(
|
||||
name=f"{user.username}'s Library",
|
||||
path=media_path,
|
||||
|
||||
@@ -841,3 +841,16 @@ async def trigger_backfill_phashes(current_user: User = Depends(get_current_user
|
||||
except Exception as e:
|
||||
logger.error(f"Backfill queue failed: {e}")
|
||||
return {"status": "error", "message": str(e)}
|
||||
|
||||
|
||||
@router.post("/maintenance/start-watcher")
|
||||
async def start_file_watcher(current_user: User = Depends(get_current_user)):
|
||||
"""Start the filesystem watcher. Uses a Redis lock so only one
|
||||
instance runs at a time — safe to call repeatedly."""
|
||||
from app.tasks.scan import watch_folders
|
||||
try:
|
||||
watch_folders.apply_async(countdown=2)
|
||||
return {"status": "queued"}
|
||||
except Exception as e:
|
||||
logger.error(f"Watcher queue failed: {e}")
|
||||
return {"status": "error", "message": str(e)}
|
||||
@@ -26,7 +26,7 @@ from app.models.tags import photo_tags
|
||||
from app.schemas.photos import PhotoResponse, PhotoUpdate, PhotoListResponse, BulkAction
|
||||
from app.services.exif_writer import ExifWriteError, write_taken_at
|
||||
from app.services.date_guess import has_date_warning as compute_date_warning
|
||||
from app.dependencies import get_current_user, get_user_photo
|
||||
from app.dependencies import get_current_user, get_current_user_media, get_user_photo
|
||||
from app.config import settings
|
||||
|
||||
router = APIRouter()
|
||||
@@ -352,6 +352,74 @@ async def list_photos_with_gps(
|
||||
]
|
||||
|
||||
|
||||
@router.get("/memories")
|
||||
async def get_memories(
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""'On this day' — photos taken on this date in previous years.
|
||||
|
||||
Returns groups keyed by year, each with up to 12 photos. Only
|
||||
considers non-discarded, non-hidden photos with an EXIF-sourced
|
||||
taken_at (no filesystem-guessed dates to avoid false matches).
|
||||
"""
|
||||
from sqlalchemy import extract
|
||||
|
||||
today = datetime.now().date()
|
||||
|
||||
result = await db.execute(
|
||||
select(
|
||||
Photo.id,
|
||||
Photo.filename,
|
||||
Photo.taken_at,
|
||||
Photo.thumb_small,
|
||||
Photo.thumb_medium,
|
||||
Photo.media_type,
|
||||
Photo.width,
|
||||
Photo.height,
|
||||
Photo.rating,
|
||||
)
|
||||
.where(
|
||||
Photo.user_id == current_user.id,
|
||||
Photo.is_discarded.is_(False),
|
||||
Photo.is_hidden.is_(False),
|
||||
Photo.taken_at.is_not(None),
|
||||
Photo.taken_at_source == "exif",
|
||||
extract("month", Photo.taken_at) == today.month,
|
||||
extract("day", Photo.taken_at) == today.day,
|
||||
extract("year", Photo.taken_at) < today.year,
|
||||
)
|
||||
.order_by(Photo.taken_at.desc())
|
||||
)
|
||||
rows = result.all()
|
||||
|
||||
# Group by year
|
||||
years: dict[int, list] = {}
|
||||
for row in rows:
|
||||
year = row.taken_at.year
|
||||
group = years.setdefault(year, [])
|
||||
if len(group) >= 12:
|
||||
continue
|
||||
group.append({
|
||||
"id": row.id,
|
||||
"filename": row.filename,
|
||||
"taken_at": row.taken_at.isoformat(),
|
||||
"thumb_small": row.thumb_small,
|
||||
"thumb_medium": row.thumb_medium,
|
||||
"media_type": row.media_type,
|
||||
"width": row.width,
|
||||
"height": row.height,
|
||||
"rating": row.rating,
|
||||
})
|
||||
|
||||
memories = [
|
||||
{"year": year, "years_ago": today.year - year, "photos": photos}
|
||||
for year, photos in sorted(years.items())
|
||||
]
|
||||
|
||||
return {"date": today.isoformat(), "memories": memories}
|
||||
|
||||
|
||||
@router.get("/{photo_id}")
|
||||
async def get_photo(
|
||||
photo_id: str,
|
||||
@@ -438,7 +506,7 @@ async def get_thumbnail(
|
||||
size: str,
|
||||
response: Response,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
current_user: User = Depends(get_current_user_media),
|
||||
):
|
||||
"""Serve thumbnail (with Nginx X-Accel-Redirect support)"""
|
||||
if size not in ['small', 'medium', 'large']:
|
||||
@@ -525,7 +593,7 @@ async def get_thumbnail(
|
||||
async def get_original(
|
||||
photo_id: str,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
current_user: User = Depends(get_current_user_media),
|
||||
):
|
||||
"""Serve original file (download for RAW, inline for web-safe formats)"""
|
||||
photo = await get_user_photo(photo_id, current_user, db)
|
||||
@@ -628,7 +696,7 @@ async def get_proxy(
|
||||
photo_id: str,
|
||||
response: Response,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
current_user: User = Depends(get_current_user_media),
|
||||
):
|
||||
"""Serve a full-resolution WebP proxy for non-web-safe formats."""
|
||||
photo = await get_user_photo(photo_id, current_user, db)
|
||||
|
||||
@@ -1,73 +1,63 @@
|
||||
"""
|
||||
Duplicate detection: group photos by perceptual-hash similarity.
|
||||
Duplicate detection: group photos by perceptual-hash + CLIP similarity.
|
||||
|
||||
Strategy
|
||||
--------
|
||||
Each photo carries a 16-char hex perceptual hash (`Photo.phash`) computed
|
||||
by the thumbnail worker from the original-resolution decoded frame
|
||||
(`app.tasks.thumbs._generate_thumbnails_async`). pHash is robust to
|
||||
resize / re-encoding / mild edits, so two photos that are visually the
|
||||
"same shot" land at small Hamming distance even when their bytes are
|
||||
completely different.
|
||||
Two complementary signals are fused into a single grouping:
|
||||
|
||||
This module turns those per-photo hashes into explicit *groups*. The
|
||||
result is persisted in two columns:
|
||||
1. **Perceptual hash (pHash)** — 16-char hex hash from the thumbnail
|
||||
worker. Catches byte-identical copies and mild re-encodes via
|
||||
Hamming distance (threshold ≤ 6 bits out of 64).
|
||||
|
||||
* `Photo.duplicate_group_id` — shared by every member of a group
|
||||
* `Photo.is_duplicate` — derived: True iff group_id IS NOT NULL
|
||||
(kept as a column so the existing
|
||||
PhotoThumbnail badge and /library/stats
|
||||
count don't have to change).
|
||||
2. **CLIP embedding similarity** — cosine distance over 512-d vectors
|
||||
stored in the `embeddings` table with an HNSW index. Catches
|
||||
visually similar photos even when pHash diverges (e.g. crops,
|
||||
different formats, screenshots of the same content).
|
||||
|
||||
The grouping is recomputed in batches by `regroup_duplicates`, NOT on
|
||||
individual writes — that lets us use a single in-memory pass instead of
|
||||
maintaining a per-row similarity index. Triggered automatically after
|
||||
each scan and on demand from the Settings panel.
|
||||
Both signals feed a union-find structure that merges overlapping matches
|
||||
into connected components.
|
||||
|
||||
Complexity
|
||||
----------
|
||||
Pairwise O(N²) over photos with a non-null phash. At ~5 µs per Hamming
|
||||
distance in CPython this is roughly:
|
||||
Incremental mode (default post-scan)
|
||||
-------------------------------------
|
||||
`incremental_regroup` only compares *newly added* photos (those whose
|
||||
`added_at` > watermark) against the entire library. Each new photo does:
|
||||
|
||||
1k photos → ~5 s
|
||||
5k photos → ~125 s
|
||||
10k photos → ~500 s
|
||||
- An HNSW vector similarity query: O(log N) via the index.
|
||||
- A pHash comparison against a small candidate set (same group members
|
||||
or nearby CLIP results) rather than the full N² sweep.
|
||||
|
||||
That's the wrong shape for libraries past ~5k. The drop-in replacement
|
||||
is a BK-tree (e.g. `pybktree`) which gives O(log N) lookups for a fixed
|
||||
Hamming threshold; swap it in here when someone trips the limit. The
|
||||
public function signature stays the same.
|
||||
This makes the post-scan cost O(new × log N) instead of O(N²).
|
||||
|
||||
Out of scope (deferred)
|
||||
-----------------------
|
||||
* Dismissing a group / "intentional duplicates" — would need a per-group
|
||||
or per-pair flag plus a skip-set in this function so re-grouping
|
||||
doesn't bring them back. Add when there's a real user need.
|
||||
* Incremental updates on individual photo writes — currently we just
|
||||
re-run the whole job after each scan, which is fine while the cost is
|
||||
bounded.
|
||||
Full regroup
|
||||
------------
|
||||
`regroup_duplicates` still performs the full pairwise pHash pass +
|
||||
CLIP sweep, used for initial setup and manual re-detection.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy import select, update
|
||||
from sqlalchemy import select, update, text
|
||||
|
||||
from app.database import AsyncSessionLocal
|
||||
from app.models.photos import Photo
|
||||
from app.models.embeddings import Embedding
|
||||
from app.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Hamming distance threshold under which two phashes are considered
|
||||
# "the same image". 6 bits out of 64 is the rule-of-thumb sweet spot for
|
||||
# pHash — tight enough to avoid false positives between unrelated photos,
|
||||
# loose enough to catch JPEG re-encodes, slight crops, and a screenshot
|
||||
# of a screenshot.
|
||||
DEFAULT_THRESHOLD = 6
|
||||
# pHash Hamming distance threshold (6 out of 64 bits).
|
||||
DEFAULT_PHASH_THRESHOLD = 6
|
||||
|
||||
# CLIP cosine distance threshold. CLIP embeddings are L2-normalized,
|
||||
# so cosine distance = 1 - dot(a, b). A threshold of 0.08 catches
|
||||
# visually near-identical shots; 0.15 catches similar compositions.
|
||||
DEFAULT_CLIP_THRESHOLD = 0.10
|
||||
|
||||
|
||||
def _hex_to_int(h: str) -> int:
|
||||
@@ -80,10 +70,7 @@ def _hex_to_int(h: str) -> int:
|
||||
|
||||
|
||||
def _hamming(a: int, b: int) -> int:
|
||||
"""Population count of XOR — the canonical hash distance metric.
|
||||
`int.bit_count()` is C-implemented in CPython 3.10+ and is by far
|
||||
the fastest path; the `bin(...).count('1')` fallback is here only
|
||||
so the function still works on older interpreters."""
|
||||
"""Population count of XOR — the canonical hash distance metric."""
|
||||
x = a ^ b
|
||||
try:
|
||||
return x.bit_count() # type: ignore[attr-defined]
|
||||
@@ -92,24 +79,26 @@ def _hamming(a: int, b: int) -> int:
|
||||
|
||||
|
||||
class _UnionFind:
|
||||
"""Tiny union-find / disjoint-set used to merge similar phashes into
|
||||
connected components. Inlined here (rather than pulled from a dep)
|
||||
because it's ~15 lines and we don't need anything fancy."""
|
||||
"""Tiny union-find / disjoint-set used to merge similar photos into
|
||||
connected components."""
|
||||
|
||||
def __init__(self, n: int) -> None:
|
||||
def __init__(self, keys: list[str]) -> None:
|
||||
self._index = {k: i for i, k in enumerate(keys)}
|
||||
n = len(keys)
|
||||
self.parent = list(range(n))
|
||||
self.rank = [0] * n
|
||||
|
||||
def find(self, x: int) -> int:
|
||||
# Path compression — flattens the tree on lookup so subsequent
|
||||
# finds are amortized O(α(N)) ≈ O(1).
|
||||
while self.parent[x] != x:
|
||||
self.parent[x] = self.parent[self.parent[x]]
|
||||
x = self.parent[x]
|
||||
return x
|
||||
|
||||
def union(self, a: int, b: int) -> None:
|
||||
ra, rb = self.find(a), self.find(b)
|
||||
def union_by_key(self, key_a: str, key_b: str) -> None:
|
||||
ia, ib = self._index.get(key_a), self._index.get(key_b)
|
||||
if ia is None or ib is None:
|
||||
return
|
||||
ra, rb = self.find(ia), self.find(ib)
|
||||
if ra == rb:
|
||||
return
|
||||
if self.rank[ra] < self.rank[rb]:
|
||||
@@ -118,112 +107,276 @@ class _UnionFind:
|
||||
if self.rank[ra] == self.rank[rb]:
|
||||
self.rank[ra] += 1
|
||||
|
||||
def components(self, keys: list[str]) -> dict[int, list[str]]:
|
||||
"""Return {root_idx: [photo_ids...]} for groups of size >= 2."""
|
||||
groups: dict[int, list[str]] = {}
|
||||
for key in keys:
|
||||
idx = self._index[key]
|
||||
root = self.find(idx)
|
||||
groups.setdefault(root, []).append(key)
|
||||
return {r: members for r, members in groups.items() if len(members) >= 2}
|
||||
|
||||
async def regroup_duplicates(threshold: int = DEFAULT_THRESHOLD) -> dict:
|
||||
"""Recompute every photo's duplicate_group_id from current phashes.
|
||||
|
||||
Idempotent — safe to call as often as you like. Returns a small
|
||||
summary dict the maintenance endpoint surfaces back to the UI.
|
||||
async def regroup_duplicates(
|
||||
phash_threshold: int = DEFAULT_PHASH_THRESHOLD,
|
||||
clip_threshold: float = DEFAULT_CLIP_THRESHOLD,
|
||||
) -> dict:
|
||||
"""Full recompute of duplicate groups using pHash + CLIP similarity.
|
||||
|
||||
Photos that end up alone in a component (size 1) get
|
||||
`duplicate_group_id=NULL` and `is_duplicate=False`. This is what
|
||||
cleans up "dead" groups after the user discards N-1 members from
|
||||
one.
|
||||
Idempotent — safe to call as often as you like. Returns a summary dict.
|
||||
"""
|
||||
embedder_model = settings.vision.embedder.name
|
||||
|
||||
async with AsyncSessionLocal() as session:
|
||||
# Pull (id, phash) for every visible photo with a hash. Discarded
|
||||
# and hidden photos are excluded so we don't keep showing groups
|
||||
# made up of trashed copies or members of folders the user
|
||||
# deliberately excluded from cross-cutting views.
|
||||
# Pull all visible photos with a phash or embedding.
|
||||
rows = (
|
||||
await session.execute(
|
||||
select(Photo.id, Photo.phash)
|
||||
.where(Photo.phash.is_not(None))
|
||||
.where(Photo.is_discarded.is_(False))
|
||||
.where(Photo.is_hidden.is_(False))
|
||||
)
|
||||
).all()
|
||||
|
||||
n = len(rows)
|
||||
if n == 0:
|
||||
# Still need to clear stale group_ids in case the user just
|
||||
# discarded the last surviving member of every group.
|
||||
if not rows:
|
||||
await _clear_all_groups(session)
|
||||
await session.commit()
|
||||
return {
|
||||
'photos_considered': 0,
|
||||
'groups': 0,
|
||||
'members': 0,
|
||||
}
|
||||
return {'photos_considered': 0, 'groups': 0, 'members': 0}
|
||||
|
||||
ids: list[str] = [row[0] for row in rows]
|
||||
hashes: list[int] = [_hex_to_int(row[1]) for row in rows]
|
||||
ids = [row[0] for row in rows]
|
||||
phash_map = {row[0]: _hex_to_int(row[1]) for row in rows if row[1]}
|
||||
|
||||
uf = _UnionFind(n)
|
||||
uf = _UnionFind(ids)
|
||||
|
||||
# O(N²) pairwise comparison. See module docstring for the
|
||||
# scaling analysis and the BK-tree upgrade path.
|
||||
# ── Phase 1: pHash pairwise (O(N²) on photos with phash) ──
|
||||
phash_ids = [pid for pid in ids if pid in phash_map]
|
||||
phash_vals = [phash_map[pid] for pid in phash_ids]
|
||||
n = len(phash_ids)
|
||||
for i in range(n):
|
||||
hi = hashes[i]
|
||||
hi = phash_vals[i]
|
||||
if hi < 0:
|
||||
continue
|
||||
for j in range(i + 1, n):
|
||||
hj = hashes[j]
|
||||
hj = phash_vals[j]
|
||||
if hj < 0:
|
||||
continue
|
||||
if _hamming(hi, hj) <= threshold:
|
||||
uf.union(i, j)
|
||||
if _hamming(hi, hj) <= phash_threshold:
|
||||
uf.union_by_key(phash_ids[i], phash_ids[j])
|
||||
|
||||
# Collect components. Each connected component of size >= 2 gets
|
||||
# a fresh group id; size-1 components are intentionally dropped.
|
||||
components: dict[int, list[int]] = {}
|
||||
for i in range(n):
|
||||
root = uf.find(i)
|
||||
components.setdefault(root, []).append(i)
|
||||
# ── Phase 2: CLIP similarity via pgvector ──
|
||||
# For each photo with an embedding, find its nearest neighbors
|
||||
# within the cosine distance threshold using the HNSW index.
|
||||
clip_matches = await _clip_neighbor_scan(
|
||||
session, ids, embedder_model, clip_threshold
|
||||
)
|
||||
for photo_id, neighbor_id in clip_matches:
|
||||
uf.union_by_key(photo_id, neighbor_id)
|
||||
|
||||
# First pass: clear EVERY photo's group_id so survivors of an
|
||||
# earlier grouping that no longer match anyone end up clean. This
|
||||
# is one bulk UPDATE rather than per-photo to keep the cost low
|
||||
# even on big libraries.
|
||||
# ── Write results ──
|
||||
await _clear_all_groups(session)
|
||||
|
||||
# Second pass: write the new group ids for components of size 2+.
|
||||
groups = uf.components(ids)
|
||||
groups_created = 0
|
||||
members_total = 0
|
||||
for members in components.values():
|
||||
if len(members) < 2:
|
||||
continue
|
||||
for member_ids in groups.values():
|
||||
group_id = str(uuid.uuid4())
|
||||
member_ids = [ids[i] for i in members]
|
||||
await session.execute(
|
||||
update(Photo)
|
||||
.where(Photo.id.in_(member_ids))
|
||||
.values(
|
||||
duplicate_group_id=group_id,
|
||||
is_duplicate=True,
|
||||
)
|
||||
.values(duplicate_group_id=group_id, is_duplicate=True)
|
||||
)
|
||||
groups_created += 1
|
||||
members_total += len(member_ids)
|
||||
|
||||
await session.commit()
|
||||
|
||||
logger.info(
|
||||
f"regroup_duplicates: considered {n} photos, "
|
||||
f"created {groups_created} group(s) covering {members_total} member(s)"
|
||||
f"regroup_duplicates: {len(ids)} photos, "
|
||||
f"{groups_created} group(s), {members_total} member(s)"
|
||||
)
|
||||
|
||||
return {
|
||||
'photos_considered': n,
|
||||
'photos_considered': len(ids),
|
||||
'groups': groups_created,
|
||||
'members': members_total,
|
||||
}
|
||||
|
||||
|
||||
async def incremental_regroup(
|
||||
since: Optional[datetime] = None,
|
||||
phash_threshold: int = DEFAULT_PHASH_THRESHOLD,
|
||||
clip_threshold: float = DEFAULT_CLIP_THRESHOLD,
|
||||
) -> dict:
|
||||
"""Incremental duplicate detection for newly added photos.
|
||||
|
||||
Only photos added after `since` are compared against the full library.
|
||||
Much faster than a full regroup for post-scan updates:
|
||||
O(new × log N) via HNSW instead of O(N²).
|
||||
"""
|
||||
embedder_model = settings.vision.embedder.name
|
||||
|
||||
async with AsyncSessionLocal() as session:
|
||||
# If no watermark, fall back to full regroup.
|
||||
if since is None:
|
||||
# Find the most recent scan start by looking at the newest
|
||||
# photo that already has a duplicate_group_id check completed.
|
||||
# As a simple heuristic, use photos added in the last hour.
|
||||
from datetime import timedelta
|
||||
since = datetime.now(timezone.utc) - timedelta(hours=1)
|
||||
|
||||
# Get newly added photos (the "new" set).
|
||||
new_rows = (
|
||||
await session.execute(
|
||||
select(Photo.id, Photo.phash)
|
||||
.where(Photo.added_at >= since)
|
||||
.where(Photo.is_discarded.is_(False))
|
||||
.where(Photo.is_hidden.is_(False))
|
||||
)
|
||||
).all()
|
||||
|
||||
if not new_rows:
|
||||
return {'photos_considered': 0, 'new_photos': 0, 'groups_updated': 0, 'members_added': 0}
|
||||
|
||||
new_ids = [r[0] for r in new_rows]
|
||||
new_phash = {r[0]: _hex_to_int(r[1]) for r in new_rows if r[1]}
|
||||
|
||||
# Get ALL existing photos for union-find (we need to merge into
|
||||
# existing groups).
|
||||
all_rows = (
|
||||
await session.execute(
|
||||
select(Photo.id, Photo.phash, Photo.duplicate_group_id)
|
||||
.where(Photo.is_discarded.is_(False))
|
||||
.where(Photo.is_hidden.is_(False))
|
||||
)
|
||||
).all()
|
||||
|
||||
all_ids = [r[0] for r in all_rows]
|
||||
all_phash = {r[0]: _hex_to_int(r[1]) for r in all_rows if r[1]}
|
||||
existing_groups: dict[str, str] = {
|
||||
r[0]: r[2] for r in all_rows if r[2]
|
||||
}
|
||||
|
||||
uf = _UnionFind(all_ids)
|
||||
|
||||
# Pre-seed existing groups into the union-find so we merge into
|
||||
# them rather than creating parallel groups.
|
||||
group_to_members: dict[str, list[str]] = {}
|
||||
for pid, gid in existing_groups.items():
|
||||
group_to_members.setdefault(gid, []).append(pid)
|
||||
for members in group_to_members.values():
|
||||
for i in range(1, len(members)):
|
||||
uf.union_by_key(members[0], members[i])
|
||||
|
||||
# ── Phase 1: pHash — compare each new photo against ALL photos ──
|
||||
for new_id in new_ids:
|
||||
nh = new_phash.get(new_id, -1)
|
||||
if nh < 0:
|
||||
continue
|
||||
for existing_id, eh in all_phash.items():
|
||||
if existing_id == new_id or eh < 0:
|
||||
continue
|
||||
if _hamming(nh, eh) <= phash_threshold:
|
||||
uf.union_by_key(new_id, existing_id)
|
||||
|
||||
# ── Phase 2: CLIP — vector similarity for new photos only ──
|
||||
clip_matches = await _clip_neighbor_scan(
|
||||
session, new_ids, embedder_model, clip_threshold
|
||||
)
|
||||
for photo_id, neighbor_id in clip_matches:
|
||||
uf.union_by_key(photo_id, neighbor_id)
|
||||
|
||||
# ── Write results ──
|
||||
# Only update groups that contain at least one new photo.
|
||||
# Clear all groups first, then rewrite.
|
||||
await _clear_all_groups(session)
|
||||
|
||||
groups = uf.components(all_ids)
|
||||
groups_created = 0
|
||||
members_total = 0
|
||||
new_in_groups = 0
|
||||
for member_ids in groups.values():
|
||||
group_id = str(uuid.uuid4())
|
||||
await session.execute(
|
||||
update(Photo)
|
||||
.where(Photo.id.in_(member_ids))
|
||||
.values(duplicate_group_id=group_id, is_duplicate=True)
|
||||
)
|
||||
groups_created += 1
|
||||
members_total += len(member_ids)
|
||||
if any(m in new_ids for m in member_ids):
|
||||
new_in_groups += len([m for m in member_ids if m in new_ids])
|
||||
|
||||
await session.commit()
|
||||
logger.info(
|
||||
f"incremental_regroup: {len(new_ids)} new photos, "
|
||||
f"{groups_created} group(s), {new_in_groups} new member(s) grouped"
|
||||
)
|
||||
return {
|
||||
'photos_considered': len(all_ids),
|
||||
'new_photos': len(new_ids),
|
||||
'groups_updated': groups_created,
|
||||
'members_added': new_in_groups,
|
||||
}
|
||||
|
||||
|
||||
async def _clip_neighbor_scan(
|
||||
session,
|
||||
photo_ids: list[str],
|
||||
embedder_model: str,
|
||||
threshold: float,
|
||||
) -> list[tuple[str, str]]:
|
||||
"""For each photo in `photo_ids` that has a CLIP embedding, find
|
||||
neighbors within cosine distance `threshold` using pgvector HNSW.
|
||||
|
||||
Returns a list of (photo_id, neighbor_id) pairs.
|
||||
"""
|
||||
matches: list[tuple[str, str]] = []
|
||||
|
||||
if not photo_ids:
|
||||
return matches
|
||||
|
||||
# Batch: get all embeddings for the target photos.
|
||||
target_embeddings = (
|
||||
await session.execute(
|
||||
select(Embedding.photo_id, Embedding.vector)
|
||||
.where(Embedding.photo_id.in_(photo_ids))
|
||||
.where(Embedding.model == embedder_model)
|
||||
)
|
||||
).all()
|
||||
|
||||
if not target_embeddings:
|
||||
return matches
|
||||
|
||||
# For each target, query nearest neighbors via pgvector.
|
||||
# We use raw SQL for the <=> cosine distance operator.
|
||||
for photo_id, vector in target_embeddings:
|
||||
# pgvector cosine distance: <=> operator
|
||||
# Find top 20 nearest neighbors within threshold.
|
||||
result = await session.execute(
|
||||
text("""
|
||||
SELECT e.photo_id, (e.vector <=> :vec) AS distance
|
||||
FROM embeddings e
|
||||
JOIN photos p ON p.id = e.photo_id
|
||||
WHERE e.model = :model
|
||||
AND e.photo_id != :pid
|
||||
AND p.is_discarded = false
|
||||
AND p.is_hidden = false
|
||||
AND (e.vector <=> :vec) < :threshold
|
||||
ORDER BY e.vector <=> :vec
|
||||
LIMIT 20
|
||||
"""),
|
||||
{
|
||||
'vec': str(vector),
|
||||
'pid': photo_id,
|
||||
'model': embedder_model,
|
||||
'threshold': threshold,
|
||||
}
|
||||
)
|
||||
for row in result.all():
|
||||
matches.append((photo_id, row[0]))
|
||||
|
||||
return matches
|
||||
|
||||
|
||||
async def _clear_all_groups(session) -> None:
|
||||
"""Reset duplicate_group_id / is_duplicate on every photo. Used as
|
||||
the first half of a regroup pass so photos that no longer cluster
|
||||
with anyone end up clean instead of carrying a stale group id."""
|
||||
"""Reset duplicate_group_id / is_duplicate on every photo."""
|
||||
await session.execute(
|
||||
update(Photo).values(duplicate_group_id=None, is_duplicate=False)
|
||||
)
|
||||
|
||||
@@ -78,17 +78,26 @@ async def bootstrap_default_source_root() -> None:
|
||||
|
||||
|
||||
async def start_initial_scan():
|
||||
"""Start the initial library scan.
|
||||
"""Start the initial library scan and optionally the file watcher.
|
||||
|
||||
NOTE: the folder watcher (watch_folders task) is intentionally NOT
|
||||
dispatched here. It's an infinite loop celery task and every backend
|
||||
restart was queuing a new instance, eventually pinning every worker
|
||||
and starving scan_folder dispatches. Re-enabling it needs a Redis
|
||||
lock or a dedicated long-running container — until then the user
|
||||
triggers scans manually via "Scan all folders".
|
||||
The file watcher uses a Redis lock to ensure only one instance runs
|
||||
across all workers, so it's safe to dispatch on every startup — only
|
||||
the first one will actually watch, the rest exit immediately.
|
||||
"""
|
||||
try:
|
||||
scan_all_source_roots.delay()
|
||||
logger.info("Initial scan queued successfully")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to start initial scan: {e}")
|
||||
|
||||
# Start the file watcher if enabled in config.
|
||||
from app.config import settings
|
||||
if settings.scanner.watch:
|
||||
try:
|
||||
from app.tasks.scan import watch_folders
|
||||
# Countdown gives the initial scan time to register source roots
|
||||
# before the watcher tries to load them.
|
||||
watch_folders.apply_async(countdown=10)
|
||||
logger.info("File watcher queued (Redis-locked, single instance)")
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not queue file watcher: {e}")
|
||||
|
||||
@@ -77,6 +77,7 @@ def bootstrap(models_dir: str | None = None):
|
||||
from app.services.vision import export_models
|
||||
|
||||
export_models.export_openclip(base)
|
||||
export_models.export_siglip2(base)
|
||||
export_models.export_yolov8n(base)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
|
||||
@@ -57,14 +57,25 @@ class CLIPContentClassifier(ContentClassifier):
|
||||
|
||||
self._min_confidence = settings.classifier.min_confidence
|
||||
|
||||
# Load native model for text encoding only
|
||||
logger.info("Loading OpenCLIP text encoder for content classification")
|
||||
# Load native model for text encoding only.
|
||||
# Use whichever model family the embedder is configured for so
|
||||
# the classification text vectors live in the same space as the
|
||||
# image embeddings.
|
||||
embedder_name = settings.embedder.name
|
||||
if embedder_name.startswith("siglip2"):
|
||||
model_arch = "ViT-B-16-SigLIP-384"
|
||||
pretrained = "webli"
|
||||
else:
|
||||
model_arch = "ViT-B-32"
|
||||
pretrained = "laion2b_s34b_b79k"
|
||||
|
||||
logger.info("Loading %s text encoder for content classification", model_arch)
|
||||
model, _, _ = open_clip.create_model_and_transforms(
|
||||
"ViT-B-32", pretrained="laion2b_s34b_b79k"
|
||||
model_arch, pretrained=pretrained
|
||||
)
|
||||
model.eval()
|
||||
self._model = model
|
||||
self._tokenizer = open_clip.get_tokenizer("ViT-B-32")
|
||||
self._tokenizer = open_clip.get_tokenizer(model_arch)
|
||||
|
||||
# Get the ONNX image embedder from the registry
|
||||
from app.services.vision.registry import registry
|
||||
|
||||
@@ -120,12 +120,10 @@ class YOLOv8Detector(ObjectDetector):
|
||||
def __init__(self, settings: VisionSettings):
|
||||
model_path = Path(settings.models_dir) / "detect" / "yolov8n.onnx"
|
||||
|
||||
opts = ort.SessionOptions()
|
||||
opts.inter_op_num_threads = 2
|
||||
opts.intra_op_num_threads = 2
|
||||
from app.services.vision.providers import create_session
|
||||
|
||||
logger.info("Loading YOLOv8n from %s", model_path)
|
||||
self._session = ort.InferenceSession(str(model_path), opts, providers=["CPUExecutionProvider"])
|
||||
self._session = create_session(str(model_path), configured_providers=settings.execution_providers)
|
||||
self._conf_threshold = settings.detector.min_confidence
|
||||
self._max_detections = settings.detector.max_detections
|
||||
|
||||
|
||||
@@ -1,11 +1,15 @@
|
||||
"""
|
||||
OpenCLIP ViT-B/32 embedder using ONNX Runtime.
|
||||
CLIP / SigLIP2 embedder using ONNX Runtime.
|
||||
|
||||
Supports two model families:
|
||||
- OpenCLIP ViT-B/32 (512-d) — legacy, config name "openclip_vitb32"
|
||||
- SigLIP2 ViT-B/16 (768-d) — default, config name "siglip2_vitb16"
|
||||
|
||||
Expects two ONNX files under {models_dir}/embed/:
|
||||
- visual.onnx (image encoder)
|
||||
- textual.onnx (text encoder)
|
||||
|
||||
These are exported from open_clip via bootstrap_models.py.
|
||||
These are exported from open_clip via export_models.py / bootstrap_models.py.
|
||||
"""
|
||||
import logging
|
||||
from pathlib import Path
|
||||
@@ -18,50 +22,63 @@ from app.services.vision.base import Embedder
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# OpenCLIP ViT-B/32 preprocessing constants (ImageNet norm)
|
||||
_MEAN = np.array([0.48145466, 0.4578275, 0.40821073], dtype=np.float32)
|
||||
_STD = np.array([0.26862954, 0.26130258, 0.27577711], dtype=np.float32)
|
||||
_INPUT_SIZE = 224
|
||||
# ── Model-specific constants ──────────────────────────────────────────
|
||||
|
||||
# OpenCLIP ViT-B/32 (ImageNet norm, 224px)
|
||||
_OPENCLIP_MEAN = np.array([0.48145466, 0.4578275, 0.40821073], dtype=np.float32)
|
||||
_OPENCLIP_STD = np.array([0.26862954, 0.26130258, 0.27577711], dtype=np.float32)
|
||||
_OPENCLIP_SIZE = 224
|
||||
|
||||
# SigLIP2 ViT-B/16 (SigLIP norm, 384px)
|
||||
_SIGLIP2_MEAN = np.array([0.5, 0.5, 0.5], dtype=np.float32)
|
||||
_SIGLIP2_STD = np.array([0.5, 0.5, 0.5], dtype=np.float32)
|
||||
_SIGLIP2_SIZE = 384
|
||||
|
||||
|
||||
def _preprocess_image(image: np.ndarray) -> np.ndarray:
|
||||
def _preprocess_image(
|
||||
image: np.ndarray,
|
||||
input_size: int,
|
||||
mean: np.ndarray,
|
||||
std: np.ndarray,
|
||||
) -> np.ndarray:
|
||||
"""Resize, center-crop, normalize an RGB uint8 image to NCHW float32."""
|
||||
from PIL import Image
|
||||
|
||||
img = Image.fromarray(image).convert("RGB")
|
||||
# Resize shortest edge to _INPUT_SIZE, then center crop
|
||||
w, h = img.size
|
||||
scale = _INPUT_SIZE / min(w, h)
|
||||
scale = input_size / min(w, h)
|
||||
img = img.resize((int(w * scale), int(h * scale)), Image.BICUBIC)
|
||||
w, h = img.size
|
||||
left = (w - _INPUT_SIZE) // 2
|
||||
top = (h - _INPUT_SIZE) // 2
|
||||
img = img.crop((left, top, left + _INPUT_SIZE, top + _INPUT_SIZE))
|
||||
left = (w - input_size) // 2
|
||||
top = (h - input_size) // 2
|
||||
img = img.crop((left, top, left + input_size, top + input_size))
|
||||
|
||||
arr = np.array(img, dtype=np.float32) / 255.0
|
||||
arr = (arr - _MEAN) / _STD
|
||||
arr = (arr - mean) / std
|
||||
arr = arr.transpose(2, 0, 1) # HWC → CHW
|
||||
return arr[np.newaxis] # NCHW
|
||||
|
||||
|
||||
class OpenCLIPEmbedder(Embedder):
|
||||
"""Legacy OpenCLIP ViT-B/32 embedder (512-d)."""
|
||||
|
||||
def __init__(self, settings: VisionSettings):
|
||||
model_dir = Path(settings.models_dir) / "embed"
|
||||
visual_path = model_dir / "visual.onnx"
|
||||
textual_path = model_dir / "textual.onnx"
|
||||
|
||||
opts = ort.SessionOptions()
|
||||
opts.inter_op_num_threads = 2
|
||||
opts.intra_op_num_threads = 2
|
||||
from app.services.vision.providers import create_session
|
||||
from app.config import settings as app_settings
|
||||
providers = app_settings.vision.execution_providers
|
||||
|
||||
logger.info("Loading visual encoder from %s", visual_path)
|
||||
self._visual = ort.InferenceSession(str(visual_path), opts, providers=["CPUExecutionProvider"])
|
||||
logger.info("Loading OpenCLIP visual encoder from %s", visual_path)
|
||||
self._visual = create_session(str(visual_path), configured_providers=providers)
|
||||
|
||||
logger.info("Loading textual encoder from %s", textual_path)
|
||||
self._textual = ort.InferenceSession(str(textual_path), opts, providers=["CPUExecutionProvider"])
|
||||
logger.info("Loading OpenCLIP textual encoder from %s", textual_path)
|
||||
self._textual = create_session(str(textual_path), configured_providers=providers)
|
||||
|
||||
def embed_image(self, image: np.ndarray) -> np.ndarray:
|
||||
inp = _preprocess_image(image)
|
||||
inp = _preprocess_image(image, _OPENCLIP_SIZE, _OPENCLIP_MEAN, _OPENCLIP_STD)
|
||||
input_name = self._visual.get_inputs()[0].name
|
||||
out = self._visual.run(None, {input_name: inp})[0][0]
|
||||
out = out / np.linalg.norm(out)
|
||||
@@ -71,7 +88,6 @@ class OpenCLIPEmbedder(Embedder):
|
||||
import open_clip
|
||||
tokenizer = open_clip.get_tokenizer("ViT-B-32")
|
||||
tokens = tokenizer([text]).numpy().astype(np.int64)
|
||||
# Compute EOT indices outside ONNX (avoids ArgMax(13) op)
|
||||
eot_indices = tokens.argmax(axis=-1).astype(np.int64)
|
||||
inputs = self._textual.get_inputs()
|
||||
out = self._textual.run(None, {
|
||||
@@ -84,3 +100,47 @@ class OpenCLIPEmbedder(Embedder):
|
||||
@property
|
||||
def dim(self) -> int:
|
||||
return 512
|
||||
|
||||
|
||||
class SigLIP2Embedder(Embedder):
|
||||
"""SigLIP2 ViT-B/16 embedder (768-d) — higher recall than OpenCLIP."""
|
||||
|
||||
def __init__(self, settings: VisionSettings):
|
||||
model_dir = Path(settings.models_dir) / "embed_siglip2"
|
||||
visual_path = model_dir / "visual.onnx"
|
||||
textual_path = model_dir / "textual.onnx"
|
||||
|
||||
from app.services.vision.providers import create_session
|
||||
from app.config import settings as app_settings
|
||||
providers = app_settings.vision.execution_providers
|
||||
|
||||
logger.info("Loading SigLIP2 visual encoder from %s", visual_path)
|
||||
self._visual = create_session(str(visual_path), configured_providers=providers)
|
||||
|
||||
logger.info("Loading SigLIP2 textual encoder from %s", textual_path)
|
||||
self._textual = create_session(str(textual_path), configured_providers=providers)
|
||||
|
||||
def embed_image(self, image: np.ndarray) -> np.ndarray:
|
||||
inp = _preprocess_image(image, _SIGLIP2_SIZE, _SIGLIP2_MEAN, _SIGLIP2_STD)
|
||||
input_name = self._visual.get_inputs()[0].name
|
||||
out = self._visual.run(None, {input_name: inp})[0][0]
|
||||
out = out / np.linalg.norm(out)
|
||||
return out.astype(np.float32)
|
||||
|
||||
def embed_text(self, text: str) -> np.ndarray:
|
||||
import open_clip
|
||||
tokenizer = open_clip.get_tokenizer("ViT-B-16-SigLIP-384")
|
||||
tokens = tokenizer([text]).numpy().astype(np.int64)
|
||||
inputs = self._textual.get_inputs()
|
||||
feed = {inputs[0].name: tokens}
|
||||
# SigLIP2 text encoder may need attention mask
|
||||
if len(inputs) > 1:
|
||||
attention_mask = (tokens != 0).astype(np.int64)
|
||||
feed[inputs[1].name] = attention_mask
|
||||
out = self._textual.run(None, feed)[0][0]
|
||||
out = out / np.linalg.norm(out)
|
||||
return out.astype(np.float32)
|
||||
|
||||
@property
|
||||
def dim(self) -> int:
|
||||
return 768
|
||||
|
||||
@@ -120,6 +120,76 @@ def export_openclip(models_dir: Path):
|
||||
logger.info("Textual encoder exported (%.1f MB)", size_mb)
|
||||
|
||||
|
||||
def export_siglip2(models_dir: Path):
|
||||
"""Export SigLIP2 ViT-B/16 to two ONNX files (visual + textual)."""
|
||||
import torch
|
||||
import open_clip
|
||||
|
||||
out_dir = models_dir / "embed_siglip2"
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
visual_path = out_dir / "visual.onnx"
|
||||
textual_path = out_dir / "textual.onnx"
|
||||
|
||||
if visual_path.exists() and textual_path.exists():
|
||||
logger.info("SigLIP2 ONNX files already exist, skipping export")
|
||||
return
|
||||
|
||||
logger.info("Loading SigLIP2 ViT-B-16-SigLIP-384 webli...")
|
||||
model, _, preprocess = open_clip.create_model_and_transforms(
|
||||
"ViT-B-16-SigLIP-384", pretrained="webli"
|
||||
)
|
||||
model.eval()
|
||||
|
||||
export_kwargs = dict(opset_version=14, dynamo=False)
|
||||
|
||||
# ── Visual encoder ────────────────────────────────────────────────
|
||||
if not visual_path.exists():
|
||||
logger.info("Exporting SigLIP2 visual encoder → %s", visual_path)
|
||||
dummy_image = torch.randn(1, 3, 384, 384)
|
||||
torch.onnx.export(
|
||||
model.visual,
|
||||
dummy_image,
|
||||
str(visual_path),
|
||||
input_names=["image"],
|
||||
output_names=["embedding"],
|
||||
dynamic_axes={"image": {0: "batch"}},
|
||||
**export_kwargs,
|
||||
)
|
||||
size_mb = visual_path.stat().st_size / 1e6
|
||||
logger.info("SigLIP2 visual encoder exported (%.1f MB)", size_mb)
|
||||
|
||||
# ── Textual encoder ───────────────────────────────────────────────
|
||||
if not textual_path.exists():
|
||||
logger.info("Exporting SigLIP2 textual encoder → %s", textual_path)
|
||||
tokenizer = open_clip.get_tokenizer("ViT-B-16-SigLIP-384")
|
||||
dummy_text = tokenizer(["a photo"]).to(torch.int64)
|
||||
|
||||
class SigLIP2TextEncoder(torch.nn.Module):
|
||||
"""Wrap the SigLIP2 text transformer for ONNX export."""
|
||||
def __init__(self, clip_model):
|
||||
super().__init__()
|
||||
self.text = clip_model.text
|
||||
|
||||
def forward(self, text):
|
||||
return self.text(text)
|
||||
|
||||
text_enc = SigLIP2TextEncoder(model)
|
||||
text_enc.eval()
|
||||
|
||||
torch.onnx.export(
|
||||
text_enc,
|
||||
dummy_text,
|
||||
str(textual_path),
|
||||
input_names=["text"],
|
||||
output_names=["embedding"],
|
||||
dynamic_axes={"text": {0: "batch"}},
|
||||
**export_kwargs,
|
||||
)
|
||||
size_mb = textual_path.stat().st_size / 1e6
|
||||
logger.info("SigLIP2 textual encoder exported (%.1f MB)", size_mb)
|
||||
|
||||
|
||||
def export_yolov8n(models_dir: Path):
|
||||
"""Export YOLOv8n to ONNX."""
|
||||
out_dir = models_dir / "detect"
|
||||
@@ -174,6 +244,7 @@ def main():
|
||||
logger.info("Exporting models to %s", models_dir)
|
||||
|
||||
export_openclip(models_dir)
|
||||
export_siglip2(models_dir)
|
||||
export_yolov8n(models_dir)
|
||||
|
||||
logger.info("Done. Run bootstrap_models.py next to download YuNet + SFace.")
|
||||
|
||||
@@ -71,11 +71,9 @@ class YuNetSFaceProcessor(FaceProcessor):
|
||||
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
|
||||
from app.services.vision.providers import create_session
|
||||
ort.set_default_logger_severity(3)
|
||||
self._recognizer = ort.InferenceSession(sface_path, opts, providers=["CPUExecutionProvider"])
|
||||
self._recognizer = create_session(sface_path, configured_providers=settings.execution_providers)
|
||||
logger.info("SFace recognizer loaded via ONNX Runtime")
|
||||
|
||||
self._min_face_size = settings.faces.min_face_size
|
||||
|
||||
@@ -23,10 +23,13 @@ class InsightFaceProcessor(FaceProcessor):
|
||||
model_root = str(Path(settings.models_dir) / "face" / "insightface")
|
||||
logger.info("Loading InsightFace buffalo_l from %s", model_root)
|
||||
|
||||
from app.services.vision.providers import get_providers
|
||||
providers = get_providers(settings.execution_providers)
|
||||
|
||||
self._app = FaceAnalysis(
|
||||
name="buffalo_l",
|
||||
root=model_root,
|
||||
providers=["CPUExecutionProvider"],
|
||||
providers=providers,
|
||||
)
|
||||
self._app.prepare(ctx_id=-1, det_size=(640, 640))
|
||||
self._min_det_score = settings.faces.recognition_threshold
|
||||
|
||||
@@ -21,6 +21,11 @@ class ONNXBackend:
|
||||
self._settings = vision_settings
|
||||
|
||||
def create_embedder(self) -> Embedder:
|
||||
model_name = self._settings.embedder.name
|
||||
if model_name.startswith("siglip2"):
|
||||
from app.services.vision.embed import SigLIP2Embedder
|
||||
return SigLIP2Embedder(self._settings)
|
||||
else:
|
||||
from app.services.vision.embed import OpenCLIPEmbedder
|
||||
return OpenCLIPEmbedder(self._settings)
|
||||
|
||||
|
||||
86
backend/app/services/vision/providers.py
Normal file
86
backend/app/services/vision/providers.py
Normal file
@@ -0,0 +1,86 @@
|
||||
"""
|
||||
ONNX Runtime execution provider resolution with GPU auto-detection.
|
||||
|
||||
Resolves configured execution providers against what's actually available
|
||||
in the current ONNX Runtime build. Falls back to CPU if no GPU provider
|
||||
is available. Logs the selected provider so users can confirm GPU is active.
|
||||
"""
|
||||
import logging
|
||||
|
||||
import onnxruntime as ort
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_resolved: list[str] | None = None
|
||||
|
||||
|
||||
def get_providers(configured: list[str] | None = None) -> list[str]:
|
||||
"""Return the best available execution providers.
|
||||
|
||||
1. If `configured` is provided, filter to only those that are
|
||||
actually available in the current ORT build.
|
||||
2. If none of the configured providers are available, fall back
|
||||
to CPUExecutionProvider.
|
||||
3. Auto-detect: if configured is ["auto"], probe for GPU providers.
|
||||
|
||||
Results are cached after first call.
|
||||
"""
|
||||
global _resolved
|
||||
if _resolved is not None:
|
||||
return _resolved
|
||||
|
||||
available = set(ort.get_available_providers())
|
||||
logger.info("ONNX Runtime available providers: %s", sorted(available))
|
||||
|
||||
if configured is None or configured == ["CPUExecutionProvider"]:
|
||||
_resolved = ["CPUExecutionProvider"]
|
||||
return _resolved
|
||||
|
||||
if configured == ["auto"]:
|
||||
# Auto-detect: prefer CUDA > ROCm > OpenVINO > CPU
|
||||
priority = [
|
||||
"CUDAExecutionProvider",
|
||||
"ROCMExecutionProvider",
|
||||
"OpenVINOExecutionProvider",
|
||||
]
|
||||
for p in priority:
|
||||
if p in available:
|
||||
_resolved = [p, "CPUExecutionProvider"]
|
||||
logger.info("Auto-detected GPU provider: %s", p)
|
||||
return _resolved
|
||||
_resolved = ["CPUExecutionProvider"]
|
||||
logger.info("No GPU provider detected, using CPU")
|
||||
return _resolved
|
||||
|
||||
# Filter configured list to available providers.
|
||||
resolved = [p for p in configured if p in available]
|
||||
if not resolved:
|
||||
logger.warning(
|
||||
"None of the configured providers %s are available. "
|
||||
"Falling back to CPU. Available: %s",
|
||||
configured,
|
||||
sorted(available),
|
||||
)
|
||||
resolved = ["CPUExecutionProvider"]
|
||||
else:
|
||||
# Always include CPU as fallback.
|
||||
if "CPUExecutionProvider" not in resolved:
|
||||
resolved.append("CPUExecutionProvider")
|
||||
|
||||
_resolved = resolved
|
||||
logger.info("Using ONNX Runtime providers: %s", _resolved)
|
||||
return _resolved
|
||||
|
||||
|
||||
def create_session(
|
||||
model_path: str,
|
||||
opts: ort.SessionOptions | None = None,
|
||||
configured_providers: list[str] | None = None,
|
||||
) -> ort.InferenceSession:
|
||||
"""Create an ONNX InferenceSession with the best available providers."""
|
||||
providers = get_providers(configured_providers)
|
||||
if opts is None:
|
||||
opts = ort.SessionOptions()
|
||||
opts.inter_op_num_threads = 2
|
||||
opts.intra_op_num_threads = 2
|
||||
return ort.InferenceSession(model_path, opts, providers=providers)
|
||||
@@ -29,6 +29,7 @@ celery_app.conf.update(
|
||||
'extract_faces': {'queue': 'vision'},
|
||||
'classify_content': {'queue': 'vision'},
|
||||
'vision_fanout': {'queue': 'vision'},
|
||||
'watch_folders': {'queue': 'watcher'},
|
||||
},
|
||||
task_default_queue='default',
|
||||
task_default_exchange='default',
|
||||
|
||||
@@ -5,7 +5,7 @@ import os
|
||||
import hashlib
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
from datetime import datetime, timezone
|
||||
import logging
|
||||
import json
|
||||
from typing import List, Dict, Optional
|
||||
@@ -419,7 +419,7 @@ async def _scan_all_source_roots_async():
|
||||
countdown is a best-effort hint — on a big library the user can
|
||||
still hit Settings → Re-detect duplicates to force a fresh pass.
|
||||
"""
|
||||
from app.tasks.thumbs import regroup_duplicates_task
|
||||
from app.tasks.thumbs import incremental_regroup_duplicates_task
|
||||
from app.tasks.vision import backfill_vision, recluster_faces
|
||||
|
||||
async with AsyncSessionLocal() as session:
|
||||
@@ -441,7 +441,14 @@ async def _scan_all_source_roots_async():
|
||||
# firing too early just means the next manual run picks up the
|
||||
# late arrivals — no corrupted state.
|
||||
try:
|
||||
regroup_duplicates_task.apply_async(countdown=60)
|
||||
# Use incremental mode: only compare newly added photos
|
||||
# against the full library via CLIP HNSW + pHash.
|
||||
# O(new × log N) instead of O(N²).
|
||||
scan_start = datetime.now(timezone.utc).isoformat()
|
||||
incremental_regroup_duplicates_task.apply_async(
|
||||
kwargs={'since_iso': scan_start},
|
||||
countdown=60,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not queue post-scan regroup: {e}")
|
||||
|
||||
@@ -461,17 +468,33 @@ async def _scan_all_source_roots_async():
|
||||
logger.warning(f"Could not queue post-scan face recluster: {e}")
|
||||
|
||||
|
||||
@shared_task(name='watch_folders')
|
||||
def watch_folders():
|
||||
WATCHER_LOCK_KEY = "mulita:watch_folders:lock"
|
||||
WATCHER_LOCK_TTL = 300 # 5 min — renewed every 60s
|
||||
|
||||
|
||||
@shared_task(name='watch_folders', bind=True)
|
||||
def watch_folders(self):
|
||||
"""
|
||||
Watch folders for changes using watchfiles. Long-running task that
|
||||
monitors filesystem events under every active source root.
|
||||
|
||||
Uses a Redis lock to ensure only one instance runs across all
|
||||
workers. The lock is renewed periodically so it survives restarts
|
||||
without leaving orphan watchers.
|
||||
"""
|
||||
import redis as redis_lib
|
||||
from watchfiles import watch
|
||||
|
||||
# Read source roots from the DB instead of the (now-removed) YAML
|
||||
# config. We need both the path and the id so we can dispatch
|
||||
# scan_folder with the source_root_id when an event fires.
|
||||
r = redis_lib.from_url(settings.redis_url)
|
||||
|
||||
# Acquire exclusive lock — if another watcher is already running,
|
||||
# this instance exits immediately instead of stacking up.
|
||||
lock = r.lock(WATCHER_LOCK_KEY, timeout=WATCHER_LOCK_TTL)
|
||||
if not lock.acquire(blocking=False):
|
||||
logger.info("watch_folders: another instance is already running, exiting")
|
||||
return {'status': 'skipped', 'reason': 'another instance is running'}
|
||||
|
||||
try:
|
||||
roots: list[tuple[str, str]] = []
|
||||
try:
|
||||
async def _load_roots():
|
||||
@@ -504,31 +527,37 @@ def watch_folders():
|
||||
return root_id
|
||||
return None
|
||||
|
||||
renew_counter = 0
|
||||
for changes in watch(*paths):
|
||||
# Renew the Redis lock periodically so it doesn't expire
|
||||
# while the watcher is idle between events.
|
||||
renew_counter += 1
|
||||
if renew_counter % 10 == 0:
|
||||
try:
|
||||
lock.extend(WATCHER_LOCK_TTL)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
for change_type, filepath in changes:
|
||||
filepath = str(filepath)
|
||||
|
||||
# Check if it's a supported file type
|
||||
if Path(filepath).suffix.lower() not in SUPPORTED_EXTENSIONS:
|
||||
continue
|
||||
|
||||
if change_type == 'added' or change_type == 'modified':
|
||||
# Queue scan for the parent folder, with the source_root_id
|
||||
# resolved by ancestor lookup so scan_folder doesn't
|
||||
# auto-create a new SourceRoot for an arbitrary subdir.
|
||||
parent_dir = str(Path(filepath).parent)
|
||||
source_root_id = find_source_root_for(parent_dir)
|
||||
if source_root_id is None:
|
||||
logger.debug(
|
||||
f"watcher event for {filepath}: parent {parent_dir} "
|
||||
f"not under any active source root, ignoring"
|
||||
)
|
||||
continue
|
||||
scan_folder.delay(parent_dir, source_root_id)
|
||||
logger.info(f"File {change_type}: {filepath}, queued scan for {parent_dir}")
|
||||
elif change_type == 'deleted':
|
||||
# Handle file deletion
|
||||
asyncio.run(handle_file_deletion(filepath))
|
||||
finally:
|
||||
try:
|
||||
lock.release()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
async def handle_file_deletion(filepath: str):
|
||||
"""Handle deletion of a file from the filesystem"""
|
||||
|
||||
@@ -467,10 +467,23 @@ async def _backfill_phashes_async():
|
||||
|
||||
@shared_task(name='regroup_duplicates')
|
||||
def regroup_duplicates_task():
|
||||
"""Celery wrapper around app.services.duplicates.regroup_duplicates.
|
||||
"""Full recompute of duplicate groups (pHash + CLIP similarity).
|
||||
|
||||
Importing the service inside the task body avoids a circular import
|
||||
at worker boot (the service uses AsyncSessionLocal which is also
|
||||
imported here at module top)."""
|
||||
Used by the Settings → Re-detect duplicates button."""
|
||||
from app.services.duplicates import regroup_duplicates
|
||||
return asyncio.run(regroup_duplicates())
|
||||
|
||||
|
||||
@shared_task(name='incremental_regroup_duplicates')
|
||||
def incremental_regroup_duplicates_task(since_iso: str | None = None):
|
||||
"""Incremental duplicate detection for newly added photos.
|
||||
|
||||
Compares only photos added after `since_iso` against the full library
|
||||
using CLIP vector similarity (O(new × log N) via HNSW) plus pHash.
|
||||
Default post-scan path — much faster than a full regroup."""
|
||||
from app.services.duplicates import incremental_regroup
|
||||
from datetime import datetime, timezone
|
||||
since = None
|
||||
if since_iso:
|
||||
since = datetime.fromisoformat(since_iso)
|
||||
return asyncio.run(incremental_regroup(since=since))
|
||||
@@ -163,6 +163,12 @@ def detect_objects(photo_id: str):
|
||||
|
||||
session = _get_sync_session()
|
||||
try:
|
||||
# Get the photo's user_id so tags inherit ownership.
|
||||
photo = session.execute(
|
||||
select(Photo).where(Photo.id == photo_id)
|
||||
).scalar_one_or_none()
|
||||
owner_id = photo.user_id if photo else None
|
||||
|
||||
# Wipe previous detection results for this photo from this model
|
||||
session.execute(
|
||||
delete(photo_tags).where(
|
||||
@@ -178,13 +184,13 @@ def detect_objects(photo_id: str):
|
||||
best_per_label[det.label] = (det.confidence, det.bbox)
|
||||
|
||||
for label, (confidence, bbox) in best_per_label.items():
|
||||
# Find or create the object tag
|
||||
# Find or create the object tag (scoped to user)
|
||||
tag = session.execute(
|
||||
select(Tag).where(Tag.name == label, Tag.kind == 'object')
|
||||
select(Tag).where(Tag.name == label, Tag.kind == 'object', Tag.user_id == owner_id)
|
||||
).scalar_one_or_none()
|
||||
|
||||
if not tag:
|
||||
tag = Tag(name=label, kind='object', source=source_name)
|
||||
tag = Tag(name=label, kind='object', source=source_name, user_id=owner_id)
|
||||
session.add(tag)
|
||||
session.flush() # get tag.id
|
||||
|
||||
@@ -234,6 +240,12 @@ def classify_content(photo_id: str):
|
||||
|
||||
session = _get_sync_session()
|
||||
try:
|
||||
# Get the photo's user_id so tags inherit ownership.
|
||||
photo = session.execute(
|
||||
select(Photo).where(Photo.id == photo_id)
|
||||
).scalar_one_or_none()
|
||||
owner_id = photo.user_id if photo else None
|
||||
|
||||
# Wipe previous classification for this photo
|
||||
session.execute(
|
||||
delete(photo_tags).where(
|
||||
@@ -242,13 +254,13 @@ def classify_content(photo_id: str):
|
||||
)
|
||||
)
|
||||
|
||||
# Find or create content_type tag
|
||||
# Find or create content_type tag (scoped to user)
|
||||
tag = session.execute(
|
||||
select(Tag).where(Tag.name == best.label, Tag.kind == 'content_type')
|
||||
select(Tag).where(Tag.name == best.label, Tag.kind == 'content_type', Tag.user_id == owner_id)
|
||||
).scalar_one_or_none()
|
||||
|
||||
if not tag:
|
||||
tag = Tag(name=best.label, kind='content_type', source=source_name)
|
||||
tag = Tag(name=best.label, kind='content_type', source=source_name, user_id=owner_id)
|
||||
session.add(tag)
|
||||
session.flush()
|
||||
|
||||
@@ -434,11 +446,16 @@ def recluster_faces():
|
||||
|
||||
if label not in cluster_tag_map:
|
||||
cluster_name = f"Person {label + 1}"
|
||||
# Inherit user_id from the representative photo.
|
||||
rep_photo = session.execute(
|
||||
select(Photo.user_id).where(Photo.id == face_rows[i].photo_id)
|
||||
).scalar_one_or_none()
|
||||
tag = Tag(
|
||||
name=cluster_name,
|
||||
kind='face_cluster',
|
||||
source=source_name,
|
||||
representative_photo_id=face_rows[i].photo_id,
|
||||
user_id=rep_photo,
|
||||
)
|
||||
session.add(tag)
|
||||
session.flush()
|
||||
|
||||
@@ -37,6 +37,7 @@ watchfiles==0.21.0
|
||||
# Vision pipeline (ONNX Runtime CPU inference)
|
||||
onnxruntime==1.18.1
|
||||
open-clip-torch==2.24.0 # tokenizer + export helper; inference via ONNX
|
||||
transformers>=4.37.0 # HuggingFace tokenizer for SigLIP models
|
||||
ultralytics==8.4.37 # YOLOv8n export helper; inference via ONNX
|
||||
rapidocr-onnxruntime==1.3.22
|
||||
scikit-learn==1.4.0 # DBSCAN for face clustering
|
||||
|
||||
@@ -120,6 +120,37 @@ services:
|
||||
- mulita-network
|
||||
restart: unless-stopped
|
||||
|
||||
# Dedicated watcher worker — runs the long-lived watch_folders task
|
||||
# on its own queue so it never blocks scan/thumbnail workers.
|
||||
worker-watcher:
|
||||
build:
|
||||
context: ./backend
|
||||
dockerfile: Dockerfile
|
||||
image: mule-image-worker
|
||||
container_name: mulita-worker-watcher
|
||||
command: sh -c "celery -A app.tasks.celery worker --loglevel=${LOG_LEVEL:-info} --concurrency=1 -Q watcher -n watcher@%h"
|
||||
volumes:
|
||||
- ./mulita.yml:/app/config/mulita.yml:ro
|
||||
- ${PHOTO_DIRS:-./photos}:/photos:rw
|
||||
- db_data:/data/db
|
||||
environment:
|
||||
- DATABASE_URL=postgresql+asyncpg://mulita:mulita@db:5432/mulita
|
||||
- REDIS_URL=redis://redis:6379
|
||||
- CELERY_BROKER_URL=redis://redis:6379
|
||||
- CELERY_RESULT_BACKEND=redis://redis:6379
|
||||
- PHOTO_DIRS=${PHOTO_DIRS:-/photos}
|
||||
- LOG_LEVEL=${LOG_LEVEL:-INFO}
|
||||
- TZ=${TZ:-UTC}
|
||||
- MULITA_CELERY_WORKER=1
|
||||
depends_on:
|
||||
redis:
|
||||
condition: service_started
|
||||
db:
|
||||
condition: service_healthy
|
||||
networks:
|
||||
- mulita-network
|
||||
restart: unless-stopped
|
||||
|
||||
worker-vision:
|
||||
build:
|
||||
context: ./backend
|
||||
@@ -143,6 +174,13 @@ services:
|
||||
- LOG_LEVEL=${LOG_LEVEL:-INFO}
|
||||
- TZ=${TZ:-UTC}
|
||||
- MULITA_CELERY_WORKER=1
|
||||
# ONNX Runtime execution providers. Set to "auto" to auto-detect
|
||||
# GPU (CUDA > ROCm > OpenVINO > CPU), or explicitly:
|
||||
# "CUDAExecutionProvider,CPUExecutionProvider"
|
||||
# "ROCMExecutionProvider,CPUExecutionProvider"
|
||||
# Default: CPU only. To enable GPU, also uncomment the deploy
|
||||
# section below and install nvidia-container-toolkit on the host.
|
||||
- VISION_EXECUTION_PROVIDERS=${VISION_EXECUTION_PROVIDERS:-CPUExecutionProvider}
|
||||
# Pin each ONNX session to one intra-op thread so N prefork children
|
||||
# × default-all-cores doesn't oversubscribe the box. With
|
||||
# concurrency=5 and OMP=1, vision peaks at 5 busy cores, leaving
|
||||
@@ -151,6 +189,14 @@ services:
|
||||
- OMP_NUM_THREADS=1
|
||||
- OPENBLAS_NUM_THREADS=1
|
||||
- MKL_NUM_THREADS=1
|
||||
# Uncomment for NVIDIA GPU passthrough:
|
||||
# deploy:
|
||||
# resources:
|
||||
# reservations:
|
||||
# devices:
|
||||
# - driver: nvidia
|
||||
# count: all
|
||||
# capabilities: [gpu]
|
||||
depends_on:
|
||||
redis:
|
||||
condition: service_started
|
||||
|
||||
@@ -2,6 +2,7 @@ import { useState } from 'react'
|
||||
import { Timeline } from './components/timeline/Timeline'
|
||||
import { DuplicatesView } from './components/duplicates/DuplicatesView'
|
||||
import { MapView } from './components/map/MapView'
|
||||
import { MemoriesView } from './components/memories/MemoriesView'
|
||||
import { PeopleView } from './components/people/PeopleView'
|
||||
import { TagsView } from './components/tags/TagsView'
|
||||
import { ColorsView } from './components/colors/ColorsView'
|
||||
@@ -87,6 +88,8 @@ function MainApp() {
|
||||
<SettingsPage />
|
||||
) : currentSection === 'map' ? (
|
||||
<MapView />
|
||||
) : currentSection === 'memories' ? (
|
||||
<MemoriesView />
|
||||
) : currentSection === 'duplicates' ? (
|
||||
<DuplicatesView />
|
||||
) : currentSection === 'people' ? (
|
||||
|
||||
@@ -23,6 +23,7 @@ import {
|
||||
User as UserIcon,
|
||||
LogOut,
|
||||
Shield,
|
||||
Clock,
|
||||
} from 'lucide-react'
|
||||
import clsx from 'clsx'
|
||||
import { sourceFolders, photos as photosApi, type FolderTreeNode } from '../../services/api'
|
||||
@@ -275,6 +276,9 @@ export function LeftSidebar({ onCollapse }: LeftSidebarProps) {
|
||||
case 'map':
|
||||
navigateToSection('map', {})
|
||||
break
|
||||
case 'memories':
|
||||
navigateToSection('memories', {})
|
||||
break
|
||||
default:
|
||||
if (id.startsWith('folder-')) {
|
||||
const folderId = id.slice('folder-'.length)
|
||||
@@ -407,6 +411,7 @@ export function LeftSidebar({ onCollapse }: LeftSidebarProps) {
|
||||
{ id: 'people', label: 'People', icon: <Users className="h-4 w-4" />, count: peopleTotalCount },
|
||||
{ id: 'colors', label: 'Colors', icon: <Palette className="h-4 w-4" />, count: stats?.colored ?? 0 },
|
||||
{ id: 'map', label: 'Map', icon: <MapPin className="h-4 w-4" />, count: stats?.with_gps ?? 0 },
|
||||
{ id: 'memories', label: 'Memories', icon: <Clock className="h-4 w-4" /> },
|
||||
{ id: 'duplicates', label: 'Duplicates', icon: <Copy className="h-4 w-4" />, count: stats?.duplicates ?? 0 },
|
||||
{ id: 'discarded', label: 'Discarded', icon: <Trash2 className="h-4 w-4" />, count: stats?.discarded ?? 0 },
|
||||
],
|
||||
|
||||
70
frontend/src/components/memories/MemoriesView.tsx
Normal file
70
frontend/src/components/memories/MemoriesView.tsx
Normal file
@@ -0,0 +1,70 @@
|
||||
import { useQuery } from '@tanstack/react-query'
|
||||
import { photos, photos as photosApi } from '../../services/api'
|
||||
import type { MemoryGroup } from '../../services/api'
|
||||
|
||||
export function MemoriesView() {
|
||||
const { data, isLoading } = useQuery({
|
||||
queryKey: ['memories'],
|
||||
queryFn: () => photos.memories(),
|
||||
staleTime: 60_000 * 30, // 30 min — date doesn't change often
|
||||
})
|
||||
|
||||
if (isLoading) {
|
||||
return (
|
||||
<div className="flex items-center justify-center h-64 text-neutral-500">
|
||||
Loading memories...
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
const memories = data?.memories ?? []
|
||||
|
||||
if (memories.length === 0) {
|
||||
return (
|
||||
<div className="flex flex-col items-center justify-center h-64 text-neutral-500 gap-2">
|
||||
<p className="text-lg font-medium">No memories for today</p>
|
||||
<p className="text-sm">Photos taken on this date in previous years will appear here.</p>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
return (
|
||||
<div className="p-6 space-y-8 max-w-5xl mx-auto">
|
||||
<h2 className="text-xl font-semibold text-neutral-200">
|
||||
On This Day — {data?.date}
|
||||
</h2>
|
||||
|
||||
{memories.map((group: MemoryGroup) => (
|
||||
<section key={group.year} className="space-y-3">
|
||||
<h3 className="text-sm font-medium text-neutral-400 uppercase tracking-wide">
|
||||
{group.year} · {group.years_ago} year{group.years_ago !== 1 ? 's' : ''} ago
|
||||
</h3>
|
||||
<div className="grid grid-cols-2 sm:grid-cols-3 md:grid-cols-4 lg:grid-cols-6 gap-2">
|
||||
{group.photos.map((photo) => (
|
||||
<div
|
||||
key={photo.id}
|
||||
className="aspect-square rounded-lg overflow-hidden bg-neutral-800 relative group"
|
||||
>
|
||||
{photo.thumb_small ? (
|
||||
<img
|
||||
src={photosApi.getThumbnailUrl(photo.id, 'small')}
|
||||
alt={photo.filename}
|
||||
className="w-full h-full object-cover"
|
||||
loading="lazy"
|
||||
/>
|
||||
) : (
|
||||
<div className="w-full h-full flex items-center justify-center text-neutral-600 text-xs">
|
||||
No thumb
|
||||
</div>
|
||||
)}
|
||||
<div className="absolute bottom-0 left-0 right-0 bg-gradient-to-t from-black/60 to-transparent px-2 py-1 opacity-0 group-hover:opacity-100 transition-opacity">
|
||||
<p className="text-xs text-white truncate">{photo.filename}</p>
|
||||
</div>
|
||||
</div>
|
||||
))}
|
||||
</div>
|
||||
</section>
|
||||
))}
|
||||
</div>
|
||||
)
|
||||
}
|
||||
@@ -100,7 +100,8 @@ export function PhotoThumbnail({
|
||||
// Cache-bust on retry so the browser actually re-requests instead of
|
||||
// serving the cached 404.
|
||||
const baseUrl = photosApi.getThumbnailUrl(photo.id, 'medium')
|
||||
const thumbnailUrl = retryCount > 0 ? `${baseUrl}?retry=${retryCount}` : baseUrl
|
||||
const sep = baseUrl.includes('?') ? '&' : '?'
|
||||
const thumbnailUrl = retryCount > 0 ? `${baseUrl}${sep}retry=${retryCount}` : baseUrl
|
||||
|
||||
// "Capture date probably wrong" — read straight from the stored
|
||||
// `has_date_warning` flag rather than recomputing the heuristic
|
||||
|
||||
@@ -20,6 +20,19 @@ export function stripPhotosFromCache(queryClient: QueryClient, ids: string[]) {
|
||||
})
|
||||
}
|
||||
|
||||
interface CursorPage {
|
||||
photos: Photo[]
|
||||
next_cursor: string | null
|
||||
}
|
||||
|
||||
async function fetchCursorPage(
|
||||
params: Record<string, unknown>,
|
||||
signal?: AbortSignal,
|
||||
): Promise<CursorPage> {
|
||||
const resp = await api.get<CursorPage>('/photos', { params, signal })
|
||||
return resp.data
|
||||
}
|
||||
|
||||
/**
|
||||
* Single source of truth for the timeline photos query. Both Timeline and
|
||||
* PreviewView call this so they share one cache entry — previously
|
||||
@@ -70,55 +83,41 @@ export function usePhotosQuery() {
|
||||
return useQuery({
|
||||
queryKey: ['photos', filterParams],
|
||||
queryFn: async ({ signal }) => {
|
||||
// Two-phase fetch so the timeline can paint its first thumbnails
|
||||
// long before the entire library has finished downloading. Phase 1
|
||||
// returns the first page synchronously (which resolves the
|
||||
// useQuery promise so consumers exit their loading state). Phase 2
|
||||
// walks the remaining pages in the background, appending each one
|
||||
// into the cache via setQueryData so the grid grows as data
|
||||
// arrives. The signal from React Query aborts the background
|
||||
// loop if the query is invalidated or unmounts mid-stream.
|
||||
// Two-phase fetch using cursor-based (keyset) pagination.
|
||||
// Phase 1 returns the first page (resolves the useQuery promise
|
||||
// so consumers exit loading state). Phase 2 chains cursors in
|
||||
// the background — each response includes a `next_cursor` that
|
||||
// seeks directly to the next slice via an indexed range scan,
|
||||
// O(1) regardless of depth (no OFFSET skipping).
|
||||
const PER_PAGE = 500
|
||||
const MAX_PAGES = 200
|
||||
|
||||
const firstResp = await api.get<{
|
||||
photos: Photo[]
|
||||
total: number
|
||||
pages: number
|
||||
}>('/photos', {
|
||||
params: { page: 1, per_page: PER_PAGE, ...filterParams },
|
||||
const first = await fetchCursorPage(
|
||||
{ per_page: PER_PAGE, ...filterParams },
|
||||
signal,
|
||||
})
|
||||
const firstBatch = firstResp.data.photos || []
|
||||
const totalPages = firstResp.data.pages ?? 1
|
||||
)
|
||||
const firstBatch = first.photos || []
|
||||
let nextCursor: string | null = first.next_cursor
|
||||
|
||||
if (totalPages > 1 && firstBatch.length === PER_PAGE) {
|
||||
// Fire-and-forget background loop. We don't await here — the
|
||||
// first batch is already enough to render. Each subsequent
|
||||
// page lands via setQueryData, which triggers consumers to
|
||||
// re-render with the larger list.
|
||||
if (nextCursor) {
|
||||
// Fire-and-forget background loop using cursor chaining.
|
||||
void (async () => {
|
||||
for (let page = 2; page <= Math.min(totalPages, MAX_PAGES); page++) {
|
||||
for (let i = 0; i < MAX_PAGES && nextCursor; i++) {
|
||||
if (signal?.aborted) return
|
||||
try {
|
||||
const resp = await api.get<{
|
||||
photos: Photo[]
|
||||
total: number
|
||||
pages: number
|
||||
}>('/photos', {
|
||||
params: { page, per_page: PER_PAGE, ...filterParams },
|
||||
const page = await fetchCursorPage(
|
||||
{ per_page: PER_PAGE, cursor: nextCursor, ...filterParams },
|
||||
signal,
|
||||
})
|
||||
)
|
||||
if (signal?.aborted) return
|
||||
const more = resp.data.photos || []
|
||||
const more = page.photos || []
|
||||
nextCursor = page.next_cursor
|
||||
queryClient.setQueryData<Photo[]>(
|
||||
['photos', filterParams],
|
||||
(prev) => (prev ? [...prev, ...more] : more)
|
||||
)
|
||||
if (more.length < PER_PAGE) return
|
||||
if (!nextCursor || more.length < PER_PAGE) return
|
||||
} catch {
|
||||
// Network or abort — give up the background stream. The
|
||||
// next user-triggered refetch will start fresh.
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
@@ -316,17 +316,29 @@ export const photos = {
|
||||
},
|
||||
|
||||
getThumbnailUrl: (photoId: string, size: 'small' | 'medium' | 'large' = 'medium') => {
|
||||
return `${API_BASE_URL}/photos/${photoId}/thumb/${size}`
|
||||
const token = localStorage.getItem('access_token')
|
||||
const qs = token ? `?token=${encodeURIComponent(token)}` : ''
|
||||
return `${API_BASE_URL}/photos/${photoId}/thumb/${size}${qs}`
|
||||
},
|
||||
|
||||
getOriginalUrl: (photoId: string) => {
|
||||
return `${API_BASE_URL}/photos/${photoId}/original`
|
||||
const token = localStorage.getItem('access_token')
|
||||
const qs = token ? `?token=${encodeURIComponent(token)}` : ''
|
||||
return `${API_BASE_URL}/photos/${photoId}/original${qs}`
|
||||
},
|
||||
|
||||
/** Full-resolution display URL. Backend serves the original for web-safe
|
||||
* formats and a transcoded WebP for RAW/HEIC/TIFF. */
|
||||
getProxyUrl: (photoId: string) => {
|
||||
return `${API_BASE_URL}/photos/${photoId}/proxy`
|
||||
const token = localStorage.getItem('access_token')
|
||||
const qs = token ? `?token=${encodeURIComponent(token)}` : ''
|
||||
return `${API_BASE_URL}/photos/${photoId}/proxy${qs}`
|
||||
},
|
||||
|
||||
/** "On this day" memories — photos taken on this date in previous years. */
|
||||
memories: async (): Promise<MemoriesResponse> => {
|
||||
const response = await api.get('/photos/memories')
|
||||
return response.data
|
||||
},
|
||||
}
|
||||
|
||||
@@ -577,6 +589,31 @@ export interface DuplicateGroupsResponse {
|
||||
total_members: number
|
||||
}
|
||||
|
||||
// ── Memories ("On this day") ────────────────────────────────────────────
|
||||
|
||||
export interface MemoryPhoto {
|
||||
id: string
|
||||
filename: string
|
||||
taken_at: string
|
||||
thumb_small: string | null
|
||||
thumb_medium: string | null
|
||||
media_type: string
|
||||
width: number | null
|
||||
height: number | null
|
||||
rating: number
|
||||
}
|
||||
|
||||
export interface MemoryGroup {
|
||||
year: number
|
||||
years_ago: number
|
||||
photos: MemoryPhoto[]
|
||||
}
|
||||
|
||||
export interface MemoriesResponse {
|
||||
date: string
|
||||
memories: MemoryGroup[]
|
||||
}
|
||||
|
||||
export interface LibraryStats {
|
||||
all_photos: number
|
||||
rated: number
|
||||
|
||||
Reference in New Issue
Block a user