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

fix: address 4 quality review findings — flush_batch order, batch finally guard, cache copy, LIMIT=0 fallthrough

Browse Source
This commit is contained in:
flan
2026-06-16 16:59:43 +00:00
parent 817fa17e41
commit f3622b8d41
3 changed files with 218 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
winnow/executor.py
+209 -207
View File
@@ -368,237 +368,239 @@ def upload_to_frigate(jobs: list[dict]) -> None:
person_has_fscores: bool = has_frigate_scores(name) person_has_fscores: bool = has_frigate_scores(name)
begin_batch(UPLOAD_TRACKER_FILE) begin_batch(UPLOAD_TRACKER_FILE)
for fname in person_files: try:
fpath = os.path.join(person_dir, fname) for fname in person_files:
fpath = os.path.join(person_dir, fname)
# If a previous replacement delete succeeded but that upload failed, # If a previous replacement delete succeeded but that upload failed,
# require the next candidate to beat the deleted file's score so the # 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. # freed slot isn't filled with something worse than what we removed.
if min_quality_score_for_slot is not None: if min_quality_score_for_slot is not None:
file_score = score_map.get(fname) file_score = score_map.get(fname)
if file_score is not None and file_score <= min_quality_score_for_slot: if file_score is not None and file_score <= min_quality_score_for_slot:
progress.console.print( progress.console.print(
f" [dim]⏭ {fname}: score {file_score:.3f} ≤ freed slot floor" f" [dim]⏭ {fname}: score {file_score:.3f} ≤ freed slot floor"
f" {min_quality_score_for_slot:.3f}, skipping[/dim]" f" {min_quality_score_for_slot:.3f}, skipping[/dim]"
) )
progress.advance(upload_task) progress.advance(upload_task)
continue continue
at_cap = effective_count >= Config.MAX_AUTO_IMAGES at_cap = effective_count >= Config.MAX_AUTO_IMAGES
# Pre-upload Frigate score — clean measurement (image not yet in training set). # 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 # 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 # 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. # 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 # 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 # 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. # as None so a wrong-person score never drives a ceiling skip or replacement.
# Frigate rebuilds its model asynchronously after any delete (clear + background # Frigate rebuilds its model asynchronously after any delete (clear + background
# thread), so the first recognize call after a deletion returns None — our code # thread), so the first recognize call after a deletion returns None — our code
# handles this conservatively by skipping that candidate until the next run. # handles this conservatively by skipping that candidate until the next run.
# LIMITATION — async rebuild during multi-replacement runs: each deletion in a # LIMITATION — async rebuild during multi-replacement runs: each deletion in a
# single run triggers a background model rebuild in Frigate. Subsequent recognize # single run triggers a background model rebuild in Frigate. Subsequent recognize
# calls in the same run may get None (rebuild in progress), causing later # 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. # 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. # 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 # TODO(frigate-api): if Frigate exposes a model generation counter or a
# rebuild-complete signal, poll it between recognize calls during replacement # rebuild-complete signal, poll it between recognize calls during replacement
# sequences rather than accepting stale/None scores. # sequences rather than accepting stale/None scores.
pre_fscore: float | None = None pre_fscore: float | None = None
if Config.ENABLE_FRIGATE_SCORES and pre_run_count > 0: if Config.ENABLE_FRIGATE_SCORES and pre_run_count > 0:
if not at_cap or person_has_fscores: if not at_cap or person_has_fscores:
_result = recognize_face(fpath) _result = recognize_face(fpath)
if _result is not None and (_result[0] or "").casefold() == name.casefold(): if _result is not None and (_result[0] or "").casefold() == name.casefold():
pre_fscore = _result[1] pre_fscore = _result[1]
# Below-cap novelty gate: skip candidates already covered by the Frigate model, # Below-cap novelty gate: skip candidates already covered by the Frigate model,
# including conditions learned from manually-added images winnow can't track. # 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 # 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. # above), so this block never fires on the first run without an extra guard.
if not at_cap and pre_fscore is not None: if not at_cap and pre_fscore is not None:
_ceiling = Config.FRIGATE_SCORE_CEILING _ceiling = Config.FRIGATE_SCORE_CEILING
if _ceiling is None: if _ceiling is None:
# Dynamic default: bar = most-redundant tracked file's Frigate score. # Dynamic default: bar = most-redundant tracked file's Frigate score.
# Falls back to uploading freely when no tracked scores exist yet. # Falls back to uploading freely when no tracked scores exist yet.
_bar = get_most_redundant_mapped_file(name) _bar = get_most_redundant_mapped_file(name)
_skip = _bar is not None and pre_fscore > _bar[2] _skip = _bar is not None and pre_fscore > _bar[2]
_bar_str = f"most redundant tracked {_bar[2]:.2f}" if _bar else "" _bar_str = f"most redundant tracked {_bar[2]:.2f}" if _bar else ""
elif _ceiling == 0.0: elif _ceiling == 0.0:
_skip = False # explicitly disabled _skip = False # explicitly disabled
_bar_str = "" _bar_str = ""
else: else:
_skip = pre_fscore > _ceiling _skip = pre_fscore > _ceiling
_bar_str = f"ceiling {_ceiling:.2f}" _bar_str = f"ceiling {_ceiling:.2f}"
if _skip: if _skip:
progress.console.print( progress.console.print(
f" [dim]⏭ {fname}: Frigate score {pre_fscore:.2f}" f" [dim]⏭ {fname}: Frigate score {pre_fscore:.2f}"
f" > {_bar_str}, already covered[/dim]" f" > {_bar_str}, already covered[/dim]"
) )
progress.advance(upload_task) progress.advance(upload_task)
continue continue
if at_cap: if at_cap:
if not quality_replacement: if not quality_replacement:
progress.console.print(f" [dim]⏭ {fname}: at cap, quality replacement disabled[/dim]") progress.console.print(f" [dim]⏭ {fname}: at cap, quality replacement disabled[/dim]")
progress.advance(upload_task) progress.advance(upload_task)
continue continue
using_fscore = person_has_fscores and Config.ENABLE_FRIGATE_SCORES using_fscore = person_has_fscores and Config.ENABLE_FRIGATE_SCORES
if using_fscore: if using_fscore:
candidate_score = pre_fscore candidate_score = pre_fscore
get_target = get_most_redundant_mapped_file get_target = get_most_redundant_mapped_file
score_label, better_note = "frigate", " (more novel)" score_label, better_note = "frigate", " (more novel)"
no_score_msg = "Frigate recognize unavailable, skipping replacement" no_score_msg = "Frigate recognize unavailable, skipping replacement"
is_better_than = lambda c, t: c < t is_better_than = lambda c, t: c < t
else: else:
candidate_score = score_map.get(fname) candidate_score = score_map.get(fname)
get_target = get_lowest_quality_mapped_file get_target = get_lowest_quality_mapped_file
score_label, better_note = "blur", "" score_label, better_note = "blur", ""
no_score_msg = "no quality score, skipping replacement" no_score_msg = "no quality score, skipping replacement"
is_better_than = lambda c, t: c > t is_better_than = lambda c, t: c > t
if candidate_score is None: if candidate_score is None:
progress.console.print(f" [dim]⏭ {fname}: {no_score_msg}[/dim]") progress.console.print(f" [dim]⏭ {fname}: {no_score_msg}[/dim]")
progress.advance(upload_task) progress.advance(upload_task)
continue continue
target = get_target(name, exclude=failed_deletes) target = get_target(name, exclude=failed_deletes)
not_better = target is None or not is_better_than(candidate_score, target[2]) not_better = target is None or not is_better_than(candidate_score, target[2])
if not_better: if not_better:
target_str = f"{target[2]:.3f}" if target is not None else "N/A" 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 ">" cmp_op = "<" if using_fscore else ">"
progress.console.print( progress.console.print(
f" [dim]⏭ {fname}: {score_label} {candidate_score:.3f}" f" 🔄 {fname}: {score_label} {candidate_score:.3f} {cmp_op} {target_score:.3f},"
f" not {cmp_op} {target_str}, skipping[/dim]" f" replacing {target_frigate_file}{better_note}"
) )
progress.advance(upload_task) if delete_frigate_person_files(name, [target_frigate_file]):
continue remove_frigate_file(name, target_frigate_file)
person_has_fscores = has_frigate_scores(name)
target_frigate_file, _target_asset_id, target_score = target effective_count -= 1
cmp_op = "<" if using_fscore else ">" min_quality_score_for_slot = None if using_fscore else target_score
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:
tracker_ok = True
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:
tracker_ok = False
# Upload to Frigate succeeded — don't retry on tracker
# failure or we'd upload a duplicate to Frigate.
logger.error(
"Tracker write failed for %s — upload succeeded"
" but asset may be re-selected next run: %s",
fname, tracker_exc,
)
if tracker_ok:
if pre_fscore is not None:
person_has_fscores = True
actually_uploaded.append((fname, asset_id))
break
else: 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:
tracker_ok = True
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:
tracker_ok = False
# Upload to Frigate succeeded — don't retry on tracker
# failure or we'd upload a duplicate to Frigate.
logger.error(
"Tracker write failed for %s — upload succeeded"
" but asset may be re-selected next run: %s",
fname, tracker_exc,
)
if tracker_ok:
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: if attempt < max_retries:
logger.warning( logger.warning(
f"Upload attempt {attempt}/{max_retries} for {fname}:" 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 continue
failed += 1 failed += 1
person_failed += 1 person_failed += 1
progress.console.print( 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)
_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" {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: if min_quality_score_for_slot is not None:
logger.warning( logger.warning(
f"{name}: freed replacement slot (floor {min_quality_score_for_slot:.3f})" f"{name}: freed replacement slot (floor {min_quality_score_for_slot:.3f})"
" was not filled this run — will be available next run" " was not filled this run — will be available next run"
) )
flush_batch(UPLOAD_TRACKER_FILE) finally:
flush_batch(UPLOAD_TRACKER_FILE)
# Batch-map Frigate filenames to asset IDs now that all uploads are done. # Batch-map Frigate filenames to asset IDs now that all uploads are done.
if actually_uploaded and not _skip_reconcile: if actually_uploaded and not _skip_reconcile:
winnow/jobs.py
+3 -3
View File
@@ -69,10 +69,10 @@ def _resolve_strategy(strategy: str, has_embedding: bool) -> tuple[int | str, st
return _getenv_int("LIMIT", 30), "time" return _getenv_int("LIMIT", 30), "time"
custom_limit = _getenv_optional_int("LIMIT") custom_limit = _getenv_optional_int("LIMIT")
if custom_limit is not None: if custom_limit is not None and custom_limit > 0:
if custom_limit == 0:
logger.warning("LIMIT=0 selects zero images — set LIMIT to a positive integer or leave unset for auto")
return custom_limit, "smart" return custom_limit, "smart"
if custom_limit == 0:
logger.warning("LIMIT=0 is invalid — ignoring and using auto strategy")
strategy_map = { strategy_map = {
"adaptive": ("auto", "smart"), "adaptive": ("auto", "smart"),
winnow/upload_tracker.py
+6 -4
View File
@@ -116,9 +116,9 @@ def flush_batch(filename: str) -> None:
"""Write the accumulated cache state for filename to disk.""" """Write the accumulated cache state for filename to disk."""
path = _tracker_path(filename) path = _tracker_path(filename)
key = str(path) key = str(path)
_deferred.discard(key)
if key in _cache: if key in _cache:
_write_to_disk(path, _cache[key]) _write_to_disk(path, _cache[key])
_deferred.discard(key)
def _flat_key(filename: str) -> str: def _flat_key(filename: str) -> str:
@@ -239,9 +239,8 @@ def remove_frigate_file(person_name: str, frigate_filename: str) -> None:
def remove_frigate_files_batch(person_name: str, frigate_filenames: list[str]) -> None: def remove_frigate_files_batch(person_name: str, frigate_filenames: list[str]) -> None:
"""Remove multiple Frigate filenames in a single load/save.""" """Remove multiple Frigate filenames in a single load/save."""
data = _load(UPLOAD_TRACKER_FILE) src = _load(UPLOAD_TRACKER_FILE)
by_person = data.get("by_person", {}) raw = src.get("by_person", {}).get(person_name)
raw = by_person.get(person_name)
if raw is None: if raw is None:
return return
entry = _migrate_entry(raw) entry = _migrate_entry(raw)
@@ -249,7 +248,10 @@ def remove_frigate_files_batch(person_name: str, frigate_filenames: list[str]) -
asset_id = entry["frigate_files"].pop(fn, None) asset_id = entry["frigate_files"].pop(fn, None)
if asset_id is not None and asset_id not in entry["frigate_files"].values(): if asset_id is not None and asset_id not in entry["frigate_files"].values():
entry["frigate_scores"].pop(asset_id, None) 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 by_person[person_name] = entry
data = dict(src)
data["by_person"] = by_person
_save(UPLOAD_TRACKER_FILE, data) _save(UPLOAD_TRACKER_FILE, data)
logger.debug(f"Removed {len(frigate_filenames)} Frigate file mapping(s) for {person_name}") logger.debug(f"Removed {len(frigate_filenames)} Frigate file mapping(s) for {person_name}")