searchi/tools/media-eval/score.py

"""Score frozen relevance without fitting an abstention threshold."""
import argparse
from collections import defaultdict
import hashlib
import json
from pathlib import Path


def temporal_hit(candidate, query, point=False):
    asset = candidate.get("asset")
    if asset not in query.get("expected", []):
        return False
    if query.get("required_modality") and candidate.get("modality") != query["required_modality"]:
        return False
    ranges = query.get("intervals", {}).get(asset, [])
    start = candidate.get("start_s")
    end = candidate.get("end_s", start)
    if start is None:
        return False
    if point:
        return any(left <= start <= right for left, right in ranges)
    return any(min(end, right) > max(start, left) for left, right in ranges)


def timestamp_distance(candidate, query):
    """Raw seconds from returned start to the nearest labelled event interval."""
    if candidate.get("asset") not in query.get("expected", []) or candidate.get("start_s") is None:
        return None
    if query.get("required_modality") and candidate.get("modality") != query["required_modality"]:
        return None
    ranges = query.get("intervals", {}).get(candidate["asset"], [])
    stamp = candidate["start_s"]
    return min((max(left - stamp, stamp - right, 0) for left, right in ranges), default=None)


def metrics(rows):
    positive = [row for row in rows if row["query"]["expected"] and not row.get("unsupported")]
    negative = [row for row in rows if not row["query"]["expected"] and not row.get("unsupported")]
    temporal = [row for row in positive if row["query"]["category"].startswith("video-")]
    distances = sorted(distance for row in temporal if row["file_rankings"] and (distance := timestamp_distance(row["file_rankings"][0]["candidate"], row["query"])) is not None)
    ranks = []
    for row in positive:
        expected = row["query"]["expected"]
        ranks.append(next((i + 1 for i, hit in enumerate(row["file_rankings"]) if hit["candidate"]["asset"] in expected), None))
    fraction = lambda count, total: count / total if total else None
    return dict(
        positive_queries=len(positive),
        unsupported_queries=sum(bool(row.get("unsupported")) for row in rows),
        recall_at1=fraction(sum(rank == 1 for rank in ranks), len(ranks)),
        recall_at5=fraction(sum(rank is not None and rank <= 5 for rank in ranks), len(ranks)),
        mrr=fraction(sum(1 / rank if rank else 0 for rank in ranks), len(ranks)),
        temporal_queries=len(temporal),
        segment_overlap_hit_at1=fraction(sum(bool(row["file_rankings"]) and temporal_hit(row["file_rankings"][0]["candidate"], row["query"]) for row in temporal), len(temporal)),
        timestamp_point_hit_at1=fraction(sum(bool(row["file_rankings"]) and temporal_hit(row["file_rankings"][0]["candidate"], row["query"], point=True) for row in temporal), len(temporal)),
        segment_overlap_hit_at5=fraction(sum(any(temporal_hit(hit["candidate"], row["query"]) for hit in row["segment_top5"]) for row in temporal), len(temporal)),
        timestamp_distance_median_s=distances[len(distances) // 2] if distances else None,
        timestamp_distance_max_s=max(distances) if distances else None,
        no_answer_queries=len(negative),
        no_answer_return_rate=fraction(sum(bool(row["file_rankings"]) for row in negative), len(negative)),
    )


def summarize(report):
    rows = report["rows"]
    by_category = defaultdict(list)
    by_language = defaultdict(list)
    failures = []
    for row in rows:
        query = row["query"]
        by_category[query["category"]].append(row)
        by_language[query["language"]].append(row)
        if row.get("unsupported"):
            failures.append(dict(query=query["id"], category="unsupported-query-capability", detail=row["unsupported"]))
            continue
        hits = row["file_rankings"]
        if not query["expected"]:
            if hits:
                failures.append(dict(query=query["id"], category="no-answer-nearest-neighbor-return", winner=hits[0]["candidate"]["asset"], score=hits[0]["score"]))
        elif not hits or hits[0]["candidate"]["asset"] not in query["expected"]:
            winner = hits[0]["candidate"]["asset"] if hits else None
            failures.append(dict(query=query["id"], category="hard-negative-confusion" if winner in query.get("hard_negatives", []) else "retrieval-top1-miss", winner=winner, expected=query["expected"]))
        elif query["category"].startswith("video-") and not temporal_hit(hits[0]["candidate"], query, point=True):
            candidate = hits[0]["candidate"]
            wrong_channel = query.get("required_modality") and candidate.get("modality") != query["required_modality"]
            failures.append(dict(query=query["id"], category="wrong-media-channel" if wrong_channel else "timestamp-localization-miss", candidate=candidate, expected_intervals=query["intervals"], timestamp_distance_s=timestamp_distance(candidate, query)))
    durations = sorted(row["query_ms"] for row in rows if "query_ms" in row)
    return dict(
        fixture_sha256=report["fixture_sha256"], worker=report["worker"],
        overall=metrics(rows), by_category={key: metrics(value) for key, value in sorted(by_category.items())},
        by_language={key: metrics(value) for key, value in sorted(by_language.items())},
        encoding_failures=report["encoding_failures"], encoding_ms=report["encoding_ms"],
        query_p50_ms=durations[len(durations) // 2] if durations else None,
        query_p95_ms=durations[min(len(durations) * 95 // 100, len(durations) - 1)] if durations else None,
        failures=failures,
        source_scope=report.get("scope", "Direct dense model baseline"),
        interpretation="Small independently labelled public-media diagnostic; source_scope identifies the measured lane. No-answer return rate measures unsuppressed output, not a calibrated confidence policy. Audio interval overlap can pass while the returned start timestamp is too early; point hits expose this. This corpus alone cannot certify production quality.",
    )


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("report", type=Path)
    parser.add_argument("output", type=Path)
    args = parser.parse_args()
    report = json.loads(args.report.read_text(encoding="utf-8"))
    fixture = Path("fixtures/media/fixture.json")
    assert report["fixture_sha256"] == hashlib.sha256(fixture.read_bytes()).hexdigest()
    summary = summarize(report)
    args.output.write_text(json.dumps(summary, indent=2), encoding="utf-8")
    print(json.dumps(summary["overall"]))


if __name__ == "__main__":
    main()