File size: 4,232 Bytes
52ecfff | 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 | """Pure binding checks for the prospective eleven-case Silero qualification.
No audio, ASR, VAD or model runtime is imported. A root-owned whole-campaign
freeze must exist before any fresh probability, boundary or transcript output.
"""
from pathlib import Path
import hashlib
import json
ROOT = Path(__file__).resolve().parents[1]
MANIFESTS = {'data/continuous-v3/manifest.json': '7ad682b2ff707832a91af12dd3550d0e45532d4bb41b56d27cb78eac07e92cbd', 'data/heldout-zh-v3/manifest.json': '52f63ea126debffc78ba0df29f904436943c070dd74d31b778993c566ab21df9'}
MODEL_SHA = '1a153a22f4509e292a94e67d6f9b85e8deb25b4988682b7e174c65279d8788e3'
POLICY = 'silero_v5_threshold0.5_silence2000ms_preroll400ms_v1'
VAD_RUNTIME_SHA = '12a3e66090507f9f6f62ed09c0c0f76407e643f121d4e0a45e655d6704ca3800'
SOURCES = (
'scripts/silero_qualification_v3.py',
'scripts/prepare_silero_qualification_v3.py',
'scripts/probe_torch_cuda_silero_case_v3.py',
'scripts/probe_torch_stream_cuda.py',
'scripts/probe_torch_stream_reference.py',
'scripts/probe_torch_cuda_silero_segments.py',
'mlx_runtime/__init__.py', 'mlx_runtime/silero_input.py', 'mlx_runtime/stream.py',
)
def require(ok, message):
if not ok:
raise ValueError(message)
def sha(path):
h = hashlib.sha256()
with Path(path).open('rb') as stream:
for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b''):
h.update(chunk)
return h.hexdigest()
def case_binding(case):
return {k: case[k] for k in ('id', 'role', 'language', 'sha256', 'samples', 'reference')}
def load_frozen(record_path, expected_sha, root=ROOT):
root = Path(root)
require(sha(record_path) == expected_sha, 'whole qualification freeze changed')
record = json.loads(Path(record_path).read_text())
require(record.get('format') == 'audio8-silero-paired-qualification-v2' and record.get('status') == 'frozen', 'whole qualification freeze missing')
require(record.get('input_manifest_sha256') == MANIFESTS, 'fresh manifest pins differ')
require(record.get('silero_model_sha256') == MODEL_SHA and record.get('silence_policy') == POLICY, 'VAD policy differs')
require(record.get('original_profile') == dict(weights='original_bf16', device='cuda', dtype='bfloat16', startup='realtime18', delay_ms=240, batch_windows=1, attention='sdpa_math_only', tf32=False), 'original reference profile differs')
# The outer freeze must pin the selected candidate, not merely this reference.
candidate = record.get('selected_candidate', {})
require(isinstance(candidate.get('profile'), dict) and candidate['profile'], 'selected candidate profile missing')
require(isinstance(candidate.get('assets'), dict) and candidate['assets'] and all(isinstance(h, str) and len(h) == 64 for h in candidate['assets'].values()), 'selected candidate asset/source hashes missing')
require(record.get('original_environment_lock_sha256') == 'bf1698431487b1415fb624d1f12e88bb9e4c2bdfaaa7cef88347bd24945420e5', 'original CUDA environment lock differs')
require(record.get('vad_runtime_identity_sha256') == VAD_RUNTIME_SHA and sha(root / 'reports/silero-ort-environment-v1/runtime-identity.json') == VAD_RUNTIME_SHA, 'VAD runtime record differs')
source_pins = record.get('source_sha256', {})
require(all(n in source_pins and sha(root / n) == source_pins[n] for n in SOURCES), 'qualification/runtime source changed')
cases = []
for name, digest in MANIFESTS.items():
path = root / name
require(sha(path) == digest, 'fresh input manifest changed')
manifest = json.loads(path.read_text())
require(manifest['format'] == 'mono_f32le' and manifest['sample_rate'] == 16000, 'input format differs')
cases.extend(manifest['cases'])
require(len(cases) == 11 and len({c['id'] for c in cases}) == 11, 'eleven-case coverage required')
require(all(c['role'] == 'heldout' for c in cases), 'fresh case role differs')
require(record.get('cases') == [case_binding(c) for c in cases], 'whole qualification case order/labels differ')
require(sum(c['language'] == 'en' for c in cases) == 3 and sum(c['language'] == 'zh' for c in cases) == 8, 'language coverage differs')
return record, cases
|