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.
277 lines
12 KiB
Python
277 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
replay.py — replay a dumped embedding HDF5 through the real KPN downstream nodes.
|
||
|
||
TRACES: VR-002 | PR-002
|
||
|
||
Reads an embedding dump (scripts/optimizer/SCHEMA.md), feeds each frame as an
|
||
EmbeddedSceneFrame into a Python-assembled KPN network wiring the *real* C++
|
||
face_tracker → identity_matcher → scene_tracker, and returns the same presence-window
|
||
JSON that scene_analyze's result_sink produces (minimal schema). No decode, no GPU
|
||
embedding — only the cheap downstream tail runs, so a sweep can vary Config knobs
|
||
freely. See [[kpn-python-replay-optimizer]].
|
||
|
||
CLI:
|
||
python scripts/optimizer/replay.py --dump film.h5 --gallery gallery.json \
|
||
--out replayed.json [--prob-threshold 0.99] [--anneal 10] ...
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import h5py
|
||
import numpy as np
|
||
|
||
REPO = Path(__file__).resolve().parent.parent.parent
|
||
sys.path.insert(0, str(REPO / "scripts"))
|
||
from sae_stamp import verify_gallery_stamp # noqa: E402
|
||
|
||
|
||
def dump_embedder_stamp(dump_path: str) -> dict:
|
||
"""The GR-004 embedder stamp recorded in an embedding dump.
|
||
|
||
A replay has no live embedder — the dump IS the embedder as far as the gallery
|
||
is concerned, so the dump's stamp is what the gallery must be checked against.
|
||
Dumps written before GR-004 have no attributes and yield an empty stamp, which
|
||
the check reports as unverifiable rather than silently accepting."""
|
||
with h5py.File(dump_path, "r") as f:
|
||
name = f.attrs.get("embedder_model", "")
|
||
sha = f.attrs.get("embedder_sha256", "")
|
||
dec = lambda v: v.decode() if isinstance(v, bytes) else ("" if v is None else str(v))
|
||
return {"model_name": dec(name), "model_sha256": dec(sha), "embed_dim": 512}
|
||
|
||
|
||
def load_frames(dump_path: str, min_conf: float = 0.0):
|
||
"""Yield EmbeddedSceneFrame dicts from the HDF5 dump, then a trailing EOF.
|
||
|
||
`min_conf` drops detections below that detector confidence before they reach the
|
||
matcher — an UPWARD-only detector_conf sweep on already-dumped faces (the dump was
|
||
made at detector_conf=0.5, so 0.5 is the floor). Lets us test whether near-threshold
|
||
detections are real faces (raising min_conf hurts recall) or phantoms (it helps
|
||
precision at no recall cost)."""
|
||
with h5py.File(dump_path, "r") as f:
|
||
ts = f["frames/timestamp_sec"][:]
|
||
fidx = f["frames/frame_idx"][:]
|
||
cut = f["frames/is_cut"][:]
|
||
off = f["frames/face_offset"][:]
|
||
cnt = f["frames/face_count"][:]
|
||
emb = f["faces/embedding"][:]
|
||
bbox = f["faces/bbox"][:]
|
||
lmk = f["faces/landmarks"][:]
|
||
conf = f["faces/confidence"][:]
|
||
movie = f.attrs.get("movie", "")
|
||
fps = float(f.attrs.get("sample_fps", 1.0))
|
||
|
||
frames = []
|
||
for i in range(len(ts)):
|
||
s, n = int(off[i]), int(cnt[i])
|
||
keep = slice(s, s + n)
|
||
c = np.ascontiguousarray(conf[keep], dtype=np.float32)
|
||
if min_conf > 0.0 and n:
|
||
m = c >= min_conf
|
||
sel = np.where(m)[0]
|
||
frames.append({
|
||
"timestamp_sec": float(ts[i]), "frame_idx": int(fidx[i]),
|
||
"is_cut": bool(cut[i]), "eof": False,
|
||
"bbox": np.ascontiguousarray(bbox[keep][sel], dtype=np.float32),
|
||
"landmarks": np.ascontiguousarray(lmk[keep][sel], dtype=np.float32),
|
||
"confidence": np.ascontiguousarray(c[sel], dtype=np.float32),
|
||
"embeddings": np.ascontiguousarray(emb[keep][sel], dtype=np.float32),
|
||
})
|
||
else:
|
||
frames.append({
|
||
"timestamp_sec": float(ts[i]), "frame_idx": int(fidx[i]),
|
||
"is_cut": bool(cut[i]), "eof": False,
|
||
"bbox": np.ascontiguousarray(bbox[keep], dtype=np.float32),
|
||
"landmarks": np.ascontiguousarray(lmk[keep], dtype=np.float32),
|
||
"confidence": c,
|
||
"embeddings": np.ascontiguousarray(emb[keep], dtype=np.float32),
|
||
})
|
||
last_ts = float(ts[-1]) if len(ts) else 0.0
|
||
frames.append({"timestamp_sec": last_ts, "eof": True})
|
||
return frames, str(movie), fps
|
||
|
||
|
||
def replay(dump_path: str, gallery: str, cfg: dict, build_dir: str, stop: bool = True,
|
||
raw_out: str | None = None) -> dict:
|
||
"""Run the dump through the real KPN chain; return minimal-schema presence JSON.
|
||
|
||
cfg may include "detector_conf" to prune dumped detections below that confidence
|
||
(upward-only from the 0.5 dump floor) before matching.
|
||
|
||
raw_out: if set, also write the raw per-frame annotations (timestamp, actor_idx,
|
||
name, bbox, similarity — one entry per input frame, before merging into windows)
|
||
as JSON lines to this path. Needed to draw bounding boxes on extracted frames;
|
||
the merged window schema returned by this function has no per-frame bbox."""
|
||
sys.path.insert(0, build_dir)
|
||
import sae_kpn
|
||
|
||
# TRACES: GR-004 | SR-001
|
||
# checked here, before any network is built, so a
|
||
# cross-model replay dies with one readable error instead of producing a
|
||
# plausible-looking score. add_identity_matcher re-checks it C++-side below;
|
||
# that is the backstop for any other caller of the binding.
|
||
stamp = dump_embedder_stamp(dump_path)
|
||
verify_gallery_stamp(gallery, stamp=stamp,
|
||
embedder_desc=f"embedding dump {Path(dump_path).name}",
|
||
require_stamp=bool(cfg.get("require_gallery_stamp", False)))
|
||
|
||
frames, movie, fps = load_frames(dump_path, min_conf=float(cfg.get("detector_conf", 0.0)))
|
||
|
||
net = sae_kpn.Network()
|
||
sae_kpn._register_types(net)
|
||
|
||
idx = [0]
|
||
eof = {"timestamp_sec": frames[-1]["timestamp_sec"], "eof": True}
|
||
|
||
def source():
|
||
# A no-input source node's run_loop calls this in a tight loop. Once frames
|
||
# are exhausted we must NOT hot-spin returning EOF — that pegs a core and
|
||
# floods the downstream channel with EOFs (livelock that wedged DE). Sleep
|
||
# briefly after the single real EOF so net.stop() can tear the thread down.
|
||
i = idx[0]
|
||
idx[0] += 1
|
||
if i < len(frames):
|
||
return frames[i]
|
||
time.sleep(0.05)
|
||
return eof
|
||
|
||
# Channel capacity must exceed the frame count so the fast source can't overflow
|
||
# a downstream FIFO before the serial reader drains it — PyNode DROPS on overflow,
|
||
# which would silently truncate the replay. Size to the whole film + slack.
|
||
# Every channel gets capacity ≥ the whole film so NOTHING can ever overflow-drop:
|
||
# the source can push all frames before any downstream node has drained, and a
|
||
# dropped frame silently corrupts the score. Memory is cheap (a few k pointers);
|
||
# correctness is not. Generous slack on top.
|
||
cap = len(frames) * 2 + 64
|
||
sae_kpn.add_node_python(net, "replay", source, [], ["EmbeddedSceneFrame"], cap)
|
||
sae_kpn.add_face_tracker(net, "tracker", cfg, cap)
|
||
sae_kpn.add_identity_matcher(net, "matcher", gallery, cfg, cap,
|
||
stamp["model_name"], stamp["model_sha256"])
|
||
sae_kpn.add_scene_tracker(net, "scene", cfg, cap)
|
||
net.connect("replay", 0, "tracker", 0)
|
||
net.connect("tracker", 0, "matcher", 0)
|
||
net.connect("matcher", 0, "scene", 0)
|
||
net.build()
|
||
net.start()
|
||
|
||
# Read exactly one annotation per input frame. The source emits EOF as an ordinary
|
||
# value AFTER the last frame, but the concurrent pipeline lets that EOF OVERTAKE
|
||
# the last few real frames still flowing tracker→matcher→scene. Breaking on the
|
||
# first eof therefore dropped a random tail (~0.5–1%, race-dependent). Instead we
|
||
# keep reading past eof until we've collected all n_frames annotations (or hit a
|
||
# run of consecutive eofs meaning the pipeline is genuinely drained).
|
||
n_expected = len(frames) - 1 # excludes the trailing eof frame
|
||
annotations = []
|
||
eof_streak = 0
|
||
max_reads = n_expected * 2 + 32
|
||
for _ in range(max_reads):
|
||
sa = net.read("scene", 0)
|
||
if sa.get("eof"):
|
||
eof_streak += 1
|
||
# stragglers can still arrive after an eof; only stop once we've either
|
||
# got everything or seen several eofs in a row (truly drained).
|
||
if len(annotations) >= n_expected or eof_streak >= 8:
|
||
break
|
||
continue
|
||
eof_streak = 0
|
||
annotations.append(sa)
|
||
if len(annotations) >= n_expected:
|
||
break
|
||
|
||
if raw_out:
|
||
with open(raw_out, "w") as f:
|
||
for sa in annotations:
|
||
f.write(json.dumps(sa) + "\n")
|
||
|
||
result = build_minimal(annotations, movie, fps, cfg)
|
||
if stop:
|
||
net.stop()
|
||
return result
|
||
|
||
|
||
def build_minimal(annotations, movie, fps, cfg) -> dict:
|
||
"""Reproduce result_sink's minimal schema: per-actor annealed [start,end] windows.
|
||
|
||
Mirrors ResultSinkFunc::build_actor_windows — merge each actor's detection
|
||
timestamps into windows, bridging gaps shorter than anneal_sec.
|
||
"""
|
||
anneal = float(cfg.get("anneal_sec", 10.0))
|
||
info = {} # actor_idx -> identity fields
|
||
times = {} # actor_idx -> [timestamps]
|
||
for sa in annotations:
|
||
for a in sa["visible_actors"]:
|
||
if a["actor_idx"] < 0:
|
||
continue
|
||
info[a["actor_idx"]] = a
|
||
times.setdefault(a["actor_idx"], []).append(sa["timestamp_sec"])
|
||
|
||
actors = []
|
||
for idx, ts in times.items():
|
||
ts.sort()
|
||
scenes = []
|
||
ws = we = ts[0]
|
||
for t in ts[1:]:
|
||
if t - we > anneal:
|
||
scenes.append([ws, we])
|
||
ws = t
|
||
we = t
|
||
scenes.append([ws, we])
|
||
a = info[idx]
|
||
actors.append({
|
||
"name": a["name"], "imdb_id": a["imdb_id"], "tmdb_id": a["tmdb_id"],
|
||
"jellyfin_id": a["jellyfin_id"], "scenes": scenes,
|
||
})
|
||
|
||
return {"schema_version": 1, "movie": movie, "sample_fps": fps,
|
||
"anneal_sec": anneal, "actors": actors}
|
||
|
||
|
||
CFG_KEYS = ["detector_conf", "prob_threshold", "match_prior", "match_threshold", "match_ratio",
|
||
"match_ratio_ceil", "track_alpha", "track_min_iou", "track_max_embed_dist",
|
||
"track_max_frames_missing", "cut_revive_sim", "cut_inactive_max_frames",
|
||
"extinction_sec", "anneal_sec"]
|
||
|
||
|
||
def main():
|
||
p = argparse.ArgumentParser(description=__doc__,
|
||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
p.add_argument("--dump", required=True, help="embedding HDF5 dump")
|
||
p.add_argument("--gallery", required=True)
|
||
p.add_argument("--out", required=True, help="output presence JSON")
|
||
p.add_argument("--raw-out", help="also write raw per-frame annotations (JSONL, with bboxes) here")
|
||
p.add_argument("--build-dir", default=str(REPO / "build"))
|
||
for k in CFG_KEYS:
|
||
p.add_argument(f"--{k.replace('_','-')}", type=float, default=None)
|
||
# per-film gallery expansion: promotes pose-varied views of confidently-identified
|
||
# actors into an in-memory annex, recovering ~+4 recall at no precision cost.
|
||
p.add_argument("--expand-gallery", action="store_true")
|
||
# TRACES: GR-004 | SR-001
|
||
# promote an unprovable gallery/dump binding from a
|
||
# loud warning to a hard error. Measurement sweeps should set this (or
|
||
# SAE_REQUIRE_GALLERY_STAMP=1) so no number comes from an unbound pair.
|
||
p.add_argument("--require-gallery-stamp", action="store_true")
|
||
args = p.parse_args()
|
||
|
||
cfg = {k: getattr(args, k) for k in CFG_KEYS if getattr(args, k) is not None}
|
||
if args.expand_gallery:
|
||
cfg["expand_gallery"] = True
|
||
if args.require_gallery_stamp:
|
||
cfg["require_gallery_stamp"] = True
|
||
# stop=True: PyNode::stop() sets stop_flag_ before joining, so the source
|
||
# thread's run_loop actually exits. stop=False skips that, leaving stop_flag_
|
||
# false forever — the PyNode destructor's jthread.join() then blocks forever
|
||
# (verified via gdb: stuck in the source node's run_loop, not the GEMM path).
|
||
result = replay(args.dump, args.gallery, cfg, args.build_dir, stop=True,
|
||
raw_out=args.raw_out)
|
||
Path(args.out).write_text(json.dumps(result, indent=2))
|
||
print(f"[replay] {len(result['actors'])} actors → {args.out}", file=sys.stderr)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|