Merge feature/insightface-crop-alignment into dev

compose.yml

@@ -26,6 +26,7 @@ services:
       # - SKIP_PEOPLE=Unknown         # Comma-separated; skip these people
       # - MIN_FACE_COUNT=5            # Skip people with fewer than N assets in Immich
       # - YEARS_FILTER=10             # Only include images from the last N years (default: 10)
+      # - MERGE_DUPLICATE_PEOPLE=true # Auto-merge Immich people with the same name (keeps most assets)
 
       # ── Image Quality ─────────────────────────────────────────────────────
       # - MIN_FACE_WIDTH=50           # Minimum face width in pixels (default: 50)

winnow/cli.py

@@ -9,7 +9,7 @@ from rich.prompt import Confirm
 
 from .config import Config, ConfigManager
 from .executor import execute_jobs, upload_to_frigate
-from .immich_api import get_people
+from .immich_api import get_people, merge_people
 from .jobs import _show_preview, auto_configure, interactive_configure
 from .log_config import console, setup_logging
 from .upload_tracker import find_by_crop_dimension, get_person_summary, reset_person
@@ -51,6 +51,85 @@ def _handle_trace_crop(size_str: str) -> None:
     sys.exit(0)
 
 
+def _handle_duplicate_people(people: list[dict]) -> list[dict]:
+    """Warn about or merge Immich people that share the same name.
+
+    Duplicates arise when Immich creates separate person records for the same
+    individual (e.g. unmerged face clusters). Without handling, winnow would
+    run multiple jobs for the same Frigate folder and overwrite its own output,
+    leaving far fewer training images than expected.
+
+    With MERGE_DUPLICATE_PEOPLE=false (default): prints a warning, skips the
+    smaller duplicates so only the person with the most assets is processed,
+    and returns a deduplicated people list.
+    With MERGE_DUPLICATE_PEOPLE=true: merges each duplicate group inside
+    Immich via its API (permanently combines the face records), then
+    re-fetches the people list so the rest of the run sees the merged state.
+    """
+    from collections import defaultdict
+
+    by_name: dict[str, list[dict]] = defaultdict(list)
+    for p in people:
+        name = (p.get("name") or "").strip()
+        if name:
+            by_name[name].append(p)
+
+    duplicates = {name: ps for name, ps in by_name.items() if len(ps) > 1}
+    if not duplicates:
+        return people
+
+    if not Config.MERGE_DUPLICATE_PEOPLE:
+        rprint("\n[bold yellow]⚠  Duplicate person names detected in Immich:[/bold yellow]")
+        for name, ps in sorted(duplicates.items()):
+            ordered = sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)
+            entries = ", ".join(
+                f"[dim]{p['id'][:8]}…[/dim] ({p.get('assetCount', 0)} assets)"
+                for p in ordered
+            )
+            rprint(f"  [yellow]{name}[/yellow] → {len(ps)} people: {entries}")
+            skipped = ordered[1:]
+            rprint(
+                f"  [dim]  Processing largest only "
+                f"({ordered[0].get('assetCount', 0)} assets). "
+                f"Skipping {len(skipped)} smaller duplicate(s) to avoid overwriting output.[/dim]"
+            )
+        rprint(
+            "  [dim]Set MERGE_DUPLICATE_PEOPLE=true to permanently merge duplicates "
+            "inside Immich (keeps the person with the most assets).[/dim]\n"
+        )
+        # Return deduplicated list — keep only the largest per name so that
+        # downstream job creation never runs two jobs for the same Frigate folder.
+        skip_ids = {
+            p["id"]
+            for ps in duplicates.values()
+            for p in sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)[1:]
+        }
+        return [p for p in people if p["id"] not in skip_ids]
+
+    # Auto-merge: survivor = largest asset count, rest merge into it inside Immich
+    merged_any = False
+    for name, ps in sorted(duplicates.items()):
+        ordered = sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)
+        survivor = ordered[0]
+        merge_ids = [p["id"] for p in ordered[1:]]
+        rprint(
+            f"  [cyan]Merging {name!r} inside Immich:[/cyan] keeping "
+            f"[dim]{survivor['id'][:8]}…[/dim] ({survivor.get('assetCount', 0)} assets), "
+            f"absorbing {len(merge_ids)} smaller duplicate(s)..."
+        )
+        if merge_people(survivor["id"], merge_ids):
+            rprint(f"  [green]✓ Merged {name!r}[/green]")
+            merged_any = True
+        else:
+            rprint(f"  [red]✗ Failed to merge {name!r}[/red]")
+
+    if merged_any:
+        rprint("  [dim]Re-fetching people after merge...[/dim]")
+        return get_people()
+
+    return people
+
+
 def main() -> None:
     """Entry point for winnow CLI."""
     try:
