karl
b8400c4fbf
Core CI / core (ubuntu-latest) (push) Failing after 3m31s
Core CI / core (macos-latest) (push) Canceled after 0s
Core CI / core (windows-latest) (push) Canceled after 0s
Core CI / desktop (ubuntu-latest) (push) Successful in 5m39s
Core CI / desktop (macos-latest) (push) Canceled after 0s
Core CI / desktop (windows-latest) (push) Canceled after 0s
115 lines
6.5 KiB
Python
115 lines
6.5 KiB
Python
"""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()
|