File size: 1,672 Bytes
63dd946 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 | # Copyright (c) 2026 The Anuvani-Diarization authors
# SPDX-License-Identifier: MIT OR Apache-2.0
"""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)
|