@@ -103,6 +182,8 @@ def main() -> None:
             rprint("[bold red]Could not fetch people from Immich. Check URL/Key.[/bold red]")
             return
 
+        people = _handle_duplicate_people(people)
+
         # Auto mode when no TTY (Docker, cron, pipes) — the primary use case.
         # A TTY means local interactive use; AUTO_MODE=true overrides that for scripting.
         auto_mode = not sys.stdin.isatty() or os.environ.get("AUTO_MODE", "").lower() in ("true", "1", "yes")

winnow/config.py

@@ -36,6 +36,7 @@ class _Config:
 
     # People filtering
     MIN_FACE_COUNT: int = 0
+    MERGE_DUPLICATE_PEOPLE: bool = False
 
     # Output quality
     FACE_MARGIN: float = 0.15
@@ -60,6 +61,7 @@ class _Config:
         self.YEARS_FILTER = int(os.getenv("YEARS_FILTER", "10"))
         self.MIN_FACE_WIDTH = int(os.getenv("MIN_FACE_WIDTH", "90"))
         self.MIN_FACE_COUNT = int(os.getenv("MIN_FACE_COUNT", "0"))
+        self.MERGE_DUPLICATE_PEOPLE = os.getenv("MERGE_DUPLICATE_PEOPLE", "false").lower() in ("true", "1", "yes")
         self.BLUR_THRESHOLD = float(os.getenv("BLUR_THRESHOLD", "120.0"))
         self.MIN_CONFIDENCE = float(os.getenv("MIN_CONFIDENCE", "0.7"))
         self.MAX_AUTO_IMAGES = int(os.getenv("MAX_AUTO_IMAGES", "80"))

winnow/executor.py

@@ -155,6 +155,18 @@ def execute_jobs(jobs: list[dict]) -> None:
 
     use_full_res = Config.USE_FULL_RESOLUTION
 
