"""Прогон набора через ядро распознавания. Пишет гипотезы в jsonl для score.py. Запуск: bench/.venv/Scripts/python.exe bench/run.py \ --manifest bench/data/common_voice_25_ru/test.jsonl \ --dataset common_voice_25_ru --slice clean_read \ --model v3_e2e_rnnt --out bench/data/hyp/cv-rnnt.jsonl Три решения, важные для достоверности цифр. 1. **Запросы строго по одному.** RTF измеряется временем ответа, а параллельные запросы делят одно и то же процессорное время: при восьми потоках каждый ответ вдвое медленнее, и цифра перестаёт что-либо значить. Пропускную способность меряют отдельным прогоном, это другая метрика. 2. **Длительность берём из ответа ядра**, а не считаем по размеру файла: ядро получает её от ffprobe. Размер зависит от кодека, час wav — 115 МБ, час opus — 30. 3. **Прогрев.** Первые вызовы после старта контейнера включают ленивую загрузку весов и прогрев ядер; они выбрасываются из замера (--warmup), иначе RTF первой сотни записей завышен, и чем меньше выборка, тем сильнее. Прогон возобновляемый: уже посчитанные id читаются из выходного файла и пропускаются. Прогон на несколько часов иначе не довести до конца. """ from __future__ import annotations import argparse import json import subprocess import time from pathlib import Path import requests CORE_DEFAULT = 'http://127.0.0.1:8011' def container_memory_mb(name: str) -> float | None: """Текущее потребление памяти контейнером ядра. Через docker stats, потому что процесс ядра живёт в контейнере и изнутри python его RSS не виден.""" try: p = subprocess.run( ['docker', 'stats', '--no-stream', '--format', '{{.MemUsage}}', name], capture_output=True, text=True, timeout=30, ) raw = (p.stdout or '').split('/')[0].strip() # «1.23GiB / 15.5GiB» for suffix, mult in (('GiB', 1024), ('MiB', 1), ('KiB', 1 / 1024)): if raw.endswith(suffix): return round(float(raw[: -len(suffix)]) * mult, 1) except Exception: # noqa: BLE001 return None return None def transcribe(core: str, path: Path, model: str, timeout: int) -> tuple[str, float, str]: """Один файл через POST /transcribe. Возвращает текст, длительность и имя модели.""" with open(path, 'rb') as fh: r = requests.post( f'{core}/transcribe', files={'file': (path.name, fh, 'application/octet-stream')}, data={'model': model}, timeout=timeout, ) r.raise_for_status() body = r.json() return body.get('text') or '', float(body.get('duration_s') or 0), body.get('model') or model def done_ids(out: Path) -> set[str]: if not out.exists(): return set() ids: set[str] = set() with open(out, encoding='utf-8') as fh: for line in fh: line = line.strip() if line: ids.add(json.loads(line).get('id')) return ids def main() -> int: ap = argparse.ArgumentParser(description='Прогон датасета через ядро распознавания') ap.add_argument('--manifest', required=True, help='jsonl: {"id","path","ref"}') ap.add_argument('--dataset', required=True) ap.add_argument('--slice', required=True, dest='slice_name', help='clean_read | spontaneous | telephony') ap.add_argument('--out', required=True) ap.add_argument('--core', default=CORE_DEFAULT) ap.add_argument('--model', default='v3_e2e_rnnt') ap.add_argument('--device', default='cpu', help='на чём считает ядро — уходит в отчёт') ap.add_argument('--system', default=None, help='имя системы в отчёте, по умолчанию модель ядра') ap.add_argument('--limit', type=int, default=0, help='ограничить число записей') ap.add_argument('--warmup', type=int, default=2, help='сколько первых записей выбросить из замера') ap.add_argument('--timeout', type=int, default=3600) ap.add_argument('--container', default='asr-core', help='контейнер ядра для замера памяти') args = ap.parse_args() manifest = [json.loads(l) for l in Path(args.manifest).read_text(encoding='utf-8').splitlines() if l.strip()] if args.limit: manifest = manifest[: args.limit] root = Path(args.manifest).parent out_path = Path(args.out) out_path.parent.mkdir(parents=True, exist_ok=True) already = done_ids(out_path) todo = [m for m in manifest if m['id'] not in already] print(f'записей {len(manifest)}, уже посчитано {len(already)}, к прогону {len(todo)}') health = requests.get(f'{args.core}/health', timeout=30).json() system = args.system or f'asr-core:{args.model}' mem_peak = container_memory_mb(args.container) or 0.0 started = time.time() ok = fail = 0 with open(out_path, 'a', encoding='utf-8') as fh: for i, item in enumerate(todo): path = Path(item['path']) if not path.is_absolute(): path = root / path row = { 'id': item['id'], 'dataset': args.dataset, 'slice': args.slice_name, 'system': system, 'device': args.device, 'ref': item.get('ref') or '', } try: t0 = time.perf_counter() text, duration, model_name = transcribe(args.core, path, args.model, args.timeout) elapsed = time.perf_counter() - t0 row |= {'hyp': text, 'duration_s': round(duration, 3), 'model': model_name} # прогрев не идёт в замер времени: гипотезу оставляем, RTF выбрасываем if i >= args.warmup: row['elapsed_s'] = round(elapsed, 3) else: row['warmup'] = True ok += 1 except Exception as e: # noqa: BLE001 row['error'] = f'{type(e).__name__}: {e}' fail += 1 fh.write(json.dumps(row, ensure_ascii=False) + '\n') fh.flush() if i % 25 == 0: mem = container_memory_mb(args.container) if mem: mem_peak = max(mem_peak, mem) done = i + 1 rate = done / max(time.time() - started, 1e-6) left = (len(todo) - done) / rate if rate else 0 print(f'{done}/{len(todo)} ok={ok} fail={fail} осталось ~{left / 60:.1f} мин', flush=True) meta = { 'dataset': args.dataset, 'slice': args.slice_name, 'system': system, 'model': args.model, 'device': args.device, 'core_health': health, 'records': len(manifest), 'ok': ok, 'failed': fail, 'memory_peak_mb': mem_peak or None, 'wall_seconds': round(time.time() - started, 1), 'finished_at': time.strftime('%Y-%m-%d %H:%M:%S'), } meta_path = out_path.with_suffix('.meta.json') meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding='utf-8') print(f'готово: ok={ok} fail={fail}, пик памяти {mem_peak or "?"} МБ, мета: {meta_path}') return 0 if fail == 0 else 1 if __name__ == '__main__': raise SystemExit(main())