diff --git a/.gitignore b/.gitignore index 12726ff..49d5592 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,6 @@ __pycache__/ dist/ *.pyc .codex -.qwen/ \ No newline at end of file +.qwen/ +tmp +.scratch/handoffs/ diff --git a/docs/adr/007-word-level-speaker-diarization.md b/docs/adr/007-word-level-speaker-diarization.md index 4d95883..3962145 100644 --- a/docs/adr/007-word-level-speaker-diarization.md +++ b/docs/adr/007-word-level-speaker-diarization.md @@ -2,6 +2,23 @@ **Статус**: Принято **Дата**: 2026-08-14 +**Обновлено**: 2026-08-30 + +## Решение + +Диаризацию реализуем как явно включаемый пост-процессинг через `sherpa-onnx`. +Говорящий назначается слову с временной привязкой, а не целому сегменту +распознавания. Модель по умолчанию остаётся прежней: Pyannote Segmentation 3.0 FP32, +WeSpeaker ResNet34 LM, порог кластеризации `0,89` и автоматическое число +кластеров. Известное число говорящих можно передать явно. + +Повторный CPU-тест FP32/INT8 × WeSpeaker/TitaNet не меняет это решение. +INT8 с WeSpeaker дал только около 2% среднего ускорения без существенного +выигрыша памяти и с худшими прокси-метриками. TitaNet оказался примерно на +четверть быстрее, но потребовал ещё 147–149 МиБ RSS и не прошёл условие приёмки +для общего автоматического порога. TitaNet можно исследовать только для режима +с явно известным числом говорящих. Такой эксперимент должен завершиться +слуховой проверкой. Код и модели по умолчанию на него не переключаем. ## Контекст @@ -21,12 +38,20 @@ ## Эксперимент Локальная связка `sherpa-onnx` с сегментацией Pyannote 3.0 и эмбеддингами -WeSpeaker ResNet34 LM проверена на трёх записях с известным составом. Для -автоматического определения числа голосовых кластеров выбран порог 0,89: это -единственное проверенное значение, которое на трёх контрольных фрагментах дало -3 / 2 / 2 кластера. На полной встрече втроём остался ложный кластер длительностью -19,1 секунды; поэтому малые кластеры нельзя молча отбрасывать, а разметку нельзя -считать эталоном точных границ и перекрывающейся речи. +WeSpeaker ResNet34 LM проверена на трёх записях с известным составом. В исходной +калибровке для автоматического определения числа голосовых кластеров выбран +порог 0,89: среди проверенных тогда значений только он дал на трёх контрольных +фрагментах `3 / 2 / 2` кластера. На полной встрече втроём остался ложный кластер +длительностью 19,1 секунды; поэтому малые кластеры нельзя молча отбрасывать, а +разметку нельзя считать эталоном точных границ и перекрывающейся речи. + +В повторном CPU-переборе на тех же пятиминутных фрагментах FP32 + WeSpeaker +прошёл условие приёмки `3 / 2 / 2` на плато `0,85–0,89`, а INT8 + WeSpeaker — +на `0,85–0,90`. +Это не отменяет порог 0,89: новый прогон не имел повторов и новой слуховой +проверки. Обе связки TitaNet провалили условие приёмки во всех проверенных +точках; число содержательных кластеров не использовалось как оправдание лишних +голосовых кластеров. Двуязычная CAMPPlus zh/en оказалась примерно на 30% быстрее и при известном числе участников улучшила прокси-метрику на двух записях, но для неё не нашлось @@ -44,6 +69,8 @@ WeSpeaker ResNet34 LM проверена на трёх записях с изв [калибровке](../benchmarks/2026-08-14-diarization-calibration.md), [смешении говорящих](../benchmarks/2026-08-14-asr-segment-speaker-mixing.md) и [производительности на Intel](../benchmarks/2026-08-14-diarization-intel-i7.md). +Сравнение облегчённых моделей и границы его применимости описаны в +[отдельном отчёте](../benchmarks/2026-08-30-lightweight-diarization-cpu.md). ### Сглаживание неизвестного говорящего @@ -78,12 +105,7 @@ speaker Markdown хранит только начало реплики. Тем невозможна, потому что в нём уже потеряны доли секунды, конец слова и причина неопределённости. -## Решение - -Диаризацию реализуем как явно включаемый пост-процессинг через `sherpa-onnx`. -Первая версия использует Pyannote segmentation 3.0, WeSpeaker ResNet34 LM, -порог кластеризации 0,89 и автоматическое число кластеров; известное число -участников можно передать явно. +## Сведение слов и разметки говорящих Говорящий назначается [слову с временной привязкой](../../CONTEXT.md#language), а не сегменту распознавания. Каждый @@ -119,6 +141,9 @@ ASR-бэкенд приводит свой результат к общему н - Последовательный режим задаёт корректный baseline. Параллельный запуск и автоматическое включение на мощных устройствах требуют отдельных измерений после стабилизации. +- Pyannote INT8 и TitaNet не становятся конфигурациями по умолчанию только на + основании CPU-теста. Для TitaNet следующий допустимый эксперимент — + режим с явно известным числом говорящих, повторами и слуховой проверкой. ## Отклонённые альтернативы diff --git a/docs/benchmarks/2026-08-30-lightweight-diarization-cpu.md b/docs/benchmarks/2026-08-30-lightweight-diarization-cpu.md new file mode 100644 index 0000000..dc8df3f --- /dev/null +++ b/docs/benchmarks/2026-08-30-lightweight-diarization-cpu.md @@ -0,0 +1,223 @@ +# Облегчённые модели диаризации на старом Intel Core i7 + +**Дата:** 2026-08-30 + +**Статус:** сравнительный CPU-тест. Конфигурация по умолчанию не менялась. + +## Решение + +Сохранить модель по умолчанию: Pyannote Segmentation 3.0 FP32, WeSpeaker +ResNet34 LM, общий порог автоматической кластеризации `0,89`. Новый прогон +подтвердил для этой связки точное число голосовых кластеров `3 / 2 / 2` на всех +трёх контрольных фрагментах. + +Pyannote INT8 с WeSpeaker не заменяет FP32. Среднее ускорение составило только +2,1%, средний пиковый RSS по ячейкам снизился примерно на 4 МиБ, а средняя +чистота сопоставленных говорящих и число слов без говорящего стали хуже. + +TitaNet small в среднем быстрее WeSpeaker на 24,3%, но требует примерно на +147–149 МиБ больше RSS и не находит правильное число кластеров ни в одной +проверенной точке общего автоматического порога. Его можно рассматривать только +как кандидата для режима с явно известным числом говорящих. Следующий эксперимент +должен завершиться отдельной слуховой проверкой. По этому тесту менять код, +модели или пороги нельзя. + +## Границы вывода + +Проверялись только четыре CPU-сочетания Pyannote FP32/INT8 × +WeSpeaker/TitaNet на трёх пятиминутных русскоязычных рабочих созвонах из +[исходной калибровки](2026-08-14-diarization-calibration.md). Число говорящих в +них известно как `3 / 2 / 2`. Это сравнительный прогон моделей диаризации, а не +новая приёмка готовых транскриптов. + +В частности: + +- новой слуховой проверки не было; +- полные записи не прогонялись; +- у каждой ячейки один запуск, без повторов и случайного перемешивания порядка; +- влияние порядка, прогрева и температуры отдельно не измерено; +- ASR выполнен один раз на запись и не входит во время ячеек диаризации; +- параллельные ASR и диаризация, GPU, NPU и другие CPU не проверялись. + +Происхождение, лицензии и закреплённые SHA-256 моделей описаны в +[исследовательской записке](../research/2026-08-29-lightweight-diarization-models.md), +а смысл `mapped speaker purity` и ограничения опорной разметки — в +[исходной калибровке](2026-08-14-diarization-calibration.md). Здесь эти сведения +не дублируются. + +## Методика + +Каждая измеряемая ячейка запускалась в новом дочернем процессе. До измеряемого +интервала загружались WAV, модели и сэмплы; полное время, CPU и пиковый RSS +считались вокруг одного вызова `diarizer.process(samples)`. Отладочный журнал +`sherpa-onnx` 1.13.5 дал время сегментации, извлечения эмбеддингов, кластеризации +и всего вызова. +Декодирование, загрузка моделей, сериализация и ASR во время не входят. + +Автоматический перебор прямоугольный: каждый порог сочетания проверен на всех +трёх записях с `num_clusters=-1`. Для сочетания выбирался один общий порог. +Условие приёмки считалось выполненным только при полном совпадении общего числа +кластеров с +`3 / 2 / 2`; число содержательных кластеров оставалось диагностикой и не могло +оправдать лишний кластер. После выбора порога каждое сочетание ещё раз прошло +три записи с известным числом кластеров. + +Итого сохранено 108 успешных ячеек без ошибок: + +- 96 с автоматическим числом кластеров: по 18 для каждого сочетания с + WeSpeaker и по 30 для каждого сочетания с TitaNet; +- 12 с известным числом кластеров: по три для каждого из четырёх сочетаний. + +Отдельно сохранены четыре результата выбора общего порога. Они описывают +агрегацию автоматических ячеек и в число 108 не входят. + +Во всех 108 ячейках совпала последовательность 1853 распознанных слов после +удаления Unicode-пробелов. Инвариантность текста не нарушена. + +## Воспроизведение + +Стенд [`diarization_calibration.py`](../../scripts/benchmarks/diarization_calibration.py) +закрепляет корпус и матрицу задачи #25. Файл manifest, ASR-sidecar, интервалы и +полный результат содержат приватные данные и хранятся отдельно от Git. +Относительные пути в manifest считаются от каталога самого файла, поэтому +приватный набор можно перенести на другую машину целиком. Готовый файл результата +при этом не переносится: продолжение допустимо только на той же платформе, CPU и +том же числе логических ядер. + +Перед дорогим запуском нужно проверить набор без декодирования и моделей: + +```bash +uv run python scripts/benchmarks/diarization_calibration.py \ + --manifest /path/to/manifest.json \ + --validate-only +``` + +Полный запуск и его продолжение используют одну команду и один файл результата: + +```bash +uv run python scripts/benchmarks/diarization_calibration.py \ + --manifest /path/to/manifest.json \ + --output /path/to/results.json \ + --work-dir /path/to/work +``` + +Совместимый файл продолжится с первой незавершенной ячейки. Код возврата 0 +означает, что завершены все автоматические ячейки и ячейки с известным числом +говорящих. Число кластеров и инвариантность текста нужно сверить с условиями +конкретного исследования. Полный контракт аргументов и режим подготовки +ASR-sidecar описаны в `--help`. Результат включает исходный manifest, +приватные пути и интервалы. Поэтому его нельзя добавлять в Git или передавать +внешнему ревьюеру. + +## Оборудование и условия + +- Intel Core i7-6820HQ, 4 ядра / 8 логических процессоров; +- Linux 7.0.0-30-generic x86_64; +- питание от сети, системный профиль `balanced`; +- Python 3.13.13, `sherpa-onnx` 1.13.5, `onnxruntime` 1.28.0, + NumPy 2.4.3; +- CPU provider и 8 потоков для segmentation и embedding; +- mono PCM16, 16 кГц, длительность каждого фрагмента 300 секунд. + +Это тот же ноутбук, что в [предыдущей Intel-приёмке](2026-08-14-diarization-intel-i7.md), +но условия нельзя считать побайтовым повтором: там отдельно зафиксированы +governor `powersave` и урезанная доступная мощность, а версия ядра была +7.0.0-29. Во время нового перебора ночная пауза произошла до сохранения текущей +ячейки. Продолжение началось после возврата профиля `balanced`; ячеек, +измеренных в энергосберегающем профиле, в итоговом результате нет. + +## Производительность полного автоматического перебора + +Числа ниже — средние по всем автоматическим ячейкам сочетания. Это не повторы +одной ячейки. «Средний пиковый RSS» — среднее пикового RSS каждой ячейки; +соседний столбец показывает наибольший такой максимум во всём переборе. + +| Сегментация + эмбеддинги | Ячеек | Время, с | RTF | CPU, ядер | Средний пиковый RSS, МиБ | Максимальный пиковый RSS, МиБ | Сегментация, с | Эмбеддинги, с | +|---|---:|---:|---:|---:|---:|---:|---:|---:| +| FP32 + WeSpeaker | 18 | 72,486 | 0,24162 | 6,84 | 254,55 | 265,61 | 15,295 | 56,837 | +| INT8 + WeSpeaker | 18 | 70,969 | 0,23656 | 6,94 | 250,40 | 263,00 | 17,094 | 53,524 | +| FP32 + TitaNet | 30 | 52,797 | 0,17599 | 6,80 | 401,80 | 428,11 | 14,724 | 37,723 | +| INT8 + TitaNet | 30 | 55,792 | 0,18597 | 6,70 | 399,10 | 419,71 | 17,125 | 38,278 | + +INT8 ускорил сочетание с WeSpeaker на 2,1%, но сегментация стала медленнее, а +выигрыш возник на извлечении эмбеддингов. Такой результат не даёт практического +основания заменять FP32. У TitaNet FP32 оказался быстрее INT8. Относительно +соответствующей связки WeSpeaker ускорение TitaNet составило 27,2% для FP32 и +21,4% для INT8. +Среднее для двух вариантов точности — 24,3%. + +Опубликованное ожидание ускорения от INT8 локально не подтвердилось. Вместо +18,9% с 3D-Speaker получено 2,1% с WeSpeaker, а с TitaNet INT8 оказался на 5,7% +медленнее FP32. Условия исходного примера и границы такого сравнения разобраны +в [исследовательской записке](../research/2026-08-29-lightweight-diarization-models.md). + +За скорость TitaNet платит памятью: средний пиковый RSS выше WeSpeaker на 147,25 +МиБ с FP32 и на 148,70 МиБ с INT8. Поэтому на слабом CPU его нельзя считать +безусловно облегчённым вариантом. + +## Автоматический порог и диагностика качества + +`Mapped speaker purity` усреднена только по T2 BDMA и Yantar, для которых есть +грубая опорная разметка. Это сравнительная прокси-метрика, не DER и не точность +по выверенной разметке. + +| Сочетание | Выбранный порог | Приёмка `3 / 2 / 2` | Кластеры | Средняя purity | Слов без говорящего, из 1853 | Максимальный остаток сверх ожидаемых | +|---|---:|:---:|---|---:|---:|---:| +| FP32 + WeSpeaker | 0,85 | да | 3 / 2 / 2 | 0,82950 | 71 | 0% | +| INT8 + WeSpeaker | 0,85 | да | 3 / 2 / 2 | 0,81028 | 104 | 0% | +| FP32 + TitaNet | 1,10 | **нет** | 5 / 3 / 3 | 0,82287 | 76 | 3,89% | +| INT8 + TitaNet | 1,10 | **нет** | 3 / 2 / 3 | 0,79315 | 109 | 12,12% | + +У FP32 + WeSpeaker условие приёмки проходит на плато `0,85–0,89`, у INT8 + +WeSpeaker — на `0,85–0,90`. Автоматический выбор дал `0,85` по общей +лексикографической оценке, но это не аргумент менять рабочий порог `0,89`: +новый перебор не содержит повторов или слуховой проверки, тогда как действующее +решение опирается также на предыдущую калибровку полных записей и слуховую +проверку плотного диалога. + +TitaNet не прошёл условие приёмки ни в одной точке проверенной сетки `0,50–1,10`. +Порог `1,10` — лишь наименее плохая точка перебора; допустимой автоматической +конфигурацией она не является. Содержательные `3 / 2 / 3` кластера у FP32 TitaNet не отменяют +лишние голосовые кластеры: диагностический фильтр не заменяет условие приёмки. + +«Максимальный остаток сверх ожидаемых» — наибольшая по трём записям доля слов, +которые попали в кластеры сверх известного числа говорящих. Метрика описывает +объём лишних кластеров, а не их количество: один крупный лишний кластер может +дать больший процент, чем несколько малых. + +## Режим с известным числом говорящих + +Ячейки с известным числом говорящих использовали тот же выбранный порог и +меняли только `num_clusters` на известные `3 / 2 / 2`. Все четыре сочетания +вернули точное число кластеров на всех трёх записях. + +| Сочетание | Точное число кластеров | Точное число содержательных кластеров | Средняя purity | +|---|---:|---:|---:| +| FP32 + WeSpeaker | 3 / 3 записей | 3 / 3 записей | 0,82950 | +| INT8 + WeSpeaker | 3 / 3 записей | 3 / 3 записей | 0,81028 | +| FP32 + TitaNet | 3 / 3 записей | 2 / 3 записей | 0,83057 | +| INT8 + TitaNet | 3 / 3 записей | 3 / 3 записей | 0,79721 | + +Близкая средняя purity FP32 TitaNet и FP32 WeSpeaker не доказывает равноценное +качество. У FP32 TitaNet на Data Test один из трёх принудительных кластеров не +достиг диагностического порога содержательности, а новой слуховой проверки не +было. Поэтому TitaNet остаётся только направлением будущего эксперимента с +обязательным известным числом говорящих. + +## Ограничения и следующий шаг + +- Три записи относятся к одному типу рабочих созвонов; результат не переносится + автоматически на другие языки, микрофоны и акустические условия. +- Две опорные разметки имеют секундную точность и не позволяют считать DER. +- Средние значения полного времени и RSS объединяют разные записи и пороги. + Без повторов они не оценивают дисперсию и не отделяют порядок, прогрев или + влияние температуры. +- Новый прогон не заменяет прежнюю слуховую проверку и не даёт слуховых данных + для TitaNet или INT8. +- Результат относится к последовательной CPU-диаризации на одном Core + i7-6820HQ и не предсказывает конкретный Core i5 11-го поколения. + +Если возвращаться к TitaNet, следующий эксперимент должен заранее требовать +известное число говорящих, включать повторные прогоны с контролем порядка и +температуры и завершаться слуховой проверкой всех трёх записей. До этого +конфигурацию и модель по умолчанию не менять. diff --git a/docs/research/2026-08-29-lightweight-diarization-models.md b/docs/research/2026-08-29-lightweight-diarization-models.md new file mode 100644 index 0000000..37849ed --- /dev/null +++ b/docs/research/2026-08-29-lightweight-diarization-models.md @@ -0,0 +1,255 @@ +# Облегчённые модели диаризации: Pyannote INT8 и TitaNet small + +**Дата:** 2026-08-29 + +**Статус:** исследовательская записка по публичным источникам. Выводы позднее +проверены локальным CPU-стендом; рабочая конфигурация не менялась. + +**Метод:** использованы только первичные публичные источники: документация, +исходный код и выпуски sherpa-onnx, карточки владельцев моделей и исходная +публикация TitaNet. Приватные медиа и опорные транскрипты Hypescribe не +открывались. + +## Краткий вывод + +Sherpa-onnx официально документирует все четыре сочетания Pyannote +Segmentation 3.0 FP32/INT8 с 3D-Speaker и `nemo_en_titanet_small.onnx`. В +опубликованном разработчиками примере производительности переход FP32 → INT8 +сокращает +RTF с 0,297 до 0,241 с 3D-Speaker и с 0,119 до 0,110 с TitaNet small. Но это +одна 56,861-секундная китайская контрольная запись, известные четыре говорящих, +CPU-провайдер и по одному потоку на сегментацию и эмбеддинги. Данных о +конкретном CPU, повторностях, разбросе и качестве на русском материале страница +не приводит. Поэтому цифры подтверждают совместимость и потенциал, но не +заменяют локальный тест. Условия и результаты взяты с [официальной +страницы sherpa-onnx](https://k2-fsa.github.io/sherpa/onnx/speaker-diarization/models.html). + +Официальные файлы выпусков доступны под следующими именами: + +- `sherpa-onnx-pyannote-segmentation-3-0.tar.bz2` содержит + `model.onnx` (FP32) и `model.int8.onnx`; +- `nemo_en_titanet_small.onnx` поставляется отдельным ONNX-файлом. + +GitHub API не публикует `digest` ни для одного из этих файлов. Ниже зафиксированы +SHA-256 байтов, скачанных 2026-08-29 с официальных URL, и точные размеры, +совпавшие с GitHub API. Эти хеши пригодны для закрепления текущей поставки, но +получены в этой проверке и не заявлены разработчиками как контрольные суммы. + +## Официальная поддержка и точные артефакты + +### Pyannote Segmentation 3.0: FP32 и INT8 в одном архиве + +Официальная документация sherpa-onnx называет модель +`sherpa-onnx-pyannote-segmentation-3-0`, указывает, что она конвертирована из +[`pyannote/segmentation-3.0`](https://huggingface.co/pyannote/segmentation-3.0), +ссылается на [официальный каталог скриптов +конвертации](https://github.com/k2-fsa/sherpa-onnx/tree/master/scripts/pyannote/segmentation) +и показывает оба файла, `model.onnx` и `model.int8.onnx`, в одном архиве +([документация и команда загрузки](https://k2-fsa.github.io/sherpa/onnx/speaker-diarization/models.html#download-the-model)). + +Точное имя и URL: + +```text +sherpa-onnx-pyannote-segmentation-3-0.tar.bz2 +https://github.com/k2-fsa/sherpa-onnx/releases/download/speaker-segmentation-models/sherpa-onnx-pyannote-segmentation-3-0.tar.bz2 +``` + +На момент проверки [GitHub Releases +API](https://api.github.com/repos/k2-fsa/sherpa-onnx/releases/tags/speaker-segmentation-models) +сообщает размер файла `6 958 444` байта, +`created_at=2024-10-08T12:54:09Z`, `updated_at=2024-10-08T12:54:10Z` и +`digest=null`; сам [официальный выпуск](https://github.com/k2-fsa/sherpa-onnx/releases/tag/speaker-segmentation-models) +содержит этот файл, а документация даёт тот же URL загрузки. + +Проверенное содержимое архива: + +| Файл | Размер | SHA-256 | +|---|---:|---| +| `model.onnx` | 5 992 913 байт | `220ad67ca923bef2fa91f2390c786097bf305bceb5e261d4af67b38e938e1079` | +| `model.int8.onnx` | 1 540 506 байт | `d582f4b4c6b48205de7e0643c57df0df5615a3c176189be3fc461e9d18827b5d` | + +Архив также содержит `LICENSE`, `README.md` и скрипты экспорта/проверки. Это +согласуется с опубликованным sherpa-onnx листингом: примерно 5,7 МБ для FP32 и +1,5 МБ для INT8 +([официальный листинг](https://k2-fsa.github.io/sherpa/onnx/speaker-diarization/models.html#download-the-model)). + +### NeMo TitaNet small: отдельная модель эмбеддингов + +Официальная страница примеров загружает `nemo_en_titanet_small.onnx` из выпуска +моделей распознавания говорящих и использует его как `--embedding.model` вместе с +обоими вариантами Pyannote +([FP32](https://k2-fsa.github.io/sherpa/onnx/speaker-diarization/models.html#nemo-model-onnx), +[INT8](https://k2-fsa.github.io/sherpa/onnx/speaker-diarization/models.html#nemo-model-int8-onnx)). +Отдельная [страница моделей NeMo в sherpa-onnx](https://k2-fsa.github.io/sherpa/onnx/nemo/index.html) +перечисляет это точное имя среди поддерживаемых моделей эмбеддингов говорящих. + +Точное имя и URL: + +```text +nemo_en_titanet_small.onnx +https://github.com/k2-fsa/sherpa-onnx/releases/download/speaker-recongition-models/nemo_en_titanet_small.onnx +``` + +Опечатка `recongition` является частью официальной метки выпуска и URL; исправлять +её на `recognition` нельзя. На момент проверки [GitHub Releases +API](https://api.github.com/repos/k2-fsa/sherpa-onnx/releases/tags/speaker-recongition-models) +сообщает размер файла +`40 257 283` байта, `created_at=2024-10-14T07:03:02Z`, +`updated_at=2024-10-14T07:03:03Z` и `digest=null`; [выпуск +sherpa-onnx](https://github.com/k2-fsa/sherpa-onnx/releases/tag/speaker-recongition-models) +предупреждает, что лицензия у каждой модели своя и её нужно искать в исходном +репозитории модели. + +## Контекст опубликованных результатов + +### Пример производительности sherpa-onnx + +Все четыре числа ниже опубликованы в одном примере sherpa-onnx. В нём используется +`0-four-speakers-zh.wav` длительностью 56,861 с, явно задано +`--clustering.num-clusters=4`, а распечатанная конфигурация показывает CPU +провайдер и `num_threads=1` отдельно для сегментации и эмбеддингов. Страница не +указывает модель CPU, не описывает прогрев, число повторов или дисперсию и не +публикует метрику качества против эталонной разметки +([полный пример](https://k2-fsa.github.io/sherpa/onnx/speaker-diarization/models.html#sherpa-onnx-pyannote-segmentation-3-0)). + +| Segmentation | Embedding | Время | RTF | +|---|---|---:|---:| +| Pyannote FP32 `model.onnx` | 3D-Speaker ERes2Net base | 16,870 с | 0,297 | +| Pyannote INT8 `model.int8.onnx` | 3D-Speaker ERes2Net base | 13,679 с | 0,241 | +| Pyannote FP32 `model.onnx` | NeMo TitaNet small | 6,756 с | 0,119 | +| Pyannote INT8 `model.int8.onnx` | NeMo TitaNet small | 6,231 с | 0,110 | + +По опубликованным RTF мы пересчитали относительное ускорение. INT8 ускоряет +сочетание с 3D-Speaker примерно на 18,9%, а сочетание с TitaNet small — +примерно на 7,6%. Замена 3D-Speaker на TitaNet small при сохранении FP32 даёт +около 59,9%, а сочетание TitaNet small + INT8 относительно исходного FP32 + +3D-Speaker — около 63,0%. Здесь сравнивается только время выполнения на +указанной записи. По этим цифрам нельзя судить о DER, переносимости ускорения +на другой CPU или качестве на русском языке +([исходные времена и RTF](https://k2-fsa.github.io/sherpa/onnx/speaker-diarization/models.html#sherpa-onnx-pyannote-segmentation-3-0)). + +### Что означает «small» в публикации TitaNet + +Исходная статья определяет TitaNet-S как вариант с 256 каналами и 6,4 млн +параметров. Для очищенного набора VoxCeleb1 в задаче верификации говорящих она +сообщает EER 1,15% при сравнении по косинусному сходству. Для диаризации авторы +используют эталонный детектор голосовой активности (oracle SAD), кластеризацию +NME-SC, допуск 0,25 с и исключают перекрытия из подсчёта DER. При +известном числе говорящих DER TitaNet-S равен 6,37 / 2,00 / 2,22 / 1,11% на +NIST-SRE-2000 / AMI-Lapel / AMI-MixHeadset / CH109; при оценённом числе — +5,49 / 2,30 / 1,97 / 1,42%. Это тест архитектуры и исходной контрольной точки +NeMo в другом конвейере диаризации, не тест ONNX-экспорта внутри +sherpa-onnx +([первичная публикация TitaNet](https://arxiv.org/abs/2110.04410)). + +Карточка NVIDIA NGC подтверждает назначение `titanet_small` для верификации +говорящих, диаризации и извлечения 192-мерных эмбеддингов, но её +раздел Performance содержит внутренне противоречивую фразу «TitaNet-L» рядом с +6,4 млн параметров. Поэтому для чисел TitaNet-S выше использована исходная +статья вместо этой фразы карточки +([официальная карточка NGC](https://catalog.ngc.nvidia.com/orgs/nvidia/teams/nemo/models/titanet_small)). + +## Происхождение моделей и лицензии + +### Pyannote Segmentation 3.0 + +Исходная карточка Pyannote описывает 10-секундный mono-вход 16 кГц и powerset +выход из семи классов: тишина, три одиночных говорящих и три пары перекрывающихся +говорящих. Модель обучил Séverin Baroudi в `pyannote.audio` 3.0.0 на сочетании +AISHELL, AliMeeting, AMI, AVA-AVD, DIHARD, Ego4D, MSDWild, REPERE и VoxConverse +([карточка исходной модели](https://huggingface.co/pyannote/segmentation-3.0)). + +Исходный репозиторий помечает модель лицензией MIT и требует принять условия +доступа к его файлам на Hugging Face. Публичный архив sherpa-onnx включает +собственный `LICENSE` с текстом MIT; в проверенном архиве copyright указан как +`Copyright (c) 2022 CNRS`, тогда как текущий LICENSE исходного репозитория +указывает 2023. Тип лицензии совпадает, но при дальнейшей поставке следует +сохранять именно `LICENSE` из закреплённого архива +([лицензия и условия исходной модели](https://huggingface.co/pyannote/segmentation-3.0), +[текущий LICENSE upstream](https://huggingface.co/pyannote/segmentation-3.0/blob/main/LICENSE)). + +### NeMo TitaNet small + +Sherpa-onnx публикует [скрипт экспорта](https://github.com/k2-fsa/sherpa-onnx/blob/master/scripts/nemo/speaker-verification/export-onnx.py), +который загружает `EncDecSpeakerLabelModel.from_pretrained(model_name="titanet_small")`, +экспортирует `nemo_en_titanet_small.onnx` и добавляет ONNX metadata с framework +`nemo`, language `English`, source URL NGC, sample rate и размерностью embedding. +[README каталога экспорта](https://github.com/k2-fsa/sherpa-onnx/tree/master/scripts/nemo/speaker-verification) +также прямо называет NGC `titanet_small` источником модели. + +Карточка NGC называет актуальную опубликованную версию `1.19.0`, обновлённую +7 июня 2023 года, и описывает обучение на VoxCeleb 1/2, RIR noise, Fisher, +Switchboard и LibriSpeech. Реализация NeMo связывает имя `titanet_small` с +checkpoint `titanet-s.nemo` версии `1.19.0` +([карточка NVIDIA NGC](https://catalog.ngc.nvidia.com/orgs/nvidia/teams/nemo/models/titanet_small), +[официальный реестр pretrained-моделей NeMo](https://github.com/NVIDIA-NeMo/Speech/blob/main/nemo/collections/asr/models/label_models.py)). + +NGC говорит, что использование модели регулируется лицензией NeMo; ссылка из +карточки ведёт на Apache License 2.0. Сам файл выпуска sherpa-onnx — одиночный +ONNX-файл без соседнего LICENSE, а описание выпуска перекладывает проверку +лицензии на пользователя. Поэтому при включении модели в приложение нужно +зафиксировать происхождение NVIDIA NeMo/TitaNet-S и приложить лицензию Apache-2.0, +а не считать общий LICENSE sherpa-onnx лицензией весов +([секция Licence карточки NGC](https://catalog.ngc.nvidia.com/orgs/nvidia/teams/nemo/models/titanet_small), +[LICENSE NeMo](https://github.com/NVIDIA-NeMo/Speech/blob/main/LICENSE), +[предупреждение release sherpa-onnx](https://github.com/k2-fsa/sherpa-onnx/releases/tag/speaker-recongition-models)). + +Происхождение нельзя воспроизвести полностью: текущий `run.sh` sherpa-onnx +устанавливает NeMo из ветки `main` без закрепления коммита или метки, затем +вызывает экспорт +для `titanet_small`. Он документирует процесс, но сам по себе не доказывает, +из какого коммита NeMo был создан файл от 2024-10-14. Для побайтовой +воспроизводимости следует считать официально скачанный ONNX плюс SHA-256 ниже +канонической единицей, пока разработчики не опубликуют данные закреплённой сборки +([официальный `run.sh`](https://github.com/k2-fsa/sherpa-onnx/blob/master/scripts/nemo/speaker-verification/run.sh)). + +## Проверенные SHA-256 + +Файлы скачаны 2026-08-29 напрямую с двух URL выше в `/tmp`; размер каждого +сверен с полем `size` GitHub Releases API. Архив дополнительно проверен командой +`bzip2 -t`. Поле `digest` в API было `null` для обоих файлов, поэтому ни один из +следующих хешей нельзя называть опубликованной контрольной суммой. + +```text +24615ee884c897d9d2ba09bb4d30da6bb1b15e685065962db5b02e76e4996488 sherpa-onnx-pyannote-segmentation-3-0.tar.bz2 +ad4a1802485d8b34c722d2a9d04249662f2ece5d28a7a039063ca22f515a789e nemo_en_titanet_small.onnx +``` + +Внутри проверенного Pyannote-архива: + +```text +220ad67ca923bef2fa91f2390c786097bf305bceb5e261d4af67b38e938e1079 model.onnx +d582f4b4c6b48205de7e0643c57df0df5615a3c176189be3fc461e9d18827b5d model.int8.onnx +``` + +Команда проверки после загрузки: + +```bash +sha256sum \ + sherpa-onnx-pyannote-segmentation-3-0.tar.bz2 \ + nemo_en_titanet_small.onnx +``` + +## Результат локальной проверки + +[Прогон на Core i7-6820HQ](../benchmarks/2026-08-30-lightweight-diarization-cpu.md) +не подтвердил опубликованное ускорение INT8: сочетание с WeSpeaker выиграло +только 2,1%, а с TitaNet стало на 5,7% медленнее. TitaNet ускорил обработку в +среднем на 24,3%, но потребовал примерно на 147–149 МиБ больше RSS и не прошёл +приёмку общего автоматического порога. + +## Практический вердикт + +1. Для минимального эксперимента можно заменить только путь Pyannote + `model.onnx` на `model.int8.onnx`: это официально поставляемый вариант того же + архива и не требует менять модель эмбеддингов + ([официальный пример INT8](https://k2-fsa.github.io/sherpa/onnx/speaker-diarization/models.html#d-speaker-model-int8-onnx)). +2. Для большего потенциального ускорения sherpa-onnx официально принимает + `nemo_en_titanet_small.onnx` с обоими вариантами Pyannote, но опубликованный + пример производительности не измеряет DER и не относится к русской речи + ([официальные примеры NeMo](https://k2-fsa.github.io/sherpa/onnx/speaker-diarization/models.html#nemo-model-onnx)). +3. Для воспроизводимой поставки следует закрепить точные URL, размеры и SHA-256, + хранить MIT LICENSE из Pyannote-архива и явно приложить сведения об авторстве + и лицензии Apache-2.0 для TitaNet/NeMo. До локальной калибровки нельзя + переносить ни коэффициент ускорения, ни порог кластеризации, ни качество с + опубликованной контрольной записи на целевые русскоязычные созвоны. diff --git a/pyproject.toml b/pyproject.toml index 50bef61..e0c55d8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,7 @@ transcribe = "local_transcriber.cli:app" [dependency-groups] dev = [ + "psutil>=7,<8", "pytest", ] diff --git a/scripts/benchmarks/diarization_calibration.py b/scripts/benchmarks/diarization_calibration.py index 150af33..9095f0e 100644 --- a/scripts/benchmarks/diarization_calibration.py +++ b/scripts/benchmarks/diarization_calibration.py @@ -1,43 +1,243 @@ -"""Воспроизводимый свип параметров офлайн-диаризации sherpa-onnx.""" +"""Воспроизводимый CPU-benchmark офлайн-диаризации sherpa-onnx.""" from __future__ import annotations import argparse +import copy import hashlib +import importlib.metadata import itertools import json +import math +import os +import platform import re import subprocess +import sys +import threading import time import wave from collections import defaultdict -from dataclasses import dataclass +from collections.abc import Callable +from datetime import UTC, datetime from pathlib import Path +from tempfile import NamedTemporaryFile from typing import Any -import numpy as np -import sherpa_onnx - +SCHEMA_VERSION = 2 +EXPECTED_SHERPA_ONNX_VERSION = "1.13.5" +SAMPLE_RATE = 16_000 +SAMPLE_WIDTH_BYTES = 2 +CHANNELS = 1 +DECODER_ID = "ffmpeg-mono-16000-pcm_s16le-v1" +AUDIO_DURATION_TOLERANCE_SECONDS = 0.05 +TOP_LEVEL_KEYS = { + "schema_version", + "segmentation_models", + "embedding_models", + "combinations", + "recordings", + "threads", + "rss_sample_interval_ms", +} +SEGMENTATION_KEYS = { + "id", + "path", + "precision", + "sha256", + "size_bytes", + "source_url", + "license", +} +EMBEDDING_KEYS = {"id", "path", "sha256", "size_bytes", "source_url", "license"} +COMBINATION_KEYS = {"id", "segmentation_id", "embedding_id", "role", "thresholds"} +RECORDING_KEYS = { + "id", + "path", + "sha256", + "start", + "duration", + "expected_speakers", + "reference", + "asr_words", +} +REFERENCE_KEYS = {"path", "sha256", "format"} +ASR_WORDS_KEYS = {"path", "sha256", "format"} +SHA256_RE = re.compile(r"^[0-9a-f]{64}$") TURN_RE = re.compile(r"^\*\*\[(\d{2}):(\d{2})(?::(\d{2}))?\] Speaker (\d+):\*\*") +NUMBER_PATTERN = r"[+-]?(?:\d+(?:\.\d*)?|\.\d+)" +STAGE_RE = re.compile( + rf"OfflineSpeakerDiarization:\s+(segmentation|embedding|clustering)\s+({NUMBER_PATTERN})\s+s" +) +TOTAL_RE = re.compile( + rf"OfflineSpeakerDiarization:\s+total\s+({NUMBER_PATTERN})\s+s,\s+" + rf"audio\s+({NUMBER_PATTERN})\s+s,\s+RTF\s+({NUMBER_PATTERN})" +) +ORIGINAL_CORPUS = { + "data-test": { + "sha256": "1057616b42e8add00e0eb975b02bdef0ec9f6cdfec6dbf488e0c60423c9b7b87", + "start": 420.0, + "duration": 300.0, + "expected_speakers": 3, + }, + "t2-bdma": { + "sha256": "51866d247fe3eda134cdd884f707b1f1db8855b492e8bd14d2b56b62476255ed", + "start": 0.0, + "duration": 300.0, + "expected_speakers": 2, + }, + "yantar": { + "sha256": "4422f04e2771091a0648e5422d14a31dd7ca2c8eef7ed9f4a5c63f43d8ca6400", + "start": 0.0, + "duration": 300.0, + "expected_speakers": 2, + }, +} -@dataclass(frozen=True) -class Recording: - name: str - path: Path - start: float - duration: float - expected_speakers: int - reference: Path | None +class PeakRssSampler: + """Сэмплирует RSS процесса только внутри измеряемого интервала.""" + + def __init__(self, process: Any, interval_seconds: float): + self._process = process + self._interval_seconds = interval_seconds + self._stop = threading.Event() + self._thread: threading.Thread | None = None + self.peak_bytes: int | None = None + + def _sample(self) -> None: + value = int(self._process.memory_info().rss) + self.peak_bytes = ( + value if self.peak_bytes is None else max(self.peak_bytes, value) + ) + + def _run(self) -> None: + while not self._stop.wait(self._interval_seconds): + self._sample() + + def start(self) -> None: + self._sample() + self._thread = threading.Thread(target=self._run, daemon=True) + self._thread.start() + + def stop(self) -> int: + self._stop.set() + if self._thread is not None: + self._thread.join() + self._sample() + if self.peak_bytes is None: + raise RuntimeError("Не удалось измерить пиковый RSS") + return self.peak_bytes -def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser() - parser.add_argument("--manifest", type=Path, required=True) - parser.add_argument("--output", type=Path, required=True) - parser.add_argument("--work-dir", type=Path, required=True) - parser.add_argument("--threads", type=int, default=8) - return parser.parse_args() +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Проверяет и запускает воспроизводимый CPU-benchmark из issue #25. " + "Приватные пути и результаты остаются вне Git." + ), + epilog=( + "Сначала выполните --validate-only. Полный запуск атомарно сохраняет " + "каждую ячейку и продолжает совместимый output после прерывания." + ), + ) + parser.add_argument( + "--manifest", + type=Path, + help="manifest v2; относительные пути считаются от его каталога", + ) + parser.add_argument( + "--output", + type=Path, + help="приватный JSON результата; существующий совместимый файл продолжится", + ) + parser.add_argument( + "--work-dir", + type=Path, + help="каталог декодированных WAV и запросов дочерним процессам", + ) + # Оставлено для воспроизводимости старых команд; значение входит в experiment_id. + parser.add_argument( + "--threads", + type=int, + help="число потоков; если задано, должно совпасть с manifest.threads", + ) + parser.add_argument( + "--validate-only", + action="store_true", + help="проверить файлы, корпус и матрицу без декодирования и запуска моделей", + ) + parser.add_argument( + "--prepare-asr", + metavar="RECORDING_ID", + help="создать ASR-sidecar одной записи и вывести его SHA-256", + ) + parser.add_argument("--asr-model", help="production ASR-модель для --prepare-asr") + parser.add_argument("--asr-device", help="устройство ASR для --prepare-asr") + parser.add_argument("--asr-compute-type", help="точность ASR для --prepare-asr") + parser.add_argument("--asr-language", help="язык ASR для --prepare-asr") + parser.add_argument("--worker", type=Path, help=argparse.SUPPRESS) + args = parser.parse_args(argv) + if args.worker is None: + validate_only_conflicts = ( + "output", + "work_dir", + "prepare_asr", + "asr_model", + "asr_device", + "asr_compute_type", + "asr_language", + ) + if args.validate_only and any( + getattr(args, name) is not None for name in validate_only_conflicts + ): + parser.error( + "--validate-only нельзя сочетать с параметрами запуска или подготовки ASR" + ) + missing = ["manifest"] if args.manifest is None else [] + if not args.validate_only and args.work_dir is None: + missing.append("work_dir") + if not args.validate_only and args.prepare_asr is None and args.output is None: + missing.append("output") + if missing: + parser.error("обязательные параметры: " + ", ".join(missing)) + if args.prepare_asr is not None: + asr_missing = [ + name + for name in ( + "asr_model", + "asr_device", + "asr_compute_type", + "asr_language", + ) + if getattr(args, name) is None + ] + if asr_missing: + parser.error("для --prepare-asr нужны: " + ", ".join(asr_missing)) + return args + + +def resolve_manifest_paths( + manifest: dict[str, Any], manifest_dir: Path +) -> dict[str, Any]: + """Разрешает приватные относительные пути от каталога manifest.""" + result = copy.deepcopy(manifest) + base = manifest_dir.resolve() + + def resolve(value: str) -> str: + path = Path(value) + return str(path if path.is_absolute() else (base / path).resolve()) + + for model in result["segmentation_models"]: + model["path"] = resolve(model["path"]) + for model in result["embedding_models"]: + model["path"] = resolve(model["path"]) + for recording in result["recordings"]: + recording["path"] = resolve(recording["path"]) + if recording["reference"] is not None: + recording["reference"]["path"] = resolve(recording["reference"]["path"]) + recording["asr_words"]["path"] = resolve(recording["asr_words"]["path"]) + return result def file_sha256(path: Path) -> str: @@ -45,14 +245,584 @@ def file_sha256(path: Path) -> str: with path.open("rb") as source: for chunk in iter(lambda: source.read(1024 * 1024), b""): digest.update(chunk) - return digest.hexdigest().upper() + return digest.hexdigest() -def decode_clip(recording: Recording, work_dir: Path) -> Path: - output = work_dir / f"{recording.name}.wav" +def _canonical_sha256(value: Any) -> str: + encoded = json.dumps( + value, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _require_exact_keys(item: dict[str, Any], expected: set[str], label: str) -> None: + if set(item) != expected: + missing = sorted(expected - set(item)) + extra = sorted(set(item) - expected) + raise ValueError( + f"{label}: неверные поля; отсутствуют={missing}, лишние={extra}" + ) + + +def _require_id(value: Any, label: str) -> str: + if not isinstance(value, str) or not value or value.strip() != value: + raise ValueError(f"{label}: id должен быть непустой стабильной строкой") + if "/" in value or "\\" in value: + raise ValueError(f"{label}: id не должен зависеть от пути") + return value + + +def _require_hash(value: Any, label: str) -> str: + if not isinstance(value, str) or SHA256_RE.fullmatch(value) is None: + raise ValueError(f"{label}: ожидается SHA-256 в нижнем регистре") + return value + + +def _require_finite(value: Any, label: str, *, positive: bool = False) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise TypeError(f"{label}: ожидается число") + result = float(value) + if not math.isfinite(result) or (positive and result <= 0): + raise ValueError(f"{label}: недопустимое число") + return result + + +def _unique_ids(items: list[dict[str, Any]], label: str) -> None: + ids = [ + _require_id(item.get("id"), f"{label}[{index}]") + for index, item in enumerate(items) + ] + if len(ids) != len(set(ids)): + raise ValueError(f"{label}: id должны быть уникальными") + + +def _embedding_family(model_id: str) -> str | None: + normalized = re.sub(r"[^a-z0-9]", "", model_id.lower()) + if "wespeaker" in normalized: + return "wespeaker" + if "titanet" in normalized: + return "titanet" + return None + + +def _recording_family(recording_id: str) -> str | None: + normalized = re.sub(r"[^a-z0-9]", "", recording_id.lower()) + if "datatest" in normalized: + return "data-test" + if "t2bdma" in normalized: + return "t2-bdma" + if "yantar" in normalized: + return "yantar" + return None + + +def validate_manifest( + manifest: dict[str, Any], *, verify_files: bool = True +) -> dict[str, Any]: + """Строго проверяет manifest v2 до запуска дорогих операций.""" + if not isinstance(manifest, dict): + raise TypeError("Manifest должен быть JSON-объектом") + _require_exact_keys(manifest, TOP_LEVEL_KEYS, "manifest") + if manifest["schema_version"] != SCHEMA_VERSION: + raise ValueError("Поддерживается только schema_version=2") + for key in ( + "segmentation_models", + "embedding_models", + "combinations", + "recordings", + ): + if ( + not isinstance(manifest[key], list) + or not manifest[key] + or not all(isinstance(item, dict) for item in manifest[key]) + ): + raise ValueError(f"manifest.{key}: ожидается непустой массив объектов") + _unique_ids(manifest[key], f"manifest.{key}") + if ( + isinstance(manifest["threads"], bool) + or not isinstance(manifest["threads"], int) + or manifest["threads"] <= 0 + ): + raise ValueError("manifest.threads должен быть положительным целым") + interval = manifest["rss_sample_interval_ms"] + if isinstance(interval, bool) or not isinstance(interval, int) or interval <= 0: + raise ValueError( + "manifest.rss_sample_interval_ms должен быть положительным целым" + ) + + for index, model in enumerate(manifest["segmentation_models"]): + label = f"segmentation_models[{index}]" + _require_exact_keys(model, SEGMENTATION_KEYS, label) + if model["precision"] not in {"fp32", "int8"}: + raise ValueError(f"{label}.precision: ожидается fp32 или int8") + _validate_artifact(model, label, verify_files) + for index, model in enumerate(manifest["embedding_models"]): + label = f"embedding_models[{index}]" + _require_exact_keys(model, EMBEDDING_KEYS, label) + if _embedding_family(model["id"]) is None: + raise ValueError(f"{label}.id: ожидается WeSpeaker или TitaNet") + _validate_artifact(model, label, verify_files) + + segmentation = {item["id"]: item for item in manifest["segmentation_models"]} + embedding = {item["id"]: item for item in manifest["embedding_models"]} + segmentation_ids_by_precision: defaultdict[str, set[str]] = defaultdict(set) + for model in manifest["segmentation_models"]: + segmentation_ids_by_precision[model["precision"]].add(model["id"]) + embedding_ids_by_family: defaultdict[str | None, set[str]] = defaultdict(set) + for model in manifest["embedding_models"]: + embedding_ids_by_family[_embedding_family(model["id"])].add(model["id"]) + if set(segmentation_ids_by_precision) != {"fp32", "int8"} or any( + len(model_ids) != 1 for model_ids in segmentation_ids_by_precision.values() + ): + raise ValueError("Нужен ровно один segmentation model id для FP32 и INT8") + if set(embedding_ids_by_family) != {"wespeaker", "titanet"} or any( + len(model_ids) != 1 for model_ids in embedding_ids_by_family.values() + ): + raise ValueError("Нужен ровно один embedding model id для WeSpeaker и TitaNet") + matrix: set[tuple[str, str | None]] = set() + baselines: list[tuple[str, str | None]] = [] + for index, combination in enumerate(manifest["combinations"]): + label = f"combinations[{index}]" + _require_exact_keys(combination, COMBINATION_KEYS, label) + if ( + combination["segmentation_id"] not in segmentation + or combination["embedding_id"] not in embedding + ): + raise ValueError(f"{label}: неизвестная модель") + if combination["role"] not in {"baseline", "candidate"}: + raise ValueError(f"{label}.role: ожидается baseline или candidate") + thresholds = combination["thresholds"] + if not isinstance(thresholds, list) or not thresholds: + raise ValueError(f"{label}.thresholds: ожидается непустой массив") + values = [_require_finite(value, f"{label}.thresholds") for value in thresholds] + if values != sorted(values) or len(values) != len(set(values)): + raise ValueError( + f"{label}.thresholds должны быть отсортированы без повторов" + ) + pair = ( + segmentation[combination["segmentation_id"]]["precision"], + _embedding_family(combination["embedding_id"]), + ) + if pair in matrix: + raise ValueError(f"{label}: сочетание моделей объявлено повторно") + matrix.add(pair) + if combination["role"] == "baseline": + baselines.append(pair) + required_matrix = { + (precision, family) + for precision in ("fp32", "int8") + for family in ("wespeaker", "titanet") + } + if matrix != required_matrix or len(manifest["combinations"]) != 4: + raise ValueError( + "Manifest должен содержать ровно четыре сочетания FP32/INT8 × WeSpeaker/TitaNet" + ) + if baselines != [("fp32", "wespeaker")]: + raise ValueError("Должен быть ровно один baseline FP32+WeSpeaker") + + recording_families: set[str | None] = set() + for index, recording in enumerate(manifest["recordings"]): + label = f"recordings[{index}]" + _require_exact_keys(recording, RECORDING_KEYS, label) + if not isinstance(recording["path"], str) or not recording["path"]: + raise ValueError(f"{label}.path: ожидается непустая строка") + _require_hash(recording["sha256"], f"{label}.sha256") + start = _require_finite(recording["start"], f"{label}.start") + if start < 0: + raise ValueError(f"{label}.start не может быть отрицательным") + _require_finite(recording["duration"], f"{label}.duration", positive=True) + expected = recording["expected_speakers"] + if isinstance(expected, bool) or not isinstance(expected, int) or expected <= 0: + raise ValueError( + f"{label}.expected_speakers должен быть положительным целым" + ) + reference = recording["reference"] + if reference is not None: + if not isinstance(reference, dict): + raise ValueError(f"{label}.reference: ожидается объект или null") + _require_exact_keys(reference, REFERENCE_KEYS, f"{label}.reference") + if not isinstance(reference["path"], str) or not reference["path"]: + raise ValueError(f"{label}.reference.path: ожидается непустая строка") + _require_hash(reference["sha256"], f"{label}.reference.sha256") + if reference["format"] != "hypescribe_markdown": + raise ValueError(f"{label}.reference.format не поддерживается") + words = recording["asr_words"] + if not isinstance(words, dict): + raise TypeError(f"{label}.asr_words: ожидается объект") + _require_exact_keys(words, ASR_WORDS_KEYS, f"{label}.asr_words") + if not isinstance(words["path"], str) or not words["path"]: + raise ValueError(f"{label}.asr_words.path: ожидается непустая строка") + _require_hash(words["sha256"], f"{label}.asr_words.sha256") + if words["format"] != "local_transcriber_words_v1": + raise ValueError(f"{label}.asr_words.format не поддерживается") + recording_families.add(_recording_family(recording["id"])) + if verify_files: + _verify_file(Path(recording["path"]), recording["sha256"], None, label) + if reference is not None: + _verify_file( + Path(reference["path"]), + reference["sha256"], + None, + f"{label}.reference", + ) + _verify_file( + Path(words["path"]), words["sha256"], None, f"{label}.asr_words" + ) + load_asr_words(Path(words["path"]), float(recording["duration"])) + if len(manifest["recordings"]) != 3 or recording_families != { + "data-test", + "t2-bdma", + "yantar", + }: + raise ValueError("Нужны ровно три исходные записи: Data Test, T2 BDMA и Yantar") + return manifest + + +def validate_original_corpus(manifest: dict[str, Any]) -> None: + """Не допускает подмену корпуса issue #25 другой локальной выборкой.""" + for recording in manifest["recordings"]: + family = _recording_family(recording["id"]) + expected = ORIGINAL_CORPUS.get(family) + if expected is None or any( + recording[key] != value for key, value in expected.items() + ): + raise ValueError( + "Корпус не совпадает с исходными Data Test, T2 BDMA и Yantar" + ) + reference_expected = family in {"t2-bdma", "yantar"} + if (recording["reference"] is not None) != reference_expected: + raise ValueError("Опорная разметка корпуса не совпадает с исходной") + + +def _validate_artifact(item: dict[str, Any], label: str, verify_files: bool) -> None: + _require_hash(item["sha256"], f"{label}.sha256") + size = item["size_bytes"] + if isinstance(size, bool) or not isinstance(size, int) or size <= 0: + raise ValueError(f"{label}.size_bytes должен быть положительным целым") + for key in ("path", "source_url", "license"): + if not isinstance(item[key], str) or not item[key]: + raise ValueError(f"{label}.{key}: ожидается непустая строка") + if verify_files: + _verify_file(Path(item["path"]), item["sha256"], size, label) + + +def _verify_file( + path: Path, expected_hash: str, expected_size: int | None, label: str +) -> None: + if not path.is_file(): + raise ValueError(f"{label}: файл не найден: {path}") + if expected_size is not None and path.stat().st_size != expected_size: + raise ValueError(f"{label}: размер файла не совпал") + if file_sha256(path) != expected_hash: + raise ValueError(f"{label}: SHA-256 не совпал") + + +def environment_snapshot() -> dict[str, Any]: + """Возвращает версии, которые влияют на семантику эксперимента.""" + + def version(distribution: str) -> str: + try: + return importlib.metadata.version(distribution) + except importlib.metadata.PackageNotFoundError: + return "unavailable" + + return { + "started_at_utc": datetime.now(UTC).isoformat(), + "platform": platform.platform(), + "cpu": platform.processor() or platform.machine(), + "logical_cpus": os.cpu_count() or 1, + "python_version": platform.python_version(), + "sherpa_onnx_version": version("sherpa-onnx"), + "onnxruntime_version": version("onnxruntime"), + "numpy_version": version("numpy"), + "psutil_version": version("psutil"), + } + + +def validate_environment(environment: dict[str, Any]) -> None: + """Отсекает окружение, которое заведомо испортит дорогой прогон.""" + missing = [ + key + for key in ( + "sherpa_onnx_version", + "onnxruntime_version", + "numpy_version", + "psutil_version", + ) + if environment.get(key) == "unavailable" + ] + if missing: + raise RuntimeError(f"Не установлены зависимости benchmark: {missing}") + if environment.get("sherpa_onnx_version") != EXPECTED_SHERPA_ONNX_VERSION: + raise RuntimeError( + "Stage timing поддержан только для sherpa-onnx " + f"{EXPECTED_SHERPA_ONNX_VERSION}" + ) + + +def semantic_manifest( + manifest: dict[str, Any], environment: dict[str, Any] +) -> dict[str, Any]: + """Исключает пути и описательные метаданные из идентичности опыта.""" + return { + "schema_version": manifest["schema_version"], + "segmentation_models": [ + {key: item[key] for key in ("id", "precision", "sha256", "size_bytes")} + for item in manifest["segmentation_models"] + ], + "embedding_models": [ + {key: item[key] for key in ("id", "sha256", "size_bytes")} + for item in manifest["embedding_models"] + ], + "combinations": manifest["combinations"], + "recordings": [ + { + "id": item["id"], + "sha256": item["sha256"], + "start": item["start"], + "duration": item["duration"], + "expected_speakers": item["expected_speakers"], + "reference_sha256": item["reference"]["sha256"] + if item["reference"] + else None, + "asr_words_sha256": item["asr_words"]["sha256"], + } + for item in manifest["recordings"] + ], + "threads": manifest["threads"], + "rss_sample_interval_ms": manifest["rss_sample_interval_ms"], + "decode": { + "decoder_id": DECODER_ID, + "channels": CHANNELS, + "sample_rate": SAMPLE_RATE, + "sample_width_bytes": SAMPLE_WIDTH_BYTES, + "codec": "pcm_s16le", + }, + "dependencies": { + key: environment[key] + for key in ("sherpa_onnx_version", "onnxruntime_version", "numpy_version") + }, + } + + +def make_experiment_id(manifest: dict[str, Any], environment: dict[str, Any]) -> str: + return _canonical_sha256(semantic_manifest(manifest, environment)) + + +def make_cell_id( + experiment_id: str, + combination_id: str, + recording_id: str, + mode: str, + threshold: float, + num_clusters: int, +) -> str: + return _canonical_sha256( + { + "experiment_id": experiment_id, + "combination_id": combination_id, + "recording_id": recording_id, + "mode": mode, + "threshold": threshold, + "num_clusters": num_clusters, + } + ) + + +def _cell_spec( + experiment_id: str, + combination_id: str, + recording_id: str, + mode: str, + threshold: float, + num_clusters: int, +) -> dict[str, Any]: + return { + "cell_id": make_cell_id( + experiment_id, combination_id, recording_id, mode, threshold, num_clusters + ), + "combination_id": combination_id, + "recording_id": recording_id, + "mode": mode, + "threshold": threshold, + "num_clusters": num_clusters, + } + + +def expand_automatic_cells( + manifest: dict[str, Any], experiment_id: str +) -> list[dict[str, Any]]: + """Разворачивает прямоугольный свип в стабильном порядке manifest.""" + return [ + _cell_spec( + experiment_id, + combination["id"], + recording["id"], + "automatic", + float(threshold), + -1, + ) + for combination in manifest["combinations"] + for threshold in combination["thresholds"] + for recording in manifest["recordings"] + ] + + +def derive_known_cells( + manifest: dict[str, Any], experiment_id: str, selection: dict[str, Any] +) -> list[dict[str, Any]]: + """Создаёт known-count ячейки с единым выбранным порогом.""" + return [ + _cell_spec( + experiment_id, + selection["combination_id"], + recording["id"], + "known", + selection["selected_threshold"], + recording["expected_speakers"], + ) + for recording in manifest["recordings"] + ] + + +def select_threshold( + combination: dict[str, Any], + recordings: list[dict[str, Any]], + cells: list[dict[str, Any]], +) -> dict[str, Any] | None: + """Выбирает один порог только после полного прямоугольного свипа.""" + expected = {item["id"]: item["expected_speakers"] for item in recordings} + cell_by_key = { + (cell["recording_id"], cell["threshold"]): cell + for cell in cells + if cell["combination_id"] == combination["id"] + and cell["mode"] == "automatic" + and cell["status"] == "complete" + } + reference_ids = {item["id"] for item in recordings if item["reference"] is not None} + rectangles = [] + for threshold in combination["thresholds"]: + rectangle = [ + cell_by_key.get((item["id"], float(threshold))) for item in recordings + ] + if any(cell is None for cell in rectangle): + return None + rectangles.append((float(threshold), rectangle)) + + candidates = [] + for threshold, rectangle in rectangles: + purities = [] + differences = [] + residuals = [] + for cell in rectangle: + assert cell is not None + differences.append( + abs(cell["diagnostics"]["clusters"] - expected[cell["recording_id"]]) + ) + residuals.append( + cell["diagnostics"]["residual_share_after_expected"] or 0.0 + ) + if cell["recording_id"] in reference_ids: + purity = cell["diagnostics"]["mapped_speaker_purity"] + if not isinstance(purity, (int, float)) or not math.isfinite(purity): + raise RuntimeError( + "Полный автоматический свип непригоден для выбора порога: " + f"нет mapped speaker purity для {cell['recording_id']} " + f"при threshold={threshold}" + ) + purities.append(float(purity)) + score: list[float | int] = [max(differences), sum(differences)] + if purities: + score.extend([-min(purities), -(sum(purities) / len(purities))]) + score.extend([sum(residuals), threshold]) + candidates.append((tuple(score), threshold, rectangle, differences)) + score, threshold, rectangle, differences = min(candidates, key=lambda item: item[0]) + return { + "combination_id": combination["id"], + "selected_threshold": threshold, + "meets_count_gate": all(value == 0 for value in differences), + "score": list(score), + "recording_cells": [cell["cell_id"] for cell in rectangle if cell is not None], + } + + +def parse_stage_timings(stderr: str, sherpa_version: str) -> dict[str, float]: + """Строго разбирает недокументированный debug-контракт sherpa 1.13.5.""" + if sherpa_version != EXPECTED_SHERPA_ONNX_VERSION: + raise ValueError( + f"Неизвестный формат stage timing sherpa-onnx {sherpa_version}" + ) + found: defaultdict[str, list[float]] = defaultdict(list) + totals: list[tuple[float, float, float]] = [] + marker = "OfflineSpeakerDiarization:" + for line in stderr.splitlines(): + if marker not in line: + continue + message = line[line.index(marker) :].strip() + stage_match = STAGE_RE.fullmatch(message) + if stage_match is not None: + found[stage_match.group(1)].append(float(stage_match.group(2))) + continue + total_match = TOTAL_RE.fullmatch(message) + if total_match is not None: + totals.append(tuple(float(value) for value in total_match.groups())) + continue + raise ValueError("Stage timing содержит повреждённую debug-строку") + required = ("segmentation", "embedding", "clustering") + if any(len(found[name]) != 1 for name in required) or len(totals) != 1: + raise ValueError( + "Debug-лог должен содержать по одной записи каждого этапа и total" + ) + total, audio_duration, logged_rtf = totals[0] + values = [found[name][0] for name in required] + [total] + if any(not math.isfinite(value) or value < 0 for value in values): + raise ValueError("Stage timing содержит отрицательное или нечисловое значение") + if ( + not math.isfinite(audio_duration) + or audio_duration <= 0 + or not math.isfinite(logged_rtf) + or logged_rtf < 0 + ): + raise ValueError("Stage timing содержит недопустимые audio/RTF значения") + if abs(sum(values[:3]) - values[3]) > max(0.05, values[3] * 0.05): + raise ValueError("Сумма этапов существенно расходится с engine total") + calculated_rtf = total / audio_duration + if abs(calculated_rtf - logged_rtf) > max(0.005, logged_rtf * 0.05): + raise ValueError("Stage timing содержит несогласованный RTF") + return { + "segmentation_seconds": values[0], + "embedding_seconds": values[1], + "clustering_seconds": values[2], + "engine_total_seconds": values[3], + "engine_audio_seconds": audio_duration, + "engine_rtf": logged_rtf, + } + + +def decode_cache_key(recording: dict[str, Any], decoder_id: str = DECODER_ID) -> str: + return _canonical_sha256( + { + "source_sha256": recording["sha256"], + "start": recording["start"], + "duration": recording["duration"], + "channels": CHANNELS, + "sample_rate": SAMPLE_RATE, + "sample_width_bytes": SAMPLE_WIDTH_BYTES, + "codec": "pcm_s16le", + "decoder_id": decoder_id, + } + ) + + +def decode_clip(recording: dict[str, Any], work_dir: Path) -> Path: + """Декодирует фрагмент один раз в контентно-адресуемый WAV-кэш.""" + cache_dir = work_dir / "decoded" + cache_dir.mkdir(parents=True, exist_ok=True) + output = cache_dir / f"{decode_cache_key(recording)}.wav" if output.exists(): + read_wav(output) return output - + temporary = output.with_suffix(".wav.tmp") command = [ "ffmpeg", "-hide_banner", @@ -60,29 +830,41 @@ def decode_clip(recording: Recording, work_dir: Path) -> Path: "error", "-y", "-ss", - str(recording.start), + str(recording["start"]), "-t", - str(recording.duration), + str(recording["duration"]), "-i", - str(recording.path), + str(recording["path"]), "-vn", "-ac", - "1", + str(CHANNELS), "-ar", - "16000", + str(SAMPLE_RATE), "-c:a", "pcm_s16le", - str(output), + "-f", + "wav", + str(temporary), ] - subprocess.run(command, check=True) + try: + subprocess.run(command, check=True) + read_wav(temporary) + temporary.replace(output) + finally: + temporary.unlink(missing_ok=True) return output -def read_wav(path: Path) -> np.ndarray: +def read_wav(path: Path) -> Any: + import numpy as np + with wave.open(str(path), "rb") as source: - if source.getnchannels() != 1 or source.getsampwidth() != 2: + if ( + source.getnchannels() != CHANNELS + or source.getsampwidth() != SAMPLE_WIDTH_BYTES + ): raise ValueError(f"Ожидался mono PCM16 WAV: {path}") - if source.getframerate() != 16000: + if source.getframerate() != SAMPLE_RATE: raise ValueError(f"Ожидалась частота 16 кГц: {path}") samples = np.frombuffer(source.readframes(source.getnframes()), np.int16) return samples.astype(np.float32) / 32768.0 @@ -90,33 +872,34 @@ def read_wav(path: Path) -> np.ndarray: def timestamp_seconds(match: re.Match[str]) -> float: first, second, third = match.group(1), match.group(2), match.group(3) - if third is None: - return int(first) * 60 + int(second) - return int(first) * 3600 + int(second) * 60 + int(third) + return ( + int(first) * 60 + int(second) + if third is None + else int(first) * 3600 + int(second) * 60 + int(third) + ) -def read_reference_turns(recording: Recording) -> list[dict[str, Any]]: - if recording.reference is None: +def read_reference_turns(recording: dict[str, Any]) -> list[dict[str, Any]]: + reference = Path(recording["reference"]["path"]) if recording["reference"] else None + start, duration = float(recording["start"]), float(recording["duration"]) + if reference is None: return [] - - starts: list[tuple[float, str]] = [] - for line in recording.reference.read_text(encoding="utf-8").splitlines(): + starts = [] + for line in reference.read_text(encoding="utf-8").splitlines(): match = TURN_RE.match(line) if match: starts.append((timestamp_seconds(match), match.group(4))) - - clip_end = recording.start + recording.duration - turns: list[dict[str, Any]] = [] - for index, (start, speaker) in enumerate(starts): - end = starts[index + 1][0] if index + 1 < len(starts) else clip_end - overlap_start = max(start, recording.start) - overlap_end = min(end, clip_end) + clip_end = start + duration + turns = [] + for index, (turn_start, speaker) in enumerate(starts): + turn_end = starts[index + 1][0] if index + 1 < len(starts) else clip_end + overlap_start, overlap_end = max(turn_start, start), min(turn_end, clip_end) if overlap_end > overlap_start: turns.append( { "speaker": speaker, - "start": overlap_start - recording.start, - "end": overlap_end - recording.start, + "start": overlap_start - start, + "end": overlap_end - start, } ) return turns @@ -127,93 +910,63 @@ def interval_overlap(left: dict[str, Any], right: dict[str, Any]) -> float: def best_mapping( - segments: list[dict[str, Any]], - reference_turns: list[dict[str, Any]], + segments: list[dict[str, Any]], reference_turns: list[dict[str, Any]] ) -> dict[str, Any] | None: if not reference_turns or not segments: return None - predicted = sorted({str(segment["speaker"]) for segment in segments}) reference = sorted({str(turn["speaker"]) for turn in reference_turns}) overlap: defaultdict[tuple[str, str], float] = defaultdict(float) total = 0.0 for segment in segments: - predicted_speaker = str(segment["speaker"]) for turn in reference_turns: value = interval_overlap(segment, turn) if value: - reference_speaker = str(turn["speaker"]) - overlap[(predicted_speaker, reference_speaker)] += value + overlap[(str(segment["speaker"]), str(turn["speaker"]))] += value total += value - - best_score = -1.0 - best_pairs: list[tuple[str, str]] = [] + best_score, best_pairs = -1.0, [] if len(predicted) >= len(reference): - for candidate in itertools.permutations(predicted, len(reference)): - pairs = list(zip(candidate, reference, strict=True)) - score = sum(overlap[pair] for pair in pairs) - if score > best_score: - best_score, best_pairs = score, pairs + candidates = ( + list(zip(candidate, reference, strict=True)) + for candidate in itertools.permutations(predicted, len(reference)) + ) else: - for candidate in itertools.permutations(reference, len(predicted)): - pairs = list(zip(predicted, candidate, strict=True)) - score = sum(overlap[pair] for pair in pairs) - if score > best_score: - best_score, best_pairs = score, pairs - + candidates = ( + list(zip(predicted, candidate, strict=True)) + for candidate in itertools.permutations(reference, len(predicted)) + ) + for pairs in candidates: + score = sum(overlap[pair] for pair in pairs) + if score > best_score: + best_score, best_pairs = score, pairs return { "mapped_speaker_purity": best_score / total if total else None, "mapped_overlap_seconds": best_score, "total_overlap_seconds": total, - "mapping": {predicted: reference for predicted, reference in best_pairs}, + "mapping": { + predicted_id: reference_id for predicted_id, reference_id in best_pairs + }, } -def make_config( - segmentation_model: Path, - embedding_model: Path, - threshold: float, - num_clusters: int, - threads: int, -) -> sherpa_onnx.OfflineSpeakerDiarizationConfig: - pyannote = sherpa_onnx.OfflineSpeakerSegmentationPyannoteModelConfig( - model=str(segmentation_model) - ) - segmentation = sherpa_onnx.OfflineSpeakerSegmentationModelConfig( - pyannote=pyannote, - num_threads=threads, - ) - embedding = sherpa_onnx.SpeakerEmbeddingExtractorConfig( - model=str(embedding_model), - num_threads=threads, - ) - clustering = sherpa_onnx.FastClusteringConfig( - num_clusters=num_clusters, - threshold=threshold, - ) - return sherpa_onnx.OfflineSpeakerDiarizationConfig( - segmentation=segmentation, - embedding=embedding, - clustering=clustering, - ) - - def summarize_segments( - segments: list[dict[str, Any]], - recording: Recording, + segments: list[dict[str, Any]], recording: dict[str, Any] ) -> dict[str, Any]: + duration = float(recording["duration"]) + expected = int(recording["expected_speakers"]) durations: defaultdict[str, float] = defaultdict(float) for segment in segments: - durations[str(segment["speaker"])] += segment["end"] - segment["start"] - - ordered = sorted(durations.items(), key=lambda item: item[1], reverse=True) + durations[str(segment["speaker"])] += max( + 0.0, float(segment["end"]) - float(segment["start"]) + ) + ordered = sorted(durations.items(), key=lambda item: (-item[1], item[0])) total = sum(durations.values()) - residual = sum(duration for _, duration in ordered[recording.expected_speakers :]) - substantial_threshold = max(5.0, recording.duration * 0.02) + residual = sum(value for _, value in ordered[expected:]) + substantial_threshold = max(5.0, duration * 0.02) return { "clusters": len(ordered), "substantial_clusters": sum( - duration >= substantial_threshold for _, duration in ordered + value >= substantial_threshold for _, value in ordered ), "substantial_threshold_seconds": substantial_threshold, "cluster_durations_seconds": dict(ordered), @@ -223,132 +976,642 @@ def summarize_segments( } -def save_output(path: Path, output: dict[str, Any]) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - temporary = path.with_suffix(path.suffix + ".tmp") - temporary.write_text( - json.dumps(output, ensure_ascii=False, indent=2), - encoding="utf-8", +def load_asr_words(path: Path, duration: float) -> list[Any]: + """Проверяет приватный sidecar и возвращает доменные слова.""" + from local_transcriber.types import Word + + raw = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(raw, list): + raise TypeError("ASR sidecar должен быть JSON-массивом") + if not raw: + raise ValueError("ASR sidecar должен содержать хотя бы одно слово") + words, previous_start = [], -math.inf + for index, item in enumerate(raw): + if not isinstance(item, dict) or set(item) != {"start", "end", "text"}: + raise ValueError(f"ASR sidecar[{index}]: неверная структура") + start = _require_finite(item["start"], f"ASR sidecar[{index}].start") + end = _require_finite(item["end"], f"ASR sidecar[{index}].end") + if start < 0 or end < start or end > duration or start < previous_start: + raise ValueError(f"ASR sidecar[{index}]: нарушены временные границы") + if not isinstance(item["text"], str): + raise TypeError(f"ASR sidecar[{index}].text: ожидается строка") + words.append(Word(start=start, end=end, text=item["text"])) + previous_start = start + return words + + +def canonical_non_whitespace_sequence(words: list[Any]) -> str: + """Сохраняет все Unicode-символы текста, кроме любых пробельных.""" + from local_transcriber.diarization import _append_word_text, _normalize_turn_text + + text = "" + for word in words: + text = _append_word_text(text, word.text) + normalized = _normalize_turn_text(text) + return "".join(character for character in normalized if not character.isspace()) + + +def text_invariance(words: list[Any], transcript: Any) -> dict[str, Any]: + from local_transcriber.diarization import _append_word_text, _normalize_turn_text + + source = canonical_non_whitespace_sequence(words) + output = "" + for turn in transcript.turns: + output = _append_word_text(output, turn.text) + output = _normalize_turn_text(output) + output = "".join(character for character in output if not character.isspace()) + mismatch = next( + ( + index + for index, pair in enumerate( + itertools.zip_longest(source, output, fillvalue=None) + ) + if pair[0] != pair[1] + ), + None, ) - temporary.replace(path) - - -def manifest_shape(manifest: dict[str, Any]) -> dict[str, Any]: - """Отделить параметры эксперимента от машинно-зависимых путей.""" return { - "models": [item["name"] for item in manifest["models"]], - "recordings": [ - { - key: item[key] - for key in ("name", "start", "duration", "expected_speakers") - } - for item in manifest["recordings"] - ], - "runs": manifest["runs"], + "input_sha256": hashlib.sha256(source.encode()).hexdigest(), + "output_sha256": hashlib.sha256(output.encode()).hexdigest(), + "equal": source == output, + "first_mismatch_index": mismatch, } -def main() -> None: - args = parse_args() - manifest = json.loads(args.manifest.read_text(encoding="utf-8")) - args.work_dir.mkdir(parents=True, exist_ok=True) +def evaluate_segments( + segments: list[dict[str, Any]], + recording: dict[str, Any], + words: list[Any], + reference_turns: list[dict[str, Any]], +) -> tuple[dict[str, Any], dict[str, Any]]: + from local_transcriber.diarization import build_speaker_transcript + from local_transcriber.types import SpeakerInterval - recordings = [ - Recording( - name=item["name"], - path=Path(item["path"]), + intervals = [ + SpeakerInterval( start=float(item["start"]), - duration=float(item["duration"]), - expected_speakers=int(item["expected_speakers"]), - reference=Path(item["reference"]) if item.get("reference") else None, + end=float(item["end"]), + cluster=int(item["speaker"]), ) - for item in manifest["recordings"] + for item in segments ] - if args.output.exists(): - output = json.loads(args.output.read_text(encoding="utf-8")) - if ( - manifest_shape(output["manifest"]) != manifest_shape(manifest) - or output["threads"] != args.threads - ): - raise ValueError("Существующий output создан с другим manifest/threads") - output["manifest"] = manifest - else: - output = { - "manifest": manifest, - "sherpa_onnx_version": sherpa_onnx.__version__, - "threads": args.threads, - "results": [], + transcript = build_speaker_transcript( + words, intervals, float(recording["duration"]) + ) + diagnostics = summarize_segments(segments, recording) + mapping = best_mapping(segments, reference_turns) + diagnostics.update( + { + "unassigned_word_count": transcript.unassigned_word_count, + "mapped_speaker_purity": mapping["mapped_speaker_purity"] + if mapping + else None, + "mapped_overlap_seconds": mapping["mapped_overlap_seconds"] + if mapping + else None, + "total_overlap_seconds": mapping["total_overlap_seconds"] + if mapping + else None, + "mapping": mapping["mapping"] if mapping else None, } - completed = { - (item["recording"], item["model"], item["run"]) for item in output["results"] + ) + return diagnostics, text_invariance(words, transcript) + + +def measure_call( + operation: Callable[[], Any], + process: Any, + *, + clock: Callable[[], float] = time.perf_counter, + sampler_factory: Callable[[Any, float], Any] = PeakRssSampler, + interval_seconds: float, + logical_cpus: int, +) -> tuple[Any, dict[str, Any]]: + """Измеряет ровно один вызов через внедряемые CPU/RSS адаптеры.""" + sampler = sampler_factory(process, interval_seconds) + sampler.start() + before = process.cpu_times() + started = clock() + operation_error: BaseException | None = None + try: + result = operation() + except BaseException as exc: + operation_error = exc + raise + finally: + stopped = clock() + after = process.cpu_times() + try: + peak_rss = sampler.stop() + except BaseException as exc: + if operation_error is None: + raise + operation_error.add_note( + f"Дополнительная ошибка остановки RSS sampler: {type(exc).__name__}: {exc}" + ) + wall = stopped - started + if wall <= 0: + raise RuntimeError("Измеренный wall time должен быть положительным") + cpu_user, cpu_system = ( + float(after.user - before.user), + float(after.system - before.system), + ) + if ( + not math.isfinite(cpu_user) + or not math.isfinite(cpu_system) + or cpu_user < 0 + or cpu_system < 0 + or logical_cpus <= 0 + or isinstance(peak_rss, bool) + or not isinstance(peak_rss, int) + or peak_rss <= 0 + ): + raise RuntimeError("CPU/RSS-измерение вернуло недопустимое значение") + cpu_total = cpu_user + cpu_system + cores = cpu_total / wall + return result, { + "wall_seconds": wall, + "cpu_user_seconds": cpu_user, + "cpu_system_seconds": cpu_system, + "cpu_total_seconds": cpu_total, + "average_cpu_cores": cores, + "average_cpu_percent_machine": 100.0 * cores / logical_cpus, + "peak_rss_bytes": peak_rss, } - segmentation_model = Path(manifest["segmentation_model"]) - for recording in recordings: - print(f"Декодирование {recording.name}", flush=True) - wav_path = decode_clip(recording, args.work_dir) - samples = read_wav(wav_path) - reference_turns = read_reference_turns(recording) - source_hash = file_sha256(recording.path) - for model in manifest["models"]: - embedding_model = Path(model["path"]) - for run in manifest["runs"]: - run_key = (recording.name, model["name"], run["name"]) - if run_key in completed: - print(f"Пропуск готового прогона: {run_key}", flush=True) - continue - num_clusters = run["num_clusters"] - if num_clusters == "expected": - num_clusters = recording.expected_speakers - threshold = float(run["threshold"]) - print( - f"{recording.name}: {model['name']} / {run['name']}", - flush=True, - ) - config = make_config( - segmentation_model=segmentation_model, - embedding_model=embedding_model, - threshold=threshold, - num_clusters=int(num_clusters), - threads=args.threads, - ) - diarizer = sherpa_onnx.OfflineSpeakerDiarization(config) - started = time.perf_counter() - result = diarizer.process(samples) - elapsed = time.perf_counter() - started - segments = [ - { - "speaker": int(segment.speaker), - "start": float(segment.start), - "end": float(segment.end), - } - for segment in result.sort_by_start_time() - ] - item = { - "recording": recording.name, - "source": recording.path.name, - "source_sha256": source_hash, - "clip_start": recording.start, - "clip_duration": recording.duration, - "expected_speakers": recording.expected_speakers, - "reference": recording.reference.name - if recording.reference - else None, - "model": model["name"], - "model_file": embedding_model.name, - "run": run["name"], - "threshold": threshold, - "num_clusters": int(num_clusters), - "elapsed_seconds": elapsed, - "rtf": elapsed / recording.duration, - "summary": summarize_segments(segments, recording), - "reference_mapping": best_mapping(segments, reference_turns), - "segments": segments, +def make_config( + segmentation_model: Path, + embedding_model: Path, + threshold: float, + num_clusters: int, + threads: int, +) -> Any: + """Создаёт строго CPU-конфигурацию и включает native stage timing.""" + import sherpa_onnx + + segmentation = sherpa_onnx.OfflineSpeakerSegmentationModelConfig( + pyannote=sherpa_onnx.OfflineSpeakerSegmentationPyannoteModelConfig( + model=str(segmentation_model) + ), + num_threads=threads, + provider="cpu", + debug=True, + ) + embedding = sherpa_onnx.SpeakerEmbeddingExtractorConfig( + model=str(embedding_model), num_threads=threads, provider="cpu", debug=True + ) + return sherpa_onnx.OfflineSpeakerDiarizationConfig( + segmentation=segmentation, + embedding=embedding, + clustering=sherpa_onnx.FastClusteringConfig( + num_clusters=num_clusters, threshold=threshold + ), + min_duration_on=0.3, + min_duration_off=0.5, + ) + + +def _worker_main(request_path: Path) -> None: + """Исполняет ровно одну ячейку в свежем процессе.""" + import psutil + import sherpa_onnx + + request = json.loads(request_path.read_text(encoding="utf-8")) + samples = read_wav(Path(request["wav_path"])) + config = make_config( + Path(request["segmentation_path"]), + Path(request["embedding_path"]), + request["threshold"], + request["num_clusters"], + request["threads"], + ) + if not config.validate(): + raise RuntimeError("Конфигурация диаризатора недействительна") + diarizer = sherpa_onnx.OfflineSpeakerDiarization(config) + result, metrics = measure_call( + lambda: diarizer.process(samples), + psutil.Process(os.getpid()), + interval_seconds=request["rss_sample_interval_ms"] / 1000.0, + logical_cpus=request["logical_cpus"], + ) + segments = [ + { + "speaker": int(item.speaker), + "start": float(item.start), + "end": float(item.end), + } + for item in result.sort_by_start_time() + ] + print( + json.dumps({"metrics": metrics, "segments": segments}, separators=(",", ":")), + flush=True, + ) + + +def run_worker( + request: dict[str, Any], + work_dir: Path, + *, + runner: Callable[..., Any] = subprocess.run, +) -> tuple[dict[str, Any], str]: + """Запускает изолированного worker; seam допускает fake protocol в тестах.""" + request_dir = work_dir / "requests" + request_dir.mkdir(parents=True, exist_ok=True) + request_path = request_dir / f"{request['cell_id']}.json" + save_output(request_path, request) + completed = runner( + [sys.executable, str(Path(__file__).resolve()), "--worker", str(request_path)], + check=False, + capture_output=True, + text=True, + ) + if completed.returncode != 0: + message = ( + completed.stderr.strip().splitlines()[-1] + if completed.stderr.strip() + else "worker завершился с ошибкой" + ) + raise RuntimeError(message) + try: + payload = json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise RuntimeError("Worker вернул некорректный JSON") from exc + if set(payload) != {"metrics", "segments"}: + raise RuntimeError("Worker нарушил протокол результата") + return payload, completed.stderr + + +def save_output(path: Path, output: Any) -> None: + """Атомарно сохраняет JSON рядом с целевым файлом.""" + path.parent.mkdir(parents=True, exist_ok=True) + temporary_path: Path | None = None + try: + with NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + delete=False, + ) as temporary: + temporary_path = Path(temporary.name) + json.dump(output, temporary, ensure_ascii=False, indent=2) + temporary.write("\n") + temporary.flush() + os.fsync(temporary.fileno()) + os.replace(temporary_path, path) + finally: + if temporary_path is not None: + temporary_path.unlink(missing_ok=True) + + +def _artifacts(manifest: dict[str, Any]) -> list[dict[str, Any]]: + result = [] + for kind, models in ( + ("segmentation", manifest["segmentation_models"]), + ("embedding", manifest["embedding_models"]), + ): + for item in models: + result.append( + { + "kind": kind, + "id": item["id"], + "filename": Path(item["path"]).name, + "sha256": item["sha256"], + "size_bytes": item["size_bytes"], + "source_url": item["source_url"], + "license": item["license"], } - output["results"].append(item) - completed.add(run_key) - save_output(args.output, output) + ) + return result + + +def _asr_baselines(manifest: dict[str, Any]) -> list[dict[str, Any]]: + baselines = [] + for recording in manifest["recordings"]: + words = load_asr_words( + Path(recording["asr_words"]["path"]), recording["duration"] + ) + baselines.append( + { + "recording_id": recording["id"], + "words_sha256": recording["asr_words"]["sha256"], + "word_count": len(words), + "canonical_text_sha256": hashlib.sha256( + canonical_non_whitespace_sequence(words).encode() + ).hexdigest(), + } + ) + return baselines + + +def _new_output( + manifest: dict[str, Any], environment: dict[str, Any], experiment_id: str +) -> dict[str, Any]: + return { + "schema_version": SCHEMA_VERSION, + "experiment_id": experiment_id, + "manifest": manifest, + "environment": environment, + "artifacts": _artifacts(manifest), + "asr_baselines": _asr_baselines(manifest), + "cells": [], + "threshold_selection": [], + } + + +def _load_or_create_output( + path: Path, + manifest: dict[str, Any], + environment: dict[str, Any], + experiment_id: str, +) -> dict[str, Any]: + if not path.exists(): + return _new_output(manifest, environment, experiment_id) + output = json.loads(path.read_text(encoding="utf-8")) + if ( + output.get("schema_version") != SCHEMA_VERSION + or output.get("experiment_id") != experiment_id + ): + raise ValueError( + "Существующий output относится к другому семантическому эксперименту" + ) + host_keys = ("platform", "cpu", "logical_cpus") + previous_environment = output.get("environment") + if not isinstance(previous_environment, dict) or any( + previous_environment.get(key) != environment.get(key) for key in host_keys + ): + raise ValueError( + "Существующий output создан на другой platform/cpu/logical_cpus" + ) + output["manifest"] = manifest + output["artifacts"] = _artifacts(manifest) + output["asr_baselines"] = _asr_baselines(manifest) + return output + + +def _replace_cell(output: dict[str, Any], cell: dict[str, Any]) -> None: + output["cells"] = [ + item for item in output["cells"] if item.get("cell_id") != cell["cell_id"] + ] + output["cells"].append(cell) + + +def _failed_cell(spec: dict[str, Any], exc: BaseException) -> dict[str, Any]: + return { + **spec, + "status": "failed", + "metrics": None, + "diagnostics": None, + "text_invariance": None, + "segments": None, + "error": {"type": type(exc).__name__, "message": str(exc)}, + } + + +def _execute_cell( + spec: dict[str, Any], + manifest: dict[str, Any], + output: dict[str, Any], + decoded: dict[str, Path], + work_dir: Path, + runner: Callable[..., Any] = subprocess.run, +) -> dict[str, Any]: + combinations = {item["id"]: item for item in manifest["combinations"]} + segmentations = {item["id"]: item for item in manifest["segmentation_models"]} + embeddings = {item["id"]: item for item in manifest["embedding_models"]} + recordings = {item["id"]: item for item in manifest["recordings"]} + combination, recording = ( + combinations[spec["combination_id"]], + recordings[spec["recording_id"]], + ) + request = { + **spec, + "wav_path": str(decoded[recording["id"]]), + "segmentation_path": segmentations[combination["segmentation_id"]]["path"], + "embedding_path": embeddings[combination["embedding_id"]]["path"], + "threads": manifest["threads"], + "rss_sample_interval_ms": manifest["rss_sample_interval_ms"], + "logical_cpus": output["environment"]["logical_cpus"], + } + payload, stderr = run_worker(request, work_dir, runner=runner) + stages = parse_stage_timings(stderr, output["environment"]["sherpa_onnx_version"]) + if not math.isclose( + stages["engine_audio_seconds"], + float(recording["duration"]), + rel_tol=0.0, + abs_tol=AUDIO_DURATION_TOLERANCE_SECONDS, + ): + raise RuntimeError( + "Длительность audio из sherpa debug-лога не совпала с manifest clip: " + f"engine_audio_seconds={stages['engine_audio_seconds']}, " + f"manifest_duration={recording['duration']}" + ) + metrics = payload["metrics"] + metrics["rtf"] = metrics["wall_seconds"] / recording["duration"] + metrics["stages"] = stages + words = load_asr_words(Path(recording["asr_words"]["path"]), recording["duration"]) + diagnostics, invariance = evaluate_segments( + payload["segments"], recording, words, read_reference_turns(recording) + ) + if not invariance["equal"]: + raise RuntimeError( + "ASR-текст изменился после сведения: " + f"input_sha256={invariance['input_sha256']}, " + f"output_sha256={invariance['output_sha256']}, " + f"first_mismatch_index={invariance['first_mismatch_index']}" + ) + return { + **spec, + "status": "complete", + "metrics": metrics, + "diagnostics": diagnostics, + "text_invariance": invariance, + "segments": payload["segments"], + "error": None, + } + + +def _run_specs( + specs: list[dict[str, Any]], + manifest: dict[str, Any], + output: dict[str, Any], + decoded: dict[str, Path], + output_path: Path, + work_dir: Path, +) -> None: + complete = { + cell["cell_id"] for cell in output["cells"] if cell.get("status") == "complete" + } + for spec in specs: + if spec["cell_id"] in complete: + continue + try: + cell = _execute_cell(spec, manifest, output, decoded, work_dir) + except Exception as exc: # noqa: BLE001 - ошибка ячейки должна сохраниться для retry + cell = _failed_cell(spec, exc) + _replace_cell(output, cell) + save_output(output_path, output) + + +def generate_asr_sidecar( + wav_path: Path, + output_path: Path, + *, + model: str, + device: str, + compute_type: str, + language: str, + threads: int, +) -> str: + """Один раз запускает production ASR и атомарно сохраняет только слова.""" + from local_transcriber.transcriber import transcribe + + result = transcribe( + wav_path, + model_name=model, + device=device, + compute_type=compute_type, + language=language, + strict_device=True, + cpu_threads=threads, + ) + save_output( + output_path, + [ + {"start": word.start, "end": word.end, "text": word.text} + for word in result.words + ], + ) + return file_sha256(output_path) + + +def _prepare_asr(args: argparse.Namespace, manifest: dict[str, Any]) -> None: + validate_manifest(manifest, verify_files=False) + validate_original_corpus(manifest) + recordings = {item["id"]: item for item in manifest["recordings"]} + if args.prepare_asr not in recordings: + raise ValueError("Неизвестный recording id для --prepare-asr") + recording = recordings[args.prepare_asr] + _verify_file(Path(recording["path"]), recording["sha256"], None, "recording") + digest = generate_asr_sidecar( + decode_clip(recording, args.work_dir), + Path(recording["asr_words"]["path"]), + model=args.asr_model, + device=args.asr_device, + compute_type=args.asr_compute_type, + language=args.asr_language, + threads=args.threads or manifest["threads"], + ) + print(digest) + + +def validation_summary( + manifest: dict[str, Any], environment: dict[str, Any] +) -> dict[str, Any]: + """Возвращает безопасный план без приватных путей и текста.""" + experiment_id = make_experiment_id(manifest, environment) + return { + "status": "valid", + "experiment_id": experiment_id, + "recordings": [item["id"] for item in manifest["recordings"]], + "combinations": [item["id"] for item in manifest["combinations"]], + "automatic_cells": len(expand_automatic_cells(manifest, experiment_id)), + "known_cells_if_all_selected": len(manifest["recordings"]) + * len(manifest["combinations"]), + "threads": manifest["threads"], + "dependencies": { + key: environment[key] + for key in ( + "sherpa_onnx_version", + "onnxruntime_version", + "numpy_version", + "psutil_version", + ) + }, + } + + +def main(argv: list[str] | None = None) -> None: + args = parse_args(argv) + if args.worker is not None: + _worker_main(args.worker) + return + raw_manifest = json.loads(args.manifest.read_text(encoding="utf-8")) + validate_manifest(raw_manifest, verify_files=False) + manifest = resolve_manifest_paths(raw_manifest, args.manifest.parent) + if args.threads is not None: + if args.threads <= 0: + raise ValueError("--threads должен быть положительным") + if args.threads != manifest.get("threads"): + raise ValueError("--threads должен совпадать с manifest.threads") + if args.prepare_asr is not None: + _prepare_asr(args, manifest) + return + validate_manifest(manifest) + validate_original_corpus(manifest) + environment = environment_snapshot() + validate_environment(environment) + if args.validate_only: + print( + json.dumps( + validation_summary(manifest, environment), + ensure_ascii=False, + indent=2, + ) + ) + return + args.work_dir.mkdir(parents=True, exist_ok=True) + experiment_id = make_experiment_id(manifest, environment) + output = _load_or_create_output(args.output, manifest, environment, experiment_id) + decoded = { + recording["id"]: decode_clip(recording, args.work_dir) + for recording in manifest["recordings"] + } + save_output(args.output, output) + automatic_specs = expand_automatic_cells(manifest, experiment_id) + _run_specs( + automatic_specs, + manifest, + output, + decoded, + args.output, + args.work_dir, + ) + expected_specs = list(automatic_specs) + for combination in manifest["combinations"]: + selection = select_threshold( + combination, manifest["recordings"], output["cells"] + ) + if selection is None: + continue + output["threshold_selection"] = [ + item + for item in output["threshold_selection"] + if item["combination_id"] != combination["id"] + ] + [selection] + save_output(args.output, output) + known_specs = derive_known_cells(manifest, experiment_id, selection) + expected_specs.extend(known_specs) + _run_specs( + known_specs, + manifest, + output, + decoded, + args.output, + args.work_dir, + ) + status_by_id = {cell.get("cell_id"): cell.get("status") for cell in output["cells"]} + incomplete = [ + spec["cell_id"] + for spec in expected_specs + if status_by_id.get(spec["cell_id"]) != "complete" + ] + if incomplete: + raise RuntimeError( + f"Benchmark не завершён: незавершённых ячеек {len(incomplete)}" + ) if __name__ == "__main__": diff --git a/tests/test_diarization_calibration.py b/tests/test_diarization_calibration.py new file mode 100644 index 0000000..f5f621e --- /dev/null +++ b/tests/test_diarization_calibration.py @@ -0,0 +1,930 @@ +import hashlib +import importlib.util +import json +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +SCRIPT = ( + Path(__file__).parents[1] / "scripts" / "benchmarks" / "diarization_calibration.py" +) +SPEC = importlib.util.spec_from_file_location("diarization_calibration", SCRIPT) +assert SPEC is not None and SPEC.loader is not None +calibration = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = calibration +SPEC.loader.exec_module(calibration) + + +def _write(path, content): + path.write_bytes(content) + return { + "path": str(path), + "sha256": hashlib.sha256(content).hexdigest(), + "size_bytes": len(content), + } + + +def _manifest(tmp_path): + fp32 = _write(tmp_path / "fp32.onnx", b"fp32") + int8 = _write(tmp_path / "int8.onnx", b"int8") + wespeaker = _write(tmp_path / "wespeaker.onnx", b"wespeaker") + titanet = _write(tmp_path / "titanet.onnx", b"titanet") + segmentations = [ + { + "id": "pyannote-fp32", + "precision": "fp32", + "source_url": "https://example/fp32", + "license": "MIT", + **fp32, + }, + { + "id": "pyannote-int8", + "precision": "int8", + "source_url": "https://example/int8", + "license": "MIT", + **int8, + }, + ] + embeddings = [ + { + "id": "wespeaker", + "source_url": "https://example/wespeaker", + "license": "CC-BY-4.0", + **wespeaker, + }, + { + "id": "titanet-small", + "source_url": "https://example/titanet", + "license": "Apache-2.0", + **titanet, + }, + ] + recordings = [] + for index, (recording_id, speakers) in enumerate( + (("data-test", 3), ("t2-bdma", 2), ("yantar", 2)) + ): + media = _write(tmp_path / f"{recording_id}.mp4", recording_id.encode()) + words_content = json.dumps( + [{"start": 0.0, "end": 0.5, "text": " Слово"}] + ).encode() + words = _write(tmp_path / f"{recording_id}.words.json", words_content) + reference = None + if index: + reference_content = b"**[00:00] Speaker 1:**\n" + ref = _write(tmp_path / f"{recording_id}.md", reference_content) + reference = { + "path": ref["path"], + "sha256": ref["sha256"], + "format": "hypescribe_markdown", + } + recordings.append( + { + "id": recording_id, + "path": media["path"], + "sha256": media["sha256"], + "start": 0.0, + "duration": 10.0, + "expected_speakers": speakers, + "reference": reference, + "asr_words": { + "path": words["path"], + "sha256": words["sha256"], + "format": "local_transcriber_words_v1", + }, + } + ) + return { + "schema_version": 2, + "segmentation_models": segmentations, + "embedding_models": embeddings, + "combinations": [ + { + "id": "fp32-wespeaker", + "segmentation_id": "pyannote-fp32", + "embedding_id": "wespeaker", + "role": "baseline", + "thresholds": [0.8, 0.9], + }, + { + "id": "fp32-titanet", + "segmentation_id": "pyannote-fp32", + "embedding_id": "titanet-small", + "role": "candidate", + "thresholds": [0.8], + }, + { + "id": "int8-wespeaker", + "segmentation_id": "pyannote-int8", + "embedding_id": "wespeaker", + "role": "candidate", + "thresholds": [0.8], + }, + { + "id": "int8-titanet", + "segmentation_id": "pyannote-int8", + "embedding_id": "titanet-small", + "role": "candidate", + "thresholds": [0.8], + }, + ], + "recordings": recordings, + "threads": 4, + "rss_sample_interval_ms": 5, + } + + +def _environment(): + return { + "sherpa_onnx_version": "1.13.5", + "onnxruntime_version": "1", + "numpy_version": "2", + "psutil_version": "7", + } + + +def test_validate_only_requires_only_manifest(): + args = calibration.parse_args(["--manifest", "manifest.json", "--validate-only"]) + + assert args.manifest == Path("manifest.json") + assert args.output is None + assert args.work_dir is None + + +@pytest.mark.parametrize( + "extra", + [ + ["--output", "unused"], + ["--work-dir", "unused"], + ["--prepare-asr", "data-test"], + ["--asr-model", "model"], + ["--asr-device", "cpu"], + ["--asr-compute-type", "float32"], + ["--asr-language", "ru"], + ], +) +def test_validate_only_rejects_unused_run_options(extra): + with pytest.raises(SystemExit): + calibration.parse_args( + ["--manifest", "manifest.json", "--validate-only", *extra] + ) + + +def test_relative_manifest_paths_resolve_from_manifest_directory(tmp_path): + manifest = _manifest(tmp_path) + for model in manifest["segmentation_models"] + manifest["embedding_models"]: + model["path"] = Path(model["path"]).name + for recording in manifest["recordings"]: + recording["path"] = Path(recording["path"]).name + recording["asr_words"]["path"] = Path(recording["asr_words"]["path"]).name + if recording["reference"] is not None: + recording["reference"]["path"] = Path(recording["reference"]["path"]).name + + resolved = calibration.resolve_manifest_paths(manifest, tmp_path) + + assert calibration.validate_manifest(resolved)["schema_version"] == 2 + assert Path(resolved["recordings"][0]["path"]).is_absolute() + + +def test_validate_only_prints_safe_plan(tmp_path, monkeypatch, capsys): + manifest_path = tmp_path / "manifest.json" + manifest = _manifest(tmp_path) + for model in manifest["segmentation_models"] + manifest["embedding_models"]: + model["path"] = Path(model["path"]).name + for recording in manifest["recordings"]: + recording["path"] = Path(recording["path"]).name + recording["asr_words"]["path"] = Path(recording["asr_words"]["path"]).name + if recording["reference"] is not None: + recording["reference"]["path"] = Path(recording["reference"]["path"]).name + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + monkeypatch.setattr(calibration, "validate_original_corpus", lambda manifest: None) + monkeypatch.setattr(calibration, "environment_snapshot", _environment) + + calibration.main(["--manifest", str(manifest_path), "--validate-only"]) + + summary = json.loads(capsys.readouterr().out) + assert summary["status"] == "valid" + assert summary["automatic_cells"] == 15 + assert summary["known_cells_if_all_selected"] == 12 + assert str(tmp_path) not in json.dumps(summary) + + +@pytest.mark.parametrize( + ("key", "value"), + [("sherpa_onnx_version", "1.14.0"), ("psutil_version", "unavailable")], +) +def test_validate_environment_rejects_incompatible_dependencies(key, value): + environment = _environment() + environment[key] = value + + with pytest.raises(RuntimeError): + calibration.validate_environment(environment) + + +def _complete_cell(spec, clusters, purity=None, residual=0.0): + return { + **spec, + "status": "complete", + "diagnostics": { + "clusters": clusters, + "mapped_speaker_purity": purity, + "residual_share_after_expected": residual, + }, + } + + +def test_manifest_validates_required_matrix_and_files(tmp_path): + assert calibration.validate_manifest(_manifest(tmp_path))["schema_version"] == 2 + + +@pytest.mark.parametrize( + "mutation", ["missing_combination", "duplicate_id", "unsorted_threshold"] +) +def test_manifest_rejects_invalid_matrix_and_ids(tmp_path, mutation): + manifest = _manifest(tmp_path) + if mutation == "missing_combination": + manifest["combinations"].pop() + elif mutation == "duplicate_id": + manifest["recordings"][1]["id"] = manifest["recordings"][0]["id"] + else: + manifest["combinations"][0]["thresholds"] = [0.9, 0.8] + with pytest.raises(ValueError): + calibration.validate_manifest(manifest, verify_files=False) + + +def test_manifest_rejects_hash_mismatch_before_work(tmp_path): + manifest = _manifest(tmp_path) + manifest["recordings"][0]["sha256"] = "0" * 64 + with pytest.raises(ValueError, match="SHA-256"): + calibration.validate_manifest(manifest) + + +@pytest.mark.parametrize("model_kind", ["segmentation", "embedding"]) +def test_manifest_rejects_multiple_exact_model_ids_per_matrix_axis( + tmp_path, model_kind +): + manifest = _manifest(tmp_path) + if model_kind == "segmentation": + duplicate = dict(manifest["segmentation_models"][0]) + duplicate["id"] = "pyannote-fp32-other" + manifest["segmentation_models"].append(duplicate) + manifest["combinations"][1]["segmentation_id"] = duplicate["id"] + else: + duplicate = dict(manifest["embedding_models"][0]) + duplicate["id"] = "wespeaker-other" + manifest["embedding_models"].append(duplicate) + manifest["combinations"][2]["embedding_id"] = duplicate["id"] + + with pytest.raises(ValueError, match="ровно один"): + calibration.validate_manifest(manifest, verify_files=False) + + +def test_complete_run_rejects_substitute_corpus(tmp_path): + manifest = _manifest(tmp_path) + with pytest.raises(ValueError, match="Корпус"): + calibration.validate_original_corpus(manifest) + for recording in manifest["recordings"]: + recording.update(calibration.ORIGINAL_CORPUS[recording["id"]]) + calibration.validate_original_corpus(manifest) + + +def test_experiment_identity_ignores_paths_but_rejects_semantic_resume(tmp_path): + manifest = _manifest(tmp_path) + first = calibration.make_experiment_id(manifest, _environment()) + moved = json.loads(json.dumps(manifest)) + moved["recordings"][0]["path"] = "/different/private/path.mp4" + assert calibration.make_experiment_id(moved, _environment()) == first + output = tmp_path / "result.json" + calibration.save_output(output, {"schema_version": 2, "experiment_id": "f" * 64}) + with pytest.raises(ValueError, match="другому семантическому"): + calibration._load_or_create_output(output, manifest, _environment(), first) + + +def test_resume_rejects_different_measurement_host_without_path_identity(tmp_path): + manifest = _manifest(tmp_path) + first_environment = { + **_environment(), + "started_at_utc": "2026-08-29T00:00:00+00:00", + "platform": "host-a", + "cpu": "cpu-a", + "logical_cpus": 16, + "python_version": "3.13", + } + experiment_id = calibration.make_experiment_id(manifest, first_environment) + output_path = tmp_path / "result.json" + calibration.save_output( + output_path, + calibration._new_output(manifest, first_environment, experiment_id), + ) + moved = json.loads(json.dumps(manifest)) + moved["recordings"][0]["path"] = "/moved/private/recording.mp4" + assert calibration.make_experiment_id(moved, first_environment) == experiment_id + second_environment = { + **first_environment, + "started_at_utc": "2026-08-30T00:00:00+00:00", + "platform": "host-b", + "cpu": "cpu-b", + "logical_cpus": 2, + } + + with pytest.raises(ValueError, match="другой platform/cpu/logical_cpus"): + calibration._load_or_create_output( + output_path, moved, second_environment, experiment_id + ) + + +def test_cell_expansion_is_rectangular_deterministic_and_path_independent(tmp_path): + manifest = _manifest(tmp_path) + experiment_id = calibration.make_experiment_id(manifest, _environment()) + cells = calibration.expand_automatic_cells(manifest, experiment_id) + assert len(cells) == 15 + assert [ + (cell["combination_id"], cell["recording_id"], cell["threshold"]) + for cell in cells[:4] + ] == [ + ("fp32-wespeaker", "data-test", 0.8), + ("fp32-wespeaker", "t2-bdma", 0.8), + ("fp32-wespeaker", "yantar", 0.8), + ("fp32-wespeaker", "data-test", 0.9), + ] + assert all(cell["num_clusters"] == -1 for cell in cells) + moved = json.loads(json.dumps(manifest)) + for model in moved["segmentation_models"] + moved["embedding_models"]: + model["path"] = f"/moved/models/{model['id']}.onnx" + for recording in moved["recordings"]: + recording["path"] = f"/moved/media/{recording['id']}.mp4" + recording["asr_words"]["path"] = f"/moved/words/{recording['id']}.json" + if recording["reference"]: + recording["reference"]["path"] = f"/moved/refs/{recording['id']}.md" + moved_experiment_id = calibration.make_experiment_id(moved, _environment()) + assert moved_experiment_id == experiment_id + assert calibration.expand_automatic_cells(moved, moved_experiment_id) == cells + + +def test_threshold_selection_requires_complete_rectangle_and_uses_shared_gate(tmp_path): + manifest = _manifest(tmp_path) + combination = manifest["combinations"][0] + specs = calibration.expand_automatic_cells(manifest, "e" * 64)[:6] + assert calibration.select_threshold(combination, manifest["recordings"], []) is None + counts = [3, 2, 3, 3, 2, 2] + purities = [None, 0.7, 0.9, None, 0.8, 0.8] + cells = [ + _complete_cell(spec, count, purity) + for spec, count, purity in zip(specs, counts, purities, strict=True) + ] + selection = calibration.select_threshold(combination, manifest["recordings"], cells) + assert selection["selected_threshold"] == 0.9 + assert selection["meets_count_gate"] is True + known = calibration.derive_known_cells(manifest, "e" * 64, selection) + assert [cell["num_clusters"] for cell in known] == [3, 2, 2] + assert {cell["threshold"] for cell in known} == {0.9} + + +def test_threshold_selection_records_failed_gate(tmp_path): + manifest = _manifest(tmp_path) + combination = manifest["combinations"][1] + specs = [ + cell + for cell in calibration.expand_automatic_cells(manifest, "e" * 64) + if cell["combination_id"] == combination["id"] + ] + cells = [ + _complete_cell(spec, count, purity) + for spec, count, purity in zip(specs, [4, 3, 3], [None, 0.8, 0.8], strict=True) + ] + assert ( + calibration.select_threshold(combination, manifest["recordings"], cells)[ + "meets_count_gate" + ] + is False + ) + + +def test_stage_timing_parser_accepts_native_lines(): + log = ( + "[I] OfflineSpeakerDiarization: segmentation 1.000 s\n" + "[I] OfflineSpeakerDiarization: embedding 2.000 s\n" + "[I] OfflineSpeakerDiarization: clustering 0.100 s\n" + "[I] OfflineSpeakerDiarization: total 3.100 s, audio 10.000 s, RTF 0.310" + ) + stages = calibration.parse_stage_timings(log, "1.13.5") + assert stages["engine_total_seconds"] == 3.1 + assert stages["engine_audio_seconds"] == 10.0 + assert stages["engine_rtf"] == 0.31 + + +@pytest.mark.parametrize( + "log", + [ + "OfflineSpeakerDiarization: segmentation 1.0 s", + "OfflineSpeakerDiarization: segmentation 1.0 s\nOfflineSpeakerDiarization: segmentation 1.0 s\nOfflineSpeakerDiarization: embedding 1.0 s\nOfflineSpeakerDiarization: clustering 1.0 s\nOfflineSpeakerDiarization: total 3.0 s, audio 4 s, RTF .75", + "OfflineSpeakerDiarization: segmentation -1.0 s\nOfflineSpeakerDiarization: embedding 1.0 s\nOfflineSpeakerDiarization: clustering 1.0 s\nOfflineSpeakerDiarization: total 1.0 s, audio 4 s, RTF .25", + ], +) +def test_stage_timing_parser_rejects_missing_duplicate_and_negative(log): + with pytest.raises(ValueError): + calibration.parse_stage_timings(log, "1.13.5") + + +def test_stage_timing_parser_rejects_extra_malformed_record(): + log = ( + "OfflineSpeakerDiarization: segmentation ??? s\n" + "OfflineSpeakerDiarization: segmentation 1.000 s\n" + "OfflineSpeakerDiarization: embedding 2.000 s\n" + "OfflineSpeakerDiarization: clustering 0.100 s\n" + "OfflineSpeakerDiarization: total 3.100 s, audio 10.000 s, RTF 0.310" + ) + with pytest.raises(ValueError, match="повреждённую"): + calibration.parse_stage_timings(log, "1.13.5") + + +def test_measurement_uses_injected_cpu_clock_and_rss_once(): + process = SimpleNamespace() + calls = [] + cpu_times = iter( + [SimpleNamespace(user=1.0, system=2.0), SimpleNamespace(user=2.0, system=2.5)] + ) + + def read_cpu_times(): + calls.append("cpu") + return next(cpu_times) + + process.cpu_times = read_cpu_times + + class Sampler: + def __init__(self, actual_process, interval): + assert actual_process is process + assert interval == 0.01 + + def start(self): + calls.append("start") + + def stop(self): + calls.append("stop") + return 1234 + + clock = iter([10.0, 12.0]).__next__ + result, metrics = calibration.measure_call( + lambda: calls.append("process") or "ok", + process, + clock=clock, + sampler_factory=Sampler, + interval_seconds=0.01, + logical_cpus=4, + ) + assert result == "ok" + assert calls == ["start", "cpu", "process", "cpu", "stop"] + assert metrics["cpu_total_seconds"] == 1.5 + assert metrics["average_cpu_percent_machine"] == 18.75 + assert metrics["peak_rss_bytes"] == 1234 + + +def test_measurement_preserves_operation_error_when_sampler_stop_also_fails(): + process = SimpleNamespace( + cpu_times=iter( + [ + SimpleNamespace(user=1.0, system=2.0), + SimpleNamespace(user=1.1, system=2.1), + ] + ).__next__ + ) + + class Sampler: + def __init__(self, process, interval): + pass + + def start(self): + pass + + def stop(self): + raise RuntimeError("sampler failed") + + def operation(): + raise ValueError("operation failed") + + with pytest.raises(ValueError, match="operation failed") as error: + calibration.measure_call( + operation, + process, + clock=iter([1.0, 2.0]).__next__, + sampler_factory=Sampler, + interval_seconds=0.01, + logical_cpus=4, + ) + assert any("sampler failed" in note for note in error.value.__notes__) + + +def test_decode_cache_identity_covers_hash_bounds_and_decoder(): + recording = {"sha256": "a" * 64, "start": 1.0, "duration": 2.0} + original = calibration.decode_cache_key(recording) + assert calibration.decode_cache_key({**recording, "start": 2.0}) != original + assert calibration.decode_cache_key({**recording, "sha256": "b" * 64}) != original + assert calibration.decode_cache_key(recording, "another-decoder") != original + + +def test_decode_clip_declares_wav_format_for_temporary_output(tmp_path, monkeypatch): + recording = { + "id": "synthetic", + "path": str(tmp_path / "source.mp4"), + "sha256": "a" * 64, + "start": 1.0, + "duration": 2.0, + } + commands = [] + + def run(command, *, check): + assert check is True + commands.append(command) + Path(command[-1]).touch() + + monkeypatch.setattr(calibration.subprocess, "run", run) + monkeypatch.setattr(calibration, "read_wav", lambda path: None) + + output = calibration.decode_clip(recording, tmp_path / "work") + + assert output.is_file() + assert not output.with_suffix(".wav.tmp").exists() + assert commands[0][-3:-1] == ["-f", "wav"] + + +def test_hypescribe_parser_mapping_and_empty_diagnostics(tmp_path): + reference = tmp_path / "reference.md" + reference.write_text( + "**[00:10] Speaker 1:**\ntext\n**[00:15] Speaker 2:**\n", encoding="utf-8" + ) + recording = { + "id": "clip", + "path": str(tmp_path / "clip.mp4"), + "sha256": "a" * 64, + "start": 12.0, + "duration": 6.0, + "expected_speakers": 2, + "reference": {"path": str(reference)}, + } + turns = calibration.read_reference_turns(recording) + assert turns == [ + {"speaker": "1", "start": 0.0, "end": 3.0}, + {"speaker": "2", "start": 3.0, "end": 6.0}, + ] + mapping = calibration.best_mapping( + [ + {"speaker": 7, "start": 0.0, "end": 3.0}, + {"speaker": 8, "start": 3.0, "end": 6.0}, + ], + turns, + ) + assert mapping["mapped_speaker_purity"] == 1.0 + assert calibration.best_mapping([], turns) is None + summary = calibration.summarize_segments([], recording) + assert summary["clusters"] == 0 and summary["residual_share_after_expected"] is None + + +def test_asr_sidecar_validation_text_invariance_and_unknown_words(tmp_path): + sidecar = tmp_path / "words.json" + sidecar.write_text( + json.dumps( + [ + {"start": 0.0, "end": 0.5, "text": "Привет"}, + {"start": 0.5, "end": 0.6, "text": "!"}, + ] + ), + encoding="utf-8", + ) + words = calibration.load_asr_words(sidecar, 2.0) + recording = {"duration": 2.0, "expected_speakers": 1} + diagnostics, invariant = calibration.evaluate_segments( + [{"speaker": 0, "start": 0.0, "end": 0.5}], recording, words, [] + ) + assert diagnostics["unassigned_word_count"] == 1 + assert diagnostics["mapped_speaker_purity"] is None + assert invariant["equal"] is True and invariant["first_mismatch_index"] is None + sidecar.write_text( + json.dumps( + [ + {"start": 1.0, "end": 1.1, "text": "a"}, + {"start": 0.5, "end": 0.6, "text": "b"}, + ] + ), + encoding="utf-8", + ) + with pytest.raises(ValueError, match="границы"): + calibration.load_asr_words(sidecar, 2.0) + + +def test_asr_sidecar_rejects_empty_baseline(tmp_path): + sidecar = tmp_path / "words.json" + sidecar.write_text("[]", encoding="utf-8") + + with pytest.raises(ValueError, match="хотя бы одно слово"): + calibration.load_asr_words(sidecar, 300.0) + + +def test_text_invariance_records_first_mismatch_without_text(): + transcript = SimpleNamespace(turns=[SimpleNamespace(text="другой")]) + words = [SimpleNamespace(text="текст")] + evidence = calibration.text_invariance(words, transcript) + assert evidence["equal"] is False + assert evidence["first_mismatch_index"] == 0 + assert "текст" not in json.dumps(evidence, ensure_ascii=False) + + +def test_text_invariance_ignores_unicode_whitespace_at_turn_boundaries(): + words = [ + SimpleNamespace(text="Вопрос?"), + SimpleNamespace(text=" — "), + SimpleNamespace(text="ответ"), + ] + transcript = SimpleNamespace( + turns=[SimpleNamespace(text="Вопрос?"), SimpleNamespace(text="— ответ")] + ) + + evidence = calibration.text_invariance(words, transcript) + + assert calibration.canonical_non_whitespace_sequence(words) == "Вопрос?—ответ" + assert evidence["equal"] is True + assert evidence["first_mismatch_index"] is None + + +@pytest.mark.parametrize("output", ["Вопрос?ответ", "Вопрос!—ответ"]) +def test_text_invariance_rejects_non_whitespace_deletion_or_substitution(output): + words = [SimpleNamespace(text="Вопрос?"), SimpleNamespace(text="— ответ")] + transcript = SimpleNamespace(turns=[SimpleNamespace(text=output)]) + + evidence = calibration.text_invariance(words, transcript) + + assert evidence["equal"] is False + assert evidence["input_sha256"] != evidence["output_sha256"] + assert evidence["first_mismatch_index"] is not None + + +def test_atomic_persistence_leaves_no_temporary_file(tmp_path): + output = tmp_path / "nested" / "result.json" + calibration.save_output(output, {"private": "value"}) + assert json.loads(output.read_text()) == {"private": "value"} + assert list(output.parent.glob("*.tmp")) == [] + + +def test_atomic_persistence_preserves_target_and_cleans_temp_on_dump_failure(tmp_path): + output = tmp_path / "result.json" + calibration.save_output(output, {"stable": True}) + + with pytest.raises(TypeError): + calibration.save_output(output, {"unserializable": object()}) + + assert json.loads(output.read_text(encoding="utf-8")) == {"stable": True} + assert list(output.parent.glob("*.tmp")) == [] + + +def test_worker_failure_is_persisted_and_retried(tmp_path, monkeypatch): + output = {"cells": []} + output_path = tmp_path / "result.json" + spec = { + "cell_id": "cell", + "combination_id": "c", + "recording_id": "r", + "mode": "automatic", + "threshold": 0.8, + "num_clusters": -1, + } + attempts = [] + + def execute(*args, **kwargs): + attempts.append(1) + if len(attempts) == 1: + raise RuntimeError("boom") + return {**spec, "status": "complete"} + + monkeypatch.setattr(calibration, "_execute_cell", execute) + calibration._run_specs([spec], {}, output, {}, output_path, tmp_path) + assert output["cells"][0]["status"] == "failed" + assert output["cells"][0]["metrics"] is None + calibration._run_specs([spec], {}, output, {}, output_path, tmp_path) + assert output["cells"][0]["status"] == "complete" + assert len(attempts) == 2 + + +def test_execute_cell_rejects_text_mismatch_without_persisting_private_text( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + output = {"environment": {"logical_cpus": 4, "sherpa_onnx_version": "1.13.5"}} + spec = calibration.expand_automatic_cells(manifest, "e" * 64)[0] + recording = manifest["recordings"][0] + private_text = " СОВЕРШЕННО-СЕКРЕТНЫЙ-ТЕКСТ" + Path(recording["asr_words"]["path"]).write_text( + json.dumps([{"start": 0.0, "end": 0.5, "text": private_text}]), + encoding="utf-8", + ) + monkeypatch.setattr( + calibration, + "run_worker", + lambda *args, **kwargs: ( + {"metrics": {"wall_seconds": 1.0}, "segments": []}, + ( + "OfflineSpeakerDiarization: segmentation 0.300 s\n" + "OfflineSpeakerDiarization: embedding 0.600 s\n" + "OfflineSpeakerDiarization: clustering 0.100 s\n" + "OfflineSpeakerDiarization: total 1.000 s, " + "audio 10.000 s, RTF 0.100" + ), + ), + ) + monkeypatch.setattr(calibration, "read_reference_turns", lambda *args: []) + + def evaluate(segments, recording, words, reference_turns): + source = calibration.canonical_non_whitespace_sequence(words) + assert private_text.strip() in source + output = source[:-1] + return {}, { + "input_sha256": hashlib.sha256(source.encode()).hexdigest(), + "output_sha256": hashlib.sha256(output.encode()).hexdigest(), + "equal": False, + "first_mismatch_index": len(output), + } + + monkeypatch.setattr(calibration, "evaluate_segments", evaluate) + + with pytest.raises(RuntimeError, match="first_mismatch_index=") as error: + calibration._execute_cell( + spec, + manifest, + output, + {recording["id"]: tmp_path / "decoded.wav"}, + tmp_path, + ) + assert private_text.strip() not in str(error.value) + + +def test_execute_cell_rejects_engine_audio_duration_mismatch(tmp_path, monkeypatch): + manifest = _manifest(tmp_path) + output = {"environment": {"logical_cpus": 4, "sherpa_onnx_version": "1.13.5"}} + spec = calibration.expand_automatic_cells(manifest, "e" * 64)[0] + recording = manifest["recordings"][0] + monkeypatch.setattr( + calibration, + "run_worker", + lambda *args, **kwargs: ( + {"metrics": {"wall_seconds": 1.0}, "segments": []}, + ( + "OfflineSpeakerDiarization: segmentation 0.300 s\n" + "OfflineSpeakerDiarization: embedding 0.600 s\n" + "OfflineSpeakerDiarization: clustering 0.100 s\n" + "OfflineSpeakerDiarization: total 1.000 s, " + "audio 9.000 s, RTF 0.111" + ), + ), + ) + + with pytest.raises(RuntimeError, match="Длительность audio"): + calibration._execute_cell( + spec, + manifest, + output, + {recording["id"]: tmp_path / "decoded.wav"}, + tmp_path, + ) + + +def test_main_fails_closed_when_a_cell_failed(tmp_path, monkeypatch): + manifest = _manifest(tmp_path) + for recording in manifest["recordings"]: + recording.update(calibration.ORIGINAL_CORPUS[recording["id"]]) + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + output_path = tmp_path / "result.json" + work_dir = tmp_path / "work" + monkeypatch.setattr(calibration, "validate_manifest", lambda *args, **kwargs: None) + monkeypatch.setattr( + calibration, + "environment_snapshot", + lambda: { + **_environment(), + "started_at_utc": "2026-08-29T00:00:00+00:00", + "platform": "test", + "cpu": "test", + "logical_cpus": 4, + "python_version": "3.13", + }, + ) + monkeypatch.setattr( + calibration, + "decode_clip", + lambda recording, target: tmp_path / f"{recording['id']}.wav", + ) + + def run_specs(specs, manifest, output, decoded, output_path, work_dir): + for spec in specs: + calibration._replace_cell( + output, calibration._failed_cell(spec, RuntimeError("boom")) + ) + calibration.save_output(output_path, output) + + monkeypatch.setattr(calibration, "_run_specs", run_specs) + + with pytest.raises(RuntimeError, match="незавершённых ячеек 15"): + calibration.main( + [ + "--manifest", + str(manifest_path), + "--output", + str(output_path), + "--work-dir", + str(work_dir), + ] + ) + + +def test_main_fails_closed_for_complete_rectangle_without_mapped_purity( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + for recording in manifest["recordings"]: + recording.update(calibration.ORIGINAL_CORPUS[recording["id"]]) + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + output_path = tmp_path / "result.json" + work_dir = tmp_path / "work" + monkeypatch.setattr(calibration, "validate_manifest", lambda *args, **kwargs: None) + monkeypatch.setattr( + calibration, + "environment_snapshot", + lambda: { + **_environment(), + "started_at_utc": "2026-08-29T00:00:00+00:00", + "platform": "test", + "cpu": "test", + "logical_cpus": 4, + "python_version": "3.13", + }, + ) + monkeypatch.setattr( + calibration, + "decode_clip", + lambda recording, target: tmp_path / f"{recording['id']}.wav", + ) + modes = [] + + def run_specs(specs, manifest, output, decoded, output_path, work_dir): + modes.extend({spec["mode"] for spec in specs}) + expected = { + recording["id"]: recording["expected_speakers"] + for recording in manifest["recordings"] + } + for spec in specs: + calibration._replace_cell( + output, + _complete_cell(spec, expected[spec["recording_id"]], purity=None), + ) + calibration.save_output(output_path, output) + + monkeypatch.setattr(calibration, "_run_specs", run_specs) + + with pytest.raises(RuntimeError, match="свип непригоден"): + calibration.main( + [ + "--manifest", + str(manifest_path), + "--output", + str(output_path), + "--work-dir", + str(work_dir), + ] + ) + + output = json.loads(output_path.read_text(encoding="utf-8")) + assert modes == ["automatic"] + assert {cell["mode"] for cell in output["cells"]} == {"automatic"} + assert output["threshold_selection"] == [] + + +def test_run_worker_fake_protocol_reports_failure_and_success(tmp_path): + request = {"cell_id": "abc"} + commands = [] + + def failed(command, **kwargs): + commands.append(command) + return SimpleNamespace(returncode=1, stdout="", stderr="worker failed\n") + + with pytest.raises(RuntimeError, match="worker failed"): + calibration.run_worker(request, tmp_path, runner=failed) + + def success(command, **kwargs): + commands.append(command) + return SimpleNamespace( + returncode=0, + stdout='{"metrics":{},"segments":[]}', + stderr="native log", + ) + + payload, stderr = calibration.run_worker(request, tmp_path, runner=success) + assert payload["segments"] == [] and stderr == "native log" + for command in commands: + assert command[:3] == [ + sys.executable, + str(SCRIPT.resolve()), + "--worker", + ] + request_path = Path(command[3]) + assert request_path.is_file() + assert json.loads(request_path.read_text(encoding="utf-8")) == request diff --git a/uv.lock b/uv.lock index 014a913..59d5d46 100644 --- a/uv.lock +++ b/uv.lock @@ -279,6 +279,7 @@ dependencies = [ [package.dev-dependencies] dev = [ + { name = "psutil" }, { name = "pytest" }, ] @@ -297,7 +298,10 @@ requires-dist = [ ] [package.metadata.requires-dev] -dev = [{ name = "pytest" }] +dev = [ + { name = "psutil", specifier = ">=7,<8" }, + { name = "pytest" }, +] [[package]] name = "markdown-it-py" @@ -521,6 +525,22 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" }, ] +[[package]] +name = "psutil" +version = "7.2.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/aa/c6/d1ddf4abb55e93cebc4f2ed8b5d6dbad109ecb8d63748dd2b20ab5e57ebe/psutil-7.2.2.tar.gz", hash = "sha256:0746f5f8d406af344fd547f1c8daa5f5c33dbc293bb8d6a16d80b4bb88f59372", size = 493740, upload-time = "2026-01-28T18:14:54.428Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e7/36/5ee6e05c9bd427237b11b3937ad82bb8ad2752d72c6969314590dd0c2f6e/psutil-7.2.2-cp36-abi3-macosx_10_9_x86_64.whl", hash = "sha256:ed0cace939114f62738d808fdcecd4c869222507e266e574799e9c0faa17d486", size = 129090, upload-time = "2026-01-28T18:15:22.169Z" }, + { url = "https://files.pythonhosted.org/packages/80/c4/f5af4c1ca8c1eeb2e92ccca14ce8effdeec651d5ab6053c589b074eda6e1/psutil-7.2.2-cp36-abi3-macosx_11_0_arm64.whl", hash = "sha256:1a7b04c10f32cc88ab39cbf606e117fd74721c831c98a27dc04578deb0c16979", size = 129859, upload-time = "2026-01-28T18:15:23.795Z" }, + { url = "https://files.pythonhosted.org/packages/b5/70/5d8df3b09e25bce090399cf48e452d25c935ab72dad19406c77f4e828045/psutil-7.2.2-cp36-abi3-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:076a2d2f923fd4821644f5ba89f059523da90dc9014e85f8e45a5774ca5bc6f9", size = 155560, upload-time = "2026-01-28T18:15:25.977Z" }, + { url = "https://files.pythonhosted.org/packages/63/65/37648c0c158dc222aba51c089eb3bdfa238e621674dc42d48706e639204f/psutil-7.2.2-cp36-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:b0726cecd84f9474419d67252add4ac0cd9811b04d61123054b9fb6f57df6e9e", size = 156997, upload-time = "2026-01-28T18:15:27.795Z" }, + { url = "https://files.pythonhosted.org/packages/8e/13/125093eadae863ce03c6ffdbae9929430d116a246ef69866dad94da3bfbc/psutil-7.2.2-cp36-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:fd04ef36b4a6d599bbdb225dd1d3f51e00105f6d48a28f006da7f9822f2606d8", size = 148972, upload-time = "2026-01-28T18:15:29.343Z" }, + { url = "https://files.pythonhosted.org/packages/04/78/0acd37ca84ce3ddffaa92ef0f571e073faa6d8ff1f0559ab1272188ea2be/psutil-7.2.2-cp36-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:b58fabe35e80b264a4e3bb23e6b96f9e45a3df7fb7eed419ac0e5947c61e47cc", size = 148266, upload-time = "2026-01-28T18:15:31.597Z" }, + { url = "https://files.pythonhosted.org/packages/b4/90/e2159492b5426be0c1fef7acba807a03511f97c5f86b3caeda6ad92351a7/psutil-7.2.2-cp37-abi3-win_amd64.whl", hash = "sha256:eb7e81434c8d223ec4a219b5fc1c47d0417b12be7ea866e24fb5ad6e84b3d988", size = 137737, upload-time = "2026-01-28T18:15:33.849Z" }, + { url = "https://files.pythonhosted.org/packages/8c/c7/7bb2e321574b10df20cbde462a94e2b71d05f9bbda251ef27d104668306a/psutil-7.2.2-cp37-abi3-win_arm64.whl", hash = "sha256:8c233660f575a5a89e6d4cb65d9f938126312bca76d8fe087b947b3a1aaac9ee", size = 134617, upload-time = "2026-01-28T18:15:36.515Z" }, +] + [[package]] name = "pytest" version = "9.0.2"