| from __future__ import annotations |
|
|
| import io |
| import struct |
| import sys |
| import unittest |
| import wave |
| from pathlib import Path |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) |
|
|
| from turn_detection.data import audit_records |
|
|
|
|
| def wav_bytes(value: int = 1) -> bytes: |
| output = io.BytesIO() |
| with wave.open(output, "wb") as handle: |
| handle.setnchannels(1) |
| handle.setsampwidth(2) |
| handle.setframerate(16_000) |
| handle.writeframes(struct.pack("<h", value) * 160) |
| return output.getvalue() |
|
|
|
|
| class DatasetAuditTests(unittest.TestCase): |
| def test_manifest_and_report_preserve_nulls_and_find_conflicts(self) -> None: |
| duplicate = wav_bytes() |
| records = [ |
| { |
| "audio": {"bytes": duplicate, "path": "one.wav"}, |
| "id": "one", |
| "language": "hi", |
| "endpoint_bool": True, |
| "midfiller": None, |
| "endfiller": False, |
| "synthetic": False, |
| "dataset": "human", |
| "__source_file": "first.parquet", |
| "__source_row": 0, |
| }, |
| { |
| "audio": {"bytes": duplicate, "path": "two.wav"}, |
| "id": "two", |
| "language": "hi", |
| "endpoint_bool": False, |
| "midfiller": True, |
| "endfiller": None, |
| "synthetic": True, |
| "dataset": "synthetic", |
| "__source_file": "second.parquet", |
| "__source_row": 3, |
| }, |
| ] |
| manifest, report = audit_records(records) |
| self.assertNotIn("audio", manifest[0]) |
| self.assertIsNone(manifest[0]["midfiller"]) |
| self.assertEqual(manifest[0]["audio_sha256"], manifest[1]["audio_sha256"]) |
| self.assertEqual(manifest[0]["group_id"], manifest[1]["group_id"]) |
| self.assertEqual(report["duplicates"]["duplicate_audio_hashes"], 1) |
| self.assertEqual(report["duplicates"]["conflicting_label_audio_hashes"], 1) |
| self.assertEqual(report["distributions"]["midfiller"]["<null>"], 1) |
| self.assertIn( |
| "conflicting_duplicate_label", |
| {warning["code"] for warning in manifest[0]["validation_warnings"]}, |
| ) |
|
|
| def test_invalid_audio_is_retained_and_counted(self) -> None: |
| manifest, report = audit_records( |
| [{"id": "bad", "audio": b"", "endpoint_bool": True, "dataset": "d"}] |
| ) |
| self.assertEqual(report["records"]["invalid"], 1) |
| self.assertIn("invalid_audio", {item["code"] for item in manifest[0]["validation_errors"]}) |
|
|
| def test_limit_applies_before_hashing_all_records(self) -> None: |
| records = ( |
| { |
| "id": str(index), |
| "audio": wav_bytes(index), |
| "endpoint_bool": bool(index % 2), |
| "dataset": "d", |
| } |
| for index in range(4) |
| ) |
| manifest, report = audit_records(records, limit=2) |
| self.assertEqual(len(manifest), 2) |
| self.assertEqual(report["records"]["total"], 2) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
