From 11e7d4aad7ecdc8fac35972ffa2c8c18145575b2 Mon Sep 17 00:00:00 2001 From: Holden Date: Sat, 13 Jun 2026 04:59:12 +0000 Subject: [PATCH] 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 --- tests/test_quality.py | 10 +++++ winnow/diversity.py | 1 + winnow/executor.py | 102 +++++++++++++++++++++++++++++++----------- winnow/quality.py | 10 +++-- 4 files changed, 93 insertions(+), 30 deletions(-) diff --git a/tests/test_quality.py b/tests/test_quality.py index 13f3e6d..323d916 100644 --- a/tests/test_quality.py +++ b/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(): diff --git a/winnow/diversity.py b/winnow/diversity.py index 0e47144..711846c 100644 --- a/winnow/diversity.py +++ b/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: diff --git a/winnow/executor.py b/winnow/executor.py index 9313403..36250dd 100644 --- a/winnow/executor.py +++ b/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 - # scores higher than the worst mapped file already in Frigate. - at_cap = len(known_frigate_files) >= Config.MAX_AUTO_IMAGES + at_cap = effective_count >= 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)) - - # 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 + actually_uploaded.append((fname, asset_id)) 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( diff --git a/winnow/quality.py b/winnow/quality.py index 6d4ad3e..4ccc2d4 100644 --- a/winnow/quality.py +++ b/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)