scene-actor-extraction/scripts/optimizer/second_score.py

#!/usr/bin/env python3
"""
second_score.py — uniform per-second agreement with X-Ray.

TRACES: VR-003 | PR-002

Unlike scene_score.py (which unions our detections over a whole X-Ray scene), this
samples EVERY SECOND of the film and asks: at second t, do we name the same actors
X-Ray says are on screen?

  GT(t)   = the cast set of the X-Ray scene containing t (scenes.csv + people_in_scenes)
  Pred(t) = actors whose presence window [start,end] covers t (the pipeline's output)

Per second we count instances:
  TPI = |Pred ∩ GT|                      true positive instances
  FPI = |Pred − GT|                      false positive instances, split into:
          FPI_misid  — actor NOT in the film's cast at all (a real misID, weighted 10×)
          FPI_incast — actor in the film but not this second (timing/boundary)
  FN  = |GT − Pred|, counting only gallery-known actors (fair recall — ~67% of X-Ray
        cast have no reference embedding and can never be recognised)
  agreement at t = Jaccard |Pred ∩ GT| / |Pred ∪ GT|  — PARTIAL credit, so naming 2
        of 3 actors scores 2/3, not 0. Averaged over sampled seconds → the
        "what fraction of the time do we agree with X-Ray" number. (Exact-set match is
        reported separately as exact_match_rate; it is far harsher and dominated by
        recall.)

Objective (DE): per-second F1 computed with the WEIGHTED FPI, so naming someone who
isn't in the film hurts 10× more than a boundary slip.

Reported: TPI, FPI (+split), FN, precision, recall, F1, and agreement_rate — the
fraction of sampled seconds where we exactly matched X-Ray.
"""
from __future__ import annotations

import argparse
import csv
import json
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(REPO / "scripts" / "validation"))
from identity import keys_for  # noqa: E402


def load_second_timeline(xray_dir: str):
    """Return (timeline, film_cast_keys, duration).

    timeline: dict second -> list of actor key-sets on screen per X-Ray.
    Each second inside a scene [start,end) inherits that scene's cast set.
    """
    d = Path(xray_dir)
    id_to_name = {}
    with open(d / "people.csv", newline="", encoding="utf-8") as f:
        for r in csv.DictReader(f):
            nm = (r.get("name_id") or "").strip()
            if nm:
                id_to_name[nm] = (r.get("person") or "").strip()

    film_cast = set()
    for nm, name in id_to_name.items():
        film_cast |= keys_for(imdb_id=nm, name=name)

    spans = {}
    with open(d / "scenes.csv", newline="", encoding="utf-8") as f:
        for r in csv.DictReader(f):
            sn = (r.get("scene") or "").strip()
            try:
                spans[sn] = (float(r["start"]) / 1000.0, float(r["end"]) / 1000.0)
            except (KeyError, ValueError):
                continue

    scene_cast: dict[str, list] = {}
    with open(d / "people_in_scenes.csv", newline="", encoding="utf-8") as f:
        for r in csv.DictReader(f):
            sn = (r.get("scene") or "").strip()
            nm = (r.get("name_id") or "").strip()
            if sn in spans and nm:
                scene_cast.setdefault(sn, []).append(
                    frozenset(keys_for(imdb_id=nm, name=id_to_name.get(nm))))

    timeline: dict[int, list] = {}
    duration = 0.0
    for sn, (t0, t1) in spans.items():
        duration = max(duration, t1)
        cast = scene_cast.get(sn, [])
        for t in range(int(t0), int(t1)):
            timeline[t] = cast
    return timeline, film_cast, duration


def load_pred_intervals(pred_json: dict):
    """[(keyset, [(t0,t1),...]), ...] for each actor the pipeline named."""
    out = []
    for a in pred_json.get("actors", []):
        keys = frozenset(keys_for(imdb_id=a.get("imdb_id"), tmdb_id=a.get("tmdb_id"),
                                  jellyfin_id=a.get("jellyfin_id"), name=a.get("name")))
        # schema_version 1: scenes is [[t0, t1], ...]; schema_version 2:
        # scenes is [{"start":…, "end":…, "belief":…, "route":…}, …].
        windows = []
        for s in a.get("scenes", []):
            if isinstance(s, dict):
                windows.append((float(s["start"]), float(s["end"])))
            else:
                windows.append((float(s[0]), float(s[1])))
        out.append((keys, windows))
    return out


def _match(P, G):
    """Greedy 1:1 match by key intersection; returns (n_matched, matched_G_mask)."""
    used = [False] * len(G)
    n = 0
    for pa in P:
        for j, ga in enumerate(G):
            if not used[j] and (pa & ga):
                used[j] = True
                n += 1
                break
    return n, used


def score_seconds(pred_json: dict, xray_dir: str, gallery_keys: set | None = None,
                  misid_weight: float = 10.0):
    timeline, film_cast, duration = load_second_timeline(xray_dir)
    pred = load_pred_intervals(pred_json)

    TPI = FPI = FN = 0
    FPI_misid = FPI_incast = 0
    FPI_w = 0.0
    jaccard_sum = 0.0   # partial-credit agreement, summed over seconds
    exact = 0
    n_sec = 0

    for t in sorted(timeline):
        G = [set(a) for a in timeline[t]]
        P = [set(k) for k, wins in pred if any(w0 <= t <= w1 for w0, w1 in wins)]
        # fair recall: only GT actors we could possibly recognise
        if gallery_keys is not None:
            G = [g for g in G if g & gallery_keys]

        tp, matched = _match(P, G)
        # classify each unmatched prediction
        fpi_w = 0.0
        n_fp = 0
        for pa in P:
            if any(pa & ga for ga in G):
                continue
            n_fp += 1
            if pa & film_cast:
                FPI_incast += 1; fpi_w += 1.0
            else:
                FPI_misid += 1; fpi_w += misid_weight
        fn = len(G) - tp

        TPI += tp; FPI += n_fp; FN += fn; FPI_w += fpi_w
        # partial-credit agreement: |∩| / |∪| at this second
        union = tp + n_fp + fn
        if union:
            jaccard_sum += tp / union
        else:
            jaccard_sum += 1.0   # both empty = agreement (nobody on screen)
        if n_fp == 0 and fn == 0:
            exact += 1
        n_sec += 1

    prec = TPI / (TPI + FPI_w) if TPI + FPI_w else 0.0   # weighted (misID hurts 10×)
    prec_raw = TPI / (TPI + FPI) if TPI + FPI else 0.0
    rec = TPI / (TPI + FN) if TPI + FN else 0.0
    f1 = 2 * prec * rec / (prec + rec) if prec + rec else 0.0
    return {"TPI": TPI, "FPI": FPI, "FPI_misid": FPI_misid, "FPI_incast": FPI_incast,
            "FN": FN, "precision": prec, "precision_raw": prec_raw, "recall": rec,
            "f1": f1,
            # partial-credit: mean per-second Jaccard = "% of actors we agree on, over time"
            "agreement_rate": jaccard_sum / n_sec if n_sec else 0.0,
            "exact_match_rate": exact / n_sec if n_sec else 0.0,
            "n_seconds": n_sec, "duration_sec": duration}


def main():
    p = argparse.ArgumentParser(description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--pred", required=True)
    p.add_argument("--xray", required=True)
    p.add_argument("--gallery")
    args = p.parse_args()
    gk = None
    if args.gallery:
        sys.path.insert(0, str(REPO / "scripts" / "validation"))
        from sample_eval import load_gallery_keys
        gk = load_gallery_keys(args.gallery)
    m = score_seconds(json.loads(Path(args.pred).read_text()), args.xray, gk)
    print(f"seconds sampled : {m['n_seconds']} (film {m['duration_sec']:.0f}s)")
    print(f"TPI/FPI/FN      : {m['TPI']}/{m['FPI']}/{m['FN']}")
    print(f"  FPI misID     : {m['FPI_misid']}  (actor not in film — weighted 10x)")
    print(f"  FPI in-cast   : {m['FPI_incast']}")
    print(f"precision (w)   : {m['precision']*100:.1f}%   raw {m['precision_raw']*100:.1f}%")
    print(f"recall          : {m['recall']*100:.1f}%")
    print(f"F1 (weighted)   : {m['f1']*100:.1f}%")
    print(f"AGREEMENT       : {m['agreement_rate']*100:.1f}%  (mean per-second % of actors "
          f"we agree on with X-Ray)")
    print(f"  exact-set match: {m['exact_match_rate']*100:.1f}% of seconds (harsher, "
          f"all-or-nothing)")


if __name__ == "__main__":
    main()