refactor: unify env var helpers and remove magic slice in cache

- Add _getenv_num as shared core for _getenv_int/_getenv_float
- Add _getenv_optional_float for FRIGATE_SCORE_CEILING (replaces 9-line inline block)
- Add _getenv_bool; replace 11 inline .lower()-in-("true","1","yes") sites
  across config.py, jobs.py, and cli.py with single call site
- _resolve_strategy no-embedding branch: inline try/except → _getenv_int("LIMIT", 30)
- cache.py: final[:-4] → final.removesuffix(".npy") — assumption is now explicit

CHANGELOG.md

@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.5.16] - 2026-06-15
+
+### Changed
+
+- **`_getenv_int` and `_getenv_float` now share a single `_getenv_num` implementation**: the two helpers were structurally identical (read env var, return typed default if absent, try-cast, warn and return default on `ValueError`) with only the cast differing. Both are now thin wrappers around a private `_getenv_num(name, default, cast)`, eliminating the duplicated warning logic.
+
+- **`FRIGATE_SCORE_CEILING` now uses `_getenv_optional_float`**: the previous 9-line inline block (`os.getenv("FRIGATE_SCORE_CEILING", "").strip()` + try/except) has been replaced with a new `_getenv_optional_float(name) -> float | None` helper that encapsulates the "empty-string means None, parse-error means None" semantics, making it consistent with the other numeric env var helpers.
+
+- **Boolean env vars now use `_getenv_bool`**: the `.lower() in ("true", "1", "yes")` pattern was repeated across 11 sites in `config.py`, `jobs.py`, and `cli.py`. A new `_getenv_bool(name, default)` helper centralises the canonical truthy-string set; all sites have been updated to call it.
+
+- **`_resolve_strategy` no-embedding branch uses `_getenv_int`**: the inline `int(custom_limit)` try/except block in `jobs.py` for the time-spread path has been replaced with `_getenv_int("LIMIT", 30)`, matching the pattern used in `config.py`. The smart-mode path retains its own try/except because its fallback is to the strategy map rather than to a numeric default.
+
+- **`cache.py` tmp path uses `str.removesuffix`**: `final[:-4] + ".tmp.npy"` replaced with `final.removesuffix(".npy") + ".tmp.npy"` — the assumption that the cache path ends in `.npy` is now explicit and self-documenting rather than expressed as a magic numeric slice.
+
 ## [0.5.15] - 2026-06-15
 
 ### Fixed

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "winnow"
-version = "0.5.15"
+version = "0.5.16"
 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

@@ -85,7 +85,7 @@ class EmbeddingCache:
         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[:-4] + ".tmp.npy"
+        tmp = final.removesuffix(".npy") + ".tmp.npy"
         try:
             np.save(tmp, embedding)
             os.replace(tmp, final)

winnow/cli.py

@@ -7,7 +7,7 @@ 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
@@ -145,7 +145,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()
@@ -232,8 +232,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")
+        auto_mode = not sys.stdin.isatty() or _getenv_bool("AUTO_MODE", False)
-        dry_run = os.environ.get("DRY_RUN", "false").lower() in ("true", "1", "yes")
+        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

@@ -12,26 +12,41 @@ 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_int(name: str, default: int) -> int:
+def _getenv_num(name: str, default, cast):
     raw = os.getenv(name)
     if raw is None:
         return default
     try:
-        return int(raw)
+        return cast(raw)
     except ValueError:
-        logging.warning("%s=%r is not a valid integer — using default %s", name, raw, default)
+        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:
-    raw = os.getenv(name)
+    return _getenv_num(name, default, float)
-    if raw is None:
+
-        return default
+
+def _getenv_optional_float(name: str) -> float | None:
+    raw = os.getenv(name, "").strip()
+    if not raw:
+        return None
     try:
         return float(raw)
     except ValueError:
-        logging.warning("%s=%r is not a valid float — using default %s", name, raw, default)
+        logging.warning("%s=%r is not a valid float — ignoring", name, raw)
+        return None
+
+
+def _getenv_bool(name: str, default: bool) -> bool:
+    raw = os.getenv(name)
+    if raw is None:
         return default
+    return raw.lower() in ("true", "1", "yes")
 
 
 class _Config:
@@ -109,25 +124,17 @@ class _Config:
         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 = os.getenv("MERGE_DUPLICATE_PEOPLE", "false").lower() in ("true", "1", "yes")
+        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 = os.getenv("QUALITY_REPLACEMENT", "true").lower() in ("true", "1", "yes")
+        self.QUALITY_REPLACEMENT = _getenv_bool("QUALITY_REPLACEMENT", True)
-        _ceiling_env = os.getenv("FRIGATE_SCORE_CEILING", "").strip()
+        self.FRIGATE_SCORE_CEILING = _getenv_optional_float("FRIGATE_SCORE_CEILING")
-        if _ceiling_env:
+        self.ENABLE_FRIGATE_SCORES = _getenv_bool("ENABLE_FRIGATE_SCORES", True)
-            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 = _getenv_float("FACE_MARGIN", 0.15)
-        self.USE_FULL_RESOLUTION = os.getenv("USE_FULL_RESOLUTION", "true").lower() in ("true", "1", "yes")
+        self.USE_FULL_RESOLUTION = _getenv_bool("USE_FULL_RESOLUTION", True)
-        self.ENABLE_FACE_ALIGNMENT = os.getenv("ENABLE_FACE_ALIGNMENT", "true").lower() in ("true", "1", "yes")
+        self.ENABLE_FACE_ALIGNMENT = _getenv_bool("ENABLE_FACE_ALIGNMENT", True)
-        self.ENABLE_CACHE = os.getenv("ENABLE_CACHE", "true").lower() in ("true", "1", "yes")
+        self.ENABLE_CACHE = _getenv_bool("ENABLE_CACHE", True)
         _data_dir = os.getenv("DATA_DIR")
         _cache_dir_legacy = os.getenv("CACHE_DIR")
         if _data_dir:

winnow/jobs.py

@@ -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
 from .diversity import select_diverse_assets
 from .embeddings import is_embedding_available, load_embedding_model
 from .frigate_api import get_frigate_face_counts
@@ -65,19 +65,10 @@ 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:
+        return _getenv_int("LIMIT", 30), "time"
-            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"
 
+    custom_limit = os.environ.get("LIMIT", "").strip()
     if custom_limit:
         try:
             return int(custom_limit), "smart"
@@ -168,7 +159,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)
@@ -317,7 +308,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]