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

refactor: collapse Config proxy, migrate tracker to SQLite, split reconcile module

Browse Source 
- Config: remove _ConfigAccessor and ConfigManager; use __getattr__ for lazy
  loading on single _Config class; re-register self as _instance in __getattr__
  so reset() always clears the correct object (item 1)
- upload_tracker: replace hand-rolled JSON store with sqlite3; auto-migrates
  existing JSON on first run; remove dead record_frigate_file function;
  connection re-opens when CACHE_DIR changes for test isolation (items 2, 8)
- diversity: move ThreadPoolExecutor import to module level; inject optional
  fetch_fn parameter for testability (items 3, 6)
- pyproject: consolidate 4 variant files into extras (gpu/rocm/intel/cpu);
  update Dockerfile to use --extra flag; delete variant pyproject/lock files;
  uv.lock needs regen with `uv lock` after this change (item 4)
- jobs: extract _build_job helper to separate business logic from terminal I/O;
  auto_configure delegates dedup/selection to _build_job (item 5)
- logging: convert f-string log calls to % interpolation throughout all winnow/
  modules (item 7)
- reconcile: new module with reconcile_frigate_mappings and
  enrich_asset_with_face_data extracted from executor.py (item 9)
- scheduler: print next scheduled run time after startup and after each run;
  fix f-string logger.error call (item 10)
This commit is contained in:
flan
2026-06-14 19:59:17 +00:00
parent ad1fbd4c2a
commit e2a1924fb0
22 changed files with 754 additions and 6613 deletions
Download Patch File Download Diff File Expand all files Collapse all files
winnow/executor.py
+10 -117
View File
@@ -3,7 +3,6 @@
import logging
import os
import shutil
import time
from io import BytesIO
from urllib.parse import quote
@@ -21,9 +20,10 @@ from .frigate_api import (
recognize_face,
)
from .image_processing import process_face_mode
from .immich_api import fetch_face_data, fetch_full_image
from .immich_api import fetch_full_image
from .log_config import console
from .quality import assess_quality
from .reconcile import enrich_asset_with_face_data, reconcile_frigate_mappings
from .upload_tracker import (
get_lowest_quality_mapped_file,
get_most_redundant_mapped_file,
@@ -32,7 +32,6 @@ from .upload_tracker import (
has_frigate_scores,
mark_rejected,
mark_uploaded,
record_frigate_files_batch,
remove_frigate_file,
)
@@ -55,112 +54,6 @@ def _safe_person_dir(output_dir: str, person_name: str) -> str:
return candidate
def _reconcile_frigate_mappings(
person_name: str,
known_files_before: set[str],
uploaded: list[tuple[str, str | None]],
) -> None:
"""Map Frigate filenames to asset IDs after a batch of uploads.
Polls until all expected new files appear in the Frigate API, then maps
them to asset IDs by filename timestamp order (Frigate processes the
upload queue in FIFO order, so earlier uploads get earlier timestamps).
KNOWN LIMITATION — race condition with external uploads:
If another client uploads a face file for this person concurrently, the
count of new files will exceed `len(uploaded)` and we bail out entirely
(the "> target" branch). That's safe — we never record a wrong mapping —
but those uploads become permanently unmapped (they won't be eligible for
quality replacement). The right fix is a Frigate API that returns the
filename in the upload response, removing the need for any post-upload
diffing. Until then, the external-upload guard keeps mappings correct at
the cost of occasionally missing them when another client is active.
"""
target = len(uploaded)
current_files: set[str] = set()
for delay in (1, 2, 4, 8):
time.sleep(delay)
fresh = get_frigate_person_files(person_name)
if fresh is None:
logger.warning(
f"{person_name}: Frigate API unreachable during mapping reconciliation"
" — quality replacement won't target these files"
)
return
current_files = set(fresh)
if len(current_files - known_files_before) >= target:
break
new_files = current_files - known_files_before
if len(new_files) == target:
def _ts(fname: str) -> float:
try:
return float(fname.rsplit("_", 1)[-1].replace(".webp", ""))
except (ValueError, IndexError):
return 0.0
mappings = {
frigate_file: asset_id
for (_, asset_id), frigate_file in zip(uploaded, sorted(new_files, key=_ts))
if asset_id
}
record_frigate_files_batch(person_name, mappings)
elif len(new_files) > target:
logger.info(
f"{person_name}: {len(new_files)} new Frigate files for {target} uploads"
" (external upload detected) — skipping file mapping"
)
else:
logger.warning(
f"{person_name}: only {len(new_files)} of {target} expected Frigate files"
" appeared after reconciliation — mapping skipped"
)
def _enrich_asset_with_face_data(asset: dict, person: dict) -> dict:
"""Enrich an asset dict with face bounding box data from the Immich faces API.
The search/metadata endpoint does not include face bounding box data,
so we fetch it from GET /api/faces?id={asset_id} and inject it into
the asset's "people" field so process_face_mode can find it.
Returns the enriched asset dict (modifies in place and returns it).
"""
person_id = person["id"]
face_data = fetch_face_data(asset["id"], person_id=person_id)
if face_data is None:
logger.debug(f"No face data returned for {person.get('name')} in asset {asset.get('id')}")
# Clean any None entries from the people list (can come from Immich API)
if "people" in asset:
asset["people"] = [p for p in asset["people"] if p is not None]
return asset
# Skip zero-area bounding boxes (face detection failed or no face found)
if face_data.bbox == (0, 0, 0, 0):
logger.debug(f"Zero-area bounding box for {person.get('name')} in asset {asset.get('id')}")
# Clean any None entries from the people list (can come from Immich API)
if "people" in asset:
asset["people"] = [p for p in asset["people"] if p is not None]
return asset
face_info = {
"boundingBoxX1": face_data.bbox[0],
"boundingBoxY1": face_data.bbox[1],
"boundingBoxX2": face_data.bbox[2],
"boundingBoxY2": face_data.bbox[3],
"imageWidth": face_data.image_width,
"imageHeight": face_data.image_height,
}
# Inject into asset so process_face_mode can find it via asset["people"]
asset["people"] = [{"id": person_id, "faces": [face_info]}]
asset["face_confidence"] = face_data.confidence
return asset
def execute_jobs(jobs: list[dict]) -> None:
"""Download and process images for all jobs.
@@ -184,7 +77,7 @@ def execute_jobs(jobs: list[dict]) -> None:
insightface_app = get_insightface_app()
except Exception as e:
logger.debug(f"InsightFace unavailable for crop alignment: {e}")
logger.debug("InsightFace unavailable for crop alignment: %s", e)
with Progress(
SpinnerColumn(),
@@ -221,7 +114,7 @@ def execute_jobs(jobs: list[dict]) -> None:
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)
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")
@@ -271,7 +164,7 @@ def execute_jobs(jobs: list[dict]) -> None:
score_img.thumbnail((1440, 1440), Image.LANCZOS)
score_map[filename] = assess_quality(score_img).blur_score
except Exception as exc:
logger.debug(f"Quality score fallback for {asset['id']}: {exc}")
logger.debug("Quality score fallback for %s: %s", asset["id"], exc)
score_map[filename] = 0.0 # unknown quality — treat as lowest
count += 1
@@ -280,7 +173,7 @@ def execute_jobs(jobs: list[dict]) -> None:
f"[yellow]Skipped {asset['id']} (no usable face data)[/yellow]"
)
except Exception as e:
logger.error(f"Failed to process asset {asset['id']}: {e}")
logger.error("Failed to process asset %s: %s", asset["id"], e)
progress.advance(job_task)
progress.advance(overall_task)
@@ -294,7 +187,7 @@ def execute_jobs(jobs: list[dict]) -> None:
# Log how many images were actually saved vs selected
if count < len(assets):
logger.info(f"{name}: saved {count}/{len(assets)} selected images")
logger.info("%s: saved %s/%s selected images", name, count, len(assets))
def upload_to_frigate(jobs: list[dict]) -> None:
@@ -558,7 +451,7 @@ def upload_to_frigate(jobs: list[dict]) -> None:
effective_count -= 1
min_quality_score_for_slot = None if using_fscore else candidate_score
else:
logger.warning(f"Failed to delete {target_frigate_file} for {name}, skipping replacement")
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
@@ -611,7 +504,7 @@ def upload_to_frigate(jobs: list[dict]) -> None:
if resp.status_code == 400:
progress.console.print(f" [dim]{error_detail}[/dim]")
else:
logger.debug(f"{fname} HTTP {resp.status_code}: {error_detail}")
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:
@@ -656,7 +549,7 @@ def upload_to_frigate(jobs: list[dict]) -> None:
# Batch-map Frigate filenames to asset IDs now that all uploads are done.
if actually_uploaded:
_reconcile_frigate_mappings(name, known_frigate_files_at_start, actually_uploaded)
reconcile_frigate_mappings(name, known_frigate_files_at_start, actually_uploaded)
# Per-person summary
if person_failed == 0: