flan/winnow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #34 from sudolulo/dev

Browse Source 
release: v0.6.0 — revert SQLite tracker to JSON backend
This commit is contained in:
flan
2026-06-15 11:53:23 -04:00
committed by GitHub
parent 86c76ba6a7 9e84e276da
commit 2de0c02c4e
19 changed files with 664 additions and 739 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+20 -122
View File
@@ -7,153 +7,51 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
## [0.5.7] - 2026-06-15
### Fixed
- **Symlink guard moved before `realpath`**: `_safe_person_dir` now checks whether the raw (unresolved) path is a symlink before calling `os.path.realpath`. The previous check in `execute_jobs` ran after `realpath` had already resolved the link, making it unreachable dead code.
- **`fetch_all_assets` returns raw item count**: the function now returns `(assets, total_raw)` where `total_raw` is the total items seen across all pages before non-dict filtering. `auto_configure` and `_configure_person` use `total_raw` for the `MIN_FACE_COUNT` guard and display, so transient non-dict API items cannot cause a person to be incorrectly skipped.
- **Pagination stop on all-non-dict page now logs a warning**: when `valid_assets` is empty but the page was non-empty (all items were non-dict), a `WARNING` is emitted explaining why pagination stopped, distinguishing it from natural end-of-data.
- **`_data_cfg.exists()` called once**: the result is cached in `_data_cfg_exists` so the dual-config warning check and the `config_file` selection always agree — previously two separate `stat()` calls created a TOCTOU window where the log could claim one file while the code loaded another.
## [0.5.6] - 2026-06-15
### Fixed
- **Pagination runaway on all-non-dict page**: the empty-page break in `fetch_all_assets` now fires after non-dict filtering rather than before, so a page whose items are all non-dict (e.g. all nulls) correctly terminates pagination instead of looping to MAX_PAGES.
- **Non-dict API items upgraded to warning**: items skipped in a paginated response are now logged at `WARNING` (previously `DEBUG`) so silent asset loss is visible at default log levels.
- **Single-pass page filtering**: `fetch_all_assets` now partitions valid and invalid items in one loop instead of iterating `page_assets` twice with inverse predicates.
- **Reconciliation checks Frigate before sleeping**: the poll loop now performs an initial check immediately after upload, then backs off with `_RECONCILE_POLL_DELAYS` only if needed. Previously the loop always slept ≥1 s before any check.
- **Reconciliation set subtraction computed once**: `current_files - known_files_before` was computed twice per poll iteration (once for the count check, once for the final mapping). It is now computed once and reused.
## [0.5.5] - 2026-06-15
## [0.6.0] - 2026-06-15
### Changed
- **Module-level constants in `diversity.py`**: magic numbers `3000` (pool cap), `20` (pool scale), and `32` (embedding batch size) extracted to named constants `_POOL_CAP`, `_POOL_SCALE`, and `_EMBEDDING_BATCH_SIZE`.
- **Upload tracker reverted to JSON storage**: the SQLite-based tracker introduced in v0.5.0 produced 17 bug-fix releases in two days due to data-loss risks in the migration layer, schema primary key conflicts, tracker isolation races, and disk-full retry storms. The JSON backend (`frigate_uploaded_ids.json` / `frigate_rejected_ids.json` in `DATA_DIR`) is restored. It is simpler, has no migration layer, and carries no external dependency. If you ran any v0.5.x version, delete `frigate_tracker.db` from your `DATA_DIR` once you confirm the JSON files look correct. JSON files from before v0.5.0 are read automatically with no changes required.
- **Reconciliation poll delays extracted**: `(1, 2, 4, 8)` back-off delays in `reconcile.py` extracted to `_RECONCILE_POLL_DELAYS` with an explanatory comment.
- **`CACHE_DIR` env var accepted as `DATA_DIR` alias**: the rename introduced in v0.5.1 is preserved — `CACHE_DIR` still works with a deprecation warning. The default data path remains `data` (Docker: `/app/data`).
- **`_VALID_SCORE_COLS` comment**: explains that the frozenset is a SQL-injection guard for dynamic column interpolation, not a runtime filter.
- **Config file now lives in `DATA_DIR`**: `.immich_config.json` resolves to `DATA_DIR/.immich_config.json` so it persists across container restarts. The legacy CWD location is still checked as a fallback for existing setups.
- **`record_frigate_files_batch` docstring**: clarifies that all mappings are written atomically — no partial failure is possible.
- **`get_person_summary()` refactored**: eliminated four repeated default-dict blocks using a local `_entry()` helper with `setdefault`.
- **`encoded` → `encoded_name` in `frigate_api.py`**: renamed the URL-encoded person name variable for clarity.
- **Dual response shape comment in `immich_api.py`**: documents that Immich ≥2.x returns `{"assets": {"items": [...]}}` while earlier versions returned `{"assets": [...]}` directly.
- **Non-dict item debug log in `fetch_all_assets`**: skipped non-dict items in a page response now emit a `logger.debug` line with the count and page number.
## [0.5.4] - 2026-06-14
- **Diversity selector receives capacity as its limit directly**: instead of selecting up to `MAX_AUTO_IMAGES` and then slicing to the remaining capacity, the selector now runs with the actual remaining slot count as its budget.
### Fixed
- **Quality replacement slot floor used wrong score**: when a blur-score replacement deleted a low-quality Frigate file but the subsequent upload failed, `min_quality_score_for_slot` was set to the failed candidate's score rather than the deleted file's score. This caused subsequent candidates that were better than the deleted file (but worse than the failed upload) to be skipped, leaving the freed slot unfilled for the rest of that run. Fixed by using `target_score` (deleted file's score) as the floor, matching the documented intent in the surrounding comment.
- **Immich v2.7.5 compatibility**: `auto_configure` no longer pre-filters people by `assetCount` from `/api/people`, which Immich v2.7.5 dropped. The `MIN_FACE_COUNT` check now runs after `fetch_all_assets` using the actual fetched count.
## [0.5.3] - 2026-06-14
- **`fetch_face_data` no longer falls back to a wrong person's bounding box**: when `person_id` is provided but not found in the Immich `/api/faces` response, the function now returns `None` instead of using `faces[0]`. Previously a group photo where the target person's face entry was missing would inject a different person's bounding box into the crop.
### Fixed
- **Corrupt thumbnail permanently rejected**: when `resp.ok=True` but `PIL.UnidentifiedImageError` is raised (Pillow cannot identify the image format), the asset is now marked rejected so it isn't re-downloaded on every future run. Transient `OSError`/truncation errors are intentionally not caught here — those are retried normally.
- **`LIMIT` env var crash**: non-integer values (e.g. `"30.5"`, `"all"`) now log a warning and fall back to the default instead of raising `ValueError` at startup.
- **`mark_uploaded` tracker failure no longer aborts the upload loop**: a tracker write failure after a successful Frigate POST is logged and the loop continues; the asset will be re-uploaded on the next run rather than the current run dying mid-job.
- **Symlink guard on person output dir**: `shutil.rmtree` is now skipped if `person_dir` resolves to a symlink, preventing traversal out of `OUTPUT_DIR` on a shared volume.
- **`progress.remove_task` now in `finally` block**: the progress bar task is cleaned up even when a job exits via an exception, preventing orphaned progress rows in the terminal.
- **`person["id"]` KeyError**: malformed Immich API responses missing the `id` field now log an error and skip that person instead of crashing the job.
- **`SKIP_PEOPLE`/`ONLY_PEOPLE` now strip whitespace**: `"Alice, Bob".split(",")` produces `[" Bob"]`; the leading space now stripped so comma-separated values with spaces work as expected.
- **Face data response type validation**: `fetch_face_data` now validates that the `/api/faces` response is a list before indexing, guarding against null or non-list API responses.
- **`FRIGATE_URL` with trailing slash no longer produces double-slash paths**: all Frigate API calls now use `_get_frigate_url()` for URL normalization rather than reading `FRIGATE_URL` inline.
- **Pagination error log includes page number**: the exception log in `fetch_all_assets` now includes the page number that failed.
- **Frigate version `v`-prefix now stripped**: `v0.16.0`-style version strings are correctly parsed.
- **`Image.open()` wrapped for non-image responses**: PIL parse errors on thumbnail fetches (e.g. reverse-proxy HTML error page returning 200) are now caught and logged instead of propagating.
- **Invalid numeric env var values warn and use defaults**: a typo such as `YEARS_FILTER=10 ` (trailing space) or `MIN_FACE_WIDTH=auto` now logs a `WARNING` and falls back to the documented default instead of raising `ValueError` at startup. Affects `YEARS_FILTER`, `MIN_FACE_WIDTH`, `MIN_FACE_COUNT`, `MAX_AUTO_IMAGES`, `BLUR_THRESHOLD`, `MIN_CONFIDENCE`, and `FACE_MARGIN`.
- **Frigate version `v`-prefix handling**: `v0.16.0`-style version strings are now correctly parsed; the leading `v` was previously misread, causing the too-old warning to never fire.
- **`IMMICH_URL` blank placeholder falls back to config file**: `IMMICH_URL=` (empty or blank) in `.env` is now treated as unset and falls through to `DATA_DIR/.immich_config.json`, matching pre-v0.5.0 behaviour.
- **`FRIGATE_SCORE_CEILING` parse guard**: a non-float value in `.env` now logs a warning and disables the ceiling instead of crashing at startup.
- **Reconciliation checks Frigate immediately before first sleep**: the poll loop now performs an immediate check after upload, then backs off with `(1, 2, 4, 8)` s delays only if needed.
- **Dual config file warning**: a log warning is emitted when both `DATA_DIR/.immich_config.json` and the legacy CWD config file exist simultaneously.
- **Dockerfile unknown `VARIANT` now fails loudly**: an unrecognised value now exits with an error instead of silently falling through to the cpu branch.
- **PID file write guard**: `OSError` on `/tmp/winnow.pid` write is now caught and logged instead of crashing the scheduler.
- **Embedding cache writes are now atomic**: `.npy` files are written to a `.tmp` sibling and renamed into place with `os.replace`, preventing truncated cache entries on process kill.
- **Scheduler sleep clamped to 60 s**: bounds recovery time after an NTP clock step.
- **`get_frigate_person_files` non-list debug log**: consistent with `get_all_frigate_person_files`.
## [0.5.2] - 2026-06-14
### Fixed
- **Immich v2.7.5 compatibility**: `auto_configure` no longer pre-filters people by `assetCount` from the `/api/people` response, which Immich v2.7.5 dropped. The `MIN_FACE_COUNT` check now runs after `fetch_all_assets` so the actual asset count is used instead of the missing field.
- **Dockerfile supply-chain**: replaced `curl | sh` uv installer with `COPY --from=ghcr.io/astral-sh/uv:0.11.21` to eliminate the network-executed script.
- **HEALTHCHECK**: replaced the static file-existence check with `kill -0 $(cat /tmp/winnow.pid)` so the container reports unhealthy when the scheduler process actually dies, not just when a script file is missing.
- **`CONFIG_FILE` volume safety**: the config file path now resolves to `DATA_DIR/.immich_config.json` so it persists across container restarts. The legacy CWD location is still read as a fallback for existing setups.
- **EmbeddingCache singleton isolation**: `get_cache()` now tracks the `cache_dir` argument and re-creates the cache when it changes, preventing test runs from sharing state across different `DATA_DIR` values.
- **File descriptor leak in `_suppress_output()`**: `devnull_fd`, `saved_out`, and `saved_err` are now all closed in a nested `finally` chain, preventing fd exhaustion on long runs.
- **Silent exception in `upload_tracker`**: `except Exception: pass` on SQLite connection close is now `except Exception as e: logger.debug(...)` so connection errors are visible in debug logs.
- **Frigate API unknown-key logging**: `get_all_frigate_person_files` now logs unexpected non-list keys at DEBUG level instead of silently skipping them.
- **Reconcile debug log**: added a debug log entry before the FIFO timestamp mapping step in `reconcile_frigate_mappings` to make the mapping assumption visible in logs.
- **CI action SHA pinning**: all five GitHub Actions workflows now pin every third-party action to a full commit SHA. Updated `setup-uv` v7→v8.2.0, `upload-artifact` v4→v7.0.1, `download-artifact` v4→v8.0.1, `ruff-action` v3→v4.0.0.
## [0.5.1] - 2026-06-14
### Changed
- **`CACHE_DIR` renamed to `DATA_DIR`**: the environment variable that sets the path for the embedding cache and SQLite tracker database is now called `DATA_DIR` (default: `data`; Docker default: `/app/data`). The old `CACHE_DIR` still works with a startup deprecation warning — rename it to `DATA_DIR` in your `.env` or `compose.yml` to silence the warning. The container-side default path changes from `/app/.if_cache` to `/app/data`; update your volume mount accordingly.
## [0.5.0] - 2026-06-14
### Changed
- **SQLite upload tracker**: `upload_tracker.py` is fully rewritten on top of SQLite (stdlib `sqlite3`). The JSON pair (`frigate_uploaded_ids.json` / `frigate_rejected_ids.json`) is replaced by a single `winnow_tracker.db` (WAL journal, `check_same_thread=False`). Existing JSON files are migrated atomically on first run and renamed to `.json.bak`. No user action required; the tracker API (`mark_uploaded`, `mark_rejected`, `filter_already_uploaded`, `get_person_summary`, etc.) is unchanged.
- **Config lazy singleton**: `_Config` now uses `__getattr__` to defer all I/O until the first attribute access. `load_dotenv()` no longer runs at module import time — it runs on the first access to any `Config` attribute. Empty-string env vars (`IMMICH_URL=`, `OUTPUT_DIR=`) are now correctly distinguished from unset ones so a `.env` file value never silently overrides an explicit `""` set in the environment. `Config.reset()` clears the loaded state for clean test isolation.
- **Reconcile module extracted**: `reconcile_frigate_mappings` and `enrich_asset_with_face_data` are extracted from `executor.py` into a new `winnow/reconcile.py` module. No behaviour change; reduces `executor.py` length and clarifies responsibility boundaries.
- **Single lockfile**: `pyproject-gpu.toml`, `pyproject-cpu.toml`, `pyproject-rocm.toml`, `pyproject-intel.toml` and their separate lockfiles are removed. GPU/ROCm/Intel/CPU variant deps are now declared as `[project.optional-dependencies]` extras in `pyproject.toml` with `[tool.uv] conflicts` for mutual exclusion. A single `uv.lock` covers all variants. The Dockerfile selects the correct extra via `uv sync --extra $VARIANT`.
- **Ubuntu base bumped**: amd64 GPU base updated from `nvidia/cuda:12.8.1-cudnn-runtime-ubuntu22.04` to `nvidia/cuda:12.8.1-cudnn-runtime-ubuntu24.04`. amd64 ROCm and CPU bases updated from Ubuntu 22.04 to Ubuntu 26.04. arm64 bases remain Ubuntu 24.04.
### Fixed
- **Frigate API unreachable at upload start no longer crashes reconciliation**: when the Frigate `GET /api/faces` call fails at upload start, reconciliation is now skipped entirely for that batch (`_skip_reconcile = True`). Previously, falling back to the tracker's known filenames as the pre-upload baseline caused the `> target` guard to fire on unmapped manual files, silently dropping all mappings.
- **Polling `== target` guards against wrong-file mapping**: the reconcile poll loop now breaks on `len(new_files) == target` and sets an "external upload detected" flag when `> target`. The old `>= target` break would have proceeded with an incorrect file set when a concurrent external upload was present, causing wrong asset-ID mappings. The poll loop now also exits early on `> target` rather than exhausting all four retry intervals (up to 15 s wasted per person with a concurrent external uploader).
- **`auto_cap` post-selection truncation removed**: the diversity selector now receives the correct upper bound (`capacity` or `min(limit, capacity)`) directly instead of selecting up to `MAX_AUTO_IMAGES` and then silently truncating the result list. The old approach produced a selection biased toward the first `capacity` items in embedding space rather than the globally optimal diverse subset.
- **Dockerfile unknown VARIANT now fails loudly**: added an explicit `elif [ "$VARIANT" = "gpu" ]` branch and an `else … exit 1` for unrecognised values. Previously, any unknown variant silently fell through to the `cpu` branch.
- **JSON migration partial-rename data loss**: if the rename of one of the two JSON files failed (e.g. a `PermissionError`), the other file's data was committed to SQLite but the `COUNT(*) > 0` guard on the next run would skip re-migration of the remaining file, permanently losing its data. The guard is removed (idempotent `INSERT OR IGNORE` makes re-running safe). Each rename is now wrapped in its own `try/except OSError` so a failure on one file is logged and does not prevent the other from completing.
- **SQL column allowlist in `_pick_mapped_file`**: the `score_col` f-string interpolation into SQL is now guarded by a `frozenset` allowlist at the function boundary, raising `ValueError` on any value outside `{"blur_score", "frigate_score"}`.
- **`load_dotenv` no longer runs at import time**: moving `load_dotenv()` to the first line of `_load()` prevents side-effects during module import (which could interfere with test environment setup) and makes the load order deterministic relative to `os.environ` overrides.
- **Empty-string env var priority fix**: `if self.IMMICH_URL or …` treated `IMMICH_URL=""` as falsy and silently fell through to the config file. Changed to `if self.IMMICH_URL is None` so an empty-string explicit env var is respected.
- **`EmbeddingCache` singleton re-creates when `DATA_DIR` changes**: prevents test runs from sharing cache state across different `DATA_DIR` values.
### Added
- **Diversity test suite expanded** (PR #11): 33 new tests covering k-medoids clustering, farthest-point sampling, adaptive threshold computation, near-duplicate deduplication, and time-spread selection. Total: 93 tests (was 60).
- **Known-limitation annotations** (PR #12): `TODO(frigate-api)` comments placed at each FIFO-ordering assumption, manual-file-invisibility note, and async-rebuild limitation in `executor.py` and `reconcile.py`. These mark spots where a richer Frigate API would allow a deeper fix.
- **Diversity test suite** (PR #11): 33 tests covering k-medoids clustering, farthest-point sampling, adaptive threshold computation, near-duplicate deduplication, and time-spread selection. Total: 93 tests.
## [0.4.11] - 2026-06-14
README.md
+2 -2
View File
@@ -74,7 +74,7 @@ Immich library
↳ at cap + QUALITY_REPLACEMENT=false — skip this person
```
Uploaded and rejected asset IDs are persisted across runs in a SQLite database (`winnow_tracker.db` in `DATA_DIR`). The same image is never processed twice; rejected assets are permanently skipped unless `RETRY_REJECTED=true`.
Uploaded and rejected asset IDs are persisted across runs in two JSON files (`frigate_uploaded_ids.json` and `frigate_rejected_ids.json` in `DATA_DIR`). The same image is never processed twice; rejected assets are permanently skipped unless `RETRY_REJECTED=true`.
---
@@ -212,7 +212,7 @@ These defaults are tuned for Frigate's ArcFace requirements. winnow will warn on
| `FORCE_CPU` | `false` | Disable GPU — fall back to CPU for all inference |
| `OPENVINO_DEVICE` | `CPU` | Intel variant only: set `GPU` to use Arc or iGPU; default runs on CPU |
| `ENABLE_CACHE` | `true` | Cache computed embeddings to disk (speeds up re-runs on the same library) |
| `DATA_DIR` | `data` | Path for embedding cache and upload tracker database (`winnow_tracker.db`) |
| `DATA_DIR` | `data` | Path for embedding cache and upload tracker JSON files |
| `INSIGHTFACE_HOME` | *(system)* | InsightFace model cache path (Buffalo_L) |
### Output
pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "winnow"
version = "0.5.7"
version = "0.6.0"
description = "Selects diverse, high-quality photos from Immich as training data for Frigate face recognition."
license = "AGPL-3.0-or-later"
requires-python = ">=3.13"
scripts/benchmark.py
+2 -4
View File
@@ -13,7 +13,6 @@ Usage inside container:
docker exec -e FORCE_CPU=true winnow python /app/scripts/benchmark.py
"""
import os
import sys
import time
@@ -22,9 +21,8 @@ from PIL import Image, ImageDraw
def _mode_label() -> str:
if os.getenv("FORCE_CPU", "").lower() in ("true", "1", "yes"):
return "CPU (FORCE_CPU=true)"
return "GPU (auto)"
from winnow.config import _getenv_bool
return "CPU (FORCE_CPU=true)" if _getenv_bool("FORCE_CPU", False) else "GPU (auto)"
def make_face_image(size: int = 640) -> Image.Image:
tests/test_upload_tracker.py
+40 -45
View File
@@ -10,19 +10,7 @@ def isolated_cache(monkeypatch, tmp_path):
monkeypatch.setenv("DATA_DIR", str(tmp_path))
from winnow.config import _Config
_Config.reset()
# Also reset the SQLite connection so the next call opens the new path
import winnow.upload_tracker as ut
ut._conn = None
ut._conn_path = None
yield tmp_path
# Teardown
if ut._conn is not None:
try:
ut._conn.close()
except Exception:
pass
ut._conn = None
ut._conn_path = None
_Config.reset()
@@ -85,12 +73,12 @@ def test_duplicate_marks_are_idempotent():
# ── frigate_files mapping ─────────────────────────────────────────────────────
def test_record_and_remove_frigate_files_batch():
from winnow.upload_tracker import get_person_summary, record_frigate_files_batch, remove_frigate_file
record_frigate_files_batch("Alice", {"Alice-1000.webp": "asset-a1"})
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().get("Alice", {}).get("frigate_files", {})
assert "Alice-1000.webp" not in get_person_summary()["Alice"]["frigate_files"]
def test_remove_nonexistent_frigate_file_is_safe():
@@ -104,11 +92,11 @@ def test_remove_frigate_file_does_not_unmark_asset():
from winnow.upload_tracker import (
filter_already_uploaded,
mark_uploaded,
record_frigate_files_batch,
record_frigate_file,
remove_frigate_file,
)
mark_uploaded("asset-a1", person_name="Alice")
record_frigate_files_batch("Alice", {"Alice-1000.webp": "asset-a1"})
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"]) == []
@@ -120,14 +108,14 @@ def test_get_tracked_frigate_file_count_zero_when_empty():
def test_get_tracked_frigate_file_count_counts_only_mapped():
"""Only files explicitly recorded via record_frigate_files_batch count toward the cap."""
from winnow.upload_tracker import get_tracked_frigate_file_count, mark_uploaded, record_frigate_files_batch
"""Only files explicitly recorded via record_frigate_file count toward the cap."""
from winnow.upload_tracker import get_tracked_frigate_file_count, mark_uploaded, record_frigate_file
mark_uploaded("asset-a", person_name="Alice")
mark_uploaded("asset-b", person_name="Alice")
record_frigate_files_batch("Alice", {"Alice-1000.webp": "asset-a"})
record_frigate_file("Alice", "Alice-1000.webp", "asset-a")
# asset-b is uploaded but not yet mapped — does not count
assert get_tracked_frigate_file_count("Alice") == 1
record_frigate_files_batch("Alice", {"Alice-1001.webp": "asset-b"})
record_frigate_file("Alice", "Alice-1001.webp", "asset-b")
assert get_tracked_frigate_file_count("Alice") == 2
@@ -140,11 +128,12 @@ def test_get_lowest_quality_mapped_file_returns_lowest():
from winnow.upload_tracker import (
get_lowest_quality_mapped_file,
mark_uploaded,
record_frigate_files_batch,
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_files_batch("Alice", {"Alice-1000.webp": "asset-hi", "Alice-1001.webp": "asset-lo"})
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
@@ -158,11 +147,12 @@ def test_get_lowest_quality_mapped_file_skips_unscored():
from winnow.upload_tracker import (
get_lowest_quality_mapped_file,
mark_uploaded,
record_frigate_files_batch,
record_frigate_file,
)
mark_uploaded("asset-scored", person_name="Alice", score=0.85)
mark_uploaded("asset-noscr", person_name="Alice")
record_frigate_files_batch("Alice", {"Alice-1000.webp": "asset-scored", "Alice-1001.webp": "asset-noscr"})
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
@@ -176,26 +166,28 @@ def test_get_tracked_frigate_filenames_empty():
def test_get_tracked_frigate_filenames_returns_mapped():
from winnow.upload_tracker import get_tracked_frigate_filenames, record_frigate_files_batch
record_frigate_files_batch("Alice", {"Alice-1000.webp": "asset-a", "Alice-1001.webp": "asset-b"})
from winnow.upload_tracker import get_tracked_frigate_filenames, record_frigate_file
record_frigate_file("Alice", "Alice-1000.webp", "asset-a")
record_frigate_file("Alice", "Alice-1001.webp", "asset-b")
assert get_tracked_frigate_filenames("Alice") == {"Alice-1000.webp", "Alice-1001.webp"}
def test_get_tracked_frigate_filenames_excludes_removed():
from winnow.upload_tracker import (
get_tracked_frigate_filenames,
record_frigate_files_batch,
record_frigate_file,
remove_frigate_file,
)
record_frigate_files_batch("Alice", {"Alice-1000.webp": "asset-a", "Alice-1001.webp": "asset-b"})
record_frigate_file("Alice", "Alice-1000.webp", "asset-a")
record_frigate_file("Alice", "Alice-1001.webp", "asset-b")
remove_frigate_file("Alice", "Alice-1000.webp")
assert get_tracked_frigate_filenames("Alice") == {"Alice-1001.webp"}
def test_get_tracked_frigate_filenames_isolated_by_person():
from winnow.upload_tracker import get_tracked_frigate_filenames, record_frigate_files_batch
record_frigate_files_batch("Alice", {"Alice-1000.webp": "asset-a"})
record_frigate_files_batch("Bob", {"Bob-2000.webp": "asset-b"})
from winnow.upload_tracker import get_tracked_frigate_filenames, record_frigate_file
record_frigate_file("Alice", "Alice-1000.webp", "asset-a")
record_frigate_file("Bob", "Bob-2000.webp", "asset-b")
assert get_tracked_frigate_filenames("Alice") == {"Alice-1000.webp"}
assert get_tracked_frigate_filenames("Bob") == {"Bob-2000.webp"}
@@ -206,11 +198,12 @@ def test_get_lowest_quality_exclude_skips_specified_file():
from winnow.upload_tracker import (
get_lowest_quality_mapped_file,
mark_uploaded,
record_frigate_files_batch,
record_frigate_file,
)
mark_uploaded("asset-lo", person_name="Alice", score=0.10)
mark_uploaded("asset-hi", person_name="Alice", score=0.90)
record_frigate_files_batch("Alice", {"Alice-lo.webp": "asset-lo", "Alice-hi.webp": "asset-hi"})
record_frigate_file("Alice", "Alice-lo.webp", "asset-lo")
record_frigate_file("Alice", "Alice-hi.webp", "asset-hi")
result = get_lowest_quality_mapped_file("Alice", exclude={"Alice-lo.webp"})
assert result is not None
assert result[1] == "asset-hi" # lo was excluded; hi is returned
@@ -220,28 +213,29 @@ def test_get_lowest_quality_exclude_all_returns_none():
from winnow.upload_tracker import (
get_lowest_quality_mapped_file,
mark_uploaded,
record_frigate_files_batch,
record_frigate_file,
)
mark_uploaded("asset-a", person_name="Alice", score=0.50)
record_frigate_files_batch("Alice", {"Alice-a.webp": "asset-a"})
record_frigate_file("Alice", "Alice-a.webp", "asset-a")
assert get_lowest_quality_mapped_file("Alice", exclude={"Alice-a.webp"}) is None
# ── get_most_redundant_mapped_file ────────────────────────────────────────────
def test_get_most_redundant_none_when_no_frigate_scores():
from winnow.upload_tracker import get_most_redundant_mapped_file, mark_uploaded, record_frigate_files_batch
from winnow.upload_tracker import get_most_redundant_mapped_file, mark_uploaded, record_frigate_file
mark_uploaded("asset-a", person_name="Alice", score=0.80)
record_frigate_files_batch("Alice", {"Alice-a.webp": "asset-a"})
record_frigate_file("Alice", "Alice-a.webp", "asset-a")
# blur score only, no frigate_score → no candidates
assert get_most_redundant_mapped_file("Alice") is None
def test_get_most_redundant_returns_highest_frigate_score():
from winnow.upload_tracker import get_most_redundant_mapped_file, mark_uploaded, record_frigate_files_batch
from winnow.upload_tracker import get_most_redundant_mapped_file, mark_uploaded, record_frigate_file
mark_uploaded("asset-novel", person_name="Alice", score=0.50, frigate_score=0.31)
mark_uploaded("asset-redundant", person_name="Alice", score=0.90, frigate_score=0.88)
record_frigate_files_batch("Alice", {"Alice-novel.webp": "asset-novel", "Alice-redundant.webp": "asset-redundant"})
record_frigate_file("Alice", "Alice-novel.webp", "asset-novel")
record_frigate_file("Alice", "Alice-redundant.webp", "asset-redundant")
result = get_most_redundant_mapped_file("Alice")
assert result is not None
frigate_filename, asset_id, score = result
@@ -251,17 +245,18 @@ def test_get_most_redundant_returns_highest_frigate_score():
def test_get_most_redundant_exclude_skips_file():
from winnow.upload_tracker import get_most_redundant_mapped_file, mark_uploaded, record_frigate_files_batch
from winnow.upload_tracker import get_most_redundant_mapped_file, mark_uploaded, record_frigate_file
mark_uploaded("asset-hi", person_name="Alice", score=0.9, frigate_score=0.85)
mark_uploaded("asset-lo", person_name="Alice", score=0.5, frigate_score=0.40)
record_frigate_files_batch("Alice", {"Alice-hi.webp": "asset-hi", "Alice-lo.webp": "asset-lo"})
record_frigate_file("Alice", "Alice-hi.webp", "asset-hi")
record_frigate_file("Alice", "Alice-lo.webp", "asset-lo")
result = get_most_redundant_mapped_file("Alice", exclude={"Alice-hi.webp"})
assert result is not None
assert result[1] == "asset-lo" # hi excluded; lo is next highest
def test_get_most_redundant_exclude_all_returns_none():
from winnow.upload_tracker import get_most_redundant_mapped_file, mark_uploaded, record_frigate_files_batch
from winnow.upload_tracker import get_most_redundant_mapped_file, mark_uploaded, record_frigate_file
mark_uploaded("asset-a", person_name="Alice", score=0.5, frigate_score=0.70)
record_frigate_files_batch("Alice", {"Alice-a.webp": "asset-a"})
record_frigate_file("Alice", "Alice-a.webp", "asset-a")
assert get_most_redundant_mapped_file("Alice", exclude={"Alice-a.webp"}) is None
uv.lock
Generated
+1 -1
View File
@@ -862,7 +862,7 @@ wheels = [
[[package]]
name = "winnow"
version = "0.5.7"
version = "0.6.0"
source = { editable = "." }
dependencies = [
{ name = "croniter" },
winnow/cache.py
+10 -1
View File
@@ -82,10 +82,19 @@ class EmbeddingCache:
def put(self, asset_id: str, embedding: np.ndarray, model: str = "insightface") -> None:
"""Store an embedding in the cache."""
self._ensure_dir()
final = self._path(asset_id, model)
# Insert .tmp before .npy so np.save doesn't auto-append another .npy extension
# (np.save appends .npy to paths that don't already end in .npy).
tmp = final.removesuffix(".npy") + ".tmp.npy"
try:
np.save(self._path(asset_id, model), embedding)
np.save(tmp, embedding)
os.replace(tmp, final)
except Exception as e:
logger.debug("Cache write failed for %s: %s", asset_id, e)
try:
os.remove(tmp)
except OSError:
pass
def clear(self) -> None:
"""Delete all cached embeddings."""
winnow/cli.py
+27 -6
View File
@@ -7,7 +7,7 @@ import sys
from rich import print as rprint
from rich.prompt import Confirm
from .config import Config
from .config import Config, _getenv_bool
from .executor import execute_jobs, upload_to_frigate
from .immich_api import get_immich_version, get_people, merge_people
from .jobs import _show_preview, auto_configure, interactive_configure
@@ -125,9 +125,30 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
if merged_any:
rprint(" [dim]Re-fetching people after merge...[/dim]")
return get_people()
fresh = get_people()
# Filter out the smaller duplicate from any group whose merge failed — those
# IDs still exist in Immich and would produce two jobs for the same folder.
# IDs from groups that merged successfully are already gone from Immich, so
# this filter is a no-op for them.
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 fresh if p.get("id") not in skip_ids]
return people
# All merges failed — fall back to local deduplication (keep largest per name) so
# downstream job creation never runs two jobs for the same Frigate folder.
rprint(
" [yellow]All merges failed — applying local deduplication"
" to avoid overwriting output.[/yellow]"
)
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]
_UNSUPPORTED_VARS = [
@@ -145,7 +166,7 @@ _UNSUPPORTED_VARS = [
def main() -> None:
"""Entry point for winnow CLI."""
try:
verbose = os.environ.get("VERBOSE", "").lower() in ("true", "1", "yes")
verbose = _getenv_bool("VERBOSE", False)
setup_logging(verbose=verbose)
trace_size = os.environ.get("TRACE_CROP_SIZE", "").strip()
@@ -232,8 +253,8 @@ def main() -> None:
# 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")
dry_run = os.environ.get("DRY_RUN", "false").lower() in ("true", "1", "yes")
auto_mode = not sys.stdin.isatty() or _getenv_bool("AUTO_MODE", False)
dry_run = _getenv_bool("DRY_RUN", False)
if dry_run:
rprint("[bold yellow]DRY RUN — no images will be downloaded or uploaded[/bold yellow]")
winnow/config.py
+62 -26
View File
@@ -12,6 +12,48 @@ from rich.prompt import Prompt
_LEGACY_CONFIG_FILE = Path(".immich_config.json") # pre-v0.6: lived in process CWD, not on a volume
def _getenv_num(name: str, default, cast):
raw = os.getenv(name)
if raw is None:
return default
raw = raw.strip()
if not raw:
return default
try:
return cast(raw)
except ValueError:
logging.warning("%s=%r is not a valid %s — using default %s", name, raw, cast.__name__, default)
return default
def _getenv_int(name: str, default: int) -> int:
return _getenv_num(name, default, int)
def _getenv_float(name: str, default: float) -> float:
return _getenv_num(name, default, float)
def _getenv_optional_float(name: str) -> float | None:
"""Return float value of env var, or None if unset/empty. Warns and returns None on invalid."""
return _getenv_num(name, None, float)
def _getenv_optional_int(name: str) -> int | None:
"""Return int value of env var, or None if unset/empty. Warns and returns None on invalid."""
return _getenv_num(name, None, int)
def _getenv_bool(name: str, default: bool) -> bool:
raw = os.getenv(name)
if raw is None:
return default
raw = raw.strip()
if not raw:
return default
return raw.lower() in ("true", "1", "yes")
class _Config:
"""Singleton configuration with lazy loading via __getattr__.
@@ -84,28 +126,20 @@ class _Config:
self.IMMICH_URL = os.getenv("IMMICH_URL")
self.API_KEY = os.getenv("API_KEY")
self.OUTPUT_DIR = os.getenv("OUTPUT_DIR", "./frigate_train")
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", "3"))
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", "20"))
self.QUALITY_REPLACEMENT = os.getenv("QUALITY_REPLACEMENT", "true").lower() in ("true", "1", "yes")
_ceiling_env = os.getenv("FRIGATE_SCORE_CEILING", "").strip()
if _ceiling_env:
try:
self.FRIGATE_SCORE_CEILING = float(_ceiling_env)
except ValueError:
logging.warning("FRIGATE_SCORE_CEILING=%r is not a valid float — ignoring", _ceiling_env)
self.FRIGATE_SCORE_CEILING = None
else:
self.FRIGATE_SCORE_CEILING = None
self.ENABLE_FRIGATE_SCORES = os.getenv("ENABLE_FRIGATE_SCORES", "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")
self.ENABLE_CACHE = os.getenv("ENABLE_CACHE", "true").lower() in ("true", "1", "yes")
self.YEARS_FILTER = _getenv_int("YEARS_FILTER", 10)
self.MIN_FACE_WIDTH = _getenv_int("MIN_FACE_WIDTH", 90)
self.MIN_FACE_COUNT = _getenv_int("MIN_FACE_COUNT", 3)
self.MERGE_DUPLICATE_PEOPLE = _getenv_bool("MERGE_DUPLICATE_PEOPLE", False)
self.BLUR_THRESHOLD = _getenv_float("BLUR_THRESHOLD", 120.0)
self.MIN_CONFIDENCE = _getenv_float("MIN_CONFIDENCE", 0.7)
self.MAX_AUTO_IMAGES = _getenv_int("MAX_AUTO_IMAGES", 20)
self.QUALITY_REPLACEMENT = _getenv_bool("QUALITY_REPLACEMENT", True)
self.FRIGATE_SCORE_CEILING = _getenv_optional_float("FRIGATE_SCORE_CEILING")
self.ENABLE_FRIGATE_SCORES = _getenv_bool("ENABLE_FRIGATE_SCORES", True)
self.FACE_MARGIN = _getenv_float("FACE_MARGIN", 0.15)
self.USE_FULL_RESOLUTION = _getenv_bool("USE_FULL_RESOLUTION", True)
self.ENABLE_FACE_ALIGNMENT = _getenv_bool("ENABLE_FACE_ALIGNMENT", True)
self.ENABLE_CACHE = _getenv_bool("ENABLE_CACHE", True)
_data_dir = os.getenv("DATA_DIR")
_cache_dir_legacy = os.getenv("CACHE_DIR")
if _data_dir:
@@ -118,8 +152,8 @@ class _Config:
else:
self.DATA_DIR = "data"
# Fall back to config file only when the env var is genuinely absent (None).
# An explicitly empty env var (IMMICH_URL="") takes priority over the file.
# Fall back to config file when the env var is absent or blank — a blank
# IMMICH_URL= placeholder in .env should not override the config file.
# Prefer DATA_DIR/.immich_config.json (volume-safe in Docker) and fall back
# to the legacy CWD path so existing installations continue to work.
_data_cfg = Path(self.DATA_DIR) / ".immich_config.json"
@@ -132,10 +166,12 @@ class _Config:
_data_cfg,
)
config_file = _data_cfg if _data_cfg_exists else _LEGACY_CONFIG_FILE
if config_file.exists():
# _data_cfg_exists already confirmed the primary path — avoid re-stat.
# The short-circuit means the legacy path is stat'd at most once here.
if _data_cfg_exists or config_file.exists():
try:
data = json.loads(config_file.read_text())
if self.IMMICH_URL is None:
if not self.IMMICH_URL:
self.IMMICH_URL = data.get("IMMICH_URL")
if os.getenv("OUTPUT_DIR") is None:
self.OUTPUT_DIR = data.get("OUTPUT_DIR", self.OUTPUT_DIR)
winnow/diversity.py
+12 -2
View File
@@ -494,8 +494,13 @@ def _cluster_aware_selection(
auto_threshold = _compute_adaptive_threshold(emb_normed) if limit == "auto" else 0.0
target = Config.MAX_AUTO_IMAGES if limit == "auto" else limit
# Short-circuit: nothing to select
if limit != "auto" and target <= 0:
return []
# --- Stage 1: K-Medoids clustering ---
k = min(max(5, target // 4), max(1, n // 3), n) # e.g., 1-20 clusters
# Cap k at target so we never seed more cluster representatives than requested.
k = min(max(5, target // 4), max(1, n // 3), n, target) # e.g., 1-20 clusters
logger.debug("Clustering %s embeddings into %s groups (K-Medoids)...", n, k)
# Compute full cosine distance matrix
@@ -547,7 +552,12 @@ def _cluster_aware_selection(
hard_count = sum(1 for c in selected_conf if c < 0.85)
logger.info("Selection complete: %s images (%s hard examples with confidence < 0.85).", len(selected), hard_count)
return [candidates[i] for i in selected]
# Slice to target: the while loop enforces this for non-auto mode, but
# guard here too in case the medoid seed already exceeded target (small target).
result = [candidates[i] for i in selected]
if limit != "auto":
result = result[:target]
return result
# =============================================================================
winnow/embeddings.py
+3 -3
View File
@@ -18,6 +18,7 @@ import numpy as np
from PIL import Image
from .cache import get_cache
from .config import _getenv_bool
logger = logging.getLogger(__name__)
@@ -50,7 +51,7 @@ _insightface_loaded = False
def _is_force_cpu() -> bool:
"""Check if CPU mode is forced via environment variable."""
return os.getenv("FORCE_CPU", "").lower() in ("true", "1", "yes")
return _getenv_bool("FORCE_CPU", False)
def _preload_cuda_libs() -> None:
@@ -234,8 +235,7 @@ def get_embedding(
def _is_module_available(module_name: str) -> bool:
"""Check if a Python module is importable without importing it fully."""
try:
importlib.util.find_spec(module_name)
return True
return importlib.util.find_spec(module_name) is not None
except (ModuleNotFoundError, ValueError):
return False
winnow/executor.py
+132 -104
View File
@@ -6,6 +6,7 @@ import shutil
from io import BytesIO
from urllib.parse import quote
import PIL
import requests
from PIL import Image
from rich import print as rprint
@@ -13,6 +14,7 @@ from rich.progress import BarColumn, Progress, SpinnerColumn, TaskProgressColumn
from .config import Config, get_headers
from .frigate_api import (
_get_frigate_url,
delete_frigate_person_files,
get_all_frigate_person_files,
get_frigate_person_files,
@@ -22,7 +24,7 @@ from .frigate_api import (
from .image_processing import process_face_mode
from .immich_api import fetch_full_image
from .log_config import console
from .quality import assess_quality
from .quality import blur_score_from_image
from .reconcile import enrich_asset_with_face_data, reconcile_frigate_mappings
from .upload_tracker import (
get_lowest_quality_mapped_file,
@@ -42,9 +44,11 @@ def _safe_person_dir(output_dir: str, person_name: str) -> str:
"""Return the output subdirectory for a person, raising ValueError on path traversal.
os.path.join silently discards output_dir when person_name is absolute,
and '../..' sequences resolve outside the tree. Both are rejected here.
Symlinks on the raw (unresolved) path are also rejected — checking after
realpath would be too late because realpath follows the link first.
and '../..' sequences resolve outside the tree. Both are rejected by the
realpath+startswith guard, which is the load-bearing security check.
The islink check below provides an earlier, cleaner error message for the
symlink sub-case; it is redundant with (not a replacement for) the
realpath+startswith traversal check.
"""
raw = os.path.join(output_dir, person_name)
if os.path.islink(raw):
@@ -100,106 +104,117 @@ def execute_jobs(jobs: list[dict]) -> None:
job_task = progress.add_task(f"Processing {name}...", total=len(assets))
try:
person_dir = _safe_person_dir(Config.OUTPUT_DIR, name)
except ValueError as e:
logger.error(str(e))
continue
# Face crops are transient (uploaded then discarded); wipe before each run.
if os.path.isdir(person_dir):
shutil.rmtree(person_dir)
os.makedirs(person_dir, exist_ok=True)
# Track filename → asset_id, filename → confidence score, filename → crop dims
asset_map: dict[str, str] = {}
score_map: dict[str, float | None] = {}
dims_map: dict[str, tuple[int, int]] = {}
count = 0
for asset in assets:
try:
# Enrich the asset with face bounding box data from the Immich
# faces API (not included in search/metadata results).
asset = enrich_asset_with_face_data(asset, person)
# Skip download if detection confidence already disqualifies
# the asset — avoids fetching a large image we'll discard.
conf = asset.get("face_confidence")
if conf is not None and conf < Config.MIN_CONFIDENCE:
progress.console.print(
f"[yellow]Skipped {asset['id']}"
f" (detection confidence {conf:.2f} < {Config.MIN_CONFIDENCE})[/yellow]"
)
mark_rejected(asset["id"], person_name=name)
progress.advance(job_task)
progress.advance(overall_task)
continue
person_dir = _safe_person_dir(Config.OUTPUT_DIR, name)
except ValueError as e:
logger.error(str(e))
continue
# Face crops are transient (uploaded then discarded); wipe before each run.
# A symlink could appear here via a TOCTOU race after _safe_person_dir
# returned — writing through it would land crops outside output_dir.
if os.path.islink(person_dir):
logger.error("person_dir %s became a symlink after path check — skipping job", person_dir)
continue
try:
if os.path.isdir(person_dir):
shutil.rmtree(person_dir)
os.makedirs(person_dir, exist_ok=True)
except OSError as e:
logger.error("Failed to prepare output dir for %s: %s", name, e)
continue
# Use full-resolution for final output when configured
if use_full_res:
img = fetch_full_image(asset["id"])
else:
resp = requests.get(
f"{Config.IMMICH_URL}/api/assets/{asset['id']}/thumbnail?size=preview&format=JPEG",
headers=get_headers(),
timeout=30,
)
if resp.ok:
try:
img = Image.open(BytesIO(resp.content))
except Exception:
logger.warning("Invalid image data for asset %s", asset["id"])
img = None
else:
img = None
# Track filename → asset_id, filename → confidence score, filename → crop dims
asset_map: dict[str, str] = {}
score_map: dict[str, float | None] = {}
dims_map: dict[str, tuple[int, int]] = {}
if img is None:
progress.console.print(f"[red]Failed download {asset['id']}[/red]")
else:
saved = process_face_mode(
img, asset, person, person_dir, count, insightface_app=insightface_app
)
if saved:
filename = f"{count}.jpg"
asset_map[filename] = asset["id"]
score_map[filename] = asset.get("quality_score")
if isinstance(saved, tuple):
dims_map[filename] = saved
# Time-spread path: compute blur score from the downloaded
# image. Cap at 1440px so the scale matches the preview
# thumbnails the embedding path uses for scoring — Laplacian
# variance grows with resolution, making full-res and
# thumbnail scores incomparable if left uncapped.
if score_map[filename] is None:
try:
score_img = img.convert("RGB") if img.mode != "RGB" else img
if score_img.width > 1440 or score_img.height > 1440:
score_img = score_img.copy()
score_img.thumbnail((1440, 1440), Image.LANCZOS)
score_map[filename] = assess_quality(score_img).blur_score
except Exception as exc:
logger.debug("Quality score fallback for %s: %s", asset["id"], exc)
score_map[filename] = 0.0 # unknown quality — treat as lowest
count += 1
else:
count = 0
for asset in assets:
try:
# Enrich the asset with face bounding box data from the Immich
# faces API (not included in search/metadata results).
asset = enrich_asset_with_face_data(asset, person)
# Skip download if detection confidence already disqualifies
# the asset — avoids fetching a large image we'll discard.
conf = asset.get("face_confidence")
if conf is not None and conf < Config.MIN_CONFIDENCE:
progress.console.print(
f"[yellow]Skipped {asset['id']} (no usable face data)[/yellow]"
f"[yellow]Skipped {asset['id']}"
f" (detection confidence {conf:.2f} < {Config.MIN_CONFIDENCE})[/yellow]"
)
except Exception as e:
logger.error("Failed to process asset %s: %s", asset["id"], e)
mark_rejected(asset["id"], person_name=name)
progress.advance(job_task)
progress.advance(overall_task)
continue
progress.advance(job_task)
progress.advance(overall_task)
# Use full-resolution for final output when configured
if use_full_res:
img = fetch_full_image(asset["id"])
if img is None:
# Both original and preview fallback failed — mark rejected
# so this asset isn't retried on every future run.
mark_rejected(asset["id"], person_name=name)
else:
resp = requests.get(
f"{Config.IMMICH_URL}/api/assets/{asset['id']}/thumbnail?size=preview&format=JPEG",
headers=get_headers(),
timeout=30,
)
if resp.ok:
try:
img = Image.open(BytesIO(resp.content))
except PIL.UnidentifiedImageError:
# Pillow cannot identify the format — genuinely corrupt
# Immich thumbnail. Mark rejected so this asset isn't
# retried indefinitely. OSError/truncation errors are
# transient and intentionally not caught here.
logger.warning("Invalid image data for asset %s — marking rejected", asset["id"])
mark_rejected(asset["id"], person_name=name)
img = None
else:
img = None
# Store maps on the job so upload_to_frigate can use them
job["asset_map"] = asset_map
job["score_map"] = score_map
job["dims_map"] = dims_map
if img is None:
progress.console.print(f"[red]Failed download {asset['id']}[/red]")
else:
saved = process_face_mode(
img, asset, person, person_dir, count, insightface_app=insightface_app
)
if saved:
filename = f"{count}.jpg"
asset_map[filename] = asset["id"]
score_map[filename] = asset.get("quality_score")
if isinstance(saved, tuple):
dims_map[filename] = saved
# Time-spread path: compute blur score from the downloaded
# image. Capped at 1440px via blur_score_from_image() so the
# scale matches the preview thumbnails the embedding path uses
# — Laplacian variance grows with resolution, making full-res
# and thumbnail scores incomparable if left uncapped.
if score_map[filename] is None:
score_map[filename] = blur_score_from_image(img)
progress.remove_task(job_task)
count += 1
else:
progress.console.print(
f"[yellow]Skipped {asset['id']} (no usable face data)[/yellow]"
)
except Exception as e:
logger.error("Failed to process asset %s: %s", asset.get("id", "<unknown>"), e)
# Log how many images were actually saved vs selected
if count < len(assets):
logger.info("%s: saved %s/%s selected images", name, count, len(assets))
progress.advance(job_task)
progress.advance(overall_task)
# Store maps on the job so upload_to_frigate can use them
job["asset_map"] = asset_map
job["score_map"] = score_map
job["dims_map"] = dims_map
# Log how many images were actually saved vs selected
if count < len(assets):
logger.info("%s: saved %s/%s selected images", name, count, len(assets))
finally:
progress.remove_task(job_task)
def upload_to_frigate(jobs: list[dict]) -> None:
@@ -212,7 +227,7 @@ def upload_to_frigate(jobs: list[dict]) -> None:
rprint("[dim]No jobs to upload.[/dim]")
return
frigate_url = os.environ.get("FRIGATE_URL", "")
frigate_url = _get_frigate_url()
if not frigate_url:
rprint("[yellow]⚠️ FRIGATE_URL not set, skipping upload.[/yellow]")
return
@@ -489,13 +504,22 @@ def upload_to_frigate(jobs: list[dict]) -> None:
asset_id = asset_map.get(fname)
if asset_id:
mark_uploaded(
asset_id,
person_name=name,
score=score_map.get(fname),
crop_dims=dims_map.get(fname),
frigate_score=pre_fscore,
)
try:
mark_uploaded(
asset_id,
person_name=name,
score=score_map.get(fname),
crop_dims=dims_map.get(fname),
frigate_score=pre_fscore,
)
except Exception as tracker_exc:
# Upload to Frigate succeeded — don't retry on tracker
# failure or we'd upload a duplicate to Frigate.
logger.error(
"Tracker write failed for %s — upload succeeded"
" but asset may be re-selected next run: %s",
fname, tracker_exc,
)
if pre_fscore is not None:
person_has_fscores = True
actually_uploaded.append((fname, asset_id))
@@ -522,7 +546,11 @@ def upload_to_frigate(jobs: list[dict]) -> None:
progress.console.print(f" [dim]{error_detail}[/dim]")
else:
logger.debug("%s HTTP %s: %s", fname, resp.status_code, error_detail)
if resp.status_code == 400 and "face" in full_body.lower():
_is_permanent = (
(resp.status_code == 400 and "face" in full_body.lower())
or resp.status_code == 422
)
if _is_permanent:
asset_id = asset_map.get(fname)
if asset_id:
mark_rejected(asset_id, person_name=name)
winnow/frigate_api.py
+9 -4
View File
@@ -8,13 +8,18 @@ import requests
logger = logging.getLogger(__name__)
def _get_frigate_url() -> str:
"""Return normalized FRIGATE_URL with whitespace and trailing slash stripped, or '' if unset."""
return os.environ.get("FRIGATE_URL", "").strip().rstrip("/")
def get_frigate_version() -> str | None:
"""Fetch Frigate's version string from GET /api/version.
Returns the version string (e.g. "0.16.0-beta4") or None if FRIGATE_URL
is unset, the endpoint is unreachable, or the response is not parseable.
"""
frigate_url = os.environ.get("FRIGATE_URL", "").rstrip("/")
frigate_url = _get_frigate_url()
if not frigate_url:
return None
try:
@@ -28,7 +33,7 @@ def get_frigate_version() -> str | None:
def _get_faces_data() -> dict | None:
"""Fetch raw GET /api/faces response. Returns None if unavailable."""
frigate_url = os.environ.get("FRIGATE_URL", "").rstrip("/")
frigate_url = _get_frigate_url()
if not frigate_url:
return None
try:
@@ -113,7 +118,7 @@ def recognize_face(file_path: str) -> tuple[str | None, float] | None:
replace mean-comparison with nearest-neighbour distance across individual
training embeddings for accurate coverage detection.
"""
frigate_url = os.environ.get("FRIGATE_URL", "").rstrip("/")
frigate_url = _get_frigate_url()
if not frigate_url:
return None
try:
@@ -140,7 +145,7 @@ def delete_frigate_person_files(person_name: str, filenames: list[str]) -> bool:
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("/")
frigate_url = _get_frigate_url()
if not frigate_url or not filenames:
return False
from urllib.parse import quote
winnow/image_processing.py
+10 -1
View File
@@ -15,7 +15,16 @@ logger = logging.getLogger(__name__)
def _save_jpeg(img: Image.Image, path: str) -> None:
if img.mode != "RGB":
img = img.convert("RGB")
img.save(path, format="JPEG")
tmp = path + ".tmp"
try:
img.save(tmp, format="JPEG")
os.replace(tmp, path)
except Exception:
try:
os.remove(tmp)
except OSError:
pass
raise
def align_face(img: Image.Image, landmarks: list[list[float]] | np.ndarray) -> Image.Image | None:
winnow/immich_api.py
+25 -10
View File
@@ -86,9 +86,12 @@ def merge_people(survivor_id: str, merge_ids: list[str]) -> bool:
def fetch_all_assets(person: dict) -> tuple[list[dict], int]:
"""Fetch all assets for a person with pagination.
Returns (assets, raw_total) where assets is the list of valid dict items
and raw_total is the total item count seen before non-dict filtering.
raw_total may exceed len(assets) if Immich returned non-dict items.
Returns (assets, total_raw) where assets is the list of valid dict items
and total_raw is the raw item count across pages that had at least one valid
dict. All-garbage pages (every item non-dict) stop pagination and are not
counted. total_raw is a lower bound in two cases: a network error interrupts
pagination (a warning is logged), or an all-garbage page terminates it early
(a warning is logged and later pages are not fetched).
"""
name = person.get("name", "Unknown")
person_id = person.get("id")
@@ -101,7 +104,7 @@ def fetch_all_assets(person: dict) -> tuple[list[dict], int]:
logger.debug("Fetching assets for %s...", name)
assets: list[dict] = []
total_raw = 0 # items seen across all pages before non-dict filtering
total_raw = 0 # raw item count across pages that yielded at least one valid dict
for page in range(1, MAX_PAGES + 1):
try:
resp = requests.post(
@@ -122,7 +125,6 @@ def fetch_all_assets(person: dict) -> tuple[list[dict], int]:
page_assets = page_assets.get("items", [])
page_count = len(page_assets) # raw count for termination check before filtering
total_raw += page_count
# Single pass: partition valid assets from unexpected non-dict items
valid_assets, skipped_count = [], 0
@@ -142,6 +144,11 @@ def fetch_all_assets(person: dict) -> tuple[list[dict], int]:
)
break
# Count page_count (not just valid items) so that non-dict items from a
# transient schema issue on a mixed page don't cause MIN_FACE_COUNT to
# skip a real person. Pages where every item is a non-dict are excluded —
# they indicate a structural problem and break above without contributing.
total_raw += page_count
assets.extend(valid_assets)
logger.debug("Fetched page %s, total: %s", page, len(assets))
@@ -150,6 +157,11 @@ def fetch_all_assets(person: dict) -> tuple[list[dict], int]:
except (requests.RequestException, ValueError) as e:
logger.error("Exception fetching assets for %s (page %s): %s", name, page, e)
if page > 1:
logger.warning(
"%s: pagination interrupted at page %s — total_raw=%s may undercount actual assets",
name, page, total_raw,
)
break
return assets, total_raw
@@ -184,14 +196,14 @@ def fetch_face_data(asset_id: str, person_id: str | None = None) -> FaceData | N
if not isinstance(faces, list) or not faces:
return None
# Match the target person if specified
# Match the target person if specified; never fall back to a different person's face.
face = None
if person_id:
face = next(
(f for f in faces if isinstance(f, dict) and (f.get("person") or {}).get("id") == person_id),
None,
)
if face is None:
else:
face = faces[0] if isinstance(faces[0], dict) else None
if face is None:
return None
@@ -256,8 +268,11 @@ def fetch_full_image(asset_id: str, timeout: int = 60) -> Image.Image | None:
def filter_recent_assets(assets: list[dict], years: int | None = None) -> list[dict]:
"""Filter assets to keep only those from the last N years."""
years = years or Config.YEARS_FILTER
"""Filter assets to keep only those from the last N years. Pass years=0 to include all."""
if years is None:
years = Config.YEARS_FILTER
if not years:
return list(assets)
cutoff = datetime.now(timezone.utc) - timedelta(days=365 * years)
logger.debug("Filtering assets older than %s years (%s)", years, cutoff)
@@ -265,7 +280,7 @@ def filter_recent_assets(assets: list[dict], years: int | None = None) -> list[d
recent, skipped = [], 0
for asset in assets:
created_at_str = asset.get("fileCreatedAt")
if not created_at_str:
if not isinstance(created_at_str, str) or not created_at_str:
continue
try:
winnow/jobs.py
+11 -24
View File
@@ -8,7 +8,7 @@ from rich.progress import BarColumn, Progress, SpinnerColumn, TaskProgressColumn
from rich.prompt import Confirm, IntPrompt, Prompt
from rich.table import Table
from .config import Config
from .config import Config, _getenv_bool, _getenv_int, _getenv_optional_int
from .diversity import select_diverse_assets
from .embeddings import is_embedding_available, load_embedding_model
from .frigate_api import get_frigate_face_counts
@@ -65,25 +65,12 @@ def _get_strategy_choice(has_embedding: bool) -> tuple[int | str, str]:
def _resolve_strategy(strategy: str, has_embedding: bool) -> tuple[int | str, str]:
"""Resolve env var strategy to (limit, selection_mode) without prompts."""
custom_limit = os.environ.get("LIMIT", "").strip()
if not has_embedding:
if custom_limit:
try:
limit = int(custom_limit)
except ValueError:
logger.warning("LIMIT=%r is not a valid integer — using default 30", custom_limit)
limit = 30
else:
limit = 30
return limit, "time"
return _getenv_int("LIMIT", 30), "time"
if custom_limit:
try:
return int(custom_limit), "smart"
except ValueError:
logger.warning("LIMIT=%r is not a valid integer — using adaptive strategy", custom_limit)
# fall through to strategy_map
custom_limit = _getenv_optional_int("LIMIT")
if custom_limit is not None:
return custom_limit, "smart"
strategy_map = {
"adaptive": ("auto", "smart"),
@@ -168,7 +155,7 @@ def _configure_person(person: dict, people: list[dict]) -> dict | None:
rprint(f" Found [bold]{total_raw}[/bold] total, [bold]{len(recent_assets)}[/bold] in range ({years} years).")
# Ask before strategy so the post-dedup count can inform the choice
retry_env = os.environ.get("RETRY_REJECTED", "false").lower() in ("true", "1", "yes")
retry_env = _getenv_bool("RETRY_REJECTED", False)
retry_rejected = Confirm.ask("Include previously rejected images?", default=retry_env)
before_dedup = len(recent_assets)
@@ -242,8 +229,8 @@ def auto_configure(people: list[dict]) -> list[dict]:
return []
strategy = os.environ.get("STRATEGY", "auto")
skip = os.environ.get("SKIP_PEOPLE", "").split(",") if os.environ.get("SKIP_PEOPLE") else []
only = os.environ.get("ONLY_PEOPLE", "").split(",") if os.environ.get("ONLY_PEOPLE") else []
skip = [s.strip() for s in os.environ.get("SKIP_PEOPLE", "").split(",") if s.strip()]
only = [s.strip() for s in os.environ.get("ONLY_PEOPLE", "").split(",") if s.strip()]
if only:
valid_people = [p for p in valid_people if p["name"] in only]
@@ -268,8 +255,8 @@ def auto_configure(people: list[dict]) -> list[dict]:
rprint(f" {name}: {total_raw} total, {len(recent_assets)} recent")
# MIN_FACE_COUNT guard: skip people with too few Immich assets.
# Uses total_raw (pre-filter count) so non-dict items from a transient
# Immich schema issue don't cause a person to be skipped incorrectly.
# Uses total_raw so that non-dict items from a transient Immich schema
# issue on a mixed page don't shrink the count below the threshold.
# Done here (after fetch) rather than upfront because Immich v2.7.5+
# dropped assetCount from the /api/people response.
if min_face_count > 0 and total_raw < min_face_count:
@@ -317,7 +304,7 @@ def auto_configure(people: list[dict]) -> list[dict]:
if selection_mode == "skip":
continue
retry_rejected = os.environ.get("RETRY_REJECTED", "false").lower() in ("true", "1", "yes")
retry_rejected = _getenv_bool("RETRY_REJECTED", False)
before_dedup = len(recent_assets)
new_asset_ids = set(filter_already_uploaded([a["id"] for a in recent_assets], retry_rejected=retry_rejected))
recent_assets = [a for a in recent_assets if a["id"] in new_asset_ids]
winnow/quality.py
+20
View File
@@ -138,3 +138,23 @@ def assess_quality(
return QualityResult(passed=len(reasons) == 0, reasons=reasons, blur_score=blur_score)
def blur_score_from_image(img: Image.Image, max_dim: int = 1440) -> float:
"""Compute Laplacian-variance blur score, capped at max_dim px to normalise scale.
Caps resolution so full-res and thumbnail scores are comparable — Laplacian
variance grows with pixel count, making uncapped full-res scores much larger
than thumbnail scores for the same perceived sharpness.
Returns 0.0 on any error so callers can treat the result as lowest quality.
"""
try:
score_img = img.convert("RGB") if img.mode != "RGB" else img
if score_img.width > max_dim or score_img.height > max_dim:
score_img = score_img.copy()
score_img.thumbnail((max_dim, max_dim), Image.LANCZOS)
return float(assess_quality(score_img).blur_score)
except Exception as exc:
logger.debug("blur_score_from_image failed: %s", exc)
return 0.0
winnow/reconcile.py
+7 -5
View File
@@ -59,7 +59,7 @@ def reconcile_frigate_mappings(
if len(new_files) == target:
def _ts(fname: str) -> float:
try:
return float(fname.rsplit("_", 1)[-1].replace(".webp", ""))
return float(fname.rsplit("_", 1)[-1].rsplit(".", 1)[0])
except (ValueError, IndexError):
return 0.0
@@ -71,14 +71,15 @@ def reconcile_frigate_mappings(
)
mappings = {
frigate_file: asset_id
for (_, asset_id), frigate_file in zip(uploaded, sorted(new_files, key=_ts))
for (_, asset_id), frigate_file in zip(uploaded, sorted(new_files, key=lambda f: (_ts(f), f)))
if asset_id
}
record_frigate_files_batch(person_name, mappings)
elif len(new_files) > target:
logger.info(
logger.warning(
"%s: %s new Frigate files for %s uploads"
" (external upload detected) — skipping file mapping",
" (external upload detected) — skipping file mapping;"
" these files are permanently unmapped",
person_name,
len(new_files),
target,
@@ -86,7 +87,8 @@ def reconcile_frigate_mappings(
else:
logger.warning(
"%s: only %s of %s expected Frigate files"
" appeared after reconciliation — mapping skipped",
" appeared after reconciliation — mapping skipped;"
" these files are permanently unmapped",
person_name,
len(new_files),
target,
winnow/upload_tracker.py
+270 -378
View File
@@ -1,213 +1,146 @@
"""Persistent tracker for Immich asset IDs uploaded/rejected by Frigate.
"""Persistent tracker for Immich asset IDs already uploaded/rejected by Frigate.
Uses a local SQLite database (frigate_tracker.db) in DATA_DIR.
Two separate JSON files in DATA_DIR:
frigate_uploaded_ids.json — successfully uploaded assets
frigate_rejected_ids.json — assets Frigate rejected (e.g. no face detected)
Schema
------
tracked_assets — one row per (asset_id, status) pair
frigate_files — Frigate filename → Immich asset_id mapping
person_metadata — last-known Frigate training image count per person
Both are excluded from future candidate pools. To reset:
- All: delete both files
- One person: call reset_person("Name") or set RESET_PERSON=Name
- Rejects only: delete frigate_rejected_ids.json, or set RETRY_REJECTED=true
Migration
---------
On first open, if the old JSON files exist and the tables are empty, their
data is migrated automatically. The JSON files are then renamed to .json.bak.
by_person schema (frigate_uploaded_ids.json):
{
"asset_ids": ["immich-id-1", ...], # all assets we attempted to upload
"scores": {"immich-id-1": 450.3}, # Laplacian blur variance at upload time
"frigate_scores": {"immich-id-1": 0.87}, # Frigate recognition confidence (0-1) pre-upload
"frigate_files": {"PersonName-123.webp": "immich-id-1"}, # Frigate filename → asset ID
"crop_dims": {"immich-id-1": [640, 480]}, # crop pixel dimensions at upload time
"frigate_count": 42 # last known Frigate training image count
}
frigate_scores stores pre-upload recognize scores (0-1 sigmoid-mapped cosine
similarity). High score = the existing training set already covers this face
condition well. Low score = a gap — novel/diverse for the training set.
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
import logging
import os
import sqlite3
from pathlib import Path
from .frigate_api import delete_frigate_person_files
logger = logging.getLogger(__name__)
# Legacy JSON filenames (for migration)
_UPLOAD_JSON = "frigate_uploaded_ids.json"
_REJECT_JSON = "frigate_rejected_ids.json"
_DB_NAME = "frigate_tracker.db"
UPLOAD_TRACKER_FILE = "frigate_uploaded_ids.json"
REJECT_TRACKER_FILE = "frigate_rejected_ids.json"
_DDL = """
CREATE TABLE IF NOT EXISTS tracked_assets (
asset_id TEXT NOT NULL,
person_name TEXT,
status TEXT NOT NULL CHECK(status IN ('uploaded', 'rejected')),
blur_score REAL,
crop_width INTEGER,
crop_height INTEGER,
frigate_score REAL,
PRIMARY KEY (asset_id, status)
);
CREATE TABLE IF NOT EXISTS frigate_files (
frigate_filename TEXT PRIMARY KEY,
person_name TEXT NOT NULL,
asset_id TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS person_metadata (
person_name TEXT PRIMARY KEY,
frigate_count INTEGER
);
"""
# Module-level connection state — re-opened when DATA_DIR changes (test isolation)
_conn: sqlite3.Connection | None = None
_conn_path: str | None = None
# Write-through in-memory cache keyed by the resolved file path.
# Reduces per-call JSON reads from O(calls) to O(1) after the first load.
# Keyed by full path so tests with isolated tmp dirs never share entries.
_cache: dict[str, dict] = {}
def _get_conn() -> sqlite3.Connection:
"""Return (or create) the module-level SQLite connection.
Re-opens the connection when Config.DATA_DIR has changed — this provides
test isolation when the isolated_cache fixture sets a new tmp directory and
calls _Config.reset().
"""
global _conn, _conn_path
from .config import Config
data_dir = Config.DATA_DIR
db_path = str(Path(data_dir) / _DB_NAME)
if _conn is not None and _conn_path != db_path:
try:
_conn.close()
except Exception as e:
logger.debug("Failed to close previous SQLite connection: %s", e)
_conn = None
if _conn is None:
Path(data_dir).mkdir(parents=True, exist_ok=True)
_conn = sqlite3.connect(db_path, check_same_thread=False)
_conn.row_factory = sqlite3.Row
_conn.execute("PRAGMA journal_mode=WAL")
_conn.execute("PRAGMA foreign_keys=ON")
_conn.executescript(_DDL)
_conn.commit()
_conn_path = db_path
_maybe_migrate(data_dir, _conn)
return _conn
# ---------------------------------------------------------------------------
# JSON → SQLite migration
# ---------------------------------------------------------------------------
def _maybe_migrate(data_dir: str, conn: sqlite3.Connection) -> None:
"""If the old JSON files exist and DB is empty, migrate and rename them."""
base = Path(data_dir)
upload_json = base / _UPLOAD_JSON
reject_json = base / _REJECT_JSON
if not upload_json.exists() and not reject_json.exists():
return
# No row-count guard here: INSERT OR IGNORE makes migration idempotent, so it
# is safe to re-run if a previous attempt renamed one file but not the other
# (e.g. a PermissionError on the second rename would have left the first file's
# data committed but the second file un-renamed and un-migrated).
logger.info("Migrating JSON tracker files to SQLite in %s", data_dir)
def _tracker_path(filename: str) -> Path:
try:
with conn:
if upload_json.exists():
_migrate_json_data(conn, json.loads(upload_json.read_text()), "uploaded")
if reject_json.exists():
_migrate_json_data(conn, json.loads(reject_json.read_text()), "rejected")
except Exception as exc:
logger.warning("JSON migration failed, will retry next run: %s", exc)
return
# Rename each file independently so a failure on one does not prevent the
# other from being marked complete on this run.
for json_path in (upload_json, reject_json):
if json_path.exists():
try:
json_path.rename(json_path.with_suffix(".json.bak"))
except OSError as exc:
logger.warning("Could not rename %s after migration: %s", json_path, exc)
logger.info("JSON → SQLite migration complete")
from .config import Config
return Path(Config.DATA_DIR) / filename
except (ImportError, AttributeError):
return Path(filename)
def _migrate_json_data(conn: sqlite3.Connection, data: dict, status: str) -> None:
"""Insert one JSON tracker file's data into SQLite tables."""
flat_key = "uploaded_asset_ids" if status == "uploaded" else "rejected_asset_ids"
flat_ids: set[str] = set(data.get(flat_key, []))
person_covered: set[str] = set()
for person_name, raw_entry in data.get("by_person", {}).items():
if isinstance(raw_entry, list):
entry: dict = {"asset_ids": raw_entry, "scores": {}, "frigate_scores": {},
"frigate_files": {}, "crop_dims": {}}
else:
entry = {
"asset_ids": raw_entry.get("asset_ids", []),
"scores": raw_entry.get("scores", {}),
"frigate_scores": raw_entry.get("frigate_scores", {}),
"frigate_files": raw_entry.get("frigate_files", {}),
"crop_dims": raw_entry.get("crop_dims", {}),
"frigate_count": raw_entry.get("frigate_count"),
}
for asset_id in entry["asset_ids"]:
person_covered.add(asset_id)
dims = entry.get("crop_dims", {}).get(asset_id)
conn.execute(
"""INSERT OR IGNORE INTO tracked_assets
(asset_id, person_name, status, blur_score,
crop_width, crop_height, frigate_score)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(
asset_id,
person_name,
status,
entry.get("scores", {}).get(asset_id),
dims[0] if dims else None,
dims[1] if dims else None,
entry.get("frigate_scores", {}).get(asset_id) if status == "uploaded" else None,
),
)
if status == "uploaded":
for ff, aid in entry.get("frigate_files", {}).items():
conn.execute(
"INSERT OR IGNORE INTO frigate_files (frigate_filename, person_name, asset_id) VALUES (?, ?, ?)",
(ff, person_name, aid),
)
fc = entry.get("frigate_count")
if fc is not None:
conn.execute(
"INSERT OR REPLACE INTO person_metadata (person_name, frigate_count) VALUES (?, ?)",
(person_name, fc),
)
# Flat IDs not covered by any by_person entry → insert with NULL person
for asset_id in flat_ids - person_covered:
conn.execute(
"INSERT OR IGNORE INTO tracked_assets (asset_id, person_name, status) VALUES (?, NULL, ?)",
(asset_id, status),
)
def _load(filename: str) -> dict:
path = _tracker_path(filename)
key = str(path)
if key in _cache:
return _cache[key]
data: dict = {}
if path.exists():
try:
with open(path) as f:
data = json.load(f)
except (json.JSONDecodeError, OSError) as e:
logger.warning(f"Could not load tracker {filename}: {e}")
_cache[key] = data
return data
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def _save(filename: str, data: dict) -> None:
path = _tracker_path(filename)
_cache[str(path)] = data # keep cache consistent with what we write
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w") as f:
json.dump(data, f, indent=2)
def _flat_key(filename: str) -> str:
return "uploaded_asset_ids" if "uploaded" in filename else "rejected_asset_ids"
def _load_flat(filename: str) -> set[str]:
return set(_load(filename).get(_flat_key(filename), []))
def _get_ids(entry: list | dict) -> list[str]:
"""Extract asset_ids from either the old list format or the new dict format."""
if isinstance(entry, list):
return entry
return entry.get("asset_ids", [])
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": {}, "frigate_scores": {}, "frigate_files": {}, "crop_dims": {}}
entry.setdefault("asset_ids", [])
entry.setdefault("scores", {})
entry.setdefault("frigate_scores", {})
entry.setdefault("frigate_files", {})
entry.setdefault("crop_dims", {})
return entry
def _mark(
filename: str,
asset_id: str,
person_name: str | None,
score: float | None = None,
crop_dims: tuple[int, int] | None = None,
frigate_score: float | None = None,
) -> None:
data = _load(filename)
flat_key = _flat_key(filename)
flat = set(data.get(flat_key, []))
flat.add(asset_id)
data[flat_key] = sorted(flat)
if person_name:
by_person = data.setdefault("by_person", {})
entry = _migrate_entry(by_person.get(person_name, {}))
ids = set(entry["asset_ids"])
ids.add(asset_id)
entry["asset_ids"] = sorted(ids)
if score is not None:
entry["scores"][asset_id] = round(score, 4)
if crop_dims is not None:
entry["crop_dims"][asset_id] = [crop_dims[0], crop_dims[1]]
if frigate_score is not None:
entry["frigate_scores"][asset_id] = round(frigate_score, 4)
by_person[person_name] = entry
_save(filename, data)
# ── Public API ────────────────────────────────────────────────────────────────
def load_uploaded_ids() -> set[str]:
conn = _get_conn()
rows = conn.execute("SELECT asset_id FROM tracked_assets WHERE status='uploaded'").fetchall()
return {r[0] for r in rows}
return _load_flat(UPLOAD_TRACKER_FILE)
def load_rejected_ids() -> set[str]:
conn = _get_conn()
rows = conn.execute("SELECT asset_id FROM tracked_assets WHERE status='rejected'").fetchall()
return {r[0] for r in rows}
return _load_flat(REJECT_TRACKER_FILE)
def mark_uploaded(
@@ -217,264 +150,223 @@ def mark_uploaded(
crop_dims: tuple[int, int] | None = None,
frigate_score: float | None = None,
) -> None:
conn = _get_conn()
with conn:
conn.execute(
"""INSERT OR REPLACE INTO tracked_assets
(asset_id, person_name, status, blur_score, crop_width, crop_height, frigate_score)
VALUES (?, ?, 'uploaded', ?, ?, ?, ?)""",
(
asset_id,
person_name,
round(score, 4) if score is not None else None,
crop_dims[0] if crop_dims else None,
crop_dims[1] if crop_dims else None,
round(frigate_score, 4) if frigate_score is not None else None,
),
)
logger.debug("Marked %s as uploaded (%s)", asset_id, person_name)
_mark(UPLOAD_TRACKER_FILE, asset_id, person_name, score=score, crop_dims=crop_dims, frigate_score=frigate_score)
logger.debug(f"Marked {asset_id} as uploaded ({person_name})")
def mark_rejected(asset_id: str, person_name: str | None = None) -> None:
conn = _get_conn()
with conn:
conn.execute(
"INSERT OR IGNORE INTO tracked_assets (asset_id, person_name, status) VALUES (?, ?, 'rejected')",
(asset_id, person_name),
)
logger.debug("Marked %s as rejected (%s)", asset_id, person_name)
_mark(REJECT_TRACKER_FILE, asset_id, person_name)
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 record_frigate_files_batch(person_name: str, mappings: dict[str, str]) -> None:
"""Record multiple Frigate filename → asset_id mappings in a single transaction.
All rows are written atomically — either every mapping is persisted or none
are (SQLite rolls back on error), so there is no risk of partial failure.
"""
"""Record multiple Frigate filename → asset_id mappings in a single load/save."""
if not mappings:
return
conn = _get_conn()
with conn:
conn.executemany(
"INSERT OR REPLACE INTO frigate_files (frigate_filename, person_name, asset_id) VALUES (?, ?, ?)",
[(ff, person_name, aid) for ff, aid in mappings.items()],
)
logger.debug("Batch-mapped %s Frigate file(s) for %s", len(mappings), person_name)
data = _load(UPLOAD_TRACKER_FILE)
by_person = data.setdefault("by_person", {})
entry = _migrate_entry(by_person.get(person_name, {}))
entry["frigate_files"].update(mappings)
by_person[person_name] = entry
_save(UPLOAD_TRACKER_FILE, data)
logger.debug(f"Batch-mapped {len(mappings)} Frigate file(s) for {person_name}")
def remove_frigate_file(person_name: str, frigate_filename: str) -> None:
"""Remove a Frigate filename mapping and clear its asset's frigate_score.
"""Remove a Frigate filename from the mapping after it has been deleted.
Does NOT unmark the source asset_id — the deletion was deliberate.
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.
"""
conn = _get_conn()
with conn:
row = conn.execute(
"SELECT asset_id FROM frigate_files WHERE frigate_filename=? AND person_name=?",
(frigate_filename, person_name),
).fetchone()
conn.execute(
"DELETE FROM frigate_files WHERE frigate_filename=? AND person_name=?",
(frigate_filename, person_name),
)
if row:
conn.execute(
"UPDATE tracked_assets SET frigate_score=NULL WHERE asset_id=? AND person_name=?",
(row["asset_id"], person_name),
)
logger.debug("Removed Frigate file mapping %s (%s)", frigate_filename, person_name)
data = _load(UPLOAD_TRACKER_FILE)
by_person = data.get("by_person", {})
entry = _migrate_entry(by_person.get(person_name, {}))
asset_id = entry["frigate_files"].pop(frigate_filename, None)
if asset_id:
entry["frigate_scores"].pop(asset_id, None)
by_person[person_name] = entry
_save(UPLOAD_TRACKER_FILE, data)
logger.debug(f"Removed Frigate file mapping {frigate_filename} ({person_name})")
def get_tracked_frigate_file_count(person_name: str) -> int:
"""Return the number of Frigate training files winnow has mapped for this person."""
conn = _get_conn()
row = conn.execute(
"SELECT COUNT(*) FROM frigate_files WHERE person_name=?", (person_name,)
).fetchone()
return row[0]
"""Return the number of Frigate training files winnow has mapped for this person.
Used as the cap baseline so that manually-added Frigate files do not
consume slots from winnow's managed quota.
"""
data = _load(UPLOAD_TRACKER_FILE)
entry = _migrate_entry(data.get("by_person", {}).get(person_name, {}))
return len(entry["frigate_files"])
def get_tracked_frigate_filenames(person_name: str) -> set[str]:
"""Return the set of Frigate filenames currently mapped for a person."""
conn = _get_conn()
rows = conn.execute(
"SELECT frigate_filename FROM frigate_files WHERE person_name=?", (person_name,)
).fetchall()
return {r[0] for r in rows}
"""Return the set of Frigate filenames currently mapped in the tracker for a person.
Used as a pre-upload baseline when the Frigate GET API is unreachable at
upload start, so reconciliation can still identify newly uploaded files.
"""
data = _load(UPLOAD_TRACKER_FILE)
entry = _migrate_entry(data.get("by_person", {}).get(person_name, {}))
return set(entry["frigate_files"].keys())
def has_frigate_scores(person_name: str) -> bool:
"""Return True if any mapped file for this person has a stored Frigate recognition score."""
conn = _get_conn()
row = conn.execute(
"""SELECT COUNT(*) FROM frigate_files ff
JOIN tracked_assets ta ON ta.asset_id=ff.asset_id AND ta.person_name=ff.person_name
WHERE ff.person_name=? AND ta.frigate_score IS NOT NULL""",
(person_name,),
).fetchone()
return row[0] > 0
# Guard against SQL injection from callers passing dynamic column names.
# Only these two columns exist and are safe to interpolate into queries.
_VALID_SCORE_COLS = frozenset({"blur_score", "frigate_score"})
data = _load(UPLOAD_TRACKER_FILE)
entry = _migrate_entry(data.get("by_person", {}).get(person_name, {}))
frigate_files = entry.get("frigate_files", {})
frigate_scores = entry.get("frigate_scores", {})
return any(asset_id in frigate_scores for asset_id in frigate_files.values())
def _pick_mapped_file(
person_name: str, score_col: str, *, highest: bool, exclude: set[str] | None = None
person_name: str, score_key: str, *, highest: bool, exclude: set[str] | None = None
) -> tuple[str, str, float] | None:
if score_col not in _VALID_SCORE_COLS:
raise ValueError(f"Invalid score column: {score_col!r}")
conn = _get_conn()
order = "DESC" if highest else "ASC"
rows = conn.execute(
f"""SELECT ff.frigate_filename, ff.asset_id, ta.{score_col}
FROM frigate_files ff
JOIN tracked_assets ta ON ta.asset_id=ff.asset_id AND ta.person_name=ff.person_name
WHERE ff.person_name=? AND ta.{score_col} IS NOT NULL
ORDER BY ta.{score_col} {order}""",
(person_name,),
).fetchall()
for row in rows:
if exclude is None or row[0] not in exclude:
return (row[0], row[1], row[2])
return None
data = _load(UPLOAD_TRACKER_FILE)
entry = _migrate_entry(data.get("by_person", {}).get(person_name, {}))
scores = entry.get(score_key, {})
candidates = [
(ff, asset_id, scores[asset_id])
for ff, asset_id in entry.get("frigate_files", {}).items()
if (exclude is None or ff not in exclude) and asset_id in scores
]
if not candidates:
return None
return max(candidates, key=lambda x: x[2]) if highest else min(candidates, key=lambda x: x[2])
def get_lowest_quality_mapped_file(
person_name: str, exclude: set[str] | None = None
) -> tuple[str, str, float] | None:
"""Return (frigate_filename, asset_id, score) for the mapped file with the lowest blur score."""
return _pick_mapped_file(person_name, "blur_score", highest=False, exclude=exclude)
"""Return (frigate_filename, asset_id, score) for the mapped file with the lowest
blur score, or None if no mapped files with known scores exist.
Used for quality replacement when no Frigate scores are available.
Pass `exclude` to skip files that failed to delete this run.
"""
return _pick_mapped_file(person_name, "scores", highest=False, exclude=exclude)
def get_most_redundant_mapped_file(
person_name: str, exclude: set[str] | None = None
) -> tuple[str, str, float] | None:
"""Return (frigate_filename, asset_id, score) for the mapped file with the highest Frigate score."""
return _pick_mapped_file(person_name, "frigate_score", highest=True, exclude=exclude)
"""Return (frigate_filename, asset_id, score) for the mapped file with the highest
Frigate recognition score, or None if no mapped files with Frigate scores exist.
def get_frigate_filename_for_asset(person_name: str, asset_id: str) -> str | None:
"""Return the Frigate training filename mapped to this asset ID, or None."""
conn = _get_conn()
row = conn.execute(
"SELECT frigate_filename FROM frigate_files WHERE person_name=? AND asset_id=?",
(person_name, asset_id),
).fetchone()
return row[0] if row else None
High Frigate score = the training set already covers this face condition well
= the most redundant file and therefore the best replacement target.
Pass `exclude` to skip files that failed to delete this run.
"""
return _pick_mapped_file(person_name, "frigate_scores", highest=True, exclude=exclude)
def find_by_crop_dimension(size: int) -> list[dict]:
"""Return all tracked crops whose width or height matches `size` pixels.
Returns a list of dicts: {person, asset_id, width, height, blur_score, frigate_score, frigate_filename}.
Returns a list of dicts: {person, asset_id, width, height, blur_score, frigate_filename}.
frigate_filename is None when the Frigate mapping was lost to a reconciliation race.
"""
conn = _get_conn()
rows = conn.execute(
"""SELECT ta.person_name, ta.asset_id, ta.crop_width, ta.crop_height,
ta.blur_score, ta.frigate_score, ff.frigate_filename
FROM tracked_assets ta
LEFT JOIN frigate_files ff ON ff.asset_id=ta.asset_id AND ff.person_name=ta.person_name
WHERE ta.status='uploaded' AND (ta.crop_width=? OR ta.crop_height=?)""",
(size, size),
).fetchall()
return [
{
"person": r["person_name"],
"asset_id": r["asset_id"],
"width": r["crop_width"],
"height": r["crop_height"],
"blur_score": r["blur_score"],
"frigate_score": r["frigate_score"],
"frigate_filename": r["frigate_filename"],
}
for r in rows
]
data = _load(UPLOAD_TRACKER_FILE)
results = []
for person_name, raw_entry in data.get("by_person", {}).items():
entry = _migrate_entry(raw_entry)
scores = entry.get("scores", {})
frigate_files = entry.get("frigate_files", {})
asset_to_frigate = {v: k for k, v in frigate_files.items()}
frigate_scores = entry.get("frigate_scores", {})
for asset_id, dims in entry.get("crop_dims", {}).items():
w, h = dims[0], dims[1]
if w == size or h == size:
results.append({
"person": person_name,
"asset_id": asset_id,
"width": w,
"height": h,
"blur_score": scores.get(asset_id),
"frigate_score": frigate_scores.get(asset_id),
"frigate_filename": asset_to_frigate.get(asset_id),
})
return results
def update_frigate_count(person_name: str, count: int) -> None:
"""Record Frigate's authoritative training image count for a person."""
conn = _get_conn()
with conn:
conn.execute(
"INSERT OR REPLACE INTO person_metadata (person_name, frigate_count) VALUES (?, ?)",
(person_name, count),
)
data = _load(UPLOAD_TRACKER_FILE)
by_person = data.setdefault("by_person", {})
entry = _migrate_entry(by_person.get(person_name, {}))
entry["frigate_count"] = count
by_person[person_name] = entry
_save(UPLOAD_TRACKER_FILE, data)
def reset_person(person_name: str) -> None:
"""Remove all uploaded and rejected records for a given person.
Also deletes winnow-managed Frigate training files so the next run starts
clean rather than uploading on top of orphaned files.
clean rather than uploading on top of orphaned files. Manually-added Frigate
files (not in frigate_files) are never touched. Proceeds with tracker reset
even if Frigate is unreachable.
"""
conn = _get_conn()
# Collect Frigate filenames before deleting
frigate_filenames = list(get_tracked_frigate_filenames(person_name))
upload_data = _load(UPLOAD_TRACKER_FILE)
entry = _migrate_entry(upload_data.get("by_person", {}).get(person_name, {}))
frigate_filenames = list(entry.get("frigate_files", {}).keys())
if frigate_filenames:
if not os.environ.get("FRIGATE_URL", "").strip():
logger.info("FRIGATE_URL not set — skipping Frigate file deletion for %s", person_name)
logger.info(f"FRIGATE_URL not set — skipping Frigate file deletion for {person_name}")
elif delete_frigate_person_files(person_name, frigate_filenames):
logger.info("Deleted %s Frigate file(s) for %s", len(frigate_filenames), person_name)
logger.info(f"Deleted {len(frigate_filenames)} Frigate file(s) for {person_name}")
else:
logger.warning(
"Could not delete Frigate files for %s — tracker reset proceeding anyway", person_name
)
logger.warning(f"Could not delete Frigate files for {person_name} — tracker reset proceeding anyway")
with conn:
conn.execute("DELETE FROM frigate_files WHERE person_name=?", (person_name,))
conn.execute("DELETE FROM tracked_assets WHERE person_name=?", (person_name,))
conn.execute("DELETE FROM person_metadata WHERE person_name=?", (person_name,))
logger.info("Reset tracking data for %s", person_name)
changed = False
tracker_files = ((UPLOAD_TRACKER_FILE, upload_data), (REJECT_TRACKER_FILE, _load(REJECT_TRACKER_FILE)))
for filename, data in tracker_files:
flat_key = _flat_key(filename)
by_person = data.get("by_person", {})
tracker_entry = by_person.pop(person_name, None)
if tracker_entry is not None:
person_ids = set(_get_ids(tracker_entry))
flat = set(data.get(flat_key, [])) - person_ids
data[flat_key] = sorted(flat)
data["by_person"] = by_person
_save(filename, data)
changed = True
if changed:
logger.info(f"Reset tracking data for {person_name}")
else:
logger.debug(f"reset_person: no tracking data found for {person_name}")
def get_person_summary() -> dict[str, dict]:
"""Return {person_name: {uploaded, rejected, frigate_count, scores, frigate_files}} for display/capacity."""
conn = _get_conn()
# Counts per person per status
rows = conn.execute(
"""SELECT person_name, status, COUNT(*) AS cnt
FROM tracked_assets WHERE person_name IS NOT NULL
GROUP BY person_name, status"""
).fetchall()
def _entry(summary: dict, name: str) -> dict:
return summary.setdefault(
name, {"uploaded": 0, "rejected": 0, "frigate_count": None, "scores": {}, "frigate_files": {}}
)
summary: dict[str, dict] = {}
for r in rows:
_entry(summary, r["person_name"])[r["status"]] = r["cnt"]
# Scores for uploaded assets
score_rows = conn.execute(
"""SELECT person_name, asset_id, blur_score
FROM tracked_assets
WHERE status='uploaded' AND person_name IS NOT NULL AND blur_score IS NOT NULL"""
).fetchall()
for r in score_rows:
_entry(summary, r["person_name"])["scores"][r["asset_id"]] = r["blur_score"]
# Frigate file mappings
ff_rows = conn.execute(
"SELECT person_name, frigate_filename, asset_id FROM frigate_files"
).fetchall()
for r in ff_rows:
_entry(summary, r["person_name"])["frigate_files"][r["frigate_filename"]] = r["asset_id"]
# Frigate counts
meta_rows = conn.execute(
"SELECT person_name, frigate_count FROM person_metadata"
).fetchall()
for r in meta_rows:
_entry(summary, r["person_name"])["frigate_count"] = r["frigate_count"]
return dict(sorted(summary.items()))
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)
result = {}
for name in sorted(names):
u_entry = uploaded_data.get(name, {})
r_entry = rejected_data.get(name, {})
result[name] = {
"uploaded": len(_get_ids(u_entry)),
"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
def filter_already_uploaded(
@@ -488,5 +380,5 @@ def filter_already_uploaded(
new_ids = [aid for aid in asset_ids if aid not in exclude]
skipped = len(asset_ids) - len(new_ids)
if skipped:
logger.info("Skipping %s assets already uploaded or rejected by Frigate", skipped)
logger.info(f"Skipping {skipped} assets already uploaded or rejected by Frigate")
return new_ids