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

release: v0.5.15 — fix cache regression and structural cleanup

Browse Source 
- cache.py: fix np.save extension bug from v0.5.13 — tmp path used
  final+".tmp" (abc.npy.tmp) but np.save auto-appends .npy to paths not
  ending in .npy, writing to abc.npy.tmp.npy instead; os.replace then
  raised FileNotFoundError silently, making every cache write a no-op
  and leaking *.npy.tmp.npy files. Fixed by inserting .tmp before .npy:
  tmp = final[:-4] + ".tmp.npy"

- config.py: remove str(default) round-trip in _getenv_int/_getenv_float
  — use raw = os.getenv(name); return default if raw is None else int(raw)
  so a future float default can't cause a spurious "not a valid integer"
  warning and return the wrong type

- executor.py: consolidate 4 progress.remove_task calls into one
  try/finally around the per-job body; continue inside try/finally
  executes the finally before the next iteration, making the invariant
  structurally enforced rather than relying on discipline across 4 sites
This commit is contained in:
flan
2026-06-15 01:45:43 +00:00
parent 5bcc5975bc
commit fe5e1574ac
5 changed files with 125 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+10
View File
@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
## [0.5.15] - 2026-06-15
### Fixed
- **Embedding cache writes were silently no-ops since v0.5.13**: the atomic-write path used `tmp = final + ".tmp"` where `final` ends in `.npy` (e.g. `abc.npy`), producing a tmp path of `abc.npy.tmp`. `np.save` auto-appends `.npy` to paths not already ending in `.npy`, so it wrote to `abc.npy.tmp.npy` instead. The subsequent `os.replace("abc.npy.tmp", "abc.npy")` then raised `FileNotFoundError` (caught silently at DEBUG), meaning no cache entry was ever committed and leaked `*.npy.tmp.npy` files accumulated on disk. The fix inserts `.tmp` before the `.npy` extension: `tmp = final[:-4] + ".tmp.npy"` so `np.save` sees a path already ending in `.npy` and does not re-append.
- **`_getenv_int`/`_getenv_float` no longer route the default through `str()` conversion**: the previous form `os.getenv(name, str(default))` converted the default to a string so it could be fed through `int()`/`float()` — an unnecessary round-trip that would cause `_getenv_int("FOO", 4.0)` to log a spurious "not a valid integer" warning and return the float. The helpers now use `raw = os.getenv(name); return default if raw is None else int(raw)`, passing the typed default through directly.
- **`execute_jobs` progress task now removed via `try/finally`**: `progress.remove_task(job_task)` was duplicated in three early-exit paths (ValueError, symlink TOCTOU, OSError) plus once at normal completion. The entire per-job body is now wrapped in `try/finally: progress.remove_task(job_task)`; the three inner `continue` statements trigger the `finally` automatically before advancing to the next job, making the invariant structurally impossible to violate by a future code path.
## [0.5.14] - 2026-06-15
### Fixed
pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "winnow"
version = "0.5.14"
version = "0.5.15"
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"
winnow/cache.py
+3 -1
View File
@@ -83,7 +83,9 @@ class EmbeddingCache:
"""Store an embedding in the cache."""
self._ensure_dir()
final = self._path(asset_id, model)
tmp = final + ".tmp"
# 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[:-4] + ".tmp.npy"
try:
np.save(tmp, embedding)
os.replace(tmp, final)
winnow/config.py
+10 -6
View File
@@ -13,20 +13,24 @@ _LEGACY_CONFIG_FILE = Path(".immich_config.json") # pre-v0.6: lived in process
def _getenv_int(name: str, default: int) -> int:
val = os.getenv(name, str(default))
raw = os.getenv(name)
if raw is None:
return default
try:
return int(val)
return int(raw)
except ValueError:
logging.warning("%s=%r is not a valid integer — using default %s", name, val, default)
logging.warning("%s=%r is not a valid integer — using default %s", name, raw, default)
return default
def _getenv_float(name: str, default: float) -> float:
val = os.getenv(name, str(default))
raw = os.getenv(name)
if raw is None:
return default
try:
return float(val)
return float(raw)
except ValueError:
logging.warning("%s=%r is not a valid float — using default %s", name, val, default)
logging.warning("%s=%r is not a valid float — using default %s", name, raw, default)
return default
winnow/executor.py
+101 -103
View File
@@ -102,118 +102,116 @@ 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))
progress.remove_task(job_task)
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)
progress.remove_task(job_task)
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)
progress.remove_task(job_task)
continue
# 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"])
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
# 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. 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
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["id"], 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: