diff --git a/abogen/webui/routes/api.py b/abogen/webui/routes/api.py index 28c1e9b..35f4f94 100644 --- a/abogen/webui/routes/api.py +++ b/abogen/webui/routes/api.py @@ -25,7 +25,7 @@ from abogen.voice_profiles import ( normalize_profile_entry, ) 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.normalization_settings import ( build_llm_configuration, diff --git a/abogen/webui/routes/utils/preview.py b/abogen/webui/routes/utils/synthesize.py similarity index 100% rename from abogen/webui/routes/utils/preview.py rename to abogen/webui/routes/utils/synthesize.py diff --git a/abogen/webui/routes/utils/voice.py b/abogen/webui/routes/utils/voice.py index 3433369..8ac1509 100644 --- a/abogen/webui/routes/utils/voice.py +++ b/abogen/webui/routes/utils/voice.py @@ -1,6 +1,4 @@ -import threading 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_analysis import analyze_speakers @@ -10,7 +8,7 @@ from abogen.voice_profiles import ( load_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 ( LANGUAGE_DESCRIPTIONS, SUBTITLE_FORMATS, @@ -20,14 +18,7 @@ from abogen.constants import ( ) from abogen.tts_plugin.utils import get_voices 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( voice: str, @@ -736,76 +727,3 @@ def pairs_to_formula(pairs: Iterable[Tuple[str, float]]) -> Optional[str]: def profiles_payload() -> Dict[str, Any]: 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) diff --git a/abogen/webui/routes/voices.py b/abogen/webui/routes/voices.py index ac39760..81c1ebb 100644 --- a/abogen/webui/routes/voices.py +++ b/abogen/webui/routes/voices.py @@ -9,7 +9,7 @@ from abogen.webui.routes.utils.voice import ( parse_voice_formula, ) 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 ( list_configs, get_config, diff --git a/tests/test_domain_imports.py b/tests/test_domain_imports.py index 7cd39dd..ed4a111 100644 --- a/tests/test_domain_imports.py +++ b/tests/test_domain_imports.py @@ -2,8 +2,15 @@ import pathlib +_MODULE_SOURCE_CACHE: dict[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(): @@ -17,56 +24,50 @@ def test_voice_module_does_not_import_conversion_runner(): ) -def test_voice_module_imports_select_device_from_domain(): - """voice.py must import select_device from abogen.domain.device.""" - import abogen.webui.routes.utils.voice as voice_mod +def test_synthesize_module_does_not_import_conversion_runner(): + """synthesize.py must not import from conversion_runner.""" + import abogen.webui.routes.utils.synthesize as synthesize_mod - with open(voice_mod.__file__, "r", encoding="utf-8") as fh: - 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() + source_text = _read_source(synthesize_mod.__file__) assert "from abogen.webui.conversion_runner import" not in source_text -def test_preview_module_imports_select_device_from_domain(): - """preview.py must import select_device from abogen.domain.device.""" - import abogen.webui.routes.utils.preview as preview_mod - - with open(preview_mod.__file__, "r", encoding="utf-8") as fh: - source_text = fh.read() +def test_synthesize_module_imports_select_device_from_domain(): + """synthesize.py must import select_device from abogen.domain.device.""" + import abogen.webui.routes.utils.synthesize as synthesize_mod + source_text = _read_source(synthesize_mod.__file__) 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 diff --git a/tests/test_preview_applies_manual_overrides.py b/tests/test_preview_applies_manual_overrides.py index 2dc786f..976e326 100644 --- a/tests/test_preview_applies_manual_overrides.py +++ b/tests/test_preview_applies_manual_overrides.py @@ -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): # Don't run real TTS/normalization; just exercise the override stage by # 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. class _Norm: @@ -42,7 +42,7 @@ def test_preview_applies_manual_override_before_normalization(monkeypatch): monkeypatch.setattr(utils, "create_pipeline", _mock_create_pipeline) try: - preview.generate_preview_audio( + synthesize.generate_preview_audio( text="He said Unfu*k loudly.", voice_spec="M1", language="en", diff --git a/tests/test_synthesize_module.py b/tests/test_synthesize_module.py new file mode 100644 index 0000000..ec1335e --- /dev/null +++ b/tests/test_synthesize_module.py @@ -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" + )