Vladimir BryzgalovandClaude Opus 5
49fb11affa
Пайплайн склеивал соседние отрезки речи и подавал склейку в модель одним куском. На длинном куске GigaAM обрывается, не договорив, и терялся конец фразы - до 15 % текста. Заметно на слух, но незаметно в счёте слов. Теперь распознавание идёт по коротким отрезкам, а склеивается уже текст: реплики остаются читаемыми, модель получает короткие куски. Turn хранит список отрезков, из которых собран. Замер на 205 минутах: слов +6,1 % к прошлой версии и +4,2 % к диаризации, при этом обработка по-прежнему в 4,3 раза быстрее. Запас по краям отрезка проверен и не подтвердился (разброс +-3 % без направления) - настройка убрана, чтобы не оставлять мёртвый код. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
82 lines
3.2 KiB
Python
82 lines
3.2 KiB
Python
"""Работа с сегментами диаризации: склейка в реплики, нарезка, статистика."""
|
|
from dataclasses import dataclass, field
|
|
|
|
__all__ = ["Segment", "Turn", "merge_turns", "chunk_ranges", "speech_stats"]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Segment:
|
|
"""Отрезок речи одного говорящего, как его вернула диаризация."""
|
|
start: float
|
|
end: float
|
|
speaker: int
|
|
|
|
|
|
@dataclass
|
|
class Turn:
|
|
"""Реплика: несколько подряд идущих сегментов одного говорящего.
|
|
|
|
Отрезки сохраняются отдельным списком: распознавание идёт по ним, а не
|
|
по реплике целиком. На длинном куске модель обрывается, не договорив,
|
|
и терялось до 15 % текста - как раз концы фраз.
|
|
"""
|
|
start: float
|
|
end: float
|
|
speaker: int
|
|
pieces: list[Segment] = field(default_factory=list)
|
|
|
|
@property
|
|
def duration(self) -> float:
|
|
return self.end - self.start
|
|
|
|
|
|
def merge_turns(segments: list[Segment], gap: float = 0.8) -> list[Turn]:
|
|
"""Склеивает соседние сегменты одного говорящего в реплики.
|
|
|
|
Диаризация режет речь на короткие куски по паузам внутри фразы;
|
|
для расшифровки нужны цельные реплики, иначе текст рассыпается.
|
|
"""
|
|
turns: list[Turn] = []
|
|
for s in segments:
|
|
if turns and turns[-1].speaker == s.speaker and s.start - turns[-1].end <= gap:
|
|
turns[-1].end = s.end
|
|
turns[-1].pieces.append(s)
|
|
else:
|
|
turns.append(Turn(start=s.start, end=s.end, speaker=s.speaker, pieces=[s]))
|
|
return turns
|
|
|
|
|
|
def chunk_ranges(start: float, end: float, max_len: float) -> list[tuple[float, float]]:
|
|
"""Режет интервал на куски не длиннее max_len.
|
|
|
|
GigaAM падает на аудио длиннее ~200 секунд: у неё позиционное кодирование
|
|
рассчитано на 5000 фреймов. Держим запас и режем по 150.
|
|
"""
|
|
chunks = []
|
|
cursor = start
|
|
while cursor < end:
|
|
stop = min(cursor + max_len, end)
|
|
chunks.append((cursor, stop))
|
|
cursor = stop
|
|
return chunks or [(start, end)]
|
|
|
|
|
|
def speech_stats(segments: list[Segment]) -> dict:
|
|
"""Считает, сколько и в каких долях говорил каждый участник."""
|
|
by_speaker: dict[int, float] = {}
|
|
for s in segments:
|
|
by_speaker[s.speaker] = by_speaker.get(s.speaker, 0.0) + (s.end - s.start)
|
|
total = sum(by_speaker.values())
|
|
return {
|
|
"speakers": len(by_speaker),
|
|
"speech_sec": round(total, 1),
|
|
"by_speaker": [
|
|
{
|
|
"speaker": spk + 1,
|
|
"speech_sec": round(dur, 1),
|
|
"share_pct": round(dur / total * 100, 1) if total else 0.0,
|
|
}
|
|
for spk, dur in sorted(by_speaker.items())
|
|
],
|
|
}
|