| """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') |
| |
| 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 |
|
|
