From 4cdd4657d6ecf2a366e3dc07caf4b409b7213604 Mon Sep 17 00:00:00 2001 From: Holden Date: Tue, 16 Jun 2026 15:44:41 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20v0.6.2=20=E2=80=94=20structural=20tracke?= =?UTF-8?q?r=20refactor,=20batch=20writes,=20multi-instance=20prep?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Drop flat list as primary storage; derive uploaded/rejected IDs from by_person (single source of truth). Legacy flat lists in existing files still read for backward compat. Removes dual-representation sync hazard. - Add begin_batch/flush_batch: per-person upload loop now does 1 os.replace instead of N (one per mark_uploaded call). Benefit on slow storage. - reset_all_people(): RESET_PERSON=* is now O(1) disk writes instead of O(P^2). - blur_score_from_image inlines cv2.Laplacian directly, removing assess_quality call overhead and decoupling from the full quality pipeline. --- CHANGELOG.md | 12 ++++++ pyproject.toml | 2 +- uv.lock | 2 +- winnow/cli.py | 5 +-- winnow/executor.py | 5 +++ winnow/quality.py | 3 +- winnow/upload_tracker.py | 91 ++++++++++++++++++++++++++++++---------- 7 files changed, 92 insertions(+), 28 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 558f920..b80c65c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.6.2] - 2026-06-16 + +### Changed + +- **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). + +- **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). + +- **`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. + +- **`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. + ## [0.6.1] - 2026-06-16 ### Fixed diff --git a/pyproject.toml b/pyproject.toml index 2889a90..fd6c77d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "winnow" -version = "0.6.1" +version = "0.6.2" 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" diff --git a/uv.lock b/uv.lock index c7fd590..3d6fbb2 100644 --- a/uv.lock +++ b/uv.lock @@ -862,7 +862,7 @@ wheels = [ [[package]] name = "winnow" -version = "0.6.0" +version = "0.6.1" source = { editable = "." } dependencies = [ { name = "croniter" }, diff --git a/winnow/cli.py b/winnow/cli.py index 986960b..c7456d4 100644 --- a/winnow/cli.py +++ b/winnow/cli.py @@ -12,7 +12,7 @@ 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__) @@ -207,8 +207,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]") diff --git a/winnow/executor.py b/winnow/executor.py index 10afd60..205eac8 100644 --- a/winnow/executor.py +++ b/winnow/executor.py @@ -34,6 +34,8 @@ from .upload_tracker import ( has_frigate_scores, mark_rejected, mark_uploaded, + begin_batch, + flush_batch, remove_frigate_file, remove_frigate_files_batch, ) @@ -364,6 +366,7 @@ def upload_to_frigate(jobs: list[dict]) -> None: min_quality_score_for_slot: float | None = None person_has_fscores: bool = has_frigate_scores(name) + begin_batch(UPLOAD_TRACKER_FILE) for fname in person_files: fpath = os.path.join(person_dir, fname) @@ -590,6 +593,8 @@ def upload_to_frigate(jobs: list[dict]) -> None: " was not filled this run — will be available next run" ) + flush_batch(UPLOAD_TRACKER_FILE) + # Batch-map Frigate filenames to asset IDs now that all uploads are done. if actually_uploaded and not _skip_reconcile: reconcile_frigate_mappings(name, known_frigate_files_at_start, actually_uploaded) diff --git a/winnow/quality.py b/winnow/quality.py index 9247f88..bd78d7a 100644 --- a/winnow/quality.py +++ b/winnow/quality.py @@ -154,7 +154,8 @@ def blur_score_from_image(img: Image.Image, max_dim: int = 1440) -> float | None if score_img.width > max_dim or score_img.height > max_dim: score_img = score_img.copy() score_img.thumbnail((max_dim, max_dim), Image.LANCZOS) - return float(assess_quality(score_img).blur_score) + gray = cv2.cvtColor(np.array(score_img), cv2.COLOR_RGB2GRAY) + return float(cv2.Laplacian(gray, cv2.CV_64F).var()) except Exception as exc: logger.debug("blur_score_from_image failed: %s", exc) return None diff --git a/winnow/upload_tracker.py b/winnow/upload_tracker.py index abf2b55..5a3a85e 100644 --- a/winnow/upload_tracker.py +++ b/winnow/upload_tracker.py @@ -43,6 +43,7 @@ REJECT_TRACKER_FILE = "frigate_rejected_ids.json" # 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() def _tracker_path(filename: str) -> Path: @@ -69,28 +70,48 @@ def _load(filename: str) -> dict: return data -def _save(filename: str, data: dict) -> None: - path = _tracker_path(filename) +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) - _cache[str(path)] = data # update only after the file is safely on disk 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() + return + _write_to_disk(path, data) + _cache[key] = data # update cache only after successful write + + +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.""" + _deferred.add(str(_tracker_path(filename))) + + +def flush_batch(filename: str) -> None: + """Write the accumulated cache state for filename to disk.""" + path = _tracker_path(filename) + key = str(path) + _deferred.discard(key) + if key in _cache: + _write_to_disk(path, _cache[key]) + + def _flat_key(filename: str) -> str: return "uploaded_asset_ids" if filename == UPLOAD_TRACKER_FILE else "rejected_asset_ids" -def _load_flat(filename: str) -> set[str]: - return set(_load(filename).get(_flat_key(filename), [])) - - def _get_ids(entry: list | dict) -> list[str]: """Extract asset_ids from either the old list format or the new dict format.""" if isinstance(entry, list): @@ -121,10 +142,6 @@ def _mark( frigate_score: float | None = None, ) -> None: data = _load(filename) - flat_key = _flat_key(filename) - flat = set(data.get(flat_key, [])) - flat.add(asset_id) - data[flat_key] = sorted(flat) if person_name: by_person = data.setdefault("by_person", {}) entry = _migrate_entry(by_person.get(person_name, {})) @@ -144,11 +161,21 @@ def _mark( # ── Public API ──────────────────────────────────────────────────────────────── def load_uploaded_ids() -> set[str]: - return _load_flat(UPLOAD_TRACKER_FILE) + """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]: - return _load_flat(REJECT_TRACKER_FILE) + """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( @@ -327,6 +354,29 @@ def update_frigate_count(person_name: str, count: int) -> None: _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) + 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 not os.environ.get("FRIGATE_URL", "").strip(): + 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(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. @@ -347,18 +397,15 @@ def reset_person(person_name: str) -> None: logger.warning(f"Could not delete Frigate files for {person_name} — tracker reset proceeding anyway") changed = False - tracker_files = ((UPLOAD_TRACKER_FILE, upload_data), (REJECT_TRACKER_FILE, _load(REJECT_TRACKER_FILE))) - for filename, data in tracker_files: - flat_key = _flat_key(filename) + for filename in (UPLOAD_TRACKER_FILE, REJECT_TRACKER_FILE): + data = upload_data if filename == UPLOAD_TRACKER_FILE else _load(REJECT_TRACKER_FILE) by_person = data.get("by_person", {}) tracker_entry = by_person.pop(person_name, None) if tracker_entry is not None: - # Rebuild from remaining entries rather than subtracting, so IDs that - # appear under another person aren't incorrectly removed from the flat list. - remaining_ids: set[str] = set() - for other_entry in by_person.values(): - remaining_ids.update(_get_ids(other_entry)) - data[flat_key] = sorted(remaining_ids) + 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) data["by_person"] = by_person _save(filename, data) changed = True