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")
|
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",
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
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
|
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"
|
||||||
|
|||||||
@@ -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:
|
||||||
|
|||||||
@@ -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())
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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)}
|
||||||
@@ -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)
|
||||||
|
|||||||
@@ -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)
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -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}")
|
||||||
|
|||||||
@@ -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(
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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.")
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
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'},
|
'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',
|
||||||
|
|||||||
@@ -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"""
|
||||||
|
|||||||
@@ -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))
|
||||||
@@ -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()
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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' ? (
|
||||||
|
|||||||
@@ -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 },
|
||||||
],
|
],
|
||||||
|
|||||||
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
|
// 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
|
||||||
|
|||||||
@@ -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
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user