File size: 4,649 Bytes
7ea1851 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """Pure helpers for the on-pod scene-pipeline validation report.
Kept dependency-free (operates on ``VideoResult.to_dict()`` dicts) so the
report logic can be unit-tested on a laptop without loading any GPU models.
The GPU orchestration lives in ``scripts/validate-scene-pipeline.py``.
"""
from __future__ import annotations
from typing import Any
def summarize_video_result(result: dict[str, Any]) -> dict[str, Any]:
"""Distill one video's pipeline output into validation metrics."""
windows = result.get("windows", []) or []
n = len(windows)
clap_windows = sum(1 for w in windows if w.get("audio_tags"))
onscreen = sum(len(w.get("onscreen_text_places") or []) for w in windows)
window_loc_hits = sum(len(w.get("window_locations") or []) for w in windows)
summary = result.get("video_summary") or {}
locations = result.get("video_locations") or {}
places = locations.get("places") or []
isos = sorted({p.get("country_iso") for p in places if p.get("country_iso")})
return {
"natural_key": result.get("natural_key", ""),
"windows": n,
"clap_windows": clap_windows,
"clap_coverage": round(clap_windows / n, 3) if n else 0.0,
"onscreen_text_count": onscreen,
"summary_parse_status": summary.get("parse_status"),
"tldr_words": len((summary.get("tldr") or "").split()),
"themes": len(summary.get("themes") or []),
"acts": len(summary.get("acts") or []),
"locations": len(places),
"location_isos": isos,
"window_location_hits": window_loc_hits,
}
def verdict(metrics: list[dict[str, Any]]) -> tuple[bool, list[str]]:
"""Smoke-test PASS/FAIL across all videos. Returns (passed, reasons).
The headline gate is CLAP: the whole point of the sigmoid fix was to turn
the old 0/175 coverage non-zero. Summaries parsing and at least one valid
location are strong-but-soft signals (a clip may genuinely name no place).
"""
reasons: list[str] = []
if not metrics:
return False, ["no videos were processed"]
clap_total = sum(m["clap_windows"] for m in metrics)
summaries_ok = sum(1 for m in metrics if m["summary_parse_status"] == "ok")
loc_videos = sum(1 for m in metrics if m["locations"] > 0)
any_valid_iso = any(m["location_isos"] for m in metrics)
passed = True
if clap_total > 0:
reasons.append(f"PASS CLAP: {clap_total} tagged windows (was 0 before the fix)")
else:
passed = False
reasons.append("FAIL CLAP: 0 tagged windows across all videos")
if summaries_ok > 0:
reasons.append(f"PASS summaries: {summaries_ok}/{len(metrics)} parsed cleanly")
else:
passed = False
reasons.append("FAIL summaries: none parsed as clean JSON")
if loc_videos > 0 and any_valid_iso:
reasons.append(f"PASS locations: {loc_videos} video(s) with valid ISO codes")
else:
reasons.append("INFO locations: none found — OK only if these clips name no places")
return passed, reasons
def format_report(metrics: list[dict[str, Any]], sweep: list[dict[str, Any]] | None = None) -> str:
"""Render a compact text report for the pod terminal."""
lines = ["", "=" * 64, "SCENE-PIPELINE VALIDATION", "=" * 64]
for m in metrics:
lines.append(f"\n▶ {m['natural_key']} ({m['windows']} windows)")
lines.append(
f" CLAP coverage : {m['clap_windows']}/{m['windows']} "
f"({m['clap_coverage']*100:.0f}%)"
)
lines.append(
f" summary : status={m['summary_parse_status']} "
f"tldr={m['tldr_words']}w themes={m['themes']} acts={m['acts']}"
)
lines.append(
f" locations : {m['locations']} places "
f"iso={m['location_isos']} window-hits={m['window_location_hits']}"
)
lines.append(f" on-screen text: {m['onscreen_text_count']} mentions")
if sweep:
lines.append("\n" + "-" * 64)
lines.append("THROUGHPUT SWEEP (--vlm-batch-size)")
lines.append(f" {'batch':>6} {'win/s':>8} {'peak VRAM GB':>13}")
for s in sweep:
vram = f"{s['peak_vram_gb']:.1f}" if s.get("peak_vram_gb") is not None else "n/a"
lines.append(f" {s['batch_size']:>6} {s['windows_per_sec']:>8.2f} {vram:>13}")
passed, reasons = verdict(metrics)
lines.append("\n" + "-" * 64)
for r in reasons:
lines.append(" " + r)
lines.append("-" * 64)
lines.append(f"OVERALL: {'PASS ✅' if passed else 'FAIL ❌'}")
lines.append("=" * 64)
return "\n".join(lines)
|