| |
| |
| """Identity check of the cached front-end records: the Rust post-processing on every cached record (1,945 files) must |
| be byte-identical to the canonical oracle (the release's own code path, oracle/cache_canonical.json; its indic mode).""" |
| import json |
| from pathlib import Path |
|
|
| import numpy as np |
| import pytest |
|
|
| import records |
|
|
| PROJ = Path(__file__).resolve().parents[1] |
| REL = PROJ.parent / "anuvani_diar_v1" |
| ORACLE = json.loads((PROJ / "oracle/cache_canonical.json").read_text()) |
| MODES = ORACLE["modes"] |
| ALL = records.all_records() |
|
|
|
|
| def run(core, set_name, path): |
| d = records.load(path) |
| sid = path.stem |
| return core.diarize_record(sid, np.ascontiguousarray(d["P"]).view(np.uint16), d["masks"], d["emb"], |
| list(map(float, d["chunk_sw"])), list(map(float, d["frames_sw"])), float(d["duration_s"]), |
| MODES["indic"]["clustering"], str(REL / "bundle" / MODES["indic"]["decoder"])) |
|
|
|
|
| @pytest.mark.slow |
| @pytest.mark.parametrize("set_name,path", ALL, ids=lambda v: getattr(v, "stem", v)) |
| def test_identity_cached(core, set_name, path): |
| rttm = run(core, set_name, path) |
| rec = ORACLE["files"][f"{set_name}/{path.stem}"] |
| out = PROJ / "artifacts/oracle_cache/canonical" / set_name |
| assert rttm.encode() == (out / "indic" / f"{path.stem}.rttm").read_bytes() |
| assert "error" not in rec["indic"] |
|
|
|
|
| @pytest.mark.parametrize("set_name,path", records.sample(40), ids=lambda v: getattr(v, "stem", v)) |
| def test_identity_cached_sample(core, set_name, path): |
| test_identity_cached(core, set_name, path) |
|
|
