searchi/tools/speech-eval/score.py

"""Score externally ranked genuine transcript hits against unchanged media v1."""
import argparse
import hashlib
import json
from pathlib import Path
from run import FIXTURE_SHA


def require_eligible(report):
    if report.get("complete") is not True or report.get("quality_score_eligible") is not True:
        raise ValueError("Cannot score incomplete runs or reduced/invalid competitor pools")
    if report.get("excluded_windows") or report.get("excluded_window_count", 0):
        raise ValueError("Cannot score rankings with excluded ASR windows")


def score(fixture, rankings):
    queries = [query for query in fixture["queries"] if query["category"] == "video-speech"]
    if len(queries) != 8:
        raise ValueError("Expected the eight frozen speech queries")
    ids = {query["id"] for query in queries}
    if set(rankings) != ids:
        raise ValueError("Rankings must cover exactly eight frozen speech queries")
    rows = []
    for query in queries:
        hits = rankings[query["id"]]
        if not isinstance(hits, list) or len(hits) > 100:
            raise ValueError("Invalid ranking length")
        hit = hits[0] if hits else None
        point = overlap = False
        if hit:
            start, end = hit["start_ms"], hit["end_ms"]
            if type(start) is not int or type(end) is not int or not 0 <= start <= end:
                raise ValueError("Invalid ranked timestamp")
            intervals = query["intervals"].get(hit["asset"], [])
            point = any(left <= start / 1000 <= right for left, right in intervals)
            overlap = any(min(end / 1000, right) > max(start / 1000, left) for left, right in intervals)
        rows.append({"query": query["id"], "point_hit": point, "overlap_hit": overlap})
    return {"scope": "Externally ranked transcript top1, no oracle segment selection",
            "queries": len(rows), "point_hits": sum(row["point_hit"] for row in rows),
            "overlap_hits": sum(row["overlap_hit"] for row in rows), "rows": rows}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--fixture", type=Path, default=Path("fixtures/media/fixture.json"))
    parser.add_argument("--rankings", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    data = args.fixture.read_bytes()
    rankings = json.loads(args.rankings.read_text(encoding="utf-8"))
    require_eligible(rankings)
    digest = hashlib.sha256(data).hexdigest()
    if digest != FIXTURE_SHA or rankings["fixture_sha256"] != digest or not rankings.get("asr_report_sha256") or not rankings.get("google_worker_sha256"):
        raise ValueError("Missing/mismatched actual ASR/Google provenance")
    report = score(json.loads(data), rankings["rankings"])
    report.update({key: rankings[key] for key in ("fixture_sha256", "asr_report_sha256", "google_worker_sha256")})
    args.output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()