Add confidence fit + quality re-score tooling (Phase 3).
CI / skip-ci-check (pull_request) Successful in 30s
CI / docker-ci (pull_request) Successful in 31s
CI / python-lint (pull_request) Failing after 32s
CI / secret-scan (pull_request) Successful in 38s
CI / admin-unit (pull_request) Successful in 50s
CI / e2e (pull_request) Successful in 3m11s
CI / viewer-unit (pull_request) Successful in 3m33s

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.
This commit is contained in:
2026-08-05 11:33:42 -04:00
parent c2ee81adce
commit b68d7d6507
10 changed files with 772 additions and 38 deletions
+88
View File
@@ -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())
+63
View File
@@ -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())