dtourolle
f891e579c5
The parser reads a tag up to end of line, so `# TRACES: GR-004 | SR-001 — prose` swallowed the prose into the tag and the row went unmatched. Splitting the comment leaves the tag greppable by the same pattern as the code tags and the commit trailers, which is the point of the house format. Mechanical throughout; no logic touched. The regenerated report reflects this session's new tags: 137 -> 148 found, and one more tagged-but-unexecuted, which is the SuperHero accuracy assertion that is documented but not yet a test.
283 lines
13 KiB
Python
283 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
optimize.py — Differential Evolution over pipeline thresholds, scored against X-Ray.
|
||
|
||
Replaces the coarse grid sweep with scipy's differential_evolution over the
|
||
continuous knob space. Each candidate config is a full-9-film replay (real KPN
|
||
nodes) scored against Amazon X-Ray presence, micro-averaged. The gallery is loaded
|
||
once per process (binding caches by path), so an evaluation is just N cheap replays.
|
||
|
||
Objective: **maximize micro-F1** (DE minimizes, so we return -F1). NOTE: X-Ray recall
|
||
is a face-vs-cast-in-scene ceiling (see [[xray-validation-results]]), so unconstrained
|
||
F1 tends to push prob_threshold DOWN to recover unreachable recall — trading real
|
||
precision for it. We therefore log precision/recall at every evaluation and print
|
||
them at the optimum so the trade-off is visible and you can pick another operating
|
||
point from the trajectory (--trajectory).
|
||
|
||
Usage:
|
||
python scripts/optimizer/optimize.py --manifest films.json \
|
||
--gallery gallery_arcface_w600k_r50.json \
|
||
--params prob_threshold:0.5:0.999 anneal_sec:1:30 extinction_sec:1:15 \
|
||
--popsize 20 --maxiter 25 --trajectory traj.json
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
from scipy.optimize import differential_evolution
|
||
|
||
REPO = Path(__file__).resolve().parent.parent.parent
|
||
sys.path.insert(0, str(REPO / "scripts" / "optimizer"))
|
||
sys.path.insert(0, str(REPO / "scripts" / "validation"))
|
||
sys.path.insert(0, str(REPO / "scripts"))
|
||
|
||
import json as _json
|
||
import os
|
||
import subprocess
|
||
import tempfile
|
||
import threading
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
|
||
# Concurrent per-eval replays. Each replay is an isolated subprocess, so parallelism
|
||
# is deadlock-safe; with 9 films/eval, 8 workers replays nearly all at once. Tune via
|
||
# REPLAY_WORKERS (8 is the measured sweet spot on this 24GB GPU).
|
||
REPLAY_WORKERS = int(os.environ.get("REPLAY_WORKERS", "8"))
|
||
|
||
# DE-level parallelism: how many population candidates get evaluated concurrently
|
||
# (each spawning its own REPLAY_WORKERS film subprocesses). Total concurrent GPU
|
||
# replay processes ≈ DE_WORKERS × min(REPLAY_WORKERS, n_films). Threads, not
|
||
# multiprocessing — each objective() call just waits on subprocess.run, so threads
|
||
# share the GIL fine and avoid pickling the objective/gallery-key cache.
|
||
DE_WORKERS = int(os.environ.get("DE_WORKERS", "1"))
|
||
|
||
from second_score import score_seconds # noqa: E402 uniform per-second TPI/FPI scoring
|
||
from sample_eval import load_gallery_keys # noqa: E402
|
||
from replay import dump_embedder_stamp # noqa: E402
|
||
from sae_stamp import EmbedderMismatch, verify_gallery_stamp # noqa: E402
|
||
|
||
_GAL_KEYS: dict = {} # gallery path → key set (fair-recall FN mask), loaded once
|
||
_REPLAY_TIMEOUT = 45 # seconds per film; a wedged replay is killed, not left to hang
|
||
|
||
REPLAY_CLI = str(Path(__file__).resolve().parent / "replay.py")
|
||
|
||
|
||
def _gallery_keys(path):
|
||
if path not in _GAL_KEYS:
|
||
_GAL_KEYS[path] = load_gallery_keys(path)
|
||
return _GAL_KEYS[path]
|
||
|
||
|
||
def _replay_subprocess(dump, gallery, cfg, build_dir):
|
||
"""Run one replay in a SUBPROCESS with a timeout, returning its presence JSON.
|
||
|
||
In-process replay intermittently DEADLOCKS at network teardown — a KPN worker
|
||
stuck mid-rocBLAS GEMM inside the ROCm driver makes ~PyNode's jthread.join() hang
|
||
forever (root-caused via gdb, 2026-07-15). Isolating each replay means a wedged
|
||
GPU thread only kills that subprocess; the sweep continues. Returns None on
|
||
timeout/failure (the caller drops that film from the average)."""
|
||
with tempfile.NamedTemporaryFile("r", suffix=".json", delete=False) as tf:
|
||
out = tf.name
|
||
argv = [sys.executable, REPLAY_CLI, "--dump", dump, "--gallery", gallery,
|
||
"--out", out, "--build-dir", build_dir]
|
||
for k, v in cfg.items():
|
||
if isinstance(v, bool): # store_true flags: pass the flag, not a value
|
||
if v:
|
||
argv.append(f"--{k.replace('_', '-')}")
|
||
else:
|
||
argv += [f"--{k.replace('_', '-')}", str(v)]
|
||
try:
|
||
subprocess.run(argv, timeout=_REPLAY_TIMEOUT, capture_output=True, check=True)
|
||
return _json.loads(Path(out).read_text())
|
||
except (subprocess.TimeoutExpired, subprocess.CalledProcessError,
|
||
FileNotFoundError, ValueError) as e:
|
||
print(f"[opt] replay failed for {Path(dump).name}: {type(e).__name__}",
|
||
file=sys.stderr)
|
||
return None
|
||
finally:
|
||
try:
|
||
Path(out).unlink()
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def evaluate(cfg, films, build_dir, step=None):
|
||
"""Objective = MACRO-mean over films of each film's duration-weighted per-scene F1.
|
||
|
||
Each film's replay runs in a subprocess (timeout-guarded) to survive the
|
||
intermittent ROCm teardown deadlock. If ANY film's replay times out, this
|
||
evaluation is scored f1=0.0 (see below) rather than averaging over the
|
||
survivors — a partial-coverage eval must never look better than a complete
|
||
one, or DE will converge onto configs that make the hardest film time out.
|
||
(An earlier version averaged over survivors, which silently rewarded
|
||
truncation; the rep4 `mbf_full_noexp` winner was one such corrupted eval.)
|
||
|
||
UNIFORM PER-SECOND scoring (second_score.py): every second of the film is sampled;
|
||
GT(t) = the cast of the X-Ray scene containing t, Pred(t) = actors whose presence
|
||
window covers t. Counts instances — TPI / FPI / FN — with FPI weighted 10× when the
|
||
named actor isn't in the film's cast at all (a real misID vs a timing slip). FN
|
||
counts only gallery-known actors (fair recall). Reports agreement_rate = mean
|
||
per-second Jaccard (the "% of on-screen actors we agree with X-Ray about, over
|
||
time"). Objective = macro-mean across films of the per-second weighted F1.
|
||
|
||
expand_gallery: controlled by env SAE_EXPAND (default on). Set SAE_EXPAND=0 to run
|
||
the no-expansion arm — the overnight matrix tests both to quantify what expansion buys.
|
||
|
||
The 9 films' replays run CONCURRENTLY (REPLAY_WORKERS) — each is an isolated
|
||
subprocess, so parallelism is safe (a wedged one only kills itself)."""
|
||
if os.environ.get("SAE_EXPAND", "1") == "1":
|
||
cfg = {**cfg, "expand_gallery": True}
|
||
|
||
def _one(film):
|
||
pj = _replay_subprocess(film["dump"], film.get("gallery"), cfg, build_dir)
|
||
if pj is None:
|
||
return None
|
||
return score_seconds(pj, film["xray"],
|
||
gallery_keys=_gallery_keys(film.get("gallery")))
|
||
|
||
with ThreadPoolExecutor(max_workers=REPLAY_WORKERS) as ex:
|
||
per_film = [m for m in ex.map(_one, films) if m is not None]
|
||
n = len(per_film)
|
||
n_expected = len(films)
|
||
# Incomplete coverage (a replay timed out) is scored as a failure, not
|
||
# averaged over survivors: dropping the hardest film would otherwise inflate
|
||
# the score and let DE reward exactly the configs that cause timeouts. We
|
||
# still record the real survivor counts so a truncated eval is diagnosable
|
||
# in the trajectory (f1=0.0, films_scored < films_expected).
|
||
if n < n_expected:
|
||
agg = {"precision": 0.0, "recall": 0.0, "f1": 0.0, "agreement": 0.0,
|
||
"TPI": sum(m["TPI"] for m in per_film),
|
||
"FPI": sum(m["FPI"] for m in per_film),
|
||
"FPI_misid": sum(m["FPI_misid"] for m in per_film),
|
||
"FN": sum(m["FN"] for m in per_film)}
|
||
agg["films_scored"] = n
|
||
agg["films_expected"] = n_expected
|
||
return agg
|
||
return {"precision": sum(m["precision"] for m in per_film) / n,
|
||
"recall": sum(m["recall"] for m in per_film) / n,
|
||
"f1": sum(m["f1"] for m in per_film) / n,
|
||
"agreement": sum(m["agreement_rate"] for m in per_film) / n,
|
||
"TPI": sum(m["TPI"] for m in per_film),
|
||
"FPI": sum(m["FPI"] for m in per_film),
|
||
"FPI_misid": sum(m["FPI_misid"] for m in per_film),
|
||
"FN": sum(m["FN"] for m in per_film),
|
||
"films_scored": n,
|
||
"films_expected": n_expected}
|
||
|
||
|
||
def main():
|
||
p = argparse.ArgumentParser(description=__doc__,
|
||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
p.add_argument("--manifest", required=True)
|
||
p.add_argument("--gallery", help="default gallery if not per-film")
|
||
p.add_argument("--params", nargs="+", required=True,
|
||
help="knob:lo:hi (e.g. prob_threshold:0.5:0.999). Int knobs kept float, rounded in cfg.")
|
||
p.add_argument("--build-dir", default=str(REPO / "build"))
|
||
p.add_argument("--step", type=float, default=5.0)
|
||
p.add_argument("--popsize", type=int, default=20)
|
||
p.add_argument("--maxiter", type=int, default=25)
|
||
p.add_argument("--seed", type=int, default=0)
|
||
p.add_argument("--trajectory", help="write every evaluation here (JSON lines)")
|
||
p.add_argument("--out", help="write best config + metrics")
|
||
# TRACES: GR-004 | SR-001
|
||
p.add_argument("--require-gallery-stamp", action="store_true",
|
||
help="unprovable gallery/dump model binding is a hard error, "
|
||
"not a warning (also via SAE_REQUIRE_GALLERY_STAMP=1)")
|
||
args = p.parse_args()
|
||
if args.require_gallery_stamp:
|
||
# Set the env var rather than threading a flag through cfg: replays run as
|
||
# subprocesses and inherit it, so strictness cannot be lost in the handoff.
|
||
os.environ["SAE_REQUIRE_GALLERY_STAMP"] = "1"
|
||
|
||
films = json.loads(Path(args.manifest).read_text())
|
||
for f in films:
|
||
f.setdefault("gallery", args.gallery)
|
||
if not Path(f["dump"]).exists():
|
||
sys.exit(f"[opt] missing dump for {f['name']}: {f['dump']}")
|
||
|
||
# TRACES: GR-004 | SR-001
|
||
# every (dump, gallery) pair is checked ONCE here,
|
||
# before the first evaluation. A DE sweep is thousands of replays; discovering
|
||
# a cross-model pair at the end (or never) means every number it produced was
|
||
# noise. Each replay subprocess re-checks its own pair anyway.
|
||
for f in films:
|
||
try:
|
||
verify_gallery_stamp(f["gallery"], stamp=dump_embedder_stamp(f["dump"]),
|
||
embedder_desc=f"embedding dump {Path(f['dump']).name}",
|
||
require_stamp=args.require_gallery_stamp)
|
||
except EmbedderMismatch as e:
|
||
sys.exit(f"[opt] {f['name']}: {e}")
|
||
|
||
names, bounds = [], []
|
||
int_knobs = {"track_max_frames_missing", "cut_inactive_max_frames"}
|
||
for spec in args.params:
|
||
k, lo, hi = spec.split(":")
|
||
names.append(k); bounds.append((float(lo), float(hi)))
|
||
print(f"[opt] DE over {names} bounds={bounds}", file=sys.stderr)
|
||
print(f"[opt] {len(films)} films, popsize={args.popsize}, maxiter={args.maxiter}", file=sys.stderr)
|
||
|
||
traj = []
|
||
evals = [0]
|
||
t0 = time.time()
|
||
traj_lock = threading.Lock()
|
||
|
||
def vec_to_cfg(x):
|
||
cfg = {}
|
||
for k, v in zip(names, x):
|
||
cfg[k] = int(round(v)) if k in int_knobs else float(v)
|
||
return cfg
|
||
|
||
def objective(x):
|
||
cfg = vec_to_cfg(x)
|
||
m = evaluate(cfg, films, args.build_dir, args.step)
|
||
with traj_lock:
|
||
evals[0] += 1
|
||
rec = {"eval": evals[0], "config": cfg, **m, "t": round(time.time() - t0, 1)}
|
||
traj.append(rec)
|
||
print(f"[opt] eval {evals[0]:3d} thr={cfg['prob_threshold']:.2f} "
|
||
f"ann={cfg['anneal_sec']:.0f} ext={cfg['extinction_sec']:.1f} → "
|
||
f"F1={m['f1']*100:.1f}% P={m['precision']*100:.1f}% R={m['recall']*100:.1f}% "
|
||
f"agree={m.get('agreement', 0)*100:.1f}% misID={m.get('FPI_misid', 0)}",
|
||
file=sys.stderr)
|
||
if args.trajectory:
|
||
with open(args.trajectory, "a") as tf:
|
||
tf.write(json.dumps(rec) + "\n")
|
||
return -m["f1"]
|
||
|
||
de_kwargs = dict(
|
||
popsize=args.popsize, maxiter=args.maxiter,
|
||
seed=args.seed, polish=False, tol=1e-4, mutation=(0.5, 1.0), recombination=0.7,
|
||
init="sobol")
|
||
if DE_WORKERS > 1:
|
||
pool = ThreadPoolExecutor(max_workers=DE_WORKERS)
|
||
de_kwargs["workers"] = pool.map
|
||
result = differential_evolution(objective, bounds, **de_kwargs)
|
||
|
||
best_cfg = vec_to_cfg(result.x)
|
||
best = evaluate(best_cfg, films, args.build_dir, args.step)
|
||
print("\n══ DE optimum (by F1) ═══════════════════════════")
|
||
print(f" config : {best_cfg}")
|
||
print(f" F1 : {best['f1']*100:.2f}%")
|
||
print(f" precision: {best['precision']*100:.2f}% recall: {best['recall']*100:.2f}%")
|
||
print(f" TP/FP/FN: {best['TP']}/{best['FP']}/{best['FN']}")
|
||
print(f" evaluations: {evals[0]} time: {time.time()-t0:.0f}s")
|
||
|
||
# Also surface the highest-precision config seen (the ship-safe operating point).
|
||
if traj:
|
||
hp = max(traj, key=lambda r: (r["precision"], r["recall"]))
|
||
print("\n── highest-precision config seen (ship-safe) ──")
|
||
print(f" config : {hp['config']}")
|
||
print(f" P={hp['precision']*100:.2f}% R={hp['recall']*100:.2f}% F1={hp['f1']*100:.2f}%")
|
||
|
||
if args.out:
|
||
Path(args.out).write_text(json.dumps(
|
||
{"best_by_f1": {"config": best_cfg, **best}, "n_evals": evals[0]}, indent=2))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|