#!/usr/bin/env python3 """ first_fpi_frames.py — for every film, find every DISTINCT out-of-cast name (misID) the raw replay stream ever reports, and render the exact second each one FIRST appears, with the proper montage renderer (dump_scene_montage.py: Onscreen/Offscreen panel, TPI/FPI/FN legend, ghosts never drawn as boxes — imported directly, not the scene-level best/worst picker, which can land on a different second within the same scene). One rule, applied uniformly across all 9 films and every distinct wrong name in each — no manual per-film picking, no stopping at the first name found. """ import csv import json import sys from pathlib import Path import cv2 REPO = Path(__file__).resolve().parent.parent.parent sys.path.insert(0, str(REPO / "scripts" / "validation")) sys.path.insert(0, str(REPO / "scripts" / "optimizer")) from identity import keys_for # noqa: E402 from sample_eval import load_gallery_keys # noqa: E402 from dump_scene_montage import ( # noqa: E402 classify_second, extract_frame, render_frame, load_scene_cast, load_dump_faces_by_second, load_raw_by_second, ) FILMS = [ ("Benny___Joon", "experiments/xray/scene_level_movie_data_XRay_US/xrays/4808_Benny__Joon"), ("Café_Society", "experiments/xray/scene_level_movie_data_XRay_US/xrays/225_Cafe_Society"), ("Downton_Abbey__A_New_Era", "experiments/xray/scene_level_movie_data_XRay_US/xrays/19_Downton_Abbey_A_New_Era"), ("Lord_of_War", "experiments/xray/scene_level_movie_data_XRay_US/xrays/2474_Lord_of_War"), ("Lovelace", "experiments/xray/scene_level_movie_data_XRay_US/xrays/4108_Lovelace"), ("Scarface", "experiments/xray/scene_level_movie_data_XRay_US/xrays/197_Scarface"), ("Sound_of_Metal", "experiments/xray/scene_level_movie_data_XRay_US/xrays/6278_Sound_of_Metal"), ("The_Many_Saints_of_Newark", "experiments/xray/scene_level_movie_data_XRay_US/xrays/900_The_Many_Saints_Of_Newark"), ("Valerian_and_the_City_of_a_Thousand_Plan", "experiments/xray/scene_level_movie_data_XRay_US/xrays/5312_Valerian_and_the_City_of_a_Thousand_Planets"), ] MOVIE_ROOT = Path("/mnt/movies") def load_film_cast_keys(xray_dir: Path) -> set: keys = set() with open(xray_dir / "people.csv", newline="", encoding="utf-8") as f: for r in csv.DictReader(f): nm = (r.get("name_id") or "").strip() person = (r.get("person") or "").strip() if nm or person: keys |= keys_for(imdb_id=nm, name=person) return keys def find_movie_file(slug: str) -> str | None: # dump HDF5 attrs carry the exact path used at dump time import h5py for model in ("LVFace-B_Glint360K",): p = REPO / f"experiments/dumps/{model}/dump_{slug}.h5" if p.exists(): with h5py.File(p, "r") as f: return f.attrs.get("movie") return None def find_scene_id(xray_dir: Path, t: int) -> str | None: with open(xray_dir / "scenes.csv", newline="", encoding="utf-8") as f: for r in csv.DictReader(f): try: t0, t1 = float(r["start"]) / 1000.0, float(r["end"]) / 1000.0 except (KeyError, ValueError): continue if t0 <= t < t1: return (r.get("scene") or "").strip() return None def main(): out_root = REPO / "experiments/results/holdout/montage_bestworst" summary = [] for slug, xray_rel in FILMS: xray_dir = REPO / xray_rel raw_path = out_root / f"raw_{slug}.jsonl" if not raw_path.exists(): print(f"SKIP {slug}: no raw file", file=sys.stderr) continue cast_keys = load_film_cast_keys(xray_dir) # every distinct out-of-cast name -> first second it appears first_seen: dict[str, int] = {} with open(raw_path) as f: for line in f: d = json.loads(line) if d.get("eof"): continue for a in d.get("visible_actors", []): name = a.get("name") if not name or name in first_seen: continue ak = keys_for(imdb_id=a.get("imdb_id"), name=name, jellyfin_id=a.get("jellyfin_id")) if not (ak & cast_keys): first_seen[name] = int(d["timestamp_sec"]) if not first_seen: print(f"{slug}: no out-of-cast FPI in the whole film", file=sys.stderr) summary.append((slug, None, None)) continue print(f"{slug}: {len(first_seen)} distinct out-of-cast name(s)", file=sys.stderr) movie = find_movie_file(slug) if not movie or not Path(movie).exists(): print(f" SKIP render: movie file not found ({movie})", file=sys.stderr) for name, t in first_seen.items(): summary.append((slug, name, t)) continue dump_path = REPO / f"experiments/dumps/LVFace-B_Glint360K/dump_{slug}.h5" gallery_path = REPO / "experiments/galleries/gallery_LVFace-B_Glint360K.h5" gallery_keys = load_gallery_keys(str(gallery_path)) raw_by_second = load_raw_by_second(str(raw_path)) dump_faces_by_second = load_dump_faces_by_second(str(dump_path)) scene_cast = load_scene_cast(str(xray_dir)) for name, t in sorted(first_seen.items(), key=lambda kv: kv[1]): scene_id = find_scene_id(xray_dir, t) gt_cast = scene_cast.get(scene_id, set()) gt_cast = {g for g in gt_cast if g & gallery_keys} score, tpi_boxes, fpi_boxes, entries, has_outofcast = classify_second( t, gt_cast, cast_keys, raw_by_second, dump_faces_by_second) slug_name = name.lower().replace(" ", "_").replace("'", "") out_dir = out_root / slug / f"first_fpi_{slug_name}" out_dir.mkdir(parents=True, exist_ok=True) out_path = out_dir / f"first_fpi_t{t:06d}.jpg" extract_frame(movie, t, out_path) canvas = render_frame(out_path, t, tpi_boxes, fpi_boxes, entries) if canvas is not None: cv2.imwrite(str(out_path), canvas) print(f" {name!r} t={t}s -> {out_path} (outofcast={has_outofcast})", file=sys.stderr) summary.append((slug, name, t)) print("\n=== summary ===", file=sys.stderr) for slug, name, t in summary: print(f" {slug:45s} {name!r:30s} t={t}", file=sys.stderr) if __name__ == "__main__": main()