Add bounded CPU VAD diagnostic with real model contract tests
This commit is contained in:
1 parent
37bb6a6e0e
commit
bc47e14971
6 files changed
+205
-2
No files matched your search
@@ -0,0 +1,53 @@
|
||||
"""Actual CPU model contract checks; explicit asset paths required."""
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import struct
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
class NativeContract(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.worker = Path(os.environ["SEARCHI_VAD_DIAGNOSTIC"])
|
||||
cls.model = Path(os.environ["SEARCHI_VAD_MODEL"])
|
||||
if hashlib.sha256(cls.model.read_bytes()).hexdigest() != "2aa269b785eeb53a82983a20501ddf7c1d9c48e33ab63a41391ac6c9f7fb6987":
|
||||
raise ValueError("Expected frozen Silero v6.2.0 export")
|
||||
|
||||
def run_pcm(self, data):
|
||||
with tempfile.TemporaryDirectory(prefix="searchi-vad-contract-") as scratch:
|
||||
pcm = Path(scratch) / "samples.f32"
|
||||
pcm.write_bytes(data)
|
||||
return subprocess.run([str(self.worker), str(self.model), str(pcm)],
|
||||
capture_output=True, timeout=30, check=False)
|
||||
|
||||
def test_real_model_silence_tail(self):
|
||||
result = self.run_pcm(b"\0" * (16001 * 4))
|
||||
self.assertEqual(result.returncode, 0, result.stderr.decode("utf-8", errors="replace"))
|
||||
report = json.loads(result.stdout)
|
||||
self.assertEqual(report["samples"], 16001)
|
||||
self.assertEqual(len(report["probabilities"]), 32)
|
||||
self.assertEqual(report["intervals_cs"], [])
|
||||
self.assertEqual(report["threads"], 1)
|
||||
self.assertIs(report["gpu"], False)
|
||||
for key in ("load_ms", "detect_ms", "segment_ms"):
|
||||
self.assertGreaterEqual(report[key], 0)
|
||||
|
||||
def test_invalid_pcm(self):
|
||||
for data in (b"", b"x", struct.pack("<f", float("nan")), struct.pack("<f", 2)):
|
||||
with self.subTest(data=data):
|
||||
result = self.run_pcm(data)
|
||||
self.assertEqual(result.returncode, 2)
|
||||
self.assertEqual(result.stdout, b"")
|
||||
|
||||
def test_unknown_arguments_fail(self):
|
||||
result = subprocess.run([str(self.worker), "--unknown"], capture_output=True, timeout=5)
|
||||
self.assertEqual(result.returncode, 2)
|
||||
self.assertEqual(result.stdout, b"")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user