+    # Load InsightFace app for landmark-based crop alignment (face mode only).
+    # The model is already resident from the diversity/embedding phase, so this
+    # is just a singleton lookup — no load cost.
+    insightface_app = None
+    if any(j["config"].get("mode", "face") == "face" for j in jobs) and Config.ENABLE_FACE_ALIGNMENT:
+        try:
+            from .embeddings import get_insightface_app
+
+            insightface_app = get_insightface_app()
+        except Exception as e:
+            logger.debug(f"InsightFace unavailable for crop alignment: {e}")
+
     with Progress(
         SpinnerColumn(),
         TextColumn("[progress.description]{task.description}"),
@@ -216,7 +228,7 @@ def execute_jobs(jobs: list[dict]) -> None:
                         progress.console.print(f"[red]Failed download {asset['id']}[/red]")
                     else:
                         saved = (
-                            process_face_mode(img, asset, person, person_dir, count)
+                            process_face_mode(img, asset, person, person_dir, count, insightface_app=insightface_app)
                             if mode == "face"
                             else process_object_mode(img, config, person_dir, count)
                             if mode == "object"
@@ -364,6 +376,8 @@ def upload_to_frigate(jobs: list[dict]) -> None:
             # Snapshot live Frigate files for post-upload reconciliation diff only.
             # effective_count is sourced from the tracker (mapped files) so that
             # manually-added Frigate files don't consume winnow's managed quota.
+            # Replacement targets also come exclusively from the tracker, so manually
+            # added files are never selected for deletion — only winnow-uploaded ones.
             _snapshot = (
                 all_frigate_files.get(name, []) if all_frigate_files is not None
                 else get_frigate_person_files(name)

winnow/image_processing.py

@@ -69,13 +69,15 @@ def process_face_mode(
     output_dir: str,
     count: int,
     min_width: int | None = None,
+    insightface_app=None,
 ) -> tuple[int, int] | None:
     """Crop face based on Immich metadata and save to output directory.
 
     Returns (width, height) of the saved crop, or None if no crop was saved.
-    If face alignment is enabled and landmarks are available, produces
-    an aligned 112x112 crop. Otherwise falls back to bounding box crop
-    with configurable margin.
+    When insightface_app is provided and ENABLE_FACE_ALIGNMENT is True,
+    re-detects the face in the Immich bbox region using InsightFace to get
+    precise landmarks for a proper 112x112 aligned crop. Falls back to
+    bounding box crop with configurable margin if alignment is unavailable.
     """
     min_width = min_width or Config.MIN_FACE_WIDTH
 
@@ -109,18 +111,51 @@ def process_face_mode(
         logger.debug(f"Face too small ({face_w:.1f}x{face_h:.1f})")
         return None
 
-    # Try face alignment if enabled and landmarks available
+    # Re-detect face with InsightFace for landmark-based alignment.
+    # Immich's /api/faces endpoint does not include landmarks, so the
+    # align_face fallback below never fires without this step.
+    if insightface_app is not None and Config.ENABLE_FACE_ALIGNMENT:
+        try:
+            # Expand the Immich bbox by 50% to give InsightFace enough context
+            # for detection and alignment, then search for the face nearest the
+            # centre of that region (handles group photos at the boundary).
+            pad_x, pad_y = face_w * 0.5, face_h * 0.5
+            search_box = (
+                max(0, x1 - pad_x),
+                max(0, y1 - pad_y),
+                min(img_w, x2 + pad_x),
+                min(img_h, y2 + pad_y),
+            )
+            search_crop = img.crop(search_box)
+            detected = insightface_app.get(np.asarray(search_crop))
+            if detected:
+                cx, cy = search_crop.width / 2, search_crop.height / 2
+                best = min(
+                    detected,
+                    key=lambda f: abs((f.bbox[0] + f.bbox[2]) / 2 - cx)
+                    + abs((f.bbox[1] + f.bbox[3]) / 2 - cy),
+                )
+                kps = getattr(best, "kps", None)
+                if kps is not None and np.asarray(kps).shape == (5, 2):
+                    aligned = align_face(search_crop, kps)
+                    if aligned is not None:
+                        _save_jpeg(aligned, os.path.join(output_dir, f"{count}.jpg"))
+                        return aligned.size
+        except Exception as e:
+            logger.debug(f"InsightFace re-detection failed for {asset.get('id')}: {e}")
+
+    # Landmark alignment from Immich metadata (Immich does not currently
+    # expose landmarks, so this path is a future-proofing fallback)
     if Config.ENABLE_FACE_ALIGNMENT:
         landmarks = face_info.get("landmarks") or face_info.get("landmark")
         if landmarks:
-            # Scale landmarks
             scaled_landmarks = [[lm[0] * scale_x, lm[1] * scale_y] for lm in landmarks]
             aligned = align_face(img, scaled_landmarks)
             if aligned is not None:
                 _save_jpeg(aligned, os.path.join(output_dir, f"{count}.jpg"))
                 return aligned.size
 
-    # Fall back to bounding box crop with configurable margin
+    # Final fallback: bounding box crop with configurable margin
     margin = Config.FACE_MARGIN
     margin_x, margin_y = face_w * margin, face_h * margin
     crop_box = (

winnow/immich_api.py

@@ -45,6 +45,26 @@ def get_people() -> list[dict]:
         return []
 
 
+def merge_people(survivor_id: str, merge_ids: list[str]) -> bool:
+    """Merge duplicate people into survivor via Immich's merge endpoint.
+
+    The survivor (identified by survivor_id) absorbs all faces and assets
+    from the people in merge_ids, which are then removed from Immich.
+    """
+    try:
+        resp = requests.put(
+            f"{Config.IMMICH_URL}/api/people/{survivor_id}/merge",
+            headers={**get_headers(), "Content-Type": "application/json"},
+            json={"ids": merge_ids},
+            timeout=30,
+        )
+        resp.raise_for_status()
+        return True
+    except requests.RequestException as e:
+        logger.error(f"Failed to merge people into {survivor_id}: {e}")
+        return False
+
+
 def fetch_all_assets(person: dict) -> list[dict]:
     """Fetch all assets for a person with pagination."""
     name = person.get("name", "Unknown")