48 lines
2.5 KiB
Python
48 lines
2.5 KiB
Python
"""Compare complete frozen CPU ASR pools; timing is not retrieval quality."""
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
ASSETS = {"clip04", "music01", "music02", "sound01", "sound02"}
|
|
|
|
|
|
def compare(baseline, gated):
|
|
for report in (baseline, gated):
|
|
if report.get("complete") is not True or report.get("quality_failed") is not False:
|
|
raise ValueError("Incomplete or structurally invalid comparison pool")
|
|
rows = report.get("rows", [])
|
|
if len(rows) != 5 or {row["asset"] for row in rows} != ASSETS:
|
|
raise ValueError("All five unique competitors required")
|
|
for field in ("fixture_sha256", "model_sha256", "worker_sha256", "ffmpeg_sha256", "ffprobe_sha256", "backend", "threads"):
|
|
if baseline[field] != gated[field]:
|
|
raise ValueError("Comparison provenance differs: " + field)
|
|
if gated.get("comparison_eligible") is not True:
|
|
raise ValueError("Gated comparison is ineligible")
|
|
original = {row["asset"]: row for row in baseline["rows"]}
|
|
vad_rows = {row["asset"]: row for row in gated["actual_vad_report"]["rows"]}
|
|
results = []
|
|
for row in gated["rows"]:
|
|
old = original[row["asset"]]
|
|
if old["source_sha256"] != row["source_sha256"]:
|
|
raise ValueError("Changed source")
|
|
def stats(windows):
|
|
if any("validation_error" in window for window in windows):
|
|
raise ValueError("Invalid window must not be scored")
|
|
return {"windows": len(windows), "segments": sum(len(w["segments"]) for w in windows),
|
|
"decode_ms": sum(w["decode_ms"] for w in windows),
|
|
"asr_process_ms": sum(w["asr_process_ms"] for w in windows)}
|
|
results.append({"asset": row["asset"], "baseline": stats(old["windows"]),
|
|
"gated": stats(row["windows"]), "status": row["status"],
|
|
"vad_decode_ms": vad_rows[row["asset"]]["decode_ms"],
|
|
"vad_process_ms": vad_rows[row["asset"]]["process_ms"]})
|
|
return {"scope": "Small sequential CPU diagnostic; process timings include model loading; not search latency or quality", "rows": results}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("baseline", type=Path)
|
|
parser.add_argument("gated", type=Path)
|
|
args = parser.parse_args()
|
|
print(json.dumps(compare(json.loads(args.baseline.read_text(encoding="utf-8")),
|
|
json.loads(args.gated.read_text(encoding="utf-8"))), indent=2))
|