scene-actor-extraction/scripts/scene_detector/de_ramp.py

#!/usr/bin/env python3
"""
de_ramp.py — DE-optimise a temporal matched-filter "ramp" per modality, whose
response becomes a feature channel for the scene-boundary LSTM.

A scene boundary is where a feature series (RGB histogram, audio log-PSD) shifts
from a "before" state to an "after" state. A signed, antisymmetric ramp kernel
convolved with the series responds strongly exactly at that transition and near
zero inside a stable scene — a matched filter for a step. Its shape is not
obvious (how wide? linear or peaked? how much centre dead-zone?), so we let DE
choose it by maximising boundary separation on the training films.

Ramp kernel over lags -H..+H seconds (1 fps → 1 sample/s):
    w(l) = sign(l) * (|l| / H) ** gamma        for |l| >= dead, else 0
  params: H (half-width), gamma (shape), dead (centre dead-zone)
Response at t = || sum_l w(l) * feat[t+l] ||   (L2 over feature bins)

DE objective: boundary-detection F1 of a top-percentile threshold on the response,
macro-averaged over the training films (±2 s tolerance). The tuned (H, gamma,
dead) is saved; train_scene_boundary.py appends the ramp response as an input
channel to each tower.

Usage:
  python scripts/scene_detector/de_ramp.py \
      --manifest experiments/manifests/films_LVFace_opencv5.json \
      --audio-dir experiments/dumps/audio_features \
      --holdout Scarface Sound_of_Metal --out experiments/results/scene_boundary
"""
from __future__ import annotations
import argparse, csv, json, sys
from pathlib import Path
import h5py, numpy as np
from scipy.optimize import differential_evolution


def xray_bounds(xray_dir):
    return sorted(float(r["start"])/1000 for r in
                  csv.DictReader(open(Path(xray_dir)/"scenes.csv"))
                  if float(r["start"]) > 500)


def load_series(dump, audio_dir, which):
    if which == "audio":
        # Audio is self-contained in the npz — no h5 needed (its ts IS the grid),
        # so the audio cutter can be tuned before/without the RGB dumps.
        slug = Path(dump).stem.replace("dump_", "")
        z = np.load(Path(audio_dir)/f"{slug}.npz")
        s = z["feat"].astype(np.float64)
        ts = z["ts"] if "ts" in z else np.arange(len(s), dtype=float)
    else:  # video
        with h5py.File(dump) as f:
            ts = f["frames/timestamp_sec"][:]
            s = f["frames/rgb_hist"][:].astype(np.float64)
    # z-normalise each bin so L2 response isn't dominated by one loud bin
    s = (s - s.mean(0)) / (s.std(0) + 1e-6)
    return s, ts


def ramp_kernel(H, gamma, dead):
    lags = np.arange(-H, H+1)
    w = np.sign(lags) * (np.abs(lags)/max(H,1))**gamma
    w[np.abs(lags) < dead] = 0.0
    return w


def response(series, w, H):
    T = series.shape[0]
    r = np.zeros(T)
    for t in range(T):
        lo, hi = max(0, t-H), min(T, t+H+1)
        wl = w[(lo-(t-H)):(hi-(t-H))]
        r[t] = np.linalg.norm((series[lo:hi]*wl[:, None]).sum(0))
    return r


def boundary_f1(resp, bounds, pct, tol=2):
    thr = np.percentile(resp, pct)
    pred = np.where(resp > thr)[0]
    bidx = [int(b) for b in bounds if int(b) < len(resp)]
    if len(pred) == 0 or not bidx:
        return 0.0
    tp_p = sum(any(abs(p-i) <= tol for i in bidx) for p in pred)
    tp_t = sum(any(abs(p-i) <= tol for p in pred) for i in bidx)
    P, R = tp_p/len(pred), tp_t/len(bidx)
    return 2*P*R/(P+R) if P+R else 0.0


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--manifest", required=True)
    ap.add_argument("--audio-dir", default="experiments/dumps/audio_features")
    ap.add_argument("--holdout", nargs="+", default=["Scarface", "Sound_of_Metal"])
    ap.add_argument("--out", default="experiments/results/scene_boundary")
    args = ap.parse_args()
    films = [f for f in json.load(open(args.manifest)) if f["slug"] not in args.holdout]

    out = {}
    for which in ("video", "audio"):
        data = [(load_series(f["dump"], args.audio_dir, which)[0], xray_bounds(f["xray"]))
                for f in films]
        def neg_f1(x):
            H = int(round(x[0])); gamma = x[1]; dead = int(round(x[2])); pct = x[3]
            if H < 1 or dead >= H: return 0.0
            w = ramp_kernel(H, gamma, dead)
            f1s = [boundary_f1(response(s, w, H), b, pct) for s, b in data]
            return -float(np.mean(f1s))
        # bounds: H 1..10s, gamma 0.3..3, dead 0..4s, threshold pct 80..98
        res = differential_evolution(
            neg_f1, [(1, 10), (0.3, 3.0), (0, 4), (80, 98)],
            seed=0, popsize=12, maxiter=25, tol=1e-4, polish=False)
        H = int(round(res.x[0])); gamma = float(res.x[1])
        dead = int(round(res.x[2])); pct = float(res.x[3])
        out[which] = {"H": H, "gamma": gamma, "dead": dead, "pct": pct,
                      "train_f1": float(-res.fun)}
        print(f"[de-ramp] {which}: H={H}s gamma={gamma:.2f} dead={dead}s "
              f"pct={pct:.0f}  train boundary-F1={-res.fun*100:.1f}%", file=sys.stderr)

    Path(args.out).mkdir(parents=True, exist_ok=True)
    json.dump(out, open(Path(args.out)/"de_ramp.json", "w"), indent=2)
    print(f"[de-ramp] → {args.out}/de_ramp.json", file=sys.stderr)


if __name__ == "__main__":
    main()