"""Подсчёт метрик по файлу гипотез (jsonl от run.py). Запуск: bench/.venv/Scripts/python.exe bench/score.py bench/data/hyp/*.jsonl \ --out bench/results/2026-08-21.json --md bench/results/2026-08-21.md Почему метрики разделены (то же изложено на странице замера, https://rechetext.ru/benchmark): 1. **WER и CER без пунктуации и чисел** — чистое распознавание. Записи, где цифры стоят в ЭТАЛОНЕ, из этого среза выброшены и посчитаны отдельно: развернуть «1990» можно двумя верными способами, и WER накажет за один из них. 2. **F1 по знакам препинания** — отдельно. Наши модели ставят знаки сами, а эталоны Common Voice и Golos их частью не содержат вовсе; если мерить одной цифрой, каждая правильная запятая уходит в ошибки, и e2e-модель выглядит хуже, чем работает. 3. **RTF** — во сколько раз обработка быстрее записи, отдельно по устройству. Публичных значений RTF по русской речи почти нет, а само число бесполезно без описания условий прогона — поэтому оно приведено вместе с ними. Считает `jiwer`, а не самописный счётчик: к самописному будут вопросы, и правильно. Лицензионное ограничение зашито в код. У Golos лицензия — переработка CC BY-SA под право РФ: коммерческое использование разрешено, но на производный материал распространяется ShareAlike. Поэтому пофайловые эталоны и гипотезы по Golos не выгружаются — только агрегаты. Попытка выгрузить пофайлово завершается ошибкой, а не предупреждением: предупреждение однажды проглядят. """ from __future__ import annotations import argparse import glob import json import statistics from collections import defaultdict from dataclasses import dataclass, field from pathlib import Path import jiwer import sys sys.path.insert(0, str(Path(__file__).parent)) from normalize import has_digits, normalize, words_with_punct # noqa: E402 from sources import SHARE_ALIKE # noqa: E402 PUNCT_MARKS = (',', '.', '?') @dataclass class Bucket: """Накопитель по одному срезу одной системы.""" refs: list[str] = field(default_factory=list) hyps: list[str] = field(default_factory=list) num_refs: list[str] = field(default_factory=list) # записи с цифрами в эталоне num_hyps: list[str] = field(default_factory=list) rtf: list[float] = field(default_factory=list) audio_s: float = 0.0 elapsed_s: float = 0.0 # пунктуация: TP/FP/FN по классам знаков punct: dict[str, list[int]] = field(default_factory=lambda: {m: [0, 0, 0] for m in PUNCT_MARKS}) punct_records: int = 0 failures: int = 0 def align_map(ref_words: list[str], hyp_words: list[str]) -> dict[int, int]: """Соответствие «индекс слова эталона → индекс слова гипотезы» по выравниванию jiwer. Нужно, чтобы сравнивать знаки после слов: без выравнивания сдвиг на одно пропущенное слово превращает всю пунктуацию дальше в ошибки.""" if not ref_words or not hyp_words: return {} out = jiwer.process_words(' '.join(ref_words), ' '.join(hyp_words)) pairs: dict[int, int] = {} for chunk in out.alignments[0]: if chunk.type not in ('equal', 'substitute'): continue for k in range(chunk.ref_end_idx - chunk.ref_start_idx): pairs[chunk.ref_start_idx + k] = chunk.hyp_start_idx + k return pairs def score_punctuation(bucket: Bucket, ref_raw: str, hyp_raw: str) -> None: """F1 по знакам. Считается только когда знаки есть в эталоне: на эталонах без пунктуации метрика не определена, и подставлять туда ноль — обман в свою пользу наоборот (модель выглядела бы плохо там, где мерить нечем).""" ref_pairs = words_with_punct(ref_raw) if not any(m for _, m in ref_pairs): return hyp_pairs = words_with_punct(hyp_raw) bucket.punct_records += 1 mapping = align_map([w for w, _ in ref_pairs], [w for w, _ in hyp_pairs]) for i, (_, ref_mark) in enumerate(ref_pairs): j = mapping.get(i) hyp_mark = hyp_pairs[j][1] if j is not None and j < len(hyp_pairs) else None if ref_mark == hyp_mark: if ref_mark: bucket.punct[ref_mark][0] += 1 # TP continue if ref_mark: bucket.punct[ref_mark][2] += 1 # FN: знак был, не поставлен if hyp_mark: bucket.punct[hyp_mark][1] += 1 # FP: поставлен там, где нет # знаки, вставленные в словах, которых в эталоне нет вовсе, тоже ложные mapped = set(mapping.values()) for j, (_, hyp_mark) in enumerate(hyp_pairs): if hyp_mark and j not in mapped: bucket.punct[hyp_mark][1] += 1 def f1(tp: int, fp: int, fn: int) -> dict[str, float | int]: prec = tp / (tp + fp) if tp + fp else 0.0 rec = tp / (tp + fn) if tp + fn else 0.0 return { 'tp': tp, 'fp': fp, 'fn': fn, 'precision': round(prec, 4), 'recall': round(rec, 4), 'f1': round(2 * prec * rec / (prec + rec), 4) if prec + rec else 0.0, } def summarize(bucket: Bucket) -> dict: out: dict = { 'records': len(bucket.refs) + len(bucket.num_refs), 'records_wer': len(bucket.refs), 'records_numeric': len(bucket.num_refs), 'failures': bucket.failures, 'audio_hours': round(bucket.audio_s / 3600, 3), } if bucket.refs: out['wer'] = round(jiwer.wer(bucket.refs, bucket.hyps), 4) out['cer'] = round(jiwer.cer(bucket.refs, bucket.hyps), 4) if bucket.num_refs: # отдельная цифра: как разворачиваются числа. Меряется тем же WER, но на # записях, где в эталоне стоят цифры — там ошибка чаще всего в разворачивании out['numeric_wer'] = round(jiwer.wer(bucket.num_refs, bucket.num_hyps), 4) if bucket.rtf: out['rtf'] = { 'total': round(bucket.elapsed_s / bucket.audio_s, 4) if bucket.audio_s else None, 'median': round(statistics.median(bucket.rtf), 4), 'p95': round(sorted(bucket.rtf)[int(len(bucket.rtf) * 0.95) - 1], 4) if len(bucket.rtf) >= 20 else None, 'speedup_x': round(bucket.audio_s / bucket.elapsed_s, 2) if bucket.elapsed_s else None, } if bucket.punct_records: tot = [0, 0, 0] marks = {} for mark, (tp, fp, fn) in bucket.punct.items(): marks[mark] = f1(tp, fp, fn) tot = [tot[0] + tp, tot[1] + fp, tot[2] + fn] out['punctuation'] = {'records': bucket.punct_records, 'micro': f1(*tot), 'by_mark': marks} return out def load(paths: list[str]) -> list[dict]: rows: list[dict] = [] for pattern in paths: for path in sorted(glob.glob(pattern)): with open(path, encoding='utf-8') as fh: for line in fh: line = line.strip() if line: rows.append(json.loads(line)) return rows def to_markdown(report: dict) -> str: """Таблица для страницы /benchmark и статьи. Пустые клетки честнее прочерка в сноске: сразу видно, где метрика не определена на этом материале.""" lines = [ f'# Бенчмарк {report["generated_at"]}', '', f'Записей: {report["totals"]["records"]}, аудио: {report["totals"]["audio_hours"]} ч.', 'Нормализатор один на эталон и все гипотезы (`bench/normalize.py`), счётчик — `jiwer`.', '', '| Система | Срез | Записей | WER | CER | Числовой WER | Пунктуация F1 | RTF | Быстрее записи |', '|---|---|---:|---:|---:|---:|---:|---:|---:|', ] for system, slices in sorted(report['systems'].items()): for name, m in sorted(slices.items()): rtf = m.get('rtf') or {} punct = (m.get('punctuation') or {}).get('micro') or {} lines.append( f'| {system} | {name} | {m["records"]} | ' f'{_pct(m.get("wer"))} | {_pct(m.get("cer"))} | {_pct(m.get("numeric_wer"))} | ' f'{punct.get("f1", "") or ""} | {rtf.get("total") or ""} | ' f'{("×" + str(rtf["speedup_x"])) if rtf.get("speedup_x") else ""} |' ) lines += ['', '## Оговорки', ''] lines += [f'- {note}' for note in report['notes']] return '\n'.join(lines) + '\n' def _pct(v: float | None) -> str: return f'{v * 100:.2f}%' if isinstance(v, float) else '' def main() -> int: ap = argparse.ArgumentParser(description='Метрики бенчмарка по файлам гипотез') ap.add_argument('inputs', nargs='+', help='jsonl от run.py (можно маску)') ap.add_argument('--out', help='куда положить агрегаты (json)') ap.add_argument('--md', help='куда положить таблицу (markdown)') ap.add_argument('--per-file', help='выгрузить пофайловые ref/hyp (запрещено для ShareAlike-наборов)') args = ap.parse_args() rows = load(args.inputs) if not rows: print('нет записей') return 1 datasets = {r.get('dataset', '?') for r in rows} share_alike = sorted(datasets & SHARE_ALIKE) if args.per_file and share_alike: print( 'ОТКАЗ: пофайловая выгрузка запрещена лицензией наборов: ' + ', '.join(share_alike) + '. ShareAlike распространяется на производный материал — публикуем только агрегаты.' ) return 2 buckets: dict[tuple[str, str], Bucket] = defaultdict(Bucket) for r in rows: key = (r.get('system', '?'), f'{r.get("dataset", "?")}/{r.get("slice", "?")}') b = buckets[key] if r.get('error'): b.failures += 1 continue ref_raw, hyp_raw = r.get('ref') or '', r.get('hyp') or '' ref, hyp = normalize(ref_raw), normalize(hyp_raw) if not ref: continue if has_digits(ref_raw): b.num_refs.append(ref) b.num_hyps.append(hyp) else: b.refs.append(ref) b.hyps.append(hyp) score_punctuation(b, ref_raw, hyp_raw) dur, el = float(r.get('duration_s') or 0), float(r.get('elapsed_s') or 0) if dur > 0 and el > 0: b.audio_s += dur b.elapsed_s += el b.rtf.append(el / dur) systems: dict[str, dict] = defaultdict(dict) for (system, slice_name), b in buckets.items(): systems[system][slice_name] = summarize(b) total_audio = sum(b.audio_s for b in buckets.values()) report = { 'generated_at': __import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M'), 'totals': { 'records': len(rows), 'audio_hours': round(total_audio / 3600, 3), 'datasets': sorted(datasets), }, 'normalizer': 'bench/normalize.py', 'scorer': 'jiwer', 'notes': [ 'WER и CER считаются на записях, где в эталоне НЕТ цифр; записи с цифрами ' 'в эталоне вынесены в числовой WER.', 'Пунктуация меряется только на эталонах, где знаки есть; на Common Voice и ' 'Golos метрика не определена.', 'RTF — отношение времени обработки к длительности записи на том же ' 'устройстве, что указано в поле device прогона.', 'Пофайловые эталоны и гипотезы по наборам с ShareAlike (' + ', '.join(sorted(SHARE_ALIKE)) + ') не публикуются.', ], 'systems': systems, } if args.out: Path(args.out).parent.mkdir(parents=True, exist_ok=True) Path(args.out).write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8') print(f'агрегаты: {args.out}') if args.md: Path(args.md).parent.mkdir(parents=True, exist_ok=True) Path(args.md).write_text(to_markdown(report), encoding='utf-8') print(f'таблица: {args.md}') if args.per_file: with open(args.per_file, 'w', encoding='utf-8') as fh: for r in rows: fh.write(json.dumps({ 'id': r.get('id'), 'dataset': r.get('dataset'), 'slice': r.get('slice'), 'ref_norm': normalize(r.get('ref') or ''), 'hyp_norm': normalize(r.get('hyp') or ''), }, ensure_ascii=False) + '\n') print(f'пофайлово: {args.per_file}') if not (args.out or args.md): print(json.dumps(report, ensure_ascii=False, indent=2)) return 0 if __name__ == '__main__': raise SystemExit(main())