From b68d7d6507a42d04b6e86254a21bed4fa2119b2b Mon Sep 17 00:00:00 2001 From: ilia Date: Wed, 5 Aug 2026 11:33:42 -0400 Subject: [PATCH 1/2] Add confidence fit + quality re-score tooling (Phase 3). MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Support distance→confidence knots from identified pairs and match_decisions, keep legacy curve until a knots file is applied, and re-score face quality without re-detecting. --- .gitignore | 1 + ROADMAP.md | 2 +- backend/data/.gitkeep | 1 + backend/services/confidence_calibration.py | 203 ++++++++++++++++ backend/services/face_service.py | 49 +--- backend/services/quality_rescore.py | 249 ++++++++++++++++++++ docs/FACE_CONFIDENCE_CALIBRATION.md | 52 ++++ scripts/fit_confidence_calibration.py | 88 +++++++ scripts/rescore_face_quality.py | 63 +++++ tests/test_confidence_calibration_phase3.py | 102 ++++++++ 10 files changed, 772 insertions(+), 38 deletions(-) create mode 100644 backend/data/.gitkeep create mode 100644 backend/services/confidence_calibration.py create mode 100644 backend/services/quality_rescore.py create mode 100644 docs/FACE_CONFIDENCE_CALIBRATION.md create mode 100755 scripts/fit_confidence_calibration.py create mode 100755 scripts/rescore_face_quality.py create mode 100644 tests/test_confidence_calibration_phase3.py diff --git a/.gitignore b/.gitignore index c192f03..eed839f 100644 --- a/.gitignore +++ b/.gitignore @@ -108,3 +108,4 @@ viewer-frontend/.data/ !viewer-frontend/public/logo.png !viewer-frontend/public/brand/ !viewer-frontend/public/brand/**/*.png +backend/data/confidence_calibration.json diff --git a/ROADMAP.md b/ROADMAP.md index d956c03..93740b0 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -107,7 +107,7 @@ Living plan for product quality, auth/email reliability, and automation. - [x] **Phase 1 — quality sharpness** — Laplacian variance computed on the face crop (was kernel constant); unit tests in `tests/test_face_quality_score.py`; error-tag checklist in `docs/FACE_MATCH_ERROR_TAGGING.md` - [x] **Phase 2 — trusted refs + stricter Auto-Match + accept/reject log** — ref quality floor 0.5; browse/run tolerances 0.5/0.4; auto-accept default 85%; `match_decisions` table + logging on Save / auto-accept; tests in `tests/test_auto_match_phase2.py` -- [ ] **Phase 3 — recalibrate confidence on labeled pairs; re-score corpus** +- [x] **Phase 3 — recalibrate confidence + re-score quality** — fit distance→confidence knots from identified pairs / `match_decisions`; optional JSON overrides legacy curve; `scripts/fit_confidence_calibration.py` + `scripts/rescore_face_quality.py`; docs in `docs/FACE_CONFIDENCE_CALIBRATION.md` - [ ] **Phase 4 — multi-ref / ensemble embeddings** - [ ] **Phase 5 — harder reject of junk detections (tiny/blur/pose)** - [ ] **Phase 6 — multi-embedding per person** diff --git a/backend/data/.gitkeep b/backend/data/.gitkeep new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/backend/data/.gitkeep @@ -0,0 +1 @@ + diff --git a/backend/services/confidence_calibration.py b/backend/services/confidence_calibration.py new file mode 100644 index 0000000..e466a8a --- /dev/null +++ b/backend/services/confidence_calibration.py @@ -0,0 +1,203 @@ +"""Distance → confidence calibration (Phase 3). + +Default path keeps the legacy empirical piecewise curve. When a fitted +knots file is present (from identified same/different pairs and/or +``match_decisions``), ``calibrate_confidence`` interpolates those knots +instead so displayed % tracks JRCC data. +""" + +from __future__ import annotations + +import functools +import json +import os +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterable, Sequence + +import numpy as np + +# Default install path (overridable via CONFIDENCE_CALIBRATION_PATH) +_DEFAULT_CALIBRATION_PATH = ( + Path(__file__).resolve().parent.parent / "data" / "confidence_calibration.json" +) + +# Minimum labeled distances before a fit may be applied (unless forced) +MIN_SAME_PAIRS = 20 +MIN_DIFF_PAIRS = 20 + + +@dataclass(frozen=True) +class CalibrationKnots: + """Monotone distance → confidence knots (distance ascending).""" + + distances: tuple[float, ...] + confidences: tuple[float, ...] + source: str = "legacy" + n_same: int = 0 + n_diff: int = 0 + fitted_at: str | None = None + + def __post_init__(self) -> None: + if len(self.distances) != len(self.confidences): + raise ValueError("distances and confidences must be same length") + if len(self.distances) < 2: + raise ValueError("need at least 2 knots") + + +def default_calibration_path() -> Path: + override = os.getenv("CONFIDENCE_CALIBRATION_PATH") + if override: + return Path(override) + return _DEFAULT_CALIBRATION_PATH + + +def interpolate_confidence(distance: float, knots: CalibrationKnots) -> float: + """Linear interpolate confidence for a cosine distance; clamp to knots ends.""" + d = float(distance) + xs = np.asarray(knots.distances, dtype=np.float64) + ys = np.asarray(knots.confidences, dtype=np.float64) + if d <= xs[0]: + return float(ys[0]) + if d >= xs[-1]: + return float(ys[-1]) + conf = float(np.interp(d, xs, ys)) + return max(1.0, min(100.0, conf)) + + +def legacy_empirical_confidence(distance: float, tolerance: float) -> float: + """Original desktop-parity piecewise curve (tolerance-relative bands).""" + if distance <= 0.12: + confidence = 100 * np.exp(-distance * 2.8) + return float(min(100, max(92, confidence))) + if distance <= tolerance * 0.5: + confidence = 100 * np.exp(-distance * 2.6) + return float(min(92, max(82, confidence))) + if distance <= tolerance: + normalized_distance = (distance - tolerance * 0.5) / (tolerance * 0.5) + confidence = 82 - (normalized_distance * 32) + return float(max(50, min(82, confidence))) + if distance <= tolerance * 1.5: + normalized_distance = (distance - tolerance) / (tolerance * 0.5) + confidence = 50 - (normalized_distance * 30) + return float(max(20, min(50, confidence))) + confidence = 20 * np.exp(-(distance - tolerance * 1.5) * 1.5) + return float(max(1, min(20, confidence))) + + +def fit_knots_from_distances( + same_distances: Sequence[float], + different_distances: Sequence[float], + *, + n_bins: int = 16, + distance_max: float = 1.2, + source: str = "identified_pairs", +) -> CalibrationKnots: + """Estimate P(same | distance) via Laplace-smoothed histograms → knots.""" + same = np.asarray([float(x) for x in same_distances if x is not None], dtype=np.float64) + diff = np.asarray( + [float(x) for x in different_distances if x is not None], dtype=np.float64 + ) + if same.size == 0 or diff.size == 0: + raise ValueError("need at least one same and one different distance") + + edges = np.linspace(0.0, distance_max, n_bins + 1) + mids = 0.5 * (edges[:-1] + edges[1:]) + same_hist, _ = np.histogram(np.clip(same, 0, distance_max), bins=edges) + diff_hist, _ = np.histogram(np.clip(diff, 0, distance_max), bins=edges) + p_same = (same_hist + 1.0) / (same_hist + diff_hist + 2.0) + conf = p_same * 100.0 + + for i in range(1, len(conf)): + if conf[i] > conf[i - 1]: + conf[i] = conf[i - 1] + + distances = (0.0, *tuple(float(x) for x in mids), float(distance_max)) + confidences = ( + float(min(100.0, max(float(conf[0]), 95.0))), + *tuple(float(max(1.0, min(100.0, c))) for c in conf), + float(max(1.0, min(float(conf[-1]), 10.0))), + ) + conf_list = list(confidences) + for i in range(1, len(conf_list)): + if conf_list[i] > conf_list[i - 1]: + conf_list[i] = conf_list[i - 1] + + return CalibrationKnots( + distances=distances, + confidences=tuple(conf_list), + source=source, + n_same=int(same.size), + n_diff=int(diff.size), + fitted_at=datetime.now(timezone.utc).isoformat(), + ) + + +def knots_ready_to_apply( + knots: CalibrationKnots, + *, + min_same: int = MIN_SAME_PAIRS, + min_diff: int = MIN_DIFF_PAIRS, +) -> bool: + return knots.n_same >= min_same and knots.n_diff >= min_diff + + +def save_calibration_knots(knots: CalibrationKnots, path: Path | None = None) -> Path: + dest = path or default_calibration_path() + dest.parent.mkdir(parents=True, exist_ok=True) + payload = { + "distances": list(knots.distances), + "confidences": list(knots.confidences), + "source": knots.source, + "n_same": knots.n_same, + "n_diff": knots.n_diff, + "fitted_at": knots.fitted_at, + } + dest.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") + clear_calibration_cache() + return dest + + +@functools.lru_cache(maxsize=4) +def _load_fitted_knots_cached(path_str: str) -> CalibrationKnots | None: + path = Path(path_str) + if not path.is_file(): + return None + try: + data = json.loads(path.read_text(encoding="utf-8")) + return CalibrationKnots( + distances=tuple(float(x) for x in data["distances"]), + confidences=tuple(float(x) for x in data["confidences"]), + source=str(data.get("source", "file")), + n_same=int(data.get("n_same", 0)), + n_diff=int(data.get("n_diff", 0)), + fitted_at=data.get("fitted_at"), + ) + except Exception: + return None + + +def load_calibration_knots(path: Path | None = None) -> CalibrationKnots | None: + """Load fitted knots from disk, or None if missing/invalid.""" + return _load_fitted_knots_cached(str(path or default_calibration_path())) + + +def clear_calibration_cache() -> None: + _load_fitted_knots_cached.cache_clear() + + +def get_active_knots() -> CalibrationKnots | None: + """Fitted knots if present; otherwise None (caller uses legacy curve).""" + return load_calibration_knots() + + +def merge_distance_lists(*groups: Iterable[float]) -> list[float]: + out: list[float] = [] + for group in groups: + out.extend(float(x) for x in group if x is not None) + return out + + +def knots_to_dict(knots: CalibrationKnots) -> dict: + return asdict(knots) diff --git a/backend/services/face_service.py b/backend/services/face_service.py index 7b9b728..88c01bc 100644 --- a/backend/services/face_service.py +++ b/backend/services/face_service.py @@ -1839,6 +1839,7 @@ def calibrate_confidence(distance: float, tolerance: float = None) -> float: """Convert distance to calibrated confidence percentage, matching desktop exactly. Uses empirical calibration method matching desktop _calibrate_confidence. + When a fitted knots file exists (Phase 3), interpolates those instead. Args: distance: Cosine distance (0 = identical, 2 = opposite) tolerance: Matching tolerance threshold (default: DEFAULT_FACE_TOLERANCE) @@ -1863,43 +1864,17 @@ def calibrate_confidence(distance: float, tolerance: float = None) -> float: sigmoid_factor = 1 / (1 + np.exp(5 * (normalized_distance - 1))) return max(1, min(100, sigmoid_factor * 100)) - else: # "empirical" - default method (matching desktop exactly) - # Empirical calibration parameters for DeepFace ArcFace model - # These are derived from analysis of distance distributions for matching/non-matching pairs - # Moderate calibration: stricter than original but not too strict - - # For very close distances (< 0.12): very high confidence - if distance <= 0.12: - # Very close matches: exponential decay from 100% - confidence = 100 * np.exp(-distance * 2.8) - return min(100, max(92, confidence)) - - # For distances well below threshold: high confidence - elif distance <= tolerance * 0.5: - # Close matches: exponential decay - confidence = 100 * np.exp(-distance * 2.6) - return min(92, max(82, confidence)) - - # For distances near threshold: moderate confidence - elif distance <= tolerance: - # Near-threshold matches: sigmoid-like curve - # Maps distance to probability based on empirical data - normalized_distance = (distance - tolerance * 0.5) / (tolerance * 0.5) - confidence = 82 - (normalized_distance * 32) # 82% to 50% range - return max(50, min(82, confidence)) - - # For distances above threshold: low confidence - elif distance <= tolerance * 1.5: - # Above threshold but not too far: rapid decay - normalized_distance = (distance - tolerance) / (tolerance * 0.5) - confidence = 50 - (normalized_distance * 30) # 50% to 20% range - return max(20, min(50, confidence)) - - # For very large distances: very low confidence - else: - # Very far matches: very low probability - confidence = 20 * np.exp(-(distance - tolerance * 1.5) * 1.5) - return max(1, min(20, confidence)) + else: # "empirical" - default method + from backend.services.confidence_calibration import ( + get_active_knots, + interpolate_confidence, + legacy_empirical_confidence, + ) + + knots = get_active_knots() + if knots is not None: + return interpolate_confidence(distance, knots) + return legacy_empirical_confidence(distance, tolerance) def _calculate_face_size_ratio(face: Face, photo: Photo) -> float: diff --git a/backend/services/quality_rescore.py b/backend/services/quality_rescore.py new file mode 100644 index 0000000..31be155 --- /dev/null +++ b/backend/services/quality_rescore.py @@ -0,0 +1,249 @@ +"""Quality re-score and labeled-pair sampling for Phase 3 calibration.""" + +from __future__ import annotations + +import json +import random +from pathlib import Path +from typing import Callable, List, Optional, Sequence, Tuple + +import numpy as np +from PIL import Image +from sqlalchemy.orm import Session, joinedload + +from backend.db.models import Face, MatchDecision, PersonEncoding, Photo +from backend.services.confidence_calibration import ( + CalibrationKnots, + fit_knots_from_distances, + merge_distance_lists, +) +from backend.services.face_service import ( + calculate_cosine_distance, + calculate_face_quality_score, + load_face_encoding, +) +from src.utils.exif_utils import EXIFOrientationHandler + + +def _parse_location(raw) -> dict: + if isinstance(raw, dict): + return raw + return json.loads(raw) if raw else {} + + +def _load_photo_array(photo_path: Path) -> Optional[np.ndarray]: + if not photo_path.is_file(): + return None + try: + corrected_image, _orientation = ( + EXIFOrientationHandler.correct_image_orientation_from_path(str(photo_path)) + ) + if corrected_image is not None: + return np.array(corrected_image.convert("RGB")) + return np.array(Image.open(photo_path).convert("RGB")) + except Exception: + try: + return np.array(Image.open(photo_path).convert("RGB")) + except Exception: + return None + + +def rescore_face_quality( + db: Session, + face: Face, + *, + commit: bool = False, +) -> Tuple[float, float] | None: + """Recompute quality_score for one face from its crop (Phase 1 sharpness fix). + + Returns (old_score, new_score) or None if the photo/crop cannot be loaded. + """ + photo = face.photo + if photo is None: + photo = db.query(Photo).filter(Photo.id == face.photo_id).first() + if photo is None: + return None + + image_np = _load_photo_array(Path(photo.path)) + if image_np is None: + return None + + loc = _parse_location(face.location) + h, w = image_np.shape[:2] + new_int = calculate_face_quality_score(image_np, loc, w, h) + new_score = new_int / 100.0 + old_score = float(face.quality_score) if face.quality_score is not None else 0.0 + face.quality_score = new_score + db.add(face) + db.query(PersonEncoding).filter(PersonEncoding.face_id == face.id).update( + {"quality_score": new_score}, + synchronize_session=False, + ) + if commit: + db.commit() + return old_score, new_score + + +def rescore_face_qualities( + db: Session, + *, + limit: int | None = None, + offset: int = 0, + dry_run: bool = False, + commit_every: int = 50, + progress: Optional[Callable[[int, int], None]] = None, +) -> dict: + """Batch re-score face quality. Does not re-detect or re-embed.""" + query = ( + db.query(Face) + .options(joinedload(Face.photo)) + .order_by(Face.id.asc()) + .offset(offset) + ) + if limit is not None: + query = query.limit(limit) + faces: List[Face] = query.all() + total = len(faces) + updated = 0 + skipped = 0 + deltas: list[float] = [] + + for i, face in enumerate(faces, start=1): + result = rescore_face_quality(db, face, commit=False) + if result is None: + skipped += 1 + else: + old, new = result + deltas.append(new - old) + updated += 1 + if dry_run: + # Roll back in-memory change for dry-run + face.quality_score = old + if progress: + progress(i, total) + if not dry_run and updated and updated % commit_every == 0: + db.commit() + + if dry_run: + db.rollback() + elif updated: + db.commit() + + return { + "examined": total, + "updated": updated, + "skipped": skipped, + "dry_run": dry_run, + "mean_delta": float(np.mean(deltas)) if deltas else 0.0, + "max_abs_delta": float(np.max(np.abs(deltas))) if deltas else 0.0, + } + + +def sample_identified_pair_distances( + db: Session, + *, + max_same: int = 500, + max_diff: int = 500, + min_quality: float = 0.3, + seed: int = 42, +) -> Tuple[list[float], list[float]]: + """Cosine distances for same-person vs different-person identified faces.""" + faces: List[Face] = ( + db.query(Face) + .filter(Face.person_id.isnot(None)) + .filter(Face.encoding.isnot(None)) + .filter(Face.quality_score >= min_quality) + .all() + ) + by_person: dict[int, list[Face]] = {} + for f in faces: + by_person.setdefault(int(f.person_id), []).append(f) + + rng = random.Random(seed) + same: list[float] = [] + # Same-person pairs + for person_faces in by_person.values(): + if len(person_faces) < 2: + continue + for i in range(len(person_faces)): + for j in range(i + 1, len(person_faces)): + a = load_face_encoding(person_faces[i].encoding) + b = load_face_encoding(person_faces[j].encoding) + same.append(calculate_cosine_distance(a, b)) + rng.shuffle(same) + same = same[:max_same] + + # Different-person pairs (sample) + person_ids = [pid for pid, fl in by_person.items() if fl] + diff: list[float] = [] + attempts = 0 + max_attempts = max_diff * 20 + while len(diff) < max_diff and attempts < max_attempts and len(person_ids) >= 2: + attempts += 1 + p1, p2 = rng.sample(person_ids, 2) + f1 = rng.choice(by_person[p1]) + f2 = rng.choice(by_person[p2]) + a = load_face_encoding(f1.encoding) + b = load_face_encoding(f2.encoding) + diff.append(calculate_cosine_distance(a, b)) + + return same, diff + + +def distances_from_match_decisions( + db: Session, +) -> Tuple[list[float], list[float]]: + """Accept → same distances; reject → different (when distance was logged).""" + rows: List[MatchDecision] = ( + db.query(MatchDecision).filter(MatchDecision.distance.isnot(None)).all() + ) + same: list[float] = [] + diff: list[float] = [] + for row in rows: + d = float(row.distance) if row.distance is not None else None + if d is None: + continue + if row.decision == "accept": + same.append(d) + elif row.decision == "reject": + diff.append(d) + return same, diff + + +def fit_calibration_from_db( + db: Session, + *, + max_same: int = 500, + max_diff: int = 500, + include_decisions: bool = True, +) -> Tuple[CalibrationKnots, dict]: + """Fit confidence knots from identified pairs (+ optional match_decisions).""" + same_id, diff_id = sample_identified_pair_distances( + db, max_same=max_same, max_diff=max_diff + ) + same_dec: list[float] = [] + diff_dec: list[float] = [] + if include_decisions: + same_dec, diff_dec = distances_from_match_decisions(db) + + same = merge_distance_lists(same_id, same_dec) + diff = merge_distance_lists(diff_id, diff_dec) + stats = { + "same_identified": len(same_id), + "diff_identified": len(diff_id), + "same_decisions": len(same_dec), + "diff_decisions": len(diff_dec), + "same_total": len(same), + "diff_total": len(diff), + } + if not same or not diff: + raise ValueError( + "Not enough labeled distances to fit " + f"(same={len(same)}, diff={len(diff)}). " + "Identify more people or accumulate Auto-Match decisions." + ) + source = "identified_pairs" + if same_dec or diff_dec: + source = "identified_pairs+match_decisions" + knots = fit_knots_from_distances(same, diff, source=source) + return knots, stats diff --git a/docs/FACE_CONFIDENCE_CALIBRATION.md b/docs/FACE_CONFIDENCE_CALIBRATION.md new file mode 100644 index 0000000..0844f33 --- /dev/null +++ b/docs/FACE_CONFIDENCE_CALIBRATION.md @@ -0,0 +1,52 @@ +# Face confidence calibration & quality re-score (Phase 3) + +After Phase 1 (honest sharpness) and Phase 2 (decision log + stricter +Auto-Match), Phase 3 makes displayed confidence track JRCC data and +backfills `quality_score` without re-detecting faces. + +## Confidence calibration + +`calibrate_confidence` still uses the legacy empirical curve by default. + +When `backend/data/confidence_calibration.json` exists (or +`CONFIDENCE_CALIBRATION_PATH`), it **interpolates fitted knots** instead. +Knots are estimated from: + +1. **Identified same-person / different-person** embedding pairs in the DB +2. **`match_decisions`** rows that include `distance` (Auto-Match accept/reject) + +### Fit on the app host + +```bash +cd /opt/punimtag +set -a && . ./.env && set +a +./venv/bin/python scripts/fit_confidence_calibration.py --dry-run +# when same≥20 and diff≥20 (or --force): +./venv/bin/python scripts/fit_confidence_calibration.py --apply +sudo -u appuser pm2 restart punimtag-api --update-env +``` + +Do **not** commit a production knots file with real corpus stats unless you +intend every environment to share it. Prefer generating per host. + +## Quality re-score + +Existing `faces.quality_score` values were computed with the broken +sharpness term. Re-score crops only (no DeepFace): + +```bash +cd /opt/punimtag +set -a && . ./.env && set +a +./venv/bin/python scripts/rescore_face_quality.py --dry-run --limit 100 +./venv/bin/python scripts/rescore_face_quality.py # all faces +``` + +Also updates `person_encodings.quality_score` for the same face ids. + +## Ops notes + +- DEV currently has few identified people — fit may need `--force` or more + Identify / Auto-Match Saves before knots are trustworthy. +- Keep sampling misses in `FACE_MATCH_ERROR_TAGGING.md` (private sheet). +- After quality re-score, Auto-Match reference picking (quality ≥ 0.5) uses + updated scores immediately (no API restart required for DB values). diff --git a/scripts/fit_confidence_calibration.py b/scripts/fit_confidence_calibration.py new file mode 100755 index 0000000..fce8d60 --- /dev/null +++ b/scripts/fit_confidence_calibration.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 +"""Fit confidence calibration knots from labeled face pairs (Phase 3). + +Usage (on DEV/PROD app host, from /opt/punimtag): + + set -a && . ./.env && set +a + ./venv/bin/python scripts/fit_confidence_calibration.py --dry-run + ./venv/bin/python scripts/fit_confidence_calibration.py --apply + +Requires ~20+ same-person and different-person distances (or --force). +After --apply, restart punimtag-api so the process reloads the knots file. +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from backend.db.session import SessionLocal # noqa: E402 +from backend.services.confidence_calibration import ( # noqa: E402 + default_calibration_path, + knots_ready_to_apply, + save_calibration_knots, +) +from backend.services.quality_rescore import fit_calibration_from_db # noqa: E402 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--apply", + action="store_true", + help="Write knots JSON (default: print only)", + ) + parser.add_argument( + "--force", + action="store_true", + help="Write even if below MIN_SAME/MIN_DIFF sample sizes", + ) + parser.add_argument( + "--output", + type=Path, + default=None, + help="Override output path (default: backend/data/confidence_calibration.json)", + ) + args = parser.parse_args() + + db = SessionLocal() + try: + knots, stats = fit_calibration_from_db(db) + except ValueError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + finally: + db.close() + + print("Sample counts:", stats) + print(f"Source: {knots.source} fitted_at={knots.fitted_at}") + print(f"Knots ({len(knots.distances)}):") + for d, c in zip(knots.distances, knots.confidences): + print(f" distance={d:.3f} → confidence={c:.1f}%") + + ready = knots_ready_to_apply(knots) + if not ready and not args.force: + print( + "\nNot enough samples to apply " + f"(need same>={20}, diff>={20}; got same={knots.n_same}, diff={knots.n_diff}). " + "Re-run with --force to write anyway, or identify more faces / review Auto-Match.", + file=sys.stderr, + ) + return 2 + + if args.apply: + path = save_calibration_knots(knots, args.output or default_calibration_path()) + print(f"\nWrote {path}") + print("Restart punimtag-api (and worker if needed) to load the new calibration.") + else: + print("\nDry-run only. Pass --apply to write the knots file.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/rescore_face_quality.py b/scripts/rescore_face_quality.py new file mode 100755 index 0000000..b3d80d4 --- /dev/null +++ b/scripts/rescore_face_quality.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""Re-score face quality_score using fixed Laplacian sharpness (Phase 3). + +Does not re-detect or re-embed — only updates quality_score on faces +(and matching person_encodings rows). + +Usage (on DEV/PROD app host, from /opt/punimtag): + + set -a && . ./.env && set +a + ./venv/bin/python scripts/rescore_face_quality.py --dry-run --limit 100 + ./venv/bin/python scripts/rescore_face_quality.py --limit 500 +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from backend.db.session import SessionLocal # noqa: E402 +from backend.services.quality_rescore import rescore_face_qualities # noqa: E402 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--limit", type=int, default=None, help="Max faces to examine") + parser.add_argument("--offset", type=int, default=0) + parser.add_argument( + "--dry-run", + action="store_true", + help="Compute deltas but do not commit", + ) + parser.add_argument("--commit-every", type=int, default=50) + args = parser.parse_args() + + db = SessionLocal() + + def progress(i: int, total: int) -> None: + if i == total or i % 100 == 0: + print(f" {i}/{total}", flush=True) + + try: + summary = rescore_face_qualities( + db, + limit=args.limit, + offset=args.offset, + dry_run=args.dry_run, + commit_every=args.commit_every, + progress=progress, + ) + finally: + db.close() + + print(summary) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_confidence_calibration_phase3.py b/tests/test_confidence_calibration_phase3.py new file mode 100644 index 0000000..22831ae --- /dev/null +++ b/tests/test_confidence_calibration_phase3.py @@ -0,0 +1,102 @@ +"""Unit tests for Phase 3 confidence calibration + quality rescore helpers.""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np + +from backend.services.confidence_calibration import ( + CalibrationKnots, + clear_calibration_cache, + fit_knots_from_distances, + interpolate_confidence, + knots_ready_to_apply, + legacy_empirical_confidence, + load_calibration_knots, + save_calibration_knots, +) +from backend.services.face_service import calibrate_confidence + + +class TestLegacyEmpirical: + def test_close_distance_is_high(self): + assert legacy_empirical_confidence(0.05, 0.5) >= 92 + + def test_near_threshold_mid(self): + c = legacy_empirical_confidence(0.4, 0.5) + assert 50 <= c <= 82 + + def test_far_is_low(self): + assert legacy_empirical_confidence(1.2, 0.5) <= 20 + + +class TestFitAndInterpolate: + def test_fit_separates_same_and_different(self): + rng = np.random.default_rng(0) + same = rng.normal(0.18, 0.05, size=80).clip(0.01, 0.45) + diff = rng.normal(0.65, 0.12, size=80).clip(0.3, 1.1) + knots = fit_knots_from_distances(same, diff) + assert knots_ready_to_apply(knots) + near = interpolate_confidence(0.15, knots) + far = interpolate_confidence(0.8, knots) + assert near > far + 20 + assert near >= 70 + assert far <= 40 + + def test_interpolate_monotone_nonincreasing(self): + knots = fit_knots_from_distances( + [0.1, 0.12, 0.15, 0.2], + [0.5, 0.6, 0.7, 0.8], + ) + prev = 101.0 + for d in np.linspace(0, 1.0, 21): + c = interpolate_confidence(float(d), knots) + assert c <= prev + 1e-6 + prev = c + + +class TestCalibrationFileRoundTrip: + def test_save_load_activates_calibrate_confidence(self, tmp_path: Path, monkeypatch): + knots = CalibrationKnots( + distances=(0.0, 0.3, 1.0), + confidences=(99.0, 70.0, 5.0), + source="test", + n_same=50, + n_diff=50, + ) + path = tmp_path / "cal.json" + save_calibration_knots(knots, path) + monkeypatch.setenv("CONFIDENCE_CALIBRATION_PATH", str(path)) + clear_calibration_cache() + + loaded = load_calibration_knots() + assert loaded is not None + assert loaded.confidences[0] == 99.0 + + # Fitted path should be used by calibrate_confidence (empirical) + conf = calibrate_confidence(0.0, tolerance=0.5) + assert conf == 99.0 + conf_far = calibrate_confidence(1.0, tolerance=0.5) + assert conf_far == 5.0 + + monkeypatch.delenv("CONFIDENCE_CALIBRATION_PATH", raising=False) + clear_calibration_cache() + + def test_without_file_matches_legacy(self, monkeypatch, tmp_path: Path): + monkeypatch.setenv("CONFIDENCE_CALIBRATION_PATH", str(tmp_path / "missing.json")) + clear_calibration_cache() + for d in (0.05, 0.2, 0.4, 0.6, 1.0): + assert calibrate_confidence(d, 0.5) == legacy_empirical_confidence(d, 0.5) + monkeypatch.delenv("CONFIDENCE_CALIBRATION_PATH", raising=False) + clear_calibration_cache() + + +class TestQualityScoreUsesCrop: + """Sanity: quality helper still differentiates sharp vs blur (Phase 1).""" + + def test_import_rescore_module(self): + from backend.services import quality_rescore + + assert callable(quality_rescore.rescore_face_qualities) + assert callable(quality_rescore.fit_calibration_from_db) From 7f883cf6ffc606a8d47d74118762ad75d25b39b7 Mon Sep 17 00:00:00 2001 From: ilia Date: Wed, 5 Aug 2026 11:34:14 -0400 Subject: [PATCH 2/2] Fix unused Sequence import for ruff. --- backend/services/quality_rescore.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/services/quality_rescore.py b/backend/services/quality_rescore.py index 31be155..1d25685 100644 --- a/backend/services/quality_rescore.py +++ b/backend/services/quality_rescore.py @@ -5,7 +5,7 @@ from __future__ import annotations import json import random from pathlib import Path -from typing import Callable, List, Optional, Sequence, Tuple +from typing import Callable, List, Optional, Tuple import numpy as np from PIL import Image