10 Commits

Author SHA1 Message Date
b7aa2aed3d fix: all users get subfolders, nobody owns the mount root
Every user — including the initial admin — now gets their own
subdirectory under PHOTO_DIRS (e.g. /photos/admin, /photos/bob).
No one's source root points to the mount root itself, eliminating
cross-user photo overlap entirely.

- Setup endpoint: admin gets /photos/{username} like everyone else
- Migration: default admin media_path set to /photos/admin
- Remove scan directory pruning (no longer needed)
- Fix thumbnail retry URL: use & separator when token query param
  already present (was producing ?token=...?retry=N)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-13 00:18:42 +02:00
180efb3eb0 fix: admin scan skips other users' source root directories
When the admin's source root is the mount root (/photos) and other
users have subdirectories (/photos/bob), the admin's scan now prunes
those directories from os.walk so photos aren't double-indexed under
the wrong user. The scanner queries all active source roots owned by
other users and excludes their paths during directory traversal.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-13 00:04:21 +02:00
fbeefb24a0 fix: vision tasks inherit user_id, admin owns mount root
- detect_objects, classify_content, recluster_faces now look up the
  photo's user_id and set it on created Tag rows — fixes tags being
  invisible to the owning user due to NULL user_id
- Initial admin setup creates source root at the mount root (/photos)
  instead of a subdirectory, since the admin owns the entire library
- Revert to OpenCLIP ViT-B/32 (512-d) as default embedder — SigLIP
  requires transformers version alignment not yet available in the
  Docker image. SigLIP2 code remains for future enablement.
- Add transformers to requirements for future SigLIP support

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-13 00:01:28 +02:00
35d87a2749 fix: dedicate watcher to own worker, fix media auth + memories nav
- Move watch_folders to dedicated 'watcher' queue with its own
  single-concurrency container so it never blocks scan/thumbnail slots
- Add get_current_user_media dependency that accepts ?token= query
  param for <img src> / <video src> media endpoints (thumb, original,
  proxy) — fixes 401 on thumbnails
- Append JWT token to all media URLs in the frontend
- Add missing 'memories' case in sidebar navigation switch

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-12 23:34:40 +02:00
d693569f59 feat: auto-start file watcher with Redis lock for live import
Re-enable the watchfiles-based folder watcher with a Redis lock to
prevent multiple instances from stacking up across restarts. The
watcher is now automatically dispatched on startup when scanner.watch
is true (default), and only one instance runs at a time.

- Redis lock (SETNX + TTL renewal) ensures single-instance execution
- Graceful exit if another watcher holds the lock
- New POST /maintenance/start-watcher endpoint for manual control
- Fix: use settings.scanner/vision properties instead of mulita_config

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-12 22:24:38 +02:00
fc8dd370c2 feat: "On this day" memories — photos from previous years
Add a Memories view that surfaces photos taken on the current date in
previous years (like Google Photos / Immich). Only uses EXIF-sourced
dates to avoid false matches from filesystem timestamps.

- Backend: GET /api/v1/photos/memories returns groups by year, up to
  12 photos each, filtered to non-discarded/non-hidden EXIF dates
- Frontend: MemoriesView with year-grouped thumbnail grid
- Sidebar: new "Memories" nav item with clock icon

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-12 22:18:28 +02:00
bbb8e4850c feat: GPU acceleration support for ONNX Runtime inference
Centralize execution provider selection in providers.py with
auto-detection and graceful fallback. All ONNX sessions (embedder,
detector, face processor, recognizer) now use the configured providers.

- New VISION_EXECUTION_PROVIDERS env var: "auto" for GPU auto-detect,
  or explicit "CUDAExecutionProvider,CPUExecutionProvider"
- Provider priority: CUDA > ROCm > OpenVINO > CPU (when set to "auto")
- docker-compose.yml includes commented-out NVIDIA GPU deploy section
- Supports onnxruntime-gpu as a drop-in replacement for onnxruntime

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-12 22:12:21 +02:00
94c07b1d0d feat: upgrade to SigLIP2 ViT-B/16 for semantic search
Replace OpenCLIP ViT-B/32 (512-d, ~78% recall) with SigLIP2 ViT-B/16
(768-d, ~84% recall) as the default embedding model for significantly
better image-text retrieval quality.

- New SigLIP2Embedder class with 384px input and SigLIP normalization
- ONNX export pipeline for SigLIP2 visual + textual encoders
- Migration 0010: resize embeddings.vector from 512 to 768 dimensions
- Config-driven model selection: "siglip2_vitb16" (default) or
  "openclip_vitb32" (legacy) — both models can coexist
- Content classifier follows the configured embedder family
- Existing embeddings cleared on migration; vision backfill regenerates

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-12 22:09:16 +02:00
c7dd03ade2 feat: CLIP-powered incremental duplicate detection
Replace O(N²) pHash-only duplicate detection with a hybrid approach:
- pHash Hamming distance for exact/near-exact copies
- CLIP embedding cosine similarity via pgvector HNSW for visually
  similar photos (crops, format changes, screenshots)

Post-scan now uses incremental mode: only newly added photos are
compared against the full library — O(new × log N) via HNSW index
instead of O(N²). Full regroup remains available from Settings.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-12 22:03:04 +02:00
8f41a23c41 feat: switch frontend to cursor-based pagination
Replace page-number walking with cursor chaining in usePhotosQuery.
Each response includes a next_cursor that seeks directly to the next
slice via an indexed range scan — O(1) regardless of depth instead of
OFFSET-based skipping that degrades on large libraries.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-12 21:56:57 +02:00
31 changed files with 1093 additions and 280 deletions

View File

@@ -80,7 +80,7 @@ def upgrade() -> None:
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
hashed = pwd_context.hash(generated_password) 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( conn.execute(
sa.text( sa.text(
"INSERT INTO users (id, username, hashed_password, role, media_path) " "INSERT INTO users (id, username, hashed_password, role, media_path) "
@@ -91,7 +91,7 @@ def upgrade() -> None:
"username": "admin", "username": "admin",
"hashed": hashed, "hashed": hashed,
"role": "admin", "role": "admin",
"media_path": "/photos", "media_path": "/photos/admin",
}, },
) )

View 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)
""")

View File

@@ -31,7 +31,8 @@ class PerformanceSettings(BaseModel):
db_pool_recycle: int = 3600 db_pool_recycle: int = 3600
class EmbedderSettings(BaseModel): 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" name: str = "openclip_vitb32"
batch_size: int = 8 batch_size: int = 8
@@ -65,6 +66,11 @@ class VisionSettings(BaseModel):
enabled: bool = True enabled: bool = True
backend: str = "onnx" # "onnx" | "rocm" (future) backend: str = "onnx" # "onnx" | "rocm" (future)
models_dir: str = "/data/models" 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() embedder: EmbedderSettings = EmbedderSettings()
ocr: OCRSettings = OCRSettings() ocr: OCRSettings = OCRSettings()
detector: DetectorSettings = DetectorSettings() detector: DetectorSettings = DetectorSettings()
@@ -182,9 +188,22 @@ class Settings(BaseSettings):
def performance(self) -> PerformanceSettings: def performance(self) -> PerformanceSettings:
return self.config.performance 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 @property
def vision(self) -> VisionSettings: 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: class Config:
env_file = ".env" env_file = ".env"

View File

@@ -1,7 +1,9 @@
""" """
FastAPI dependencies for authentication and user-scoped data access. 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 fastapi.security import OAuth2PasswordBearer
from jose import JWTError from jose import JWTError
from sqlalchemy import select from sqlalchemy import select
@@ -45,6 +47,54 @@ async def get_current_user(
return 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( async def require_admin(
user: User = Depends(get_current_user), user: User = Depends(get_current_user),
) -> User: ) -> User:

View File

@@ -3,6 +3,11 @@ Embedding model — stores CLIP/SigLIP image embeddings via pgvector.
Composite PK (photo_id, model) allows re-embedding with newer models Composite PK (photo_id, model) allows re-embedding with newer models
without clobbering old vectors. 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 sqlalchemy import Column, String, ForeignKey, DateTime, func
from pgvector.sqlalchemy import Vector from pgvector.sqlalchemy import Vector
@@ -14,6 +19,6 @@ class Embedding(Base):
__tablename__ = 'embeddings' __tablename__ = 'embeddings'
photo_id = Column(String, ForeignKey('photos.id', ondelete='CASCADE'), primary_key=True) 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 vector = Column(Vector(512)) # OpenCLIP ViT-B/32 → 512-d
created_at = Column(DateTime(timezone=True), server_default=func.now()) created_at = Column(DateTime(timezone=True), server_default=func.now())

View File

@@ -161,6 +161,8 @@ async def setup(body: SetupRequest, db: AsyncSession = Depends(get_db)):
if len(body.password) < 6: if len(body.password) < 6:
raise HTTPException(status_code=400, detail="Password must be at least 6 characters") 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()) media_path = os.path.join(settings.photo_dirs, body.username.strip())
os.makedirs(media_path, exist_ok=True) 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, media_path=media_path,
) )
db.add(user) 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( source_root = SourceRoot(
name=f"{user.username}'s Library", name=f"{user.username}'s Library",
path=media_path, path=media_path,

View File

@@ -841,3 +841,16 @@ async def trigger_backfill_phashes(current_user: User = Depends(get_current_user
except Exception as e: except Exception as e:
logger.error(f"Backfill queue failed: {e}") logger.error(f"Backfill queue failed: {e}")
return {"status": "error", "message": str(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)}

View File

@@ -26,7 +26,7 @@ from app.models.tags import photo_tags
from app.schemas.photos import PhotoResponse, PhotoUpdate, PhotoListResponse, BulkAction from app.schemas.photos import PhotoResponse, PhotoUpdate, PhotoListResponse, BulkAction
from app.services.exif_writer import ExifWriteError, write_taken_at 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.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 from app.config import settings
router = APIRouter() 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}") @router.get("/{photo_id}")
async def get_photo( async def get_photo(
photo_id: str, photo_id: str,
@@ -438,7 +506,7 @@ async def get_thumbnail(
size: str, size: str,
response: Response, response: Response,
db: AsyncSession = Depends(get_db), 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)""" """Serve thumbnail (with Nginx X-Accel-Redirect support)"""
if size not in ['small', 'medium', 'large']: if size not in ['small', 'medium', 'large']:
@@ -525,7 +593,7 @@ async def get_thumbnail(
async def get_original( async def get_original(
photo_id: str, photo_id: str,
db: AsyncSession = Depends(get_db), 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)""" """Serve original file (download for RAW, inline for web-safe formats)"""
photo = await get_user_photo(photo_id, current_user, db) photo = await get_user_photo(photo_id, current_user, db)
@@ -628,7 +696,7 @@ async def get_proxy(
photo_id: str, photo_id: str,
response: Response, response: Response,
db: AsyncSession = Depends(get_db), 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.""" """Serve a full-resolution WebP proxy for non-web-safe formats."""
photo = await get_user_photo(photo_id, current_user, db) photo = await get_user_photo(photo_id, current_user, db)

View File

@@ -1,73 +1,63 @@
""" """
Duplicate detection: group photos by perceptual-hash similarity. Duplicate detection: group photos by perceptual-hash + CLIP similarity.
Strategy Strategy
-------- --------
Each photo carries a 16-char hex perceptual hash (`Photo.phash`) computed Two complementary signals are fused into a single grouping:
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.
This module turns those per-photo hashes into explicit *groups*. The 1. **Perceptual hash (pHash)** — 16-char hex hash from the thumbnail
result is persisted in two columns: 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 2. **CLIP embedding similarity** — cosine distance over 512-d vectors
* `Photo.is_duplicate` — derived: True iff group_id IS NOT NULL stored in the `embeddings` table with an HNSW index. Catches
(kept as a column so the existing visually similar photos even when pHash diverges (e.g. crops,
PhotoThumbnail badge and /library/stats different formats, screenshots of the same content).
count don't have to change).
The grouping is recomputed in batches by `regroup_duplicates`, NOT on Both signals feed a union-find structure that merges overlapping matches
individual writes — that lets us use a single in-memory pass instead of into connected components.
maintaining a per-row similarity index. Triggered automatically after
each scan and on demand from the Settings panel.
Complexity Incremental mode (default post-scan)
---------- -------------------------------------
Pairwise O(N²) over photos with a non-null phash. At ~5 µs per Hamming `incremental_regroup` only compares *newly added* photos (those whose
distance in CPython this is roughly: `added_at` > watermark) against the entire library. Each new photo does:
1k photos → ~5 s - An HNSW vector similarity query: O(log N) via the index.
5k photos → ~125 s - A pHash comparison against a small candidate set (same group members
10k photos → ~500 s or nearby CLIP results) rather than the full N² sweep.
That's the wrong shape for libraries past ~5k. The drop-in replacement This makes the post-scan cost O(new × log N) instead of O(N²).
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.
Out of scope (deferred) Full regroup
----------------------- ------------
* Dismissing a group / "intentional duplicates" — would need a per-group `regroup_duplicates` still performs the full pairwise pHash pass +
or per-pair flag plus a skip-set in this function so re-grouping CLIP sweep, used for initial setup and manual re-detection.
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.
""" """
from __future__ import annotations from __future__ import annotations
import logging import logging
import uuid import uuid
from datetime import datetime, timezone
from typing import Optional from typing import Optional
from sqlalchemy import select, update from sqlalchemy import select, update, text
from app.database import AsyncSessionLocal from app.database import AsyncSessionLocal
from app.models.photos import Photo from app.models.photos import Photo
from app.models.embeddings import Embedding
from app.config import settings
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Hamming distance threshold under which two phashes are considered # pHash Hamming distance threshold (6 out of 64 bits).
# "the same image". 6 bits out of 64 is the rule-of-thumb sweet spot for DEFAULT_PHASH_THRESHOLD = 6
# pHash — tight enough to avoid false positives between unrelated photos,
# loose enough to catch JPEG re-encodes, slight crops, and a screenshot # CLIP cosine distance threshold. CLIP embeddings are L2-normalized,
# of a screenshot. # so cosine distance = 1 - dot(a, b). A threshold of 0.08 catches
DEFAULT_THRESHOLD = 6 # visually near-identical shots; 0.15 catches similar compositions.
DEFAULT_CLIP_THRESHOLD = 0.10
def _hex_to_int(h: str) -> int: 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: def _hamming(a: int, b: int) -> int:
"""Population count of XOR — the canonical hash distance metric. """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."""
x = a ^ b x = a ^ b
try: try:
return x.bit_count() # type: ignore[attr-defined] return x.bit_count() # type: ignore[attr-defined]
@@ -92,24 +79,26 @@ def _hamming(a: int, b: int) -> int:
class _UnionFind: class _UnionFind:
"""Tiny union-find / disjoint-set used to merge similar phashes into """Tiny union-find / disjoint-set used to merge similar photos into
connected components. Inlined here (rather than pulled from a dep) connected components."""
because it's ~15 lines and we don't need anything fancy."""
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.parent = list(range(n))
self.rank = [0] * n self.rank = [0] * n
def find(self, x: int) -> int: 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: while self.parent[x] != x:
self.parent[x] = self.parent[self.parent[x]] self.parent[x] = self.parent[self.parent[x]]
x = self.parent[x] x = self.parent[x]
return x return x
def union(self, a: int, b: int) -> None: def union_by_key(self, key_a: str, key_b: str) -> None:
ra, rb = self.find(a), self.find(b) 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: if ra == rb:
return return
if self.rank[ra] < self.rank[rb]: if self.rank[ra] < self.rank[rb]:
@@ -118,112 +107,276 @@ class _UnionFind:
if self.rank[ra] == self.rank[rb]: if self.rank[ra] == self.rank[rb]:
self.rank[ra] += 1 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 async def regroup_duplicates(
summary dict the maintenance endpoint surfaces back to the UI. 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 Idempotent — safe to call as often as you like. Returns a summary dict.
`duplicate_group_id=NULL` and `is_duplicate=False`. This is what
cleans up "dead" groups after the user discards N-1 members from
one.
""" """
embedder_model = settings.vision.embedder.name
async with AsyncSessionLocal() as session: async with AsyncSessionLocal() as session:
# Pull (id, phash) for every visible photo with a hash. Discarded # Pull all visible photos with a phash or embedding.
# 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.
rows = ( rows = (
await session.execute( await session.execute(
select(Photo.id, Photo.phash) select(Photo.id, Photo.phash)
.where(Photo.phash.is_not(None))
.where(Photo.is_discarded.is_(False)) .where(Photo.is_discarded.is_(False))
.where(Photo.is_hidden.is_(False)) .where(Photo.is_hidden.is_(False))
) )
).all() ).all()
n = len(rows) if not rows:
if n == 0:
# Still need to clear stale group_ids in case the user just
# discarded the last surviving member of every group.
await _clear_all_groups(session) await _clear_all_groups(session)
await session.commit() await session.commit()
return { return {'photos_considered': 0, 'groups': 0, 'members': 0}
'photos_considered': 0,
'groups': 0,
'members': 0,
}
ids: list[str] = [row[0] for row in rows] ids = [row[0] for row in rows]
hashes: list[int] = [_hex_to_int(row[1]) 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 # ── Phase 1: pHash pairwise (O(N²) on photos with phash) ──
# scaling analysis and the BK-tree upgrade path. 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): for i in range(n):
hi = hashes[i] hi = phash_vals[i]
if hi < 0: if hi < 0:
continue continue
for j in range(i + 1, n): for j in range(i + 1, n):
hj = hashes[j] hj = phash_vals[j]
if hj < 0: if hj < 0:
continue continue
if _hamming(hi, hj) <= threshold: if _hamming(hi, hj) <= phash_threshold:
uf.union(i, j) uf.union_by_key(phash_ids[i], phash_ids[j])
# Collect components. Each connected component of size >= 2 gets # ── Phase 2: CLIP similarity via pgvector ──
# a fresh group id; size-1 components are intentionally dropped. # For each photo with an embedding, find its nearest neighbors
components: dict[int, list[int]] = {} # within the cosine distance threshold using the HNSW index.
for i in range(n): clip_matches = await _clip_neighbor_scan(
root = uf.find(i) session, ids, embedder_model, clip_threshold
components.setdefault(root, []).append(i) )
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 # ── Write results ──
# 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.
await _clear_all_groups(session) await _clear_all_groups(session)
# Second pass: write the new group ids for components of size 2+. groups = uf.components(ids)
groups_created = 0 groups_created = 0
members_total = 0 members_total = 0
for members in components.values(): for member_ids in groups.values():
if len(members) < 2:
continue
group_id = str(uuid.uuid4()) group_id = str(uuid.uuid4())
member_ids = [ids[i] for i in members]
await session.execute( await session.execute(
update(Photo) update(Photo)
.where(Photo.id.in_(member_ids)) .where(Photo.id.in_(member_ids))
.values( .values(duplicate_group_id=group_id, is_duplicate=True)
duplicate_group_id=group_id,
is_duplicate=True,
)
) )
groups_created += 1 groups_created += 1
members_total += len(member_ids) members_total += len(member_ids)
await session.commit() await session.commit()
logger.info( logger.info(
f"regroup_duplicates: considered {n} photos, " f"regroup_duplicates: {len(ids)} photos, "
f"created {groups_created} group(s) covering {members_total} member(s)" f"{groups_created} group(s), {members_total} member(s)"
) )
return { return {
'photos_considered': n, 'photos_considered': len(ids),
'groups': groups_created, 'groups': groups_created,
'members': members_total, '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: async def _clear_all_groups(session) -> None:
"""Reset duplicate_group_id / is_duplicate on every photo. Used as """Reset duplicate_group_id / is_duplicate on every photo."""
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."""
await session.execute( await session.execute(
update(Photo).values(duplicate_group_id=None, is_duplicate=False) update(Photo).values(duplicate_group_id=None, is_duplicate=False)
) )

View File

@@ -78,17 +78,26 @@ async def bootstrap_default_source_root() -> None:
async def start_initial_scan(): 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 The file watcher uses a Redis lock to ensure only one instance runs
dispatched here. It's an infinite loop celery task and every backend across all workers, so it's safe to dispatch on every startup — only
restart was queuing a new instance, eventually pinning every worker the first one will actually watch, the rest exit immediately.
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".
""" """
try: try:
scan_all_source_roots.delay() scan_all_source_roots.delay()
logger.info("Initial scan queued successfully") logger.info("Initial scan queued successfully")
except Exception as e: except Exception as e:
logger.error(f"Failed to start initial scan: {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}")

View File

@@ -77,6 +77,7 @@ def bootstrap(models_dir: str | None = None):
from app.services.vision import export_models from app.services.vision import export_models
export_models.export_openclip(base) export_models.export_openclip(base)
export_models.export_siglip2(base)
export_models.export_yolov8n(base) export_models.export_yolov8n(base)
except Exception as e: except Exception as e:
logger.error( logger.error(

View File

@@ -57,14 +57,25 @@ class CLIPContentClassifier(ContentClassifier):
self._min_confidence = settings.classifier.min_confidence self._min_confidence = settings.classifier.min_confidence
# Load native model for text encoding only # Load native model for text encoding only.
logger.info("Loading OpenCLIP text encoder for content classification") # 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( model, _, _ = open_clip.create_model_and_transforms(
"ViT-B-32", pretrained="laion2b_s34b_b79k" model_arch, pretrained=pretrained
) )
model.eval() model.eval()
self._model = model 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 # Get the ONNX image embedder from the registry
from app.services.vision.registry import registry from app.services.vision.registry import registry

View File

@@ -120,12 +120,10 @@ class YOLOv8Detector(ObjectDetector):
def __init__(self, settings: VisionSettings): def __init__(self, settings: VisionSettings):
model_path = Path(settings.models_dir) / "detect" / "yolov8n.onnx" model_path = Path(settings.models_dir) / "detect" / "yolov8n.onnx"
opts = ort.SessionOptions() from app.services.vision.providers import create_session
opts.inter_op_num_threads = 2
opts.intra_op_num_threads = 2
logger.info("Loading YOLOv8n from %s", model_path) 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._conf_threshold = settings.detector.min_confidence
self._max_detections = settings.detector.max_detections self._max_detections = settings.detector.max_detections

View File

@@ -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/: Expects two ONNX files under {models_dir}/embed/:
- visual.onnx (image encoder) - visual.onnx (image encoder)
- textual.onnx (text 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 import logging
from pathlib import Path from pathlib import Path
@@ -18,50 +22,63 @@ from app.services.vision.base import Embedder
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# OpenCLIP ViT-B/32 preprocessing constants (ImageNet norm) # ── Model-specific constants ──────────────────────────────────────────
_MEAN = np.array([0.48145466, 0.4578275, 0.40821073], dtype=np.float32)
_STD = np.array([0.26862954, 0.26130258, 0.27577711], dtype=np.float32) # OpenCLIP ViT-B/32 (ImageNet norm, 224px)
_INPUT_SIZE = 224 _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.""" """Resize, center-crop, normalize an RGB uint8 image to NCHW float32."""
from PIL import Image from PIL import Image
img = Image.fromarray(image).convert("RGB") img = Image.fromarray(image).convert("RGB")
# Resize shortest edge to _INPUT_SIZE, then center crop
w, h = img.size 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) img = img.resize((int(w * scale), int(h * scale)), Image.BICUBIC)
w, h = img.size w, h = img.size
left = (w - _INPUT_SIZE) // 2 left = (w - input_size) // 2
top = (h - _INPUT_SIZE) // 2 top = (h - input_size) // 2
img = img.crop((left, top, left + _INPUT_SIZE, top + _INPUT_SIZE)) img = img.crop((left, top, left + input_size, top + input_size))
arr = np.array(img, dtype=np.float32) / 255.0 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 arr = arr.transpose(2, 0, 1) # HWC → CHW
return arr[np.newaxis] # NCHW return arr[np.newaxis] # NCHW
class OpenCLIPEmbedder(Embedder): class OpenCLIPEmbedder(Embedder):
"""Legacy OpenCLIP ViT-B/32 embedder (512-d)."""
def __init__(self, settings: VisionSettings): def __init__(self, settings: VisionSettings):
model_dir = Path(settings.models_dir) / "embed" model_dir = Path(settings.models_dir) / "embed"
visual_path = model_dir / "visual.onnx" visual_path = model_dir / "visual.onnx"
textual_path = model_dir / "textual.onnx" textual_path = model_dir / "textual.onnx"
opts = ort.SessionOptions() from app.services.vision.providers import create_session
opts.inter_op_num_threads = 2 from app.config import settings as app_settings
opts.intra_op_num_threads = 2 providers = app_settings.vision.execution_providers
logger.info("Loading visual encoder from %s", visual_path) logger.info("Loading OpenCLIP visual encoder from %s", visual_path)
self._visual = ort.InferenceSession(str(visual_path), opts, providers=["CPUExecutionProvider"]) self._visual = create_session(str(visual_path), configured_providers=providers)
logger.info("Loading textual encoder from %s", textual_path) logger.info("Loading OpenCLIP textual encoder from %s", textual_path)
self._textual = ort.InferenceSession(str(textual_path), opts, providers=["CPUExecutionProvider"]) self._textual = create_session(str(textual_path), configured_providers=providers)
def embed_image(self, image: np.ndarray) -> np.ndarray: 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 input_name = self._visual.get_inputs()[0].name
out = self._visual.run(None, {input_name: inp})[0][0] out = self._visual.run(None, {input_name: inp})[0][0]
out = out / np.linalg.norm(out) out = out / np.linalg.norm(out)
@@ -71,7 +88,6 @@ class OpenCLIPEmbedder(Embedder):
import open_clip import open_clip
tokenizer = open_clip.get_tokenizer("ViT-B-32") tokenizer = open_clip.get_tokenizer("ViT-B-32")
tokens = tokenizer([text]).numpy().astype(np.int64) 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) eot_indices = tokens.argmax(axis=-1).astype(np.int64)
inputs = self._textual.get_inputs() inputs = self._textual.get_inputs()
out = self._textual.run(None, { out = self._textual.run(None, {
@@ -84,3 +100,47 @@ class OpenCLIPEmbedder(Embedder):
@property @property
def dim(self) -> int: def dim(self) -> int:
return 512 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

View File

@@ -120,6 +120,76 @@ def export_openclip(models_dir: Path):
logger.info("Textual encoder exported (%.1f MB)", size_mb) 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): def export_yolov8n(models_dir: Path):
"""Export YOLOv8n to ONNX.""" """Export YOLOv8n to ONNX."""
out_dir = models_dir / "detect" out_dir = models_dir / "detect"
@@ -174,6 +244,7 @@ def main():
logger.info("Exporting models to %s", models_dir) logger.info("Exporting models to %s", models_dir)
export_openclip(models_dir) export_openclip(models_dir)
export_siglip2(models_dir)
export_yolov8n(models_dir) export_yolov8n(models_dir)
logger.info("Done. Run bootstrap_models.py next to download YuNet + SFace.") logger.info("Done. Run bootstrap_models.py next to download YuNet + SFace.")

View File

@@ -71,11 +71,9 @@ class YuNetSFaceProcessor(FaceProcessor):
logger.info("YuNet face detector loaded via OpenCV") logger.info("YuNet face detector loaded via OpenCV")
# SFace via ONNX Runtime # SFace via ONNX Runtime
opts = ort.SessionOptions() from app.services.vision.providers import create_session
opts.inter_op_num_threads = 2
opts.intra_op_num_threads = 2
ort.set_default_logger_severity(3) 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") logger.info("SFace recognizer loaded via ONNX Runtime")
self._min_face_size = settings.faces.min_face_size self._min_face_size = settings.faces.min_face_size

View File

@@ -23,10 +23,13 @@ class InsightFaceProcessor(FaceProcessor):
model_root = str(Path(settings.models_dir) / "face" / "insightface") model_root = str(Path(settings.models_dir) / "face" / "insightface")
logger.info("Loading InsightFace buffalo_l from %s", model_root) 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( self._app = FaceAnalysis(
name="buffalo_l", name="buffalo_l",
root=model_root, root=model_root,
providers=["CPUExecutionProvider"], providers=providers,
) )
self._app.prepare(ctx_id=-1, det_size=(640, 640)) self._app.prepare(ctx_id=-1, det_size=(640, 640))
self._min_det_score = settings.faces.recognition_threshold self._min_det_score = settings.faces.recognition_threshold

View File

@@ -21,8 +21,13 @@ class ONNXBackend:
self._settings = vision_settings self._settings = vision_settings
def create_embedder(self) -> Embedder: def create_embedder(self) -> Embedder:
from app.services.vision.embed import OpenCLIPEmbedder model_name = self._settings.embedder.name
return OpenCLIPEmbedder(self._settings) 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)
def create_ocr(self) -> OCREngine: def create_ocr(self) -> OCREngine:
from app.services.vision.ocr import RapidOCREngine from app.services.vision.ocr import RapidOCREngine

View 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)

View File

@@ -29,6 +29,7 @@ celery_app.conf.update(
'extract_faces': {'queue': 'vision'}, 'extract_faces': {'queue': 'vision'},
'classify_content': {'queue': 'vision'}, 'classify_content': {'queue': 'vision'},
'vision_fanout': {'queue': 'vision'}, 'vision_fanout': {'queue': 'vision'},
'watch_folders': {'queue': 'watcher'},
}, },
task_default_queue='default', task_default_queue='default',
task_default_exchange='default', task_default_exchange='default',

View File

@@ -5,7 +5,7 @@ import os
import hashlib import hashlib
import asyncio import asyncio
from pathlib import Path from pathlib import Path
from datetime import datetime from datetime import datetime, timezone
import logging import logging
import json import json
from typing import List, Dict, Optional 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 countdown is a best-effort hint — on a big library the user can
still hit Settings → Re-detect duplicates to force a fresh pass. 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 from app.tasks.vision import backfill_vision, recluster_faces
async with AsyncSessionLocal() as session: 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 # firing too early just means the next manual run picks up the
# late arrivals — no corrupted state. # late arrivals — no corrupted state.
try: 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: except Exception as e:
logger.warning(f"Could not queue post-scan regroup: {e}") logger.warning(f"Could not queue post-scan regroup: {e}")
@@ -461,74 +468,96 @@ async def _scan_all_source_roots_async():
logger.warning(f"Could not queue post-scan face recluster: {e}") logger.warning(f"Could not queue post-scan face recluster: {e}")
@shared_task(name='watch_folders') WATCHER_LOCK_KEY = "mulita:watch_folders:lock"
def watch_folders(): 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 Watch folders for changes using watchfiles. Long-running task that
monitors filesystem events under every active source root. 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 from watchfiles import watch
# Read source roots from the DB instead of the (now-removed) YAML r = redis_lib.from_url(settings.redis_url)
# config. We need both the path and the id so we can dispatch
# scan_folder with the source_root_id when an event fires. # Acquire exclusive lock — if another watcher is already running,
roots: list[tuple[str, str]] = [] # 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: try:
async def _load_roots(): roots: list[tuple[str, str]] = []
async with AsyncSessionLocal() as session: try:
result = await session.execute( async def _load_roots():
select(SourceRoot).where(SourceRoot.is_active == True) # noqa: E712 async with AsyncSessionLocal() as session:
) result = await session.execute(
return [ select(SourceRoot).where(SourceRoot.is_active == True) # noqa: E712
(os.path.normpath(sr.path), sr.id)
for sr in result.scalars().all()
if os.path.exists(sr.path)
]
roots = asyncio.run(_load_roots())
except Exception as e:
logger.error(f"watch_folders could not load source roots: {e}")
return
if not roots:
logger.warning("No valid source roots to watch")
return
paths = [p for p, _ in roots]
logger.info(f"Starting folder watcher for: {paths}")
def find_source_root_for(path: str) -> Optional[str]:
"""Return the source_root id whose path contains `path`, or None."""
normalized = os.path.normpath(path)
for root_path, root_id in roots:
if normalized == root_path or normalized.startswith(root_path + os.sep):
return root_id
return None
for changes in watch(*paths):
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"
) )
return [
(os.path.normpath(sr.path), sr.id)
for sr in result.scalars().all()
if os.path.exists(sr.path)
]
roots = asyncio.run(_load_roots())
except Exception as e:
logger.error(f"watch_folders could not load source roots: {e}")
return
if not roots:
logger.warning("No valid source roots to watch")
return
paths = [p for p, _ in roots]
logger.info(f"Starting folder watcher for: {paths}")
def find_source_root_for(path: str) -> Optional[str]:
"""Return the source_root id whose path contains `path`, or None."""
normalized = os.path.normpath(path)
for root_path, root_id in roots:
if normalized == root_path or normalized.startswith(root_path + os.sep):
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)
if Path(filepath).suffix.lower() not in SUPPORTED_EXTENSIONS:
continue continue
scan_folder.delay(parent_dir, source_root_id)
logger.info(f"File {change_type}: {filepath}, queued scan for {parent_dir}") if change_type == 'added' or change_type == 'modified':
elif change_type == 'deleted': parent_dir = str(Path(filepath).parent)
# Handle file deletion source_root_id = find_source_root_for(parent_dir)
asyncio.run(handle_file_deletion(filepath)) if source_root_id is None:
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':
asyncio.run(handle_file_deletion(filepath))
finally:
try:
lock.release()
except Exception:
pass
async def handle_file_deletion(filepath: str): async def handle_file_deletion(filepath: str):
"""Handle deletion of a file from the filesystem""" """Handle deletion of a file from the filesystem"""

View File

@@ -467,10 +467,23 @@ async def _backfill_phashes_async():
@shared_task(name='regroup_duplicates') @shared_task(name='regroup_duplicates')
def regroup_duplicates_task(): 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 Used by the Settings → Re-detect duplicates button."""
at worker boot (the service uses AsyncSessionLocal which is also
imported here at module top)."""
from app.services.duplicates import regroup_duplicates from app.services.duplicates import regroup_duplicates
return asyncio.run(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))

View File

@@ -163,6 +163,12 @@ def detect_objects(photo_id: str):
session = _get_sync_session() session = _get_sync_session()
try: 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 # Wipe previous detection results for this photo from this model
session.execute( session.execute(
delete(photo_tags).where( delete(photo_tags).where(
@@ -178,13 +184,13 @@ def detect_objects(photo_id: str):
best_per_label[det.label] = (det.confidence, det.bbox) best_per_label[det.label] = (det.confidence, det.bbox)
for label, (confidence, bbox) in best_per_label.items(): 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( 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() ).scalar_one_or_none()
if not tag: 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.add(tag)
session.flush() # get tag.id session.flush() # get tag.id
@@ -234,6 +240,12 @@ def classify_content(photo_id: str):
session = _get_sync_session() session = _get_sync_session()
try: 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 # Wipe previous classification for this photo
session.execute( session.execute(
delete(photo_tags).where( 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( 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() ).scalar_one_or_none()
if not tag: 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.add(tag)
session.flush() session.flush()
@@ -434,11 +446,16 @@ def recluster_faces():
if label not in cluster_tag_map: if label not in cluster_tag_map:
cluster_name = f"Person {label + 1}" 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( tag = Tag(
name=cluster_name, name=cluster_name,
kind='face_cluster', kind='face_cluster',
source=source_name, source=source_name,
representative_photo_id=face_rows[i].photo_id, representative_photo_id=face_rows[i].photo_id,
user_id=rep_photo,
) )
session.add(tag) session.add(tag)
session.flush() session.flush()

View File

@@ -37,6 +37,7 @@ watchfiles==0.21.0
# Vision pipeline (ONNX Runtime CPU inference) # Vision pipeline (ONNX Runtime CPU inference)
onnxruntime==1.18.1 onnxruntime==1.18.1
open-clip-torch==2.24.0 # tokenizer + export helper; inference via ONNX 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 ultralytics==8.4.37 # YOLOv8n export helper; inference via ONNX
rapidocr-onnxruntime==1.3.22 rapidocr-onnxruntime==1.3.22
scikit-learn==1.4.0 # DBSCAN for face clustering scikit-learn==1.4.0 # DBSCAN for face clustering

View File

@@ -120,6 +120,37 @@ services:
- mulita-network - mulita-network
restart: unless-stopped 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: worker-vision:
build: build:
context: ./backend context: ./backend
@@ -143,6 +174,13 @@ services:
- LOG_LEVEL=${LOG_LEVEL:-INFO} - LOG_LEVEL=${LOG_LEVEL:-INFO}
- TZ=${TZ:-UTC} - TZ=${TZ:-UTC}
- MULITA_CELERY_WORKER=1 - 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 # Pin each ONNX session to one intra-op thread so N prefork children
# × default-all-cores doesn't oversubscribe the box. With # × default-all-cores doesn't oversubscribe the box. With
# concurrency=5 and OMP=1, vision peaks at 5 busy cores, leaving # concurrency=5 and OMP=1, vision peaks at 5 busy cores, leaving
@@ -151,6 +189,14 @@ services:
- OMP_NUM_THREADS=1 - OMP_NUM_THREADS=1
- OPENBLAS_NUM_THREADS=1 - OPENBLAS_NUM_THREADS=1
- MKL_NUM_THREADS=1 - MKL_NUM_THREADS=1
# Uncomment for NVIDIA GPU passthrough:
# deploy:
# resources:
# reservations:
# devices:
# - driver: nvidia
# count: all
# capabilities: [gpu]
depends_on: depends_on:
redis: redis:
condition: service_started condition: service_started

View File

@@ -2,6 +2,7 @@ import { useState } from 'react'
import { Timeline } from './components/timeline/Timeline' import { Timeline } from './components/timeline/Timeline'
import { DuplicatesView } from './components/duplicates/DuplicatesView' import { DuplicatesView } from './components/duplicates/DuplicatesView'
import { MapView } from './components/map/MapView' import { MapView } from './components/map/MapView'
import { MemoriesView } from './components/memories/MemoriesView'
import { PeopleView } from './components/people/PeopleView' import { PeopleView } from './components/people/PeopleView'
import { TagsView } from './components/tags/TagsView' import { TagsView } from './components/tags/TagsView'
import { ColorsView } from './components/colors/ColorsView' import { ColorsView } from './components/colors/ColorsView'
@@ -87,6 +88,8 @@ function MainApp() {
<SettingsPage /> <SettingsPage />
) : currentSection === 'map' ? ( ) : currentSection === 'map' ? (
<MapView /> <MapView />
) : currentSection === 'memories' ? (
<MemoriesView />
) : currentSection === 'duplicates' ? ( ) : currentSection === 'duplicates' ? (
<DuplicatesView /> <DuplicatesView />
) : currentSection === 'people' ? ( ) : currentSection === 'people' ? (

View File

@@ -23,6 +23,7 @@ import {
User as UserIcon, User as UserIcon,
LogOut, LogOut,
Shield, Shield,
Clock,
} from 'lucide-react' } from 'lucide-react'
import clsx from 'clsx' import clsx from 'clsx'
import { sourceFolders, photos as photosApi, type FolderTreeNode } from '../../services/api' import { sourceFolders, photos as photosApi, type FolderTreeNode } from '../../services/api'
@@ -275,6 +276,9 @@ export function LeftSidebar({ onCollapse }: LeftSidebarProps) {
case 'map': case 'map':
navigateToSection('map', {}) navigateToSection('map', {})
break break
case 'memories':
navigateToSection('memories', {})
break
default: default:
if (id.startsWith('folder-')) { if (id.startsWith('folder-')) {
const folderId = id.slice('folder-'.length) 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: '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: '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: '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: '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 }, { id: 'discarded', label: 'Discarded', icon: <Trash2 className="h-4 w-4" />, count: stats?.discarded ?? 0 },
], ],

View 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 &mdash; {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} &middot; {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>
)
}

View File

@@ -100,7 +100,8 @@ export function PhotoThumbnail({
// Cache-bust on retry so the browser actually re-requests instead of // Cache-bust on retry so the browser actually re-requests instead of
// serving the cached 404. // serving the cached 404.
const baseUrl = photosApi.getThumbnailUrl(photo.id, 'medium') 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 // "Capture date probably wrong" — read straight from the stored
// `has_date_warning` flag rather than recomputing the heuristic // `has_date_warning` flag rather than recomputing the heuristic

View File

@@ -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 * Single source of truth for the timeline photos query. Both Timeline and
* PreviewView call this so they share one cache entry — previously * PreviewView call this so they share one cache entry — previously
@@ -70,55 +83,41 @@ export function usePhotosQuery() {
return useQuery({ return useQuery({
queryKey: ['photos', filterParams], queryKey: ['photos', filterParams],
queryFn: async ({ signal }) => { queryFn: async ({ signal }) => {
// Two-phase fetch so the timeline can paint its first thumbnails // Two-phase fetch using cursor-based (keyset) pagination.
// long before the entire library has finished downloading. Phase 1 // Phase 1 returns the first page (resolves the useQuery promise
// returns the first page synchronously (which resolves the // so consumers exit loading state). Phase 2 chains cursors in
// useQuery promise so consumers exit their loading state). Phase 2 // the background — each response includes a `next_cursor` that
// walks the remaining pages in the background, appending each one // seeks directly to the next slice via an indexed range scan,
// into the cache via setQueryData so the grid grows as data // O(1) regardless of depth (no OFFSET skipping).
// arrives. The signal from React Query aborts the background
// loop if the query is invalidated or unmounts mid-stream.
const PER_PAGE = 500 const PER_PAGE = 500
const MAX_PAGES = 200 const MAX_PAGES = 200
const firstResp = await api.get<{ const first = await fetchCursorPage(
photos: Photo[] { per_page: PER_PAGE, ...filterParams },
total: number
pages: number
}>('/photos', {
params: { page: 1, per_page: PER_PAGE, ...filterParams },
signal, signal,
}) )
const firstBatch = firstResp.data.photos || [] const firstBatch = first.photos || []
const totalPages = firstResp.data.pages ?? 1 let nextCursor: string | null = first.next_cursor
if (totalPages > 1 && firstBatch.length === PER_PAGE) { if (nextCursor) {
// Fire-and-forget background loop. We don't await here — the // Fire-and-forget background loop using cursor chaining.
// first batch is already enough to render. Each subsequent
// page lands via setQueryData, which triggers consumers to
// re-render with the larger list.
void (async () => { 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 if (signal?.aborted) return
try { try {
const resp = await api.get<{ const page = await fetchCursorPage(
photos: Photo[] { per_page: PER_PAGE, cursor: nextCursor, ...filterParams },
total: number
pages: number
}>('/photos', {
params: { page, per_page: PER_PAGE, ...filterParams },
signal, signal,
}) )
if (signal?.aborted) return if (signal?.aborted) return
const more = resp.data.photos || [] const more = page.photos || []
nextCursor = page.next_cursor
queryClient.setQueryData<Photo[]>( queryClient.setQueryData<Photo[]>(
['photos', filterParams], ['photos', filterParams],
(prev) => (prev ? [...prev, ...more] : more) (prev) => (prev ? [...prev, ...more] : more)
) )
if (more.length < PER_PAGE) return if (!nextCursor || more.length < PER_PAGE) return
} catch { } catch {
// Network or abort — give up the background stream. The
// next user-triggered refetch will start fresh.
return return
} }
} }

View File

@@ -316,17 +316,29 @@ export const photos = {
}, },
getThumbnailUrl: (photoId: string, size: 'small' | 'medium' | 'large' = 'medium') => { 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) => { 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 /** Full-resolution display URL. Backend serves the original for web-safe
* formats and a transcoded WebP for RAW/HEIC/TIFF. */ * formats and a transcoded WebP for RAW/HEIC/TIFF. */
getProxyUrl: (photoId: string) => { 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 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 { export interface LibraryStats {
all_photos: number all_photos: number
rated: number rated: number