From 129c74720e0102e32b0e8da9c07a70617aabe559 Mon Sep 17 00:00:00 2001 From: Holden Date: Sat, 13 Jun 2026 03:44:36 +0000 Subject: [PATCH] Add quality replacement for Frigate face training images MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a person is at MAX_AUTO_IMAGES, winnow now replaces the lowest-quality mapped training image in Frigate if a higher-confidence candidate is available, keeping the training set always optimised. Only files winnow uploaded (tracked via frigate_files mapping) are ever replaced — manually added Frigate training images are never touched. A concurrent-upload race condition is detected per-file: if N>1 new files appear after one upload, the mapping is skipped rather than guessed, logging at INFO level. The per-file snapshot approach is retained over a batch approach because wrong mappings (which a batch approach risks on race) are worse than no mapping. Co-Authored-By: Claude Sonnet 4.6 --- CHANGELOG.md | 6 +++ README.md | 3 +- tests/test_upload_tracker.py | 70 ++++++++++++++++++++++++++++++++ winnow/config.py | 2 + winnow/executor.py | 63 ++++++++++++++++++++++++++++- winnow/frigate_api.py | 77 +++++++++++++++++++++++++++++------- winnow/jobs.py | 32 +++++++++------ winnow/upload_tracker.py | 59 ++++++++++++++++++++++++--- 8 files changed, 279 insertions(+), 33 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index eac3b3a..a9900cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **Quality replacement**: when a person is at `MAX_AUTO_IMAGES`, winnow now checks each new candidate against the lowest-quality image already in Frigate and swaps it in if the new image scores higher. Only images winnow uploaded (tracked in `frigate_files`) are ever replaced — files added manually through Frigate's UI are left untouched permanently. Enabled by default; set `QUALITY_REPLACEMENT=false` to revert to the previous behaviour of skipping people at cap. +- **Frigate filename mapping**: each successful upload now records the mapping from Frigate's assigned filename to the originating Immich asset ID and face confidence score in the tracker (`frigate_files` field). This is the foundation for quality replacement and future management of the Frigate training set. +- **`QUALITY_REPLACEMENT` env var** (default `true`): controls whether at-cap people are eligible for quality replacement. When disabled, people at `MAX_AUTO_IMAGES` are skipped as before. + ## [0.2.12] - 2026-06-13 ### Added diff --git a/README.md b/README.md index 54177d6..2d5fd9b 100644 --- a/README.md +++ b/README.md @@ -177,7 +177,8 @@ In scheduled mode the process (and loaded models) stays resident between runs. T | `USE_FULL_RESOLUTION` | `true` | Download full-resolution originals rather than preview thumbnails | | `MIN_CONFIDENCE` | `0.7` | Minimum Immich face detection confidence | | `BLUR_THRESHOLD` | `100.0` | Laplacian variance threshold — lower accepts more blur | -| `MAX_AUTO_IMAGES` | `80` | Maximum images selected in auto mode | +| `MAX_AUTO_IMAGES` | `80` | Maximum training images per person in Frigate | +| `QUALITY_REPLACEMENT` | `true` | When at cap, replace the lowest-quality mapped training image if a better candidate is found. Only affects images winnow uploaded — manually added Frigate training files are never touched. Set `false` to disable and skip people already at cap | ### GPU & Models diff --git a/tests/test_upload_tracker.py b/tests/test_upload_tracker.py index 5c59883..070a591 100644 --- a/tests/test_upload_tracker.py +++ b/tests/test_upload_tracker.py @@ -69,3 +69,73 @@ def test_duplicate_marks_are_idempotent(): mark_uploaded("dup", person_name="Alice") mark_uploaded("dup", person_name="Alice") assert filter_already_uploaded(["dup", "new"]) == ["new"] + + +# ── frigate_files mapping ───────────────────────────────────────────────────── + +def test_record_and_remove_frigate_file(): + from winnow.upload_tracker import get_person_summary, record_frigate_file, remove_frigate_file + record_frigate_file("Alice", "Alice-1000.webp", "asset-a1") + assert "Alice-1000.webp" in get_person_summary()["Alice"]["frigate_files"] + remove_frigate_file("Alice", "Alice-1000.webp") + assert "Alice-1000.webp" not in get_person_summary()["Alice"]["frigate_files"] + + +def test_remove_nonexistent_frigate_file_is_safe(): + from winnow.upload_tracker import remove_frigate_file + # Should not raise even if the file was never recorded + remove_frigate_file("Alice", "Alice-ghost.webp") + + +def test_remove_frigate_file_does_not_unmark_asset(): + """Deleting a Frigate file should not re-expose the source asset for upload.""" + from winnow.upload_tracker import ( + filter_already_uploaded, + mark_uploaded, + record_frigate_file, + remove_frigate_file, + ) + mark_uploaded("asset-a1", person_name="Alice") + record_frigate_file("Alice", "Alice-1000.webp", "asset-a1") + remove_frigate_file("Alice", "Alice-1000.webp") + # Asset must still be excluded — it was deliberately replaced, not lost + assert filter_already_uploaded(["asset-a1"]) == [] + + +def test_get_lowest_quality_mapped_file_none_when_empty(): + from winnow.upload_tracker import get_lowest_quality_mapped_file + assert get_lowest_quality_mapped_file("Alice") is None + + +def test_get_lowest_quality_mapped_file_returns_lowest(): + from winnow.upload_tracker import ( + get_lowest_quality_mapped_file, + mark_uploaded, + record_frigate_file, + ) + mark_uploaded("asset-hi", person_name="Alice", score=0.95) + mark_uploaded("asset-lo", person_name="Alice", score=0.71) + record_frigate_file("Alice", "Alice-1000.webp", "asset-hi") + record_frigate_file("Alice", "Alice-1001.webp", "asset-lo") + result = get_lowest_quality_mapped_file("Alice") + assert result is not None + frigate_filename, asset_id, score = result + assert frigate_filename == "Alice-1001.webp" + assert asset_id == "asset-lo" + assert score == pytest.approx(0.71, abs=0.001) + + +def test_get_lowest_quality_mapped_file_skips_unscored(): + """Files mapped without a score should not be returned as candidates.""" + from winnow.upload_tracker import ( + get_lowest_quality_mapped_file, + mark_uploaded, + record_frigate_file, + ) + mark_uploaded("asset-scored", person_name="Alice", score=0.85) + mark_uploaded("asset-noscr", person_name="Alice") + record_frigate_file("Alice", "Alice-1000.webp", "asset-scored") + record_frigate_file("Alice", "Alice-1001.webp", "asset-noscr") + result = get_lowest_quality_mapped_file("Alice") + assert result is not None + assert result[1] == "asset-scored" # only scored file is a candidate diff --git a/winnow/config.py b/winnow/config.py index 481f87f..ad3be2f 100644 --- a/winnow/config.py +++ b/winnow/config.py @@ -30,6 +30,7 @@ class _Config: BLUR_THRESHOLD: float = 100.0 MIN_CONFIDENCE: float = 0.7 MAX_AUTO_IMAGES: int = 80 + QUALITY_REPLACEMENT: bool = True # People filtering MIN_FACE_COUNT: int = 0 @@ -60,6 +61,7 @@ class _Config: self.BLUR_THRESHOLD = float(os.getenv("BLUR_THRESHOLD", "100.0")) self.MIN_CONFIDENCE = float(os.getenv("MIN_CONFIDENCE", "0.7")) self.MAX_AUTO_IMAGES = int(os.getenv("MAX_AUTO_IMAGES", "80")) + self.QUALITY_REPLACEMENT = os.getenv("QUALITY_REPLACEMENT", "true").lower() in ("true", "1", "yes") self.FACE_MARGIN = float(os.getenv("FACE_MARGIN", "0.15")) self.USE_FULL_RESOLUTION = os.getenv("USE_FULL_RESOLUTION", "true").lower() in ("true", "1", "yes") self.ENABLE_FACE_ALIGNMENT = os.getenv("ENABLE_FACE_ALIGNMENT", "true").lower() in ("true", "1", "yes") diff --git a/winnow/executor.py b/winnow/executor.py index 4bc608f..3da9f80 100644 --- a/winnow/executor.py +++ b/winnow/executor.py @@ -15,7 +15,14 @@ from .config import Config, get_headers from .image_processing import process_face_mode, process_full_mode, process_object_mode from .immich_api import fetch_face_data, fetch_full_image from .log_config import console -from .upload_tracker import mark_rejected, mark_uploaded +from .frigate_api import delete_frigate_person_files, get_frigate_person_files +from .upload_tracker import ( + get_lowest_quality_mapped_file, + mark_rejected, + mark_uploaded, + record_frigate_file, + remove_frigate_file, +) logger = logging.getLogger(__name__) @@ -246,8 +253,49 @@ 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 []) + quality_replacement = job.get("config", {}).get("quality_replacement", False) + 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 + if at_cap: + if not quality_replacement: + progress.console.print(f" [dim]⏭ {fname}: at cap, quality replacement disabled[/dim]") + progress.advance(upload_task) + continue + new_score = score_map.get(fname) + if new_score is None: + progress.console.print(f" [dim]⏭ {fname}: no confidence score, skipping replacement[/dim]") + progress.advance(upload_task) + continue + worst = get_lowest_quality_mapped_file(name) + if worst is None or new_score <= worst[2]: + progress.console.print( + f" [dim]⏭ {fname}: score {new_score:.3f} ≤ worst mapped" + f" {worst[2]:.3f if worst else 'N/A'}, skipping[/dim]" + ) + progress.advance(upload_task) + continue + # Delete the worst mapped file to make room for the better one + worst_frigate_file, _worst_asset_id, worst_score = worst + progress.console.print( + f" 🔄 {fname}: score {new_score:.3f} > {worst_score:.3f}," + f" replacing {worst_frigate_file}" + ) + if delete_frigate_person_files(name, [worst_frigate_file]): + remove_frigate_file(name, worst_frigate_file) + known_frigate_files.discard(worst_frigate_file) + else: + logger.warning(f"Failed to delete {worst_frigate_file} for {name}, skipping replacement") + progress.advance(upload_task) + continue + for attempt in range(1, max_retries + 1): try: with open(fpath, "rb") as f: @@ -265,6 +313,19 @@ def upload_to_frigate(jobs: list[dict]) -> None: 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. + current_files = set(get_frigate_person_files(name) or known_frigate_files) + 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: if attempt < max_retries: diff --git a/winnow/frigate_api.py b/winnow/frigate_api.py index 1988bb4..c51115e 100644 --- a/winnow/frigate_api.py +++ b/winnow/frigate_api.py @@ -8,26 +8,73 @@ import requests logger = logging.getLogger(__name__) -def get_frigate_face_counts() -> dict[str, int] | None: - """Return {person_name: training_image_count} from Frigate's train directory. - - Returns None if FRIGATE_URL is not set or the API is unreachable, so callers - can distinguish "API unavailable" from "person has 0 images." - """ +def _get_faces_data() -> dict | None: + """Fetch raw GET /api/faces response. Returns None if unavailable.""" frigate_url = os.environ.get("FRIGATE_URL", "").rstrip("/") if not frigate_url: return None try: resp = requests.get(f"{frigate_url}/api/faces", timeout=10) resp.raise_for_status() - data = resp.json() - # Response: {person_name: [file, ...], "train": [...], ...} - # "train" is a flat pending list, not a person — skip it. - return { - name: len(files) - for name, files in data.items() - if name != "train" and isinstance(files, list) - } + return resp.json() except Exception as e: - logger.warning(f"Could not query Frigate face counts: {e}") + logger.warning(f"Could not query Frigate faces API: {e}") return None + + +def get_frigate_face_counts() -> dict[str, int] | None: + """Return {person_name: training_image_count} from Frigate's train directory. + + Returns None if FRIGATE_URL is not set or the API is unreachable, so callers + can distinguish "API unavailable" from "person has 0 images." + """ + data = _get_faces_data() + if data is None: + return None + # Response: {person_name: [file, ...], "train": [...], ...} + # "train" is a flat pending list, not a person — skip it. + return { + name: len(files) + for name, files in data.items() + if name != "train" and isinstance(files, list) + } + + +def get_frigate_person_files(person_name: str) -> list[str] | None: + """Return the list of training filenames for a person in Frigate. + + Returns None if the API is unreachable. Returns an empty list if the + person exists but has no training images yet. + """ + data = _get_faces_data() + if data is None: + return None + files = data.get(person_name) + return files if isinstance(files, list) else [] + + +def delete_frigate_person_files(person_name: str, filenames: list[str]) -> bool: + """Delete specific training files for a person from Frigate. + + Uses POST /api/faces/{name}/delete with body {"ids": [filename, ...]}. + Returns True on success, False if unreachable or the request fails. + """ + frigate_url = os.environ.get("FRIGATE_URL", "").rstrip("/") + if not frigate_url or not filenames: + return False + from urllib.parse import quote + encoded = quote(person_name, safe="") + try: + resp = requests.post( + f"{frigate_url}/api/faces/{encoded}/delete", + json={"ids": filenames}, + timeout=10, + ) + if resp.ok: + logger.debug(f"Deleted {len(filenames)} Frigate file(s) for {person_name}") + return True + logger.warning(f"Frigate delete returned {resp.status_code} for {person_name}") + return False + except Exception as e: + logger.warning(f"Failed to delete Frigate files for {person_name}: {e}") + return False diff --git a/winnow/jobs.py b/winnow/jobs.py index ced0b04..954e66e 100644 --- a/winnow/jobs.py +++ b/winnow/jobs.py @@ -293,24 +293,34 @@ def auto_configure(people: list[dict]) -> list[dict]: already_uploaded = fc if fc is not None else person_summary.get("uploaded", 0) capacity = Config.MAX_AUTO_IMAGES - already_uploaded if capacity <= 0: + if not Config.QUALITY_REPLACEMENT: + rprint( + f" [dim]Skipping {name} (at cap:" + f" {already_uploaded}/{Config.MAX_AUTO_IMAGES}, quality replacement disabled).[/dim]" + ) + continue rprint( - f" [dim]Skipping {name} (at lifetime cap:" - f" {already_uploaded}/{Config.MAX_AUTO_IMAGES} trained).[/dim]" + f" [cyan]{name}: at cap ({already_uploaded}/{Config.MAX_AUTO_IMAGES})," + f" checking for quality improvements...[/cyan]" ) - continue + quality_replacement_only = True + else: + quality_replacement_only = False + + config["quality_replacement"] = quality_replacement_only or Config.QUALITY_REPLACEMENT has_embedding = is_embedding_available(entity_type) limit, selection_mode = _resolve_strategy(strategy, has_embedding) - # Cap selection to remaining capacity. - # For auto mode with partial training, keep "auto" so adaptive stopping - # still runs — just trim the result to the remaining capacity afterward. + # Cap selection to remaining capacity (no cap when replacement-only — executor + # decides per-image whether to swap; any candidate could be an improvement). auto_cap = None - if limit == "auto": - if already_uploaded > 0: - auto_cap = capacity - else: - limit = min(limit, capacity) + if not quality_replacement_only: + if limit == "auto": + if already_uploaded > 0: + auto_cap = capacity + else: + limit = min(limit, capacity) if selection_mode == "skip": continue diff --git a/winnow/upload_tracker.py b/winnow/upload_tracker.py index 0317639..ccaf97d 100644 --- a/winnow/upload_tracker.py +++ b/winnow/upload_tracker.py @@ -11,10 +11,14 @@ Both are excluded from future candidate pools. To reset: by_person schema (frigate_uploaded_ids.json): { - "asset_ids": ["immich-id-1", ...], # all assets we attempted to upload - "scores": {"immich-id-1": 0.953}, # Immich face confidence at upload time - "frigate_count": 42 # last known Frigate training image count + "asset_ids": ["immich-id-1", ...], # all assets we attempted to upload + "scores": {"immich-id-1": 0.953}, # Immich face confidence at upload time + "frigate_files": {"PersonName-123.webp": "immich-id-1"}, # Frigate filename → asset ID + "frigate_count": 42 # last known Frigate training image count } + +frigate_files only contains files winnow uploaded — files added manually through +Frigate's UI are never mapped here and are never touched by quality replacement. """ import json @@ -72,9 +76,10 @@ def _get_ids(entry: list | dict) -> list[str]: def _migrate_entry(entry: list | dict) -> dict: """Ensure by_person entry is in the current dict format.""" if isinstance(entry, list): - return {"asset_ids": sorted(entry), "scores": {}} + return {"asset_ids": sorted(entry), "scores": {}, "frigate_files": {}} entry.setdefault("asset_ids", []) entry.setdefault("scores", {}) + entry.setdefault("frigate_files", {}) return entry @@ -116,6 +121,49 @@ def mark_rejected(asset_id: str, person_name: str | None = None) -> None: logger.debug(f"Marked {asset_id} as rejected ({person_name})") +def record_frigate_file(person_name: str, frigate_filename: str, asset_id: str) -> None: + """Record the mapping from a Frigate training filename to an Immich asset ID.""" + data = _load(UPLOAD_TRACKER_FILE) + by_person = data.setdefault("by_person", {}) + entry = _migrate_entry(by_person.get(person_name, {})) + entry["frigate_files"][frigate_filename] = asset_id + by_person[person_name] = entry + _save(UPLOAD_TRACKER_FILE, data) + logger.debug(f"Mapped Frigate file {frigate_filename} → {asset_id} ({person_name})") + + +def remove_frigate_file(person_name: str, frigate_filename: str) -> None: + """Remove a Frigate filename from the mapping after it has been deleted. + + Does NOT unmark the source asset_id — the deletion was deliberate and + we don't want to re-upload the inferior image on the next run. + """ + data = _load(UPLOAD_TRACKER_FILE) + by_person = data.get("by_person", {}) + entry = _migrate_entry(by_person.get(person_name, {})) + entry["frigate_files"].pop(frigate_filename, None) + by_person[person_name] = entry + _save(UPLOAD_TRACKER_FILE, data) + logger.debug(f"Removed Frigate file mapping {frigate_filename} ({person_name})") + + +def get_lowest_quality_mapped_file(person_name: str) -> tuple[str, str, float] | None: + """Return (frigate_filename, asset_id, score) for the mapped file with the lowest + confidence score, or None if no mapped files with known scores exist.""" + data = _load(UPLOAD_TRACKER_FILE) + entry = _migrate_entry(data.get("by_person", {}).get(person_name, {})) + frigate_files = entry.get("frigate_files", {}) + scores = entry.get("scores", {}) + candidates = [ + (frigate_filename, asset_id, scores[asset_id]) + for frigate_filename, asset_id in frigate_files.items() + if asset_id in scores + ] + if not candidates: + return None + return min(candidates, key=lambda x: x[2]) + + def update_frigate_count(person_name: str, count: int) -> None: """Record Frigate's authoritative training image count for a person.""" data = _load(UPLOAD_TRACKER_FILE) @@ -143,7 +191,7 @@ def reset_person(person_name: str) -> None: def get_person_summary() -> dict[str, dict]: - """Return {person_name: {uploaded, rejected, frigate_count, scores}} for display/capacity.""" + """Return {person_name: {uploaded, rejected, frigate_count, scores, frigate_files}} for display/capacity.""" uploaded_data = _load(UPLOAD_TRACKER_FILE).get("by_person", {}) rejected_data = _load(REJECT_TRACKER_FILE).get("by_person", {}) names = set(uploaded_data) | set(rejected_data) @@ -156,6 +204,7 @@ def get_person_summary() -> dict[str, dict]: "rejected": len(_get_ids(r_entry)), "frigate_count": u_entry.get("frigate_count") if isinstance(u_entry, dict) else None, "scores": u_entry.get("scores", {}) if isinstance(u_entry, dict) else {}, + "frigate_files": u_entry.get("frigate_files", {}) if isinstance(u_entry, dict) else {}, } return result