| |
| |
| """The release, unchanged, from audio to RTTM with the stand-in networks of tests/fake_nn.py (CPU only): for each |
| input, `load_audio` -> `analyse` -> `indic_mode`, saving the record and the RTTM, or the error. |
| |
| python -s tests/ref_fake_run.py INPUTS_JSON OUT_DIR |
| """ |
| import json |
| import os |
| import sys |
| from pathlib import Path |
|
|
| os.environ["CUDA_VISIBLE_DEVICES"] = "" |
| import numpy as np |
|
|
| HERE = Path(__file__).resolve().parent |
| sys.path.insert(0, str(HERE)) |
| import refpkg |
|
|
| refpkg.load() |
| import fake_nn |
| from anuvani_diar_ref.frontend import load_bundle, load_audio, analyse |
| from anuvani_diar_ref.modes import indic_mode, write_rttm_indic |
| from anuvani_diar_ref.decoder import load_decoder |
|
|
| REL = refpkg.REL |
|
|
|
|
| def main(): |
| items = json.loads(Path(sys.argv[1]).read_text()) |
| out = Path(sys.argv[2]); out.mkdir(parents=True, exist_ok=True) |
| modes = json.loads((REL / "bundle/MANIFEST.json").read_text())["modes"] |
| model = load_bundle(REL / "bundle", "cpu", 8, 2) |
| fake_nn.patch_release(model) |
| decoder = load_decoder(REL / "bundle" / modes["indic"]["decoder"]) |
| log = {} |
| for it in items: |
| sid, rec = it["sid"], {} |
| try: |
| audio, sr = load_audio(Path(it["path"])) |
| d = analyse(model, audio, sr) |
| np.savez(out / f"{sid}.npz", **d) |
| write_rttm_indic(out / f"{sid}.rttm", sid, indic_mode(d, modes["indic"]["clustering"], decoder)) |
| except Exception as e: |
| rec["error"] = f"{type(e).__name__}: {e}" |
| log[sid] = rec |
| (out / "log.json").write_text(json.dumps(log, indent=1)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
