File size: 3,584 Bytes
35d483e | 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 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | from __future__ import annotations
import json
import struct
import subprocess
import sys
import tempfile
import unittest
import wave
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PROJECT_ROOT / "src"))
from turn_detection.data import read_manifest # noqa: E402
def write_wav(path: Path, value: int) -> None:
with wave.open(str(path), "wb") as handle:
handle.setnchannels(1)
handle.setsampwidth(2)
handle.setframerate(16_000)
handle.writeframes(struct.pack("<h", value) * 160)
class DataCliTests(unittest.TestCase):
def test_audit_then_prepare_splits_without_optional_dependencies(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
for index in range(6):
write_wav(root / f"{index}.wav", index + 1)
source = root / "records.jsonl"
with source.open("w", encoding="utf-8") as handle:
for index in range(6):
handle.write(
json.dumps(
{
"audio": {"path": f"{index}.wav"},
"id": f"id-{index}",
"language": "hi" if index % 2 else "en",
"endpoint_bool": bool(index % 2),
"midfiller": None,
"endfiller": False,
"synthetic": False,
"dataset": "test",
}
)
+ "\n"
)
manifest = root / "manifest.jsonl"
audit_report = root / "audit.json"
audit = subprocess.run(
[
sys.executable,
str(PROJECT_ROOT / "scripts" / "audit_dataset.py"),
str(source),
"--output",
str(manifest),
"--report",
str(audit_report),
"--progress-every",
"0",
],
cwd=PROJECT_ROOT,
capture_output=True,
text=True,
check=False,
)
self.assertEqual(audit.returncode, 0, audit.stderr)
self.assertEqual(len(list(read_manifest(manifest))), 6)
splits = root / "splits.jsonl"
split_report = root / "splits.json"
prepare = subprocess.run(
[
sys.executable,
str(PROJECT_ROOT / "scripts" / "prepare_splits.py"),
"--input",
str(manifest),
"--output",
str(splits),
"--report",
str(split_report),
"--split",
"train=2",
"--split",
"validation=1",
"--limit",
"5",
],
cwd=PROJECT_ROOT,
capture_output=True,
text=True,
check=False,
)
self.assertEqual(prepare.returncode, 0, prepare.stderr)
split_rows = list(read_manifest(splits))
self.assertEqual(len(split_rows), 5)
self.assertTrue(all(row.get("split") for row in split_rows))
if __name__ == "__main__":
unittest.main()
|