Files
audita/tests/test_normalization.py

154 lines
4.3 KiB
Python

from audita.normalization import normalize_transcript
from audita.schemas import parse_source_transcript_json
class WordEstimator:
def estimate_json(self, value):
return len(value[0]["original_text"].split())
def _normalize(raw, **overrides):
defaults = {
"max_segment_gap": 5.0,
"ellipsis_gap": 2.0,
"max_segment_duration": 60.0,
"max_segment_tokens": 2048,
}
defaults.update(overrides)
return normalize_transcript(parse_source_transcript_json(raw), **defaults)
def test_same_speaker_short_gap_merges_with_space():
result = _normalize(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Hello"},
{"speaker": "Eric", "start": 2.0, "end": 3.0, "text": "there"}
]
"""
)
assert len(result.transcript) == 1
assert result.transcript[0].id == 1
assert result.transcript[0].text == "Hello there"
assert result.transcript[0].start == 0.0
assert result.transcript[0].end == 3.0
assert result.summary.merge_count == 1
def test_same_speaker_larger_allowed_gap_merges_with_ellipsis():
result = _normalize(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Hello"},
{"speaker": "Eric", "start": 4.0, "end": 5.0, "text": "there"}
]
"""
)
assert result.transcript[0].text == "Hello ... there"
def test_different_speakers_do_not_merge():
result = _normalize(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Hello"},
{"speaker": "Mike", "start": 1.5, "end": 2.0, "text": "there"}
]
"""
)
assert [segment.text for segment in result.transcript] == ["Hello", "there"]
assert result.summary.merge_count == 0
def test_gap_above_max_does_not_merge():
result = _normalize(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Hello"},
{"speaker": "Eric", "start": 7.0, "end": 8.0, "text": "there"}
]
"""
)
assert [segment.text for segment in result.transcript] == ["Hello", "there"]
def test_overlapping_segments_do_not_merge():
result = _normalize(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 2.0, "text": "Hello"},
{"speaker": "Eric", "start": 1.5, "end": 3.0, "text": "there"}
]
"""
)
assert [segment.text for segment in result.transcript] == ["Hello", "there"]
def test_max_duration_prevents_merge():
result = _normalize(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 40.0, "text": "Hello"},
{"speaker": "Eric", "start": 45.0, "end": 50.0, "text": "there"}
]
""",
max_segment_duration=45.0,
)
assert [segment.text for segment in result.transcript] == ["Hello", "there"]
def test_max_token_limit_prevents_merge():
segments = parse_source_transcript_json(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "one two"},
{"speaker": "Eric", "start": 2.0, "end": 3.0, "text": "three four"}
]
"""
)
result = normalize_transcript(
segments,
max_segment_gap=5.0,
ellipsis_gap=2.0,
max_segment_duration=60.0,
max_segment_tokens=3,
estimator=WordEstimator(),
)
assert [segment.text for segment in result.transcript] == ["one two", "three four"]
def test_shortest_gap_merges_first():
result = _normalize(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "A"},
{"speaker": "Eric", "start": 3.0, "end": 4.0, "text": "B"},
{"speaker": "Eric", "start": 4.5, "end": 5.0, "text": "C"}
]
""",
max_segment_duration=4.0,
)
assert [segment.text for segment in result.transcript] == ["A", "B C"]
def test_fresh_ids_are_assigned_chronologically_and_source_ids_are_discarded():
result = _normalize(
"""
[
{"id": 99, "speaker": "Mike", "start": 10.0, "end": 11.0, "text": "Second"},
{"id": 42, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "First"}
]
"""
)
assert [(segment.id, segment.text) for segment in result.transcript] == [(1, "First"), (2, "Second")]