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

release: v0.6.3

Browse Source
This commit is contained in:
flan
2026-06-16 18:13:42 +00:00
parent 00e378b375 cf7660595d
commit 3c0ef47fdc
19 changed files with 1011 additions and 919 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+52 -96
View File
@@ -7,153 +7,109 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
## [0.5.7] - 2026-06-15
## [0.6.3] - 2026-06-16
### 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.
- **`record_frigate_files_batch` no longer mutates the tracker cache before write**: the function shared the same cache-corruption-on-write-failure bug that was fixed in `remove_frigate_files_batch` in v0.6.1 — `data.setdefault("by_person", {})` mutated the cached dict in-place, so a disk-full or permission error left the in-memory cache ahead of the on-disk file. Now uses the same copy-before-mutate pattern (shallow copies of the top-level dict and `by_person` sub-dict) so a failed write leaves cache and disk in sync.
- **`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.
- **`tracker_ok` boolean flag replaced with try/else**: the intermediate boolean was a misleading placeholder — the `True` initial value suggested success before the operation ran. The control flow is now expressed directly with a try/except/else block.
- **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.
- **`LIMIT` env var guard simplified**: the two adjacent `if custom_limit is not None` checks in `_resolve_strategy` are collapsed into a single `if custom_limit is not None:` with nested branches, removing redundant evaluation.
- **`_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.2] - 2026-06-16
### 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`.
- **Flat `uploaded_asset_ids` / `rejected_asset_ids` lists dropped as primary storage**: asset IDs are now derived on read from `by_person` entries, which are the single source of truth. The legacy flat lists in existing tracker files are still read (union) so no assets become re-eligible after upgrading. New writes no longer maintain the flat lists. This removes the dual-representation sync hazard and paves the way for multi-instance support (per-instance `by_person` keying in a future release).
- **Reconciliation poll delays extracted**: `(1, 2, 4, 8)` back-off delays in `reconcile.py` extracted to `_RECONCILE_POLL_DELAYS` with an explanatory comment.
- **Tracker writes batched per person**: `mark_uploaded` calls inside the per-person upload loop are now accumulated in memory (`begin_batch`) and flushed in a single `os.replace` write at the end of each person's loop (`flush_batch`), reducing N tracker writes per person to 1. Benefits users on slow storage (NAS, SD card, spinning disks).
- **`_VALID_SCORE_COLS` comment**: explains that the frozenset is a SQL-injection guard for dynamic column interpolation, not a runtime filter.
- **`RESET_PERSON=*` is now O(1) disk writes**: replaced the per-person `reset_person` loop with `reset_all_people()`, which makes one Frigate API call per person for file deletion and then clears both tracker files in two writes. Previously it was O(P²) iterations and 2P writes.
- **`record_frigate_files_batch` docstring**: clarifies that all mappings are written atomically — no partial failure is possible.
- **`blur_score_from_image` inlines Laplacian computation**: replaced the `assess_quality()` call (which ran grayscale, exposure, and confidence checks whose results were discarded) with a direct `cv2.Laplacian` computation. The function is now self-contained and does not silently inherit future costs added to the full quality pipeline.
- **`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
## [0.6.1] - 2026-06-16
### 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.
- **Corrupt or truncated full-res thumbnails now marked rejected**: `OSError` (truncated file) is caught alongside `PIL.UnidentifiedImageError` in the thumbnail path so persistently bad assets are tombstoned instead of retried forever. Full-res download failures (`USE_FULL_RESOLUTION=true`) remain transient — not marked rejected — so a Immich blip doesn't permanently blacklist valid assets.
## [0.5.3] - 2026-06-14
- **Quality replacement mode no longer flips mid-loop**: `person_has_fscores` was re-evaluated after each file deletion, which could switch the remaining replacements from Frigate-score mode to blur-score mode if the deleted file was the last scored one. The mode is now fixed for the duration of the upload loop.
### Fixed
- **`reset_person` no longer removes shared asset IDs**: the flat `uploaded_asset_ids` list is now rebuilt from all remaining `by_person` entries rather than subtracting the reset person's IDs. Previously, resetting Alice could remove an asset ID that also appeared under Bob, making it re-eligible for upload.
- **`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.
- **`_save` cache updated only after successful write**: the in-memory tracker cache is now updated after `os.replace` succeeds rather than before. A disk-full or permission error no longer leaves the cache permanently ahead of the on-disk file.
- **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.
- **Stale Frigate file cleanup batched**: the per-file `remove_frigate_file` loop is replaced with a single `remove_frigate_files_batch` call, reducing N tracker writes to 1 when stale mappings are cleaned up.
- **`person["id"]` KeyError**: malformed Immich API responses missing the `id` field now log an error and skip that person instead of crashing the job.
- **`_migrate_entry` no longer mutates the cache through nested dict aliases**: all five nested dicts (`asset_ids`, `scores`, `frigate_scores`, `frigate_files`, `crop_dims`) are now individually copied so `.pop()` calls in write paths cannot reach the in-memory cache.
- **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.
- **`find_by_crop_dimension` and `_pick_mapped_file` now agree on duplicate asset→file handling**: both use first-seen-wins when the same `asset_id` maps to multiple Frigate filenames, preventing inconsistent replacement decisions.
- **Pagination error log includes page number**: the exception log in `fetch_all_assets` now includes the page number that failed.
- **Non-atomic JSON write**: tracker files are written to a `.tmp` sibling then renamed with `os.replace` so a crash mid-write never leaves a truncated file.
- **`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.
- **`get_person_summary` uses `_migrate_entry`**: replaced three ad-hoc `isinstance` guards with a single `_migrate_entry` call, making old-format (list) entries consistent with every other read path.
- **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.
- **Quality replacement floor check**: a candidate with a `None` blur score (PIL error during scoring) no longer blocks a freed slot — the `<=` floor comparison is only applied when a score is actually available.
- **`FRIGATE_SCORE_CEILING` parse guard**: a non-float value in `.env` now logs a warning and disables the ceiling instead of crashing at startup.
- **`executor.py` syntax error**: the `if img is None:` block in the full-res download path was comment-only and would have raised `IndentationError` on import. Added `pass`.
- **Dual config file warning**: a log warning is emitted when both `DATA_DIR/.immich_config.json` and the legacy CWD config file exist simultaneously.
- **Duplicate `if stale:` guard**: two consecutive identical guards around stale-cleanup and its log print were merged into one.
- **PID file write guard**: `OSError` on `/tmp/winnow.pid` write is now caught and logged instead of crashing the scheduler.
- **`_flat_key` uses constant equality** instead of substring match, removing a latent routing bug for any filename that happens to contain "uploaded".
- **Scheduler sleep clamped to 60 s**: bounds recovery time after an NTP clock step.
- **`remove_frigate_file` no longer creates ghost entries**: returns early when the person is absent rather than writing an empty stub.
- **`get_frigate_person_files` non-list debug log**: consistent with `get_all_frigate_person_files`.
- **`skip_ids` extracted to helper**: the identical set comprehension in `_handle_duplicate_people` that appeared in three branches is now a single `_smaller_duplicate_ids()` inner function.
## [0.5.2] - 2026-06-14
- **`blur_score_from_image` returns `None` on error** instead of `0.0`, so callers can distinguish a failed measurement from a legitimately near-zero Laplacian variance score.
### 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
## [0.6.0] - 2026-06-15
### 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.
- **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.
## [0.5.0] - 2026-06-14
- **`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`).
### Changed
- **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.
- **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.
- **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
- **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.
- **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.
- **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).
- **`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.
- **`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.
- **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.
- **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.
- **`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.
- **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.
- **`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.
- **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"}`.
- **`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.
- **`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.
- **`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.
- **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.
- **Frigate version `v`-prefix now stripped**: `v0.16.0`-style version strings are correctly parsed.
- **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`.
- **`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.
- **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.
- **Dockerfile unknown `VARIANT` now fails loudly**: an unrecognised value now exits with an error instead of silently falling through to the cpu branch.
- **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.
- **`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
+3 -3
View File
@@ -5,7 +5,7 @@
> **Note:** winnow's approach to training Frigate face recognition is not an officially documented workflow — results may vary.
> **Early Development — Use With Caution**
> winnow is functional but still maturing. Features that modify your Frigate training data — quality replacement, stale mapping cleanup — can remove images from your dataset and are not yet battle-tested at scale. Review the logs after each run and keep backups of your Frigate face training directory until you are confident in the results.
> winnow is in an unfinished state and maturing. Features that modify your Frigate training data — quality replacement, stale mapping cleanup — can remove images from your dataset and are not yet battle-tested at scale. Review the logs after each run and keep backups of your Frigate face training directory until you are confident in the results.
**Docs:** [Setup](https://github.com/sudolulo/winnow/wiki/Setup) · [Troubleshooting](https://github.com/sudolulo/winnow/wiki/Troubleshooting) · [FAQ](https://github.com/sudolulo/winnow/wiki/FAQ)
@@ -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.3"
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.2"
source = { editable = "." }
dependencies = [
{ name = "croniter" },
winnow/cache.py
+11 -2
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)
logger.warning("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
+29 -14
View File
@@ -7,12 +7,12 @@ 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
from .log_config import console, setup_logging
from .upload_tracker import find_by_crop_dimension, get_person_summary, reset_person
from .upload_tracker import find_by_crop_dimension, get_person_summary, reset_all_people, reset_person
logger = logging.getLogger(__name__)
@@ -78,6 +78,16 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
if not duplicates:
return people
def _smaller_duplicate_ids(groups: dict) -> set[str]:
"""IDs of all but the largest person in each duplicate group."""
return {
p["id"]
for ps in groups.values()
for p in sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)[1:]
}
skip_ids = _smaller_duplicate_ids(duplicates)
if not Config.MERGE_DUPLICATE_PEOPLE:
rprint("\n[bold yellow]⚠ Duplicate person names detected in Immich:[/bold yellow]")
for name, ps in sorted(duplicates.items()):
@@ -99,11 +109,6 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
)
# Return deduplicated list — keep only the largest per name so that
# downstream job creation never runs two jobs for the same Frigate folder.
skip_ids = {
p["id"]
for ps in duplicates.values()
for p in sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)[1:]
}
return [p for p in people if p["id"] not in skip_ids]
# Auto-merge: survivor = largest asset count, rest merge into it inside Immich
@@ -125,9 +130,20 @@ 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.
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]"
)
return [p for p in people if p["id"] not in skip_ids]
_UNSUPPORTED_VARS = [
@@ -145,7 +161,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()
@@ -192,8 +208,7 @@ def main() -> None:
"and will be reset along with everyone else.[/yellow]"
)
if names:
for name in names:
reset_person(name)
reset_all_people()
rprint(f"[bold yellow]Reset tracking data for all {len(names)} people.[/bold yellow]")
else:
rprint("[dim]No tracking data to reset.[/dim]")
@@ -232,8 +247,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
+341 -296
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,9 +24,11 @@ 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 (
UPLOAD_TRACKER_FILE,
REJECT_TRACKER_FILE,
get_lowest_quality_mapped_file,
get_most_redundant_mapped_file,
get_tracked_frigate_file_count,
@@ -32,7 +36,10 @@ from .upload_tracker import (
has_frigate_scores,
mark_rejected,
mark_uploaded,
begin_batch,
flush_batch,
remove_frigate_file,
remove_frigate_files_batch,
)
logger = logging.getLogger(__name__)
@@ -42,9 +49,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 +109,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:
# Full-res download failed — could be a transient network
# error, so don't mark rejected; it will be retried next run.
pass
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, OSError):
# Pillow cannot identify the format or the content is
# truncated. The download already succeeded (resp.ok),
# so this is a data problem, not a transient network
# error — mark rejected so it isn't retried forever.
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 +232,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
@@ -330,9 +350,8 @@ def upload_to_frigate(jobs: list[dict]) -> None:
# (manually deleted, or cleaned up outside winnow). This corrects the
# effective_count so those slots are available for new uploads.
stale = get_tracked_frigate_filenames(name) - known_frigate_files_at_start
for stale_fn in stale:
remove_frigate_file(name, stale_fn)
if stale:
remove_frigate_files_batch(name, list(stale))
progress.console.print(
f" [dim]{name}: cleared {len(stale)} stale mapping(s)"
" (file(s) no longer in Frigate)[/dim]"
@@ -349,220 +368,246 @@ def upload_to_frigate(jobs: list[dict]) -> None:
min_quality_score_for_slot: float | None = None
person_has_fscores: bool = has_frigate_scores(name)
for fname in person_files:
fpath = os.path.join(person_dir, fname)
begin_batch(UPLOAD_TRACKER_FILE)
begin_batch(REJECT_TRACKER_FILE)
try:
for fname in person_files:
fpath = os.path.join(person_dir, fname)
# If a previous replacement delete succeeded but that upload failed,
# require the next candidate to beat the deleted file's score so the
# freed slot isn't filled with something worse than what we removed.
if min_quality_score_for_slot is not None:
file_score = score_map.get(fname)
if file_score is None or file_score <= min_quality_score_for_slot:
score_str = f"{file_score:.3f}" if file_score is not None else "N/A"
progress.console.print(
f" [dim]⏭ {fname}: score {score_str} ≤ freed slot floor"
f" {min_quality_score_for_slot:.3f}, skipping[/dim]"
)
progress.advance(upload_task)
continue
at_cap = effective_count >= Config.MAX_AUTO_IMAGES
# Pre-upload Frigate score — clean measurement (image not yet in training set).
# Called for all below-cap uploads (seeds frigate_scores for future at-cap
# replacement) and for at-cap uploads when scores already exist. Skipped on
# the first run (pre_run_count == 0) since Frigate has no model yet.
# recognize_face returns (face_name, score); we only use the score when the
# best match is for the correct person. Mismatches (or "unknown") are treated
# as None so a wrong-person score never drives a ceiling skip or replacement.
# Frigate rebuilds its model asynchronously after any delete (clear + background
# thread), so the first recognize call after a deletion returns None — our code
# handles this conservatively by skipping that candidate until the next run.
# LIMITATION — async rebuild during multi-replacement runs: each deletion in a
# single run triggers a background model rebuild in Frigate. Subsequent recognize
# calls in the same run may get None (rebuild in progress), causing later
# candidates to fall back to blur-score replacement or be skipped entirely.
# The more replacements that happen in one run, the worse the scoring gets.
# TODO(frigate-api): if Frigate exposes a model generation counter or a
# rebuild-complete signal, poll it between recognize calls during replacement
# sequences rather than accepting stale/None scores.
pre_fscore: float | None = None
if Config.ENABLE_FRIGATE_SCORES and pre_run_count > 0:
if not at_cap or person_has_fscores:
_result = recognize_face(fpath)
if _result is not None and (_result[0] or "").casefold() == name.casefold():
pre_fscore = _result[1]
# Below-cap novelty gate: skip candidates already covered by the Frigate model,
# including conditions learned from manually-added images winnow can't track.
# pre_fscore is None on the first run (pre_run_count == 0 skips recognize_face
# above), so this block never fires on the first run without an extra guard.
if not at_cap and pre_fscore is not None:
_ceiling = Config.FRIGATE_SCORE_CEILING
if _ceiling is None:
# Dynamic default: bar = most-redundant tracked file's Frigate score.
# Falls back to uploading freely when no tracked scores exist yet.
_bar = get_most_redundant_mapped_file(name)
_skip = _bar is not None and pre_fscore > _bar[2]
_bar_str = f"most redundant tracked {_bar[2]:.2f}" if _bar else ""
elif _ceiling == 0.0:
_skip = False # explicitly disabled
_bar_str = ""
else:
_skip = pre_fscore > _ceiling
_bar_str = f"ceiling {_ceiling:.2f}"
if _skip:
progress.console.print(
f" [dim]⏭ {fname}: Frigate score {pre_fscore:.2f}"
f" > {_bar_str}, already covered[/dim]"
)
progress.advance(upload_task)
continue
if at_cap:
if not quality_replacement:
progress.console.print(f" [dim]⏭ {fname}: at cap, quality replacement disabled[/dim]")
progress.advance(upload_task)
continue
using_fscore = person_has_fscores and Config.ENABLE_FRIGATE_SCORES
if using_fscore:
candidate_score = pre_fscore
get_target = get_most_redundant_mapped_file
score_label, better_note = "frigate", " (more novel)"
no_score_msg = "Frigate recognize unavailable, skipping replacement"
else:
candidate_score = score_map.get(fname)
get_target = get_lowest_quality_mapped_file
score_label, better_note = "blur", ""
no_score_msg = "no quality score, skipping replacement"
if candidate_score is None:
progress.console.print(f" [dim]⏭ {fname}: {no_score_msg}[/dim]")
progress.advance(upload_task)
continue
target = get_target(name, exclude=failed_deletes)
not_better = target is None or (
candidate_score >= target[2] if using_fscore else candidate_score <= target[2]
)
if not_better:
target_str = f"{target[2]:.3f}" if target is not None else "N/A"
op = "<" if using_fscore else ">"
progress.console.print(
f" [dim]⏭ {fname}: {score_label} {candidate_score:.3f}"
f" not {op} {target_str}, skipping[/dim]"
)
progress.advance(upload_task)
continue
target_frigate_file, _target_asset_id, target_score = target
op = "<" if using_fscore else ">"
progress.console.print(
f" 🔄 {fname}: {score_label} {candidate_score:.3f} {op} {target_score:.3f},"
f" replacing {target_frigate_file}{better_note}"
)
if delete_frigate_person_files(name, [target_frigate_file]):
remove_frigate_file(name, target_frigate_file)
person_has_fscores = has_frigate_scores(name)
effective_count -= 1
min_quality_score_for_slot = None if using_fscore else target_score
else:
logger.warning("Failed to delete %s for %s, skipping replacement", target_frigate_file, name)
failed_deletes.add(target_frigate_file)
progress.advance(upload_task)
continue
for attempt in range(1, max_retries + 1):
try:
with open(fpath, "rb") as f:
resp = requests.post(
f"{frigate_url}/api/faces/{encoded_name}/register",
files={"file": (fname, f, "image/jpeg")},
timeout=30,
# If a previous replacement delete succeeded but that upload failed,
# require the next candidate to beat the deleted file's score so the
# freed slot isn't filled with something worse than what we removed.
if min_quality_score_for_slot is not None:
file_score = score_map.get(fname)
if file_score is not None and file_score <= min_quality_score_for_slot:
progress.console.print(
f" [dim]⏭ {fname}: score {file_score:.3f} ≤ freed slot floor"
f" {min_quality_score_for_slot:.3f}, skipping[/dim]"
)
if resp.status_code == 200:
uploaded += 1
person_uploaded += 1
effective_count += 1
min_quality_score_for_slot = None
progress.advance(upload_task)
continue
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,
)
if pre_fscore is not None:
person_has_fscores = True
actually_uploaded.append((fname, asset_id))
at_cap = effective_count >= Config.MAX_AUTO_IMAGES
break
# Pre-upload Frigate score — clean measurement (image not yet in training set).
# Called for all below-cap uploads (seeds frigate_scores for future at-cap
# replacement) and for at-cap uploads when scores already exist. Skipped on
# the first run (pre_run_count == 0) since Frigate has no model yet.
# recognize_face returns (face_name, score); we only use the score when the
# best match is for the correct person. Mismatches (or "unknown") are treated
# as None so a wrong-person score never drives a ceiling skip or replacement.
# Frigate rebuilds its model asynchronously after any delete (clear + background
# thread), so the first recognize call after a deletion returns None — our code
# handles this conservatively by skipping that candidate until the next run.
# LIMITATION — async rebuild during multi-replacement runs: each deletion in a
# single run triggers a background model rebuild in Frigate. Subsequent recognize
# calls in the same run may get None (rebuild in progress), causing later
# candidates to fall back to blur-score replacement or be skipped entirely.
# The more replacements that happen in one run, the worse the scoring gets.
# TODO(frigate-api): if Frigate exposes a model generation counter or a
# rebuild-complete signal, poll it between recognize calls during replacement
# sequences rather than accepting stale/None scores.
pre_fscore: float | None = None
if Config.ENABLE_FRIGATE_SCORES and pre_run_count > 0:
if not at_cap or person_has_fscores:
_result = recognize_face(fpath)
if _result is not None and (_result[0] or "").casefold() == name.casefold():
pre_fscore = _result[1]
# Below-cap novelty gate: skip candidates already covered by the Frigate model,
# including conditions learned from manually-added images winnow can't track.
# pre_fscore is None on the first run (pre_run_count == 0 skips recognize_face
# above), so this block never fires on the first run without an extra guard.
if not at_cap and pre_fscore is not None:
_ceiling = Config.FRIGATE_SCORE_CEILING
if _ceiling is None:
# Dynamic default: bar = most-redundant tracked file's Frigate score.
# Falls back to uploading freely when no tracked scores exist yet.
_bar = get_most_redundant_mapped_file(name)
_skip = _bar is not None and pre_fscore > _bar[2]
_bar_str = f"most redundant tracked {_bar[2]:.2f}" if _bar else ""
elif _ceiling == 0.0:
_skip = False # explicitly disabled
_bar_str = ""
else:
_skip = pre_fscore > _ceiling
_bar_str = f"ceiling {_ceiling:.2f}"
if _skip:
progress.console.print(
f" [dim]⏭ {fname}: Frigate score {pre_fscore:.2f}"
f" > {_bar_str}, already covered[/dim]"
)
progress.advance(upload_task)
continue
if at_cap:
if not quality_replacement:
progress.console.print(f" [dim]⏭ {fname}: at cap, quality replacement disabled[/dim]")
progress.advance(upload_task)
continue
using_fscore = person_has_fscores and Config.ENABLE_FRIGATE_SCORES
if using_fscore:
candidate_score = pre_fscore
get_target = get_most_redundant_mapped_file
score_label, better_note = "frigate", " (more novel)"
no_score_msg = "Frigate recognize unavailable, skipping replacement"
is_better_than = lambda c, t: c < t
else:
candidate_score = score_map.get(fname)
get_target = get_lowest_quality_mapped_file
score_label, better_note = "blur", ""
no_score_msg = "no quality score, skipping replacement"
is_better_than = lambda c, t: c > t
if candidate_score is None:
progress.console.print(f" [dim]⏭ {fname}: {no_score_msg}[/dim]")
progress.advance(upload_task)
continue
target = get_target(name, exclude=failed_deletes)
not_better = target is None or not is_better_than(candidate_score, target[2])
if not_better:
target_str = f"{target[2]:.3f}" if target is not None else "N/A"
cmp_op = "<" if using_fscore else ">"
progress.console.print(
f" [dim]⏭ {fname}: {score_label} {candidate_score:.3f}"
f" not {cmp_op} {target_str}, skipping[/dim]"
)
progress.advance(upload_task)
continue
target_frigate_file, _target_asset_id, target_score = target
cmp_op = "<" if using_fscore else ">"
progress.console.print(
f" 🔄 {fname}: {score_label} {candidate_score:.3f} {cmp_op} {target_score:.3f},"
f" replacing {target_frigate_file}{better_note}"
)
if delete_frigate_person_files(name, [target_frigate_file]):
remove_frigate_file(name, target_frigate_file)
person_has_fscores = has_frigate_scores(name)
effective_count -= 1
min_quality_score_for_slot = None if using_fscore else target_score
else:
logger.warning("Failed to delete %s for %s, skipping replacement", target_frigate_file, name)
failed_deletes.add(target_frigate_file)
progress.advance(upload_task)
continue
for attempt in range(1, max_retries + 1):
try:
with open(fpath, "rb") as f:
resp = requests.post(
f"{frigate_url}/api/faces/{encoded_name}/register",
files={"file": (fname, f, "image/jpeg")},
timeout=30,
)
if resp.status_code == 200:
uploaded += 1
person_uploaded += 1
effective_count += 1
min_quality_score_for_slot = None
asset_id = asset_map.get(fname)
if asset_id:
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,
)
else:
if pre_fscore is not None:
person_has_fscores = True
actually_uploaded.append((fname, asset_id))
break
else:
if attempt < max_retries:
logger.warning(
f"Upload attempt {attempt}/{max_retries} for {fname}:"
f" HTTP {resp.status_code}, retrying..."
)
continue
failed += 1
person_failed += 1
progress.console.print(
f" [red]✗ {fname}: HTTP {resp.status_code} (after {max_retries} attempts)[/red]"
)
full_body = resp.text
try:
error_detail = resp.json().get("message", full_body[:100])
except Exception:
error_detail = full_body[:100]
if resp.status_code == 400:
progress.console.print(f" [dim]{error_detail}[/dim]")
else:
logger.debug("%s HTTP %s: %s", fname, resp.status_code, error_detail)
_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)
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as exc:
if attempt < max_retries:
logger.warning(
f"Upload attempt {attempt}/{max_retries} for {fname}:"
f" HTTP {resp.status_code}, retrying..."
f" {type(exc).__name__}, retrying..."
)
continue
failed += 1
person_failed += 1
label = (
"Connection refused"
if isinstance(exc, requests.exceptions.ConnectionError)
else "Request timed out (30s)"
)
progress.console.print(
f" [red]✗ {fname}: {label} (after {max_retries} attempts)[/red]"
)
except Exception as e:
if attempt < max_retries:
logger.warning(
f"Upload attempt {attempt}/{max_retries} for {fname}:"
f" {type(e).__name__}, retrying..."
)
continue
failed += 1
person_failed += 1
progress.console.print(
f" [red]✗ {fname}: HTTP {resp.status_code} (after {max_retries} attempts)[/red]"
f" [red]✗ {fname}: {type(e).__name__} - {e} (after {max_retries} attempts)[/red]"
)
full_body = resp.text
try:
error_detail = resp.json().get("message", full_body[:100])
except Exception:
error_detail = full_body[:100]
if resp.status_code == 400:
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():
asset_id = asset_map.get(fname)
if asset_id:
mark_rejected(asset_id, person_name=name)
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as exc:
if attempt < max_retries:
logger.warning(
f"Upload attempt {attempt}/{max_retries} for {fname}:"
f" {type(exc).__name__}, retrying..."
)
continue
failed += 1
person_failed += 1
label = (
"Connection refused"
if isinstance(exc, requests.exceptions.ConnectionError)
else "Request timed out (30s)"
)
progress.console.print(
f" [red]✗ {fname}: {label} (after {max_retries} attempts)[/red]"
)
except Exception as e:
if attempt < max_retries:
logger.warning(
f"Upload attempt {attempt}/{max_retries} for {fname}:"
f" {type(e).__name__}, retrying..."
)
continue
failed += 1
person_failed += 1
progress.console.print(
f" [red]✗ {fname}: {type(e).__name__} - {e} (after {max_retries} attempts)[/red]"
)
progress.advance(upload_task)
progress.advance(upload_task)
if min_quality_score_for_slot is not None:
logger.warning(
f"{name}: freed replacement slot (floor {min_quality_score_for_slot:.3f})"
" was not filled this run — will be available next run"
)
if min_quality_score_for_slot is not None:
logger.warning(
f"{name}: freed replacement slot (floor {min_quality_score_for_slot:.3f})"
" was not filled this run — will be available next run"
)
finally:
try:
flush_batch(UPLOAD_TRACKER_FILE)
except Exception as _flush_exc:
logger.warning("flush_batch failed during cleanup — batch will be recovered on next begin_batch: %s", _flush_exc)
try:
flush_batch(REJECT_TRACKER_FILE)
except Exception as _flush_exc:
logger.warning("flush_batch failed during cleanup — batch will be recovered on next begin_batch: %s", _flush_exc)
# Batch-map Frigate filenames to asset IDs now that all uploads are done.
if actually_uploaded and not _skip_reconcile:
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
+13 -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,14 @@ 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:
if custom_limit > 0:
return custom_limit, "smart"
logger.warning("LIMIT=%s is invalid — ignoring and using auto strategy", custom_limit)
strategy_map = {
"adaptive": ("auto", "smart"),
@@ -168,7 +157,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 +231,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 +257,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 +306,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
+28 -4
View File
@@ -14,6 +14,11 @@ from PIL import Image
logger = logging.getLogger(__name__)
def _laplacian_var(img_np: np.ndarray) -> float:
gray = cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY) if img_np.ndim == 3 else img_np
return float(cv2.Laplacian(gray, cv2.CV_64F).var())
@dataclass
class QualityResult:
"""Result of quality assessment on a face/image crop."""
@@ -32,8 +37,7 @@ def check_blur(img_np: np.ndarray, threshold: float = 100.0) -> tuple[bool, str]
Lower variance = blurrier image. ArcFace needs clear facial features.
"""
gray = cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY) if img_np.ndim == 3 else img_np
variance = cv2.Laplacian(gray, cv2.CV_64F).var()
variance = _laplacian_var(img_np)
if variance < threshold:
return False, f"Blurry (laplacian={variance:.1f}, threshold={threshold})"
return True, ""
@@ -115,8 +119,7 @@ def assess_quality(
reasons = []
# Compute laplacian variance once (used by check_blur and stored as blur_score)
gray = cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY) if img_np.ndim == 3 else img_np
blur_score = float(cv2.Laplacian(gray, cv2.CV_64F).var())
blur_score = _laplacian_var(img_np)
checks = [
(
@@ -138,3 +141,24 @@ 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 | None:
"""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 None on error so callers can distinguish a failed measurement from a
legitimately low (near-zero) score.
"""
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 _laplacian_var(np.array(score_img))
except Exception as exc:
logger.debug("blur_score_from_image failed: %s", exc)
return None
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
+362 -378
View File
@@ -1,213 +1,201 @@
"""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
from .frigate_api import _get_frigate_url, 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] = {}
_deferred: set[str] = set() # paths whose disk writes are batched until flush_batch()
_dirty: set[str] = set() # deferred paths that received at least one _save during the batch
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)
from .config import Config
return Path(Config.DATA_DIR) / filename
except (ImportError, AttributeError):
return Path(filename)
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
def _write_to_disk(path: Path, data: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(".tmp")
try:
with open(tmp, "w") as f:
json.dump(data, f, indent=2)
os.replace(tmp, path)
except Exception:
tmp.unlink(missing_ok=True)
raise
def _save(filename: str, data: dict) -> None:
path = _tracker_path(filename)
key = str(path)
if key in _deferred:
_cache[key] = data # accumulate in cache; disk write deferred until flush_batch()
_dirty.add(key)
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")
_write_to_disk(path, data)
_cache[key] = data # update cache only after successful write
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()
def begin_batch(filename: str) -> None:
"""Defer tracker disk writes for filename. All _save calls accumulate in the
in-memory cache until flush_batch() is called. Use around per-person upload loops
to reduce N writes to 1.
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),
)
If a previous batch for this file was interrupted before flush_batch() was called
(e.g. an exception escaped the upload loop), the leftover cache state is flushed
to disk here before starting fresh so that partial progress is not silently lost.
"""
path = _tracker_path(filename)
key = str(path)
if key in _deferred and key in _dirty:
try:
_write_to_disk(path, _cache[key])
except Exception:
logger.warning("begin_batch: could not flush leftover deferred state for %s — partial progress may be lost", path)
_deferred.discard(key)
_dirty.discard(key)
_deferred.add(key)
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def flush_batch(filename: str) -> None:
"""Write the accumulated cache state for filename to disk."""
path = _tracker_path(filename)
key = str(path)
if key in _dirty and key in _cache:
_write_to_disk(path, _cache[key])
_deferred.discard(key)
_dirty.discard(key)
def _flat_key(filename: str) -> str:
return "uploaded_asset_ids" if filename == UPLOAD_TRACKER_FILE else "rejected_asset_ids"
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": {}}
# Copy top-level and all nested dicts so callers' mutations never reach the cache.
result = dict(entry)
result["asset_ids"] = list(result.get("asset_ids", []))
result["scores"] = dict(result.get("scores", {}))
result["frigate_scores"] = dict(result.get("frigate_scores", {}))
result["frigate_files"] = dict(result.get("frigate_files", {}))
result["crop_dims"] = dict(result.get("crop_dims", {}))
return result
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:
if not person_name:
logger.warning("_mark called with empty person_name for asset %s — asset not recorded", asset_id)
return
data = _load(filename)
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)
logger.debug("Marked %s in %s (%s)", asset_id, filename, person_name)
# ── 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 all asset IDs recorded as uploaded. Derives from by_person (primary)
plus any legacy flat list still present in old tracker files."""
data = _load(UPLOAD_TRACKER_FILE)
ids = {aid for e in data.get("by_person", {}).values() for aid in _get_ids(e)}
ids.update(data.get("uploaded_asset_ids", [])) # backward compat with pre-0.6.1 files
return ids
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 all asset IDs recorded as rejected. Derives from by_person (primary)
plus any legacy flat list still present in old tracker files."""
data = _load(REJECT_TRACKER_FILE)
ids = {aid for e in data.get("by_person", {}).values() for aid in _get_ids(e)}
ids.update(data.get("rejected_asset_ids", [])) # backward compat with pre-0.6.1 files
return ids
def mark_uploaded(
@@ -217,264 +205,260 @@ 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)
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)
def record_frigate_file(person_name: str, frigate_filename: str, asset_id: str) -> None:
"""Record a single Frigate filename → asset_id mapping."""
record_frigate_files_batch(person_name, {frigate_filename: asset_id})
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)
src = _load(UPLOAD_TRACKER_FILE)
by_person = dict(src.get("by_person", {}))
entry = _migrate_entry(by_person.get(person_name, {}))
entry["frigate_files"].update(mappings)
by_person[person_name] = entry
data = dict(src)
data["by_person"] = by_person
_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)
remove_frigate_files_batch(person_name, [frigate_filename])
def remove_frigate_files_batch(person_name: str, frigate_filenames: list[str]) -> None:
"""Remove multiple Frigate filenames in a single load/save."""
src = _load(UPLOAD_TRACKER_FILE)
raw = src.get("by_person", {}).get(person_name)
if raw is None:
return
entry = _migrate_entry(raw)
for fn in frigate_filenames:
asset_id = entry["frigate_files"].pop(fn, None)
if asset_id is not None and asset_id not in entry["frigate_files"].values():
entry["frigate_scores"].pop(asset_id, None)
by_person = dict(src.get("by_person", {})) # copy so assignment does not mutate the cache
by_person[person_name] = entry
data = dict(src)
data["by_person"] = by_person
_save(UPLOAD_TRACKER_FILE, data)
logger.debug(f"Removed {len(frigate_filenames)} Frigate file mapping(s) for {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)
raw = data.get("by_person", {}).get(person_name)
if not raw or isinstance(raw, list):
return False
frigate_files = raw.get("frigate_files", {})
frigate_scores = raw.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, {})
seen_assets: set[str] = set()
candidates = []
for ff, asset_id in entry.get("frigate_files", {}).items():
if (exclude is None or ff not in exclude) and asset_id in scores and asset_id not in seen_assets:
seen_assets.add(asset_id)
candidates.append((ff, asset_id, scores[asset_id]))
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: dict[str, str] = {}
for fn, aid in frigate_files.items():
asset_to_frigate.setdefault(aid, fn) # first-seen wins; plain inversion silently drops duplicates
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_all_people() -> None:
"""Reset all tracking data in two writes (O(P) Frigate API calls, O(1) disk writes).
Preferred over calling reset_person() in a loop when RESET_PERSON=* — that
approach is O(P²) because each call rebuilds the flat list from all remaining entries.
"""
upload_data = _load(UPLOAD_TRACKER_FILE)
frigate_url = _get_frigate_url()
if not frigate_url:
logger.info("FRIGATE_URL not set — skipping Frigate file deletion")
for person_name, raw_entry in upload_data.get("by_person", {}).items():
entry = _migrate_entry(raw_entry)
frigate_filenames = list(entry.get("frigate_files", {}).keys())
if not frigate_filenames:
continue
if frigate_url:
if delete_frigate_person_files(person_name, frigate_filenames):
logger.info(f"Deleted {len(frigate_filenames)} Frigate file(s) for {person_name}")
else:
logger.warning(f"Could not delete Frigate files for {person_name} — tracker reset proceeding anyway")
_save(UPLOAD_TRACKER_FILE, {})
_save(REJECT_TRACKER_FILE, {})
logger.info("Reset all tracking 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)
if not _get_frigate_url():
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
for filename in (UPLOAD_TRACKER_FILE, REJECT_TRACKER_FILE):
src = upload_data if filename == UPLOAD_TRACKER_FILE else _load(REJECT_TRACKER_FILE)
by_person = dict(src.get("by_person", {})) # copy so pop() does not mutate the cache
tracker_entry = by_person.pop(person_name, None)
if tracker_entry is not None:
data = dict(src)
data["by_person"] = by_person
flat_key = _flat_key(filename)
person_ids = set(_get_ids(tracker_entry))
if person_ids and flat_key in data:
data[flat_key] = sorted(set(data[flat_key]) - person_ids)
_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 = _migrate_entry(uploaded_data.get(name, {}))
r_entry = _migrate_entry(rejected_data.get(name, {}))
result[name] = {
"uploaded": len(u_entry["asset_ids"]),
"rejected": len(r_entry["asset_ids"]),
"frigate_count": u_entry.get("frigate_count"),
"scores": u_entry["scores"],
"frigate_files": u_entry["frigate_files"],
}
return result
def filter_already_uploaded(
@@ -488,5 +472,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