Implement quality score tracking and batch Frigate file mapping

- Track laplacian blur score through quality filtering pipeline
  (quality.py: blur_score on QualityResult; diversity.py: store on asset;
   executor.py: read via quality_score key)
- Replace per-file polling with post-person batch reconciliation:
  after all uploads for a person complete, poll Frigate (up to 15s)
  until the expected number of new files appear, then map by filename
  timestamp order (Frigate FIFO queue = upload order = timestamp order)
- Document race condition limitation: concurrent external uploads cause
  the batch to be skipped entirely (safe but files go unmapped); noted
  in code as requiring a Frigate API fix (return filename on upload)
- Add two assess_quality integration tests for blur_score

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

tests/test_quality.py

@@ -134,6 +134,16 @@ def test_assess_quality_passes_good_image():
     img = _noisy_color_image()
     result = assess_quality(img, face_bbox=(10, 10, 110, 110), confidence=0.9)
     assert result.passed
+    assert result.blur_score is not None
+    assert result.blur_score > 0
+
+
+def test_assess_quality_blur_score_is_low_for_flat_image():
+    from winnow.quality import assess_quality
+    flat = _rgb_image(128, 128, 128)
+    result = assess_quality(flat)
+    assert result.blur_score is not None
+    assert result.blur_score < 1.0
 
 
 def test_assess_quality_collects_multiple_failures():

winnow/diversity.py

@@ -258,6 +258,7 @@ def _select_by_embedding(
                     logger.debug(f"Quality filtered {asset['id']}: {quality.reason}")
                     continue
 
+                asset["quality_score"] = quality.blur_score
                 face_crop = _crop_face_from_thumbnail(img, asset, person_id=person_id)
                 embed_img = face_crop if face_crop is not None else img
             else:

winnow/executor.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import shutil
+import time
 from io import BytesIO
 from urllib.parse import quote
 
@@ -27,6 +28,68 @@ from .upload_tracker import (
 logger = logging.getLogger(__name__)
 
 
+def _reconcile_frigate_mappings(
+    person_name: str,
+    known_files_before: set[str],
+    uploaded: list[tuple[str, str | None]],
+) -> None:
+    """Map Frigate filenames to asset IDs after a batch of uploads.
+
+    Polls until all expected new files appear in the Frigate API, then maps
+    them to asset IDs by filename timestamp order (Frigate processes the
+    upload queue in FIFO order, so earlier uploads get earlier timestamps).
+
+    KNOWN LIMITATION — race condition with external uploads:
+    If another client uploads a face file for this person concurrently, the
+    count of new files will exceed `len(uploaded)` and we bail out entirely
+    (the "> target" branch). That's safe — we never record a wrong mapping —
+    but those uploads become permanently unmapped (they won't be eligible for
+    quality replacement). The right fix is a Frigate API that returns the
+    filename in the upload response, removing the need for any post-upload
+    diffing. Until then, the external-upload guard keeps mappings correct at
+    the cost of occasionally missing them when another client is active.
+    """
+    target = len(uploaded)
+    current_files: set[str] = set()
+
+    for delay in (1, 2, 4, 8):
+        time.sleep(delay)
+        fresh = get_frigate_person_files(person_name)
+        if fresh is None:
+            logger.warning(
+                f"{person_name}: Frigate API unreachable during mapping reconciliation"
+                " — quality replacement won't target these files"
+            )
+            return
+        current_files = set(fresh)
+        if len(current_files - known_files_before) >= target:
+            break
+
+    new_files = current_files - known_files_before
+
+    if len(new_files) == target:
+        def _ts(fname: str) -> float:
+            try:
+                return float(fname.rsplit("_", 1)[-1].replace(".webp", ""))
+            except (ValueError, IndexError):
+                return 0.0
+
+        for (fname, asset_id), frigate_file in zip(uploaded, sorted(new_files, key=_ts)):
+            if asset_id:
+                record_frigate_file(person_name, frigate_file, asset_id)
+        logger.debug(f"{person_name}: batch-mapped {target} Frigate file(s)")
+    elif len(new_files) > target:
+        logger.info(
+            f"{person_name}: {len(new_files)} new Frigate files for {target} uploads"
+            " (external upload detected) — skipping file mapping"
+        )
+    else:
+        logger.warning(
+            f"{person_name}: only {len(new_files)} of {target} expected Frigate files"
+            " appeared after reconciliation — mapping skipped"
+        )
+
+
 def _enrich_asset_with_face_data(asset: dict, person: dict) -> dict:
     """Enrich an asset dict with face bounding box data from the Immich faces API.
 
@@ -141,7 +204,7 @@ def execute_jobs(jobs: list[dict]) -> None:
                             # Record which asset produced which output file
                             filename = f"{count}.jpg"
                             asset_map[filename] = asset["id"]
-                            score_map[filename] = asset.get("face_confidence")
+                            score_map[filename] = asset.get("quality_score") or asset.get("face_confidence")
                             # Also record object-mode variant filenames
                             if mode == "object":
                                 for f in sorted(os.listdir(person_dir)):
@@ -253,17 +316,16 @@ def upload_to_frigate(jobs: list[dict]) -> None:
             person_uploaded = 0
             person_failed = 0
 
-            # Snapshot current Frigate filenames so we can identify which file
-            # each upload produces (Frigate assigns its own filename on ingest).
             known_frigate_files: set[str] = set(get_frigate_person_files(name) or [])
+            known_frigate_files_at_start = set(known_frigate_files)
+            effective_count = len(known_frigate_files)
             quality_replacement = job.get("config", {}).get("quality_replacement", False)
+            actually_uploaded: list[tuple[str, str | None]] = []
 
             for fname in person_files:
                 fpath = os.path.join(person_dir, fname)
 
-                # Quality replacement gate: when at cap, only upload if this image
+                at_cap = effective_count >= Config.MAX_AUTO_IMAGES
-                # scores higher than the worst mapped file already in Frigate.
-                at_cap = len(known_frigate_files) >= Config.MAX_AUTO_IMAGES
                 if at_cap:
                     if not quality_replacement:
                         progress.console.print(f"    [dim]⏭  {fname}: at cap, quality replacement disabled[/dim]")
@@ -291,6 +353,7 @@ def upload_to_frigate(jobs: list[dict]) -> None:
                     if delete_frigate_person_files(name, [worst_frigate_file]):
                         remove_frigate_file(name, worst_frigate_file)
                         known_frigate_files.discard(worst_frigate_file)
+                        effective_count -= 1
                     else:
                         logger.warning(f"Failed to delete {worst_frigate_file} for {name}, skipping replacement")
                         # Remove from tracker so the next candidate targets a different file.
@@ -310,31 +373,12 @@ def upload_to_frigate(jobs: list[dict]) -> None:
                         if resp.status_code == 200:
                             uploaded += 1
                             person_uploaded += 1
+                            effective_count += 1
 
-                            # Mark this asset as uploaded so it's skipped on future runs
                             asset_id = asset_map.get(fname)
                             if asset_id:
                                 mark_uploaded(asset_id, person_name=name, score=score_map.get(fname))
-
+                                actually_uploaded.append((fname, asset_id))
-                            # Identify the Frigate filename assigned to this upload
-                            # and record the mapping for future quality management.
-                            fresh = get_frigate_person_files(name)
-                            if fresh is None:
-                                logger.warning(
-                                    f"{name}: Frigate API unreachable after uploading {fname}"
-                                    f" — file mapping skipped, quality replacement won't target this file"
-                                )
-                            else:
-                                current_files = set(fresh)
-                                new_files = current_files - known_frigate_files
-                                if len(new_files) == 1 and asset_id:
-                                    record_frigate_file(name, next(iter(new_files)), asset_id)
-                                elif len(new_files) > 1:
-                                    logger.info(
-                                        f"{name}: {len(new_files)} new Frigate files after uploading {fname}"
-                                        f" (concurrent upload detected) — skipping file mapping"
-                                    )
-                                known_frigate_files = current_files
 
                             break
                         else:
@@ -391,6 +435,10 @@ def upload_to_frigate(jobs: list[dict]) -> None:
 
                 progress.advance(upload_task)
 
+            # Batch-map Frigate filenames to asset IDs now that all uploads are done
+            if actually_uploaded:
+                _reconcile_frigate_mappings(name, known_frigate_files_at_start, actually_uploaded)
+
             # Per-person summary
             if person_failed == 0:
                 progress.console.print(

winnow/quality.py

@@ -20,6 +20,7 @@ class QualityResult:
 
     passed: bool
     reasons: list[str] = field(default_factory=list)
+    blur_score: float | None = None
 
     @property
     def reason(self) -> str:
@@ -113,9 +114,12 @@ def assess_quality(
     img_np = np.asarray(img)
     reasons = []
 
-    # Run all checks, collect failures
+    # Compute laplacian variance once (used by check_blur and stored as blur_score)
+    gray = cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY) if img_np.ndim == 3 else img_np
+    blur_score = float(cv2.Laplacian(gray, cv2.CV_64F).var())
+
     checks = [
-        check_blur(img_np, blur_threshold),
+        (blur_score >= blur_threshold, f"Blurry (laplacian={blur_score:.1f}, threshold={blur_threshold})" if blur_score < blur_threshold else ""),
         check_grayscale(img_np),
         check_exposure(img_np),
         check_confidence(confidence, min_confidence),
@@ -129,5 +133,5 @@ def assess_quality(
         if not passed:
             reasons.append(reason)
 
-    return QualityResult(passed=len(reasons) == 0, reasons=reasons)
+    return QualityResult(passed=len(reasons) == 0, reasons=reasons, blur_score=blur_score)