79 lines
4.8 KiB
Python
79 lines
4.8 KiB
Python
"""Recompute frozen service retrieval and negative controls; never invent point scores."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
|
|
def score(report, fixture):
|
|
if report.get("complete") is not True or report.get("strict_point_score_eligible") is not False:
|
|
raise ValueError("Incomplete report or invented point scoring")
|
|
core = report["core"]
|
|
if {source["id"] for source in core["sources"]} != {"speech-en", "speech-de", "music-control", "silence-control"}:
|
|
raise ValueError("Incomplete competitor pool")
|
|
queries = {query["id"]: query for query in fixture["queries"]}
|
|
if len(core["rows"]) != len(queries) or {row["query"]["id"] for row in core["rows"]} != set(queries):
|
|
raise ValueError("Missing or duplicated frozen query")
|
|
rows = []
|
|
for row in core["rows"]:
|
|
query = row["query"]
|
|
if query != queries[query["id"]] or row.get("strict_point_score_eligible") is not False:
|
|
raise ValueError("Frozen query changed")
|
|
hits = row["hits"]
|
|
ids = [hit["asset"] for hit in hits]
|
|
if len(set(ids)) != len(ids) or not set(ids) <= {source["id"] for source in core["sources"]}:
|
|
raise ValueError("Unexpected result pool")
|
|
rank = next((index + 1 for index, id in enumerate(ids) if id in query["expected"]), None)
|
|
rows.append({"id": query["id"], "language": query["language"], "positive": bool(query["expected"]),
|
|
"first_relevant_rank": rank, "returned_count": len(hits),
|
|
"first_hit": None if not hits else {"asset": hits[0]["asset"], "match_kind": hits[0]["result"]["match_kind"], "snippet": hits[0]["result"]["snippet"], "source": hits[0]["result"]["source"]}})
|
|
positive = [row for row in rows if row["positive"]]
|
|
negative = [row for row in rows if not row["positive"]]
|
|
native_gate_counts = Counter()
|
|
complete_calls = True
|
|
for case in (core, report["seams"]):
|
|
expected = Counter(window["pcm_sha256"] for source in case["sources"] for window in source["windows"])
|
|
actual = Counter(call["pcm_sha256"] for call in case["native_calls"])
|
|
complete_calls &= expected == actual and all("error" not in call for call in case["native_calls"])
|
|
for call in core["native_calls"]:
|
|
ids = {window["asset"] for window in call["expected_source_windows"]}
|
|
if len(ids) == 1:
|
|
native_gate_counts[next(iter(ids))] += len(call.get("transcription", {}).get("passages", []))
|
|
concepts = {}
|
|
for asset in fixture["assets"]:
|
|
text = " ".join(chunk["text"] for chunk in core["published_speech"]
|
|
if Path(chunk["path"]).stem == asset["id"]).casefold()
|
|
concepts[asset["id"]] = {keyword: keyword.casefold() in text for keyword in asset["reference_keywords"]}
|
|
no_false_speech = all(native_gate_counts[id] == 0 for id in ("music-control", "silence-control"))
|
|
eligible = complete_calls and core["comparison_eligible"] and report["seams"]["comparison_eligible"]
|
|
return {"comparison_eligible": eligible, "strict_point_score_eligible": False,
|
|
"positive_queries": len(positive), "top1": sum(row["first_relevant_rank"] == 1 for row in positive) if eligible else None,
|
|
"recall_at_5": sum(row["first_relevant_rank"] is not None and row["first_relevant_rank"] <= 5 for row in positive) if eligible else None,
|
|
"mrr": sum(1 / row["first_relevant_rank"] if row["first_relevant_rank"] else 0 for row in positive) / len(positive) if eligible else None,
|
|
"negative_queries": len(negative), "negative_returned_any": sum(row["returned_count"] > 0 for row in negative),
|
|
"native_passage_counts": dict(native_gate_counts), "negative_gate_acceptance": no_false_speech,
|
|
"reference_concept_substring_coverage": concepts, "rows": rows,
|
|
"scope": "Source-level retrieval only, substring concept diagnostic not WER; no timed publisher phrase labels or semantic confidence; retain hallucination failures"}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("report", type=Path)
|
|
parser.add_argument("fixture", type=Path)
|
|
parser.add_argument("output", type=Path)
|
|
args = parser.parse_args()
|
|
raw = args.report.read_bytes()
|
|
fixture_raw = args.fixture.read_bytes()
|
|
lock = json.loads(args.fixture.with_name("labels.lock.json").read_text())
|
|
if hashlib.sha256(fixture_raw).hexdigest() != lock["fixture_sha256"]:
|
|
raise ValueError("Frozen labels changed")
|
|
result = score(json.loads(raw), json.loads(fixture_raw))
|
|
result.update(raw_report_sha256=hashlib.sha256(raw).hexdigest(), fixture_sha256=lock["fixture_sha256"])
|
|
with args.output.open("x", encoding="utf-8") as output:
|
|
json.dump(result, output, ensure_ascii=False, indent=2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|