mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-18 13:40:27 +02:00
refactor(webui): rename preview.py to synthesize.py and remove dead code
- Rename abogen/webui/routes/utils/preview.py → synthesize.py The file contains the core TTS synthesis pipeline (generate_preview_audio, synthesize_preview), not just preview logic. Name now matches responsibility. - Remove dead code from voice.py: get_preview_pipeline(), synthesize_audio_from_normalized(), _preview_pipeline_lock, _preview_pipelines, and unused imports (threading, numpy, create_pipeline, get_new_voice, _select_device, _to_float32, SAMPLE_RATE, SPLIT_PATTERN). These were never called — identical logic lives in synthesize.py. - Update imports in api.py, voices.py, and test_preview_applies_manual_overrides.py - 7 new tests in test_synthesize_module.py enforce file naming and import rules - 7 tests in test_domain_imports.py updated for renamed module
This commit is contained in:
@@ -25,7 +25,7 @@ from abogen.voice_profiles import (
|
|||||||
normalize_profile_entry,
|
normalize_profile_entry,
|
||||||
)
|
)
|
||||||
from abogen.webui.routes.utils.common import split_profile_spec
|
from abogen.webui.routes.utils.common import split_profile_spec
|
||||||
from abogen.webui.routes.utils.preview import synthesize_preview, generate_preview_audio
|
from abogen.webui.routes.utils.synthesize import synthesize_preview, generate_preview_audio
|
||||||
from abogen.webui.routes.utils.voice import formula_from_profile
|
from abogen.webui.routes.utils.voice import formula_from_profile
|
||||||
from abogen.normalization_settings import (
|
from abogen.normalization_settings import (
|
||||||
build_llm_configuration,
|
build_llm_configuration,
|
||||||
|
|||||||
@@ -1,6 +1,4 @@
|
|||||||
import threading
|
|
||||||
from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, cast
|
from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple, cast
|
||||||
import numpy as np
|
|
||||||
|
|
||||||
from abogen.speaker_configs import slugify_label
|
from abogen.speaker_configs import slugify_label
|
||||||
from abogen.speaker_analysis import analyze_speakers
|
from abogen.speaker_analysis import analyze_speakers
|
||||||
@@ -10,7 +8,7 @@ from abogen.voice_profiles import (
|
|||||||
load_profiles,
|
load_profiles,
|
||||||
serialize_profiles,
|
serialize_profiles,
|
||||||
)
|
)
|
||||||
from abogen.voice_formulas import get_new_voice, parse_formula_terms
|
from abogen.voice_formulas import parse_formula_terms
|
||||||
from abogen.constants import (
|
from abogen.constants import (
|
||||||
LANGUAGE_DESCRIPTIONS,
|
LANGUAGE_DESCRIPTIONS,
|
||||||
SUBTITLE_FORMATS,
|
SUBTITLE_FORMATS,
|
||||||
@@ -20,14 +18,7 @@ from abogen.constants import (
|
|||||||
)
|
)
|
||||||
from abogen.tts_plugin.utils import get_voices
|
from abogen.tts_plugin.utils import get_voices
|
||||||
from abogen.speaker_configs import list_configs
|
from abogen.speaker_configs import list_configs
|
||||||
from abogen.tts_plugin.utils import create_pipeline
|
|
||||||
from abogen.domain.device import select_device as _select_device
|
|
||||||
from abogen.domain.audio_helpers import to_float32 as _to_float32, SAMPLE_RATE
|
|
||||||
|
|
||||||
SPLIT_PATTERN = r"\n+"
|
|
||||||
|
|
||||||
_preview_pipeline_lock = threading.RLock()
|
|
||||||
_preview_pipelines: Dict[Tuple[str, str], Any] = {}
|
|
||||||
|
|
||||||
def build_narrator_roster(
|
def build_narrator_roster(
|
||||||
voice: str,
|
voice: str,
|
||||||
@@ -736,76 +727,3 @@ def pairs_to_formula(pairs: Iterable[Tuple[str, float]]) -> Optional[str]:
|
|||||||
|
|
||||||
def profiles_payload() -> Dict[str, Any]:
|
def profiles_payload() -> Dict[str, Any]:
|
||||||
return {"profiles": serialize_profiles()}
|
return {"profiles": serialize_profiles()}
|
||||||
|
|
||||||
|
|
||||||
def get_preview_pipeline(language: str, device: str):
|
|
||||||
key = (language, device)
|
|
||||||
with _preview_pipeline_lock:
|
|
||||||
pipeline = _preview_pipelines.get(key)
|
|
||||||
if pipeline is not None:
|
|
||||||
return pipeline
|
|
||||||
pipeline = create_pipeline("kokoro", lang_code=language, device=device)
|
|
||||||
_preview_pipelines[key] = pipeline
|
|
||||||
return pipeline
|
|
||||||
|
|
||||||
|
|
||||||
def synthesize_audio_from_normalized(
|
|
||||||
*,
|
|
||||||
normalized_text: str,
|
|
||||||
voice_spec: str,
|
|
||||||
language: str,
|
|
||||||
speed: float,
|
|
||||||
use_gpu: bool,
|
|
||||||
max_seconds: float,
|
|
||||||
) -> np.ndarray:
|
|
||||||
if not normalized_text.strip():
|
|
||||||
raise ValueError("Preview text is required")
|
|
||||||
|
|
||||||
device = "cpu"
|
|
||||||
if use_gpu:
|
|
||||||
try:
|
|
||||||
device = _select_device()
|
|
||||||
except Exception:
|
|
||||||
device = "cpu"
|
|
||||||
use_gpu = False
|
|
||||||
|
|
||||||
pipeline = get_preview_pipeline(language, device)
|
|
||||||
if pipeline is None:
|
|
||||||
raise RuntimeError("Preview pipeline is unavailable")
|
|
||||||
|
|
||||||
voice_choice: Any = voice_spec
|
|
||||||
if voice_spec and "*" in voice_spec:
|
|
||||||
voice_choice = get_new_voice(pipeline, voice_spec, use_gpu)
|
|
||||||
|
|
||||||
segments = pipeline(
|
|
||||||
normalized_text,
|
|
||||||
voice=voice_choice,
|
|
||||||
speed=speed,
|
|
||||||
split_pattern=SPLIT_PATTERN,
|
|
||||||
)
|
|
||||||
|
|
||||||
audio_chunks: List[np.ndarray] = []
|
|
||||||
accumulated = 0
|
|
||||||
max_samples = int(max(1.0, max_seconds) * SAMPLE_RATE)
|
|
||||||
|
|
||||||
for segment in segments:
|
|
||||||
graphemes = getattr(segment, "graphemes", "").strip()
|
|
||||||
if not graphemes:
|
|
||||||
continue
|
|
||||||
audio = _to_float32(getattr(segment, "audio", None))
|
|
||||||
if audio.size == 0:
|
|
||||||
continue
|
|
||||||
remaining = max_samples - accumulated
|
|
||||||
if remaining <= 0:
|
|
||||||
break
|
|
||||||
if audio.shape[0] > remaining:
|
|
||||||
audio = audio[:remaining]
|
|
||||||
audio_chunks.append(audio)
|
|
||||||
accumulated += audio.shape[0]
|
|
||||||
if accumulated >= max_samples:
|
|
||||||
break
|
|
||||||
|
|
||||||
if not audio_chunks:
|
|
||||||
raise RuntimeError("Preview could not be generated")
|
|
||||||
|
|
||||||
return np.concatenate(audio_chunks)
|
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ from abogen.webui.routes.utils.voice import (
|
|||||||
parse_voice_formula,
|
parse_voice_formula,
|
||||||
)
|
)
|
||||||
from abogen.webui.routes.utils.settings import load_settings, coerce_bool
|
from abogen.webui.routes.utils.settings import load_settings, coerce_bool
|
||||||
from abogen.webui.routes.utils.preview import synthesize_preview
|
from abogen.webui.routes.utils.synthesize import synthesize_preview
|
||||||
from abogen.speaker_configs import (
|
from abogen.speaker_configs import (
|
||||||
list_configs,
|
list_configs,
|
||||||
get_config,
|
get_config,
|
||||||
|
|||||||
@@ -2,8 +2,15 @@
|
|||||||
import pathlib
|
import pathlib
|
||||||
|
|
||||||
|
|
||||||
|
_MODULE_SOURCE_CACHE: dict[str, str] = {}
|
||||||
|
|
||||||
|
|
||||||
def _read_source(module_file: str) -> str:
|
def _read_source(module_file: str) -> str:
|
||||||
return pathlib.Path(module_file).read_text(encoding="utf-8")
|
if module_file not in _MODULE_SOURCE_CACHE:
|
||||||
|
_MODULE_SOURCE_CACHE[module_file] = pathlib.Path(module_file).read_text(
|
||||||
|
encoding="utf-8"
|
||||||
|
)
|
||||||
|
return _MODULE_SOURCE_CACHE[module_file]
|
||||||
|
|
||||||
|
|
||||||
def test_voice_module_does_not_import_conversion_runner():
|
def test_voice_module_does_not_import_conversion_runner():
|
||||||
@@ -17,56 +24,50 @@ def test_voice_module_does_not_import_conversion_runner():
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_voice_module_imports_select_device_from_domain():
|
def test_synthesize_module_does_not_import_conversion_runner():
|
||||||
"""voice.py must import select_device from abogen.domain.device."""
|
"""synthesize.py must not import from conversion_runner."""
|
||||||
import abogen.webui.routes.utils.voice as voice_mod
|
import abogen.webui.routes.utils.synthesize as synthesize_mod
|
||||||
|
|
||||||
with open(voice_mod.__file__, "r", encoding="utf-8") as fh:
|
source_text = _read_source(synthesize_mod.__file__)
|
||||||
source_text = fh.read()
|
|
||||||
|
|
||||||
assert "from abogen.domain.device import" in source_text
|
|
||||||
|
|
||||||
|
|
||||||
def test_voice_module_imports_to_float32_from_domain():
|
|
||||||
"""voice.py must import to_float32 from abogen.domain.audio_helpers."""
|
|
||||||
import abogen.webui.routes.utils.voice as voice_mod
|
|
||||||
|
|
||||||
with open(voice_mod.__file__, "r", encoding="utf-8") as fh:
|
|
||||||
source_text = fh.read()
|
|
||||||
|
|
||||||
assert "from abogen.domain.audio_helpers import" in source_text
|
|
||||||
|
|
||||||
|
|
||||||
def test_voice_module_has_sample_rate():
|
|
||||||
"""voice.py must define SAMPLE_RATE = 24000."""
|
|
||||||
from abogen.webui.routes.utils.voice import SAMPLE_RATE
|
|
||||||
|
|
||||||
assert SAMPLE_RATE == 24000
|
|
||||||
|
|
||||||
|
|
||||||
def test_voice_module_has_split_pattern():
|
|
||||||
"""voice.py must define SPLIT_PATTERN."""
|
|
||||||
from abogen.webui.routes.utils.voice import SPLIT_PATTERN
|
|
||||||
|
|
||||||
assert isinstance(SPLIT_PATTERN, str)
|
|
||||||
assert len(SPLIT_PATTERN) > 0
|
|
||||||
|
|
||||||
|
|
||||||
def test_preview_module_does_not_import_conversion_runner():
|
|
||||||
"""preview.py must not import from conversion_runner."""
|
|
||||||
import abogen.webui.routes.utils.preview as preview_mod
|
|
||||||
|
|
||||||
with open(preview_mod.__file__, "r", encoding="utf-8") as fh:
|
|
||||||
source_text = fh.read()
|
|
||||||
|
|
||||||
assert "from abogen.webui.conversion_runner import" not in source_text
|
assert "from abogen.webui.conversion_runner import" not in source_text
|
||||||
|
|
||||||
|
|
||||||
def test_preview_module_imports_select_device_from_domain():
|
def test_synthesize_module_imports_select_device_from_domain():
|
||||||
"""preview.py must import select_device from abogen.domain.device."""
|
"""synthesize.py must import select_device from abogen.domain.device."""
|
||||||
import abogen.webui.routes.utils.preview as preview_mod
|
import abogen.webui.routes.utils.synthesize as synthesize_mod
|
||||||
|
|
||||||
with open(preview_mod.__file__, "r", encoding="utf-8") as fh:
|
|
||||||
source_text = fh.read()
|
|
||||||
|
|
||||||
|
source_text = _read_source(synthesize_mod.__file__)
|
||||||
assert "from abogen.domain.device import" in source_text
|
assert "from abogen.domain.device import" in source_text
|
||||||
|
|
||||||
|
|
||||||
|
def test_voice_module_does_not_define_synthesize_audio_from_normalized():
|
||||||
|
"""Dead code: synthesize_audio_from_normalized must be removed from voice.py."""
|
||||||
|
import abogen.webui.routes.utils.voice as voice_mod
|
||||||
|
|
||||||
|
source_text = _read_source(voice_mod.__file__)
|
||||||
|
assert "def synthesize_audio_from_normalized(" not in source_text
|
||||||
|
|
||||||
|
|
||||||
|
def test_voice_module_does_not_define_get_preview_pipeline():
|
||||||
|
"""Dead code: get_preview_pipeline must be removed from voice.py."""
|
||||||
|
import abogen.webui.routes.utils.voice as voice_mod
|
||||||
|
|
||||||
|
source_text = _read_source(voice_mod.__file__)
|
||||||
|
assert "def get_preview_pipeline(" not in source_text
|
||||||
|
|
||||||
|
|
||||||
|
def test_voice_module_does_not_import_domain_audio_helpers():
|
||||||
|
"""After removing dead code, voice.py no longer needs audio_helpers imports."""
|
||||||
|
import abogen.webui.routes.utils.voice as voice_mod
|
||||||
|
|
||||||
|
source_text = _read_source(voice_mod.__file__)
|
||||||
|
assert "from abogen.domain.audio_helpers import" not in source_text
|
||||||
|
|
||||||
|
|
||||||
|
def test_voice_module_does_not_import_domain_device():
|
||||||
|
"""After removing dead code, voice.py no longer needs device imports."""
|
||||||
|
import abogen.webui.routes.utils.voice as voice_mod
|
||||||
|
|
||||||
|
source_text = _read_source(voice_mod.__file__)
|
||||||
|
assert "from abogen.domain.device import" not in source_text
|
||||||
|
|||||||
@@ -1,11 +1,11 @@
|
|||||||
from abogen.webui.routes.utils import preview
|
from abogen.webui.routes.utils import synthesize
|
||||||
|
|
||||||
|
|
||||||
def test_preview_applies_manual_override_before_normalization(monkeypatch):
|
def test_preview_applies_manual_override_before_normalization(monkeypatch):
|
||||||
# Don't run real TTS/normalization; just exercise the override stage by
|
# Don't run real TTS/normalization; just exercise the override stage by
|
||||||
# forcing provider=kokoro and then stubbing normalize_for_pipeline.
|
# forcing provider=kokoro and then stubbing normalize_for_pipeline.
|
||||||
|
|
||||||
monkeypatch.setattr(preview, "get_preview_pipeline", lambda language, device: None)
|
monkeypatch.setattr(synthesize, "get_preview_pipeline", lambda language, device: None)
|
||||||
|
|
||||||
# Stub normalize_for_pipeline to be identity; we only care that overrides run.
|
# Stub normalize_for_pipeline to be identity; we only care that overrides run.
|
||||||
class _Norm:
|
class _Norm:
|
||||||
@@ -42,7 +42,7 @@ def test_preview_applies_manual_override_before_normalization(monkeypatch):
|
|||||||
monkeypatch.setattr(utils, "create_pipeline", _mock_create_pipeline)
|
monkeypatch.setattr(utils, "create_pipeline", _mock_create_pipeline)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
preview.generate_preview_audio(
|
synthesize.generate_preview_audio(
|
||||||
text="He said Unfu*k loudly.",
|
text="He said Unfu*k loudly.",
|
||||||
voice_spec="M1",
|
voice_spec="M1",
|
||||||
language="en",
|
language="en",
|
||||||
|
|||||||
@@ -0,0 +1,70 @@
|
|||||||
|
"""Tests that preview/synthesis module is correctly named and importable."""
|
||||||
|
import pathlib
|
||||||
|
|
||||||
|
|
||||||
|
def _read_source(module_file: str) -> str:
|
||||||
|
return pathlib.Path(module_file).read_text(encoding="utf-8")
|
||||||
|
|
||||||
|
|
||||||
|
def test_preview_file_renamed_to_synthesize():
|
||||||
|
"""preview.py must be renamed to synthesize.py."""
|
||||||
|
import abogen.webui.routes.utils.synthesize as synthesize_mod
|
||||||
|
|
||||||
|
synthesize_path = pathlib.Path(synthesize_mod.__file__)
|
||||||
|
assert synthesize_path.name == "synthesize.py"
|
||||||
|
assert synthesize_path.exists()
|
||||||
|
assert not synthesize_path.with_name("preview.py").exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_synthesize_module_has_generate_preview_audio():
|
||||||
|
"""synthesize.py must export generate_preview_audio."""
|
||||||
|
from abogen.webui.routes.utils.synthesize import generate_preview_audio
|
||||||
|
|
||||||
|
assert callable(generate_preview_audio)
|
||||||
|
|
||||||
|
|
||||||
|
def test_synthesize_module_has_synthesize_preview():
|
||||||
|
"""synthesize.py must export synthesize_preview."""
|
||||||
|
from abogen.webui.routes.utils.synthesize import synthesize_preview
|
||||||
|
|
||||||
|
assert callable(synthesize_preview)
|
||||||
|
|
||||||
|
|
||||||
|
def test_synthesize_module_has_get_preview_pipeline():
|
||||||
|
"""synthesize.py must export get_preview_pipeline."""
|
||||||
|
from abogen.webui.routes.utils.synthesize import get_preview_pipeline
|
||||||
|
|
||||||
|
assert callable(get_preview_pipeline)
|
||||||
|
|
||||||
|
|
||||||
|
def test_api_imports_from_synthesize():
|
||||||
|
"""api.py must import from synthesize, not preview."""
|
||||||
|
import abogen.webui.routes.api as api_mod
|
||||||
|
|
||||||
|
source = _read_source(api_mod.__file__)
|
||||||
|
assert "from abogen.webui.routes.utils.synthesize import" in source
|
||||||
|
assert "from abogen.webui.routes.utils.preview import" not in source
|
||||||
|
|
||||||
|
|
||||||
|
def test_voices_imports_from_synthesize():
|
||||||
|
"""voices.py must import from synthesize, not preview."""
|
||||||
|
import abogen.webui.routes.voices as voices_mod
|
||||||
|
|
||||||
|
source = _read_source(voices_mod.__file__)
|
||||||
|
assert "from abogen.webui.routes.utils.synthesize import" in source
|
||||||
|
assert "from abogen.webui.routes.utils.preview import" not in source
|
||||||
|
|
||||||
|
|
||||||
|
def test_no_module_imports_preview():
|
||||||
|
"""No module should import from the old preview path."""
|
||||||
|
import glob
|
||||||
|
import os
|
||||||
|
|
||||||
|
project_root = pathlib.Path(__file__).parent.parent
|
||||||
|
py_files = glob.glob(str(project_root / "abogen" / "**" / "*.py"), recursive=True)
|
||||||
|
|
||||||
|
for py_file in py_files:
|
||||||
|
content = pathlib.Path(py_file).read_text(encoding="utf-8")
|
||||||
|
assert "from abogen.webui.routes.utils.preview import" not in content, (
|
||||||
|
f"{py_file} still imports from preview"
|
||||||
|
)
|
||||||
Reference in New Issue
Block a user