refactor: both UIs use shared tts_segments() from domain/conversion_pipeline.py

- domain/conversion_pipeline.py: add tts_segments() for pre-normalized text;
  emit_text_segments() now delegates to tts_segments() internally
- pyqt/conversion.py: inner TTS loop replaced with tts_segments() iterator;
  removed FakeToken import (handled by domain)
- webui/conversion_runner.py: emit_text() inner loop replaced with
  tts_segments() iterator; removed FakeToken import
- Both UIs now share the same TTS emission logic — normalization + backend
  invocation + segment iteration + token extraction
- +3 tests for tts_segments (no-normalization, chunk_start, basic)
- 1188 tests pass
This commit is contained in:
Artem Akymenko
2026-07-19 11:47:14 +00:00
parent 1193185833
commit aec3462f1f
4 changed files with 176 additions and 158 deletions
+48 -2
View File
@@ -1,4 +1,4 @@
"""Tests for domain/conversion_pipeline.py — emit_text_segments."""
"""Tests for domain/conversion_pipeline.py — tts_segments, emit_text_segments."""
from dataclasses import dataclass, field
from typing import Any, List, Optional
@@ -7,7 +7,7 @@ from unittest.mock import MagicMock
import numpy as np
import pytest
from abogen.domain.conversion_pipeline import emit_text_segments, SegmentResult
from abogen.domain.conversion_pipeline import tts_segments, emit_text_segments, SegmentResult
@dataclass
@@ -135,3 +135,49 @@ class TestEmitTextSegments:
assert seg.graphemes == "Test"
assert seg.duration == 2.0
assert len(seg.audio) == 48000
class TestTtsSegments:
def test_yields_segments_from_normalized_text(self):
audio = np.ones(24000, dtype="float32")
segments = [FakeSegment("Hello", audio)]
results = list(tts_segments(
"Already normalized text",
backend=make_backend(segments),
voice="A",
speed=1.0,
split_pattern=r"\s+",
))
assert len(results) == 1
assert results[0].graphemes == "Hello"
def test_no_normalization_performed(self):
"""tts_segments should NOT normalize — it passes text directly to backend."""
received_texts = []
def capture_backend(text, voice=None, speed=1.0, split_pattern=None):
received_texts.append(text)
yield FakeSegment("ok", np.ones(24000, dtype="float32"))
list(tts_segments(
"Raw unnormalized text",
backend=capture_backend,
voice="A",
speed=1.0,
split_pattern=r"\s+",
))
assert received_texts[0] == "Raw unnormalized text"
def test_chunk_start_increments(self):
audio = np.ones(24000, dtype="float32")
segments = [FakeSegment("A", audio), FakeSegment("B", audio)]
results = list(tts_segments(
"test",
backend=make_backend(segments),
voice="A",
speed=1.0,
split_pattern=r"\s+",
current_time=10.0,
))
assert results[0].chunk_start == 10.0
assert results[1].chunk_start == 11.0