| """Small helpers shared by adapters (stdlib only).""" |
|
|
| from __future__ import annotations |
|
|
| import importlib |
| from collections.abc import Mapping, Sequence |
| from typing import Any |
|
|
| from nida.bench.protocol import WorkRequest |
|
|
| __all__ = ["HF_WEIGHT_PATTERNS", "LANGUAGES", "language_for", "load_mono_16k", "snapshot"] |
|
|
| |
| |
| HF_WEIGHT_PATTERNS: tuple[str, ...] = ( |
| "*.json", |
| "*.txt", |
| "*.model", |
| "*.tiktoken", |
| "model.safetensors", |
| "model-*.safetensors", |
| ) |
|
|
| |
| LANGUAGES: Mapping[str, Mapping[str, str]] = { |
| "tr": {"whisper": "turkish", "qwen": "Turkish", "omni": "tur_Latn"}, |
| "en": {"whisper": "english", "qwen": "English", "omni": "eng_Latn"}, |
| } |
|
|
|
|
| def language_for(request: WorkRequest, family: str) -> str: |
| """Model-specific language name for ``request.params["language"]`` (required). |
| |
| Raises: |
| ValueError: if the language parameter is missing or unsupported. |
| """ |
| code = request.params.get("language") |
| if code is None: |
| raise ValueError(f"{request.clip_id}: missing params.language") |
| try: |
| return LANGUAGES[code][family] |
| except KeyError as err: |
| raise ValueError(f"{request.clip_id}: unsupported language {code!r} for {family}") from err |
|
|
|
|
| def snapshot( |
| model_id: str, revision: str, allow_patterns: Sequence[str] = HF_WEIGHT_PATTERNS |
| ) -> str: |
| """Download (or reuse) a pinned Hugging Face snapshot and return its local path. |
| |
| An empty ``revision`` is refused: benchmark runs must be reproducible. Only files |
| matching ``allow_patterns`` are fetched. |
| """ |
| if not revision: |
| raise ValueError(f"{model_id}: a pinned revision is required") |
| hub = importlib.import_module("huggingface_hub") |
| path: str = hub.snapshot_download( |
| repo_id=model_id, revision=revision, allow_patterns=list(allow_patterns) |
| ) |
| return path |
|
|
|
|
| |
| def load_mono_16k(path: str) -> Any: |
| """Read a 16 kHz mono file as float32 (the orchestrator guarantees this format). |
| |
| Raises: |
| ValueError: if the file is not 16 kHz mono. |
| """ |
| soundfile = importlib.import_module("soundfile") |
| audio, sample_rate_hz = soundfile.read(path, dtype="float32", always_2d=False) |
| if sample_rate_hz != 16000 or audio.ndim != 1: |
| raise ValueError(f"{path}: expected 16 kHz mono, got {sample_rate_hz} Hz, {audio.ndim}-D") |
| return audio |
|
|
