Dymas/abogen
1
0
Fork 0
mirror of https://github.com/denizsafak/abogen.git synced 2026-09-20 11:40:57 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: unify Language enum across all layers

Browse Source 
- EngineConfig.language: Language (was lang_code: str = 'a')
- Engine owns _KOKORO_LANG_MAP, engine_language(), supported_languages()
- Engine provides language_for_voice_id() for voice catalog
- Plugins/kokoro/__init__.py calls engine_language() internally
- create_pipeline(plugin_id, language=Language) — no kokoro codes
- pipeline_factory.py clean of kokoro-specific code
- Domain functions raise TypeError if non-enum passed
- WebUI api.py: _parse_language() helper at API boundary
- Voice catalog returns ISO codes (lang.value)
- Constants: LANGUAGE_DESCRIPTIONS keyed by Language enum
- All tests updated for Language enum
- 1414 tests pass
This commit is contained in:
Artem Akymenko
2026-07-27 07:36:36 +00:00
parent 713abdfd73
commit 0dc491e420
35 changed files with 339 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files
abogen/application/conversion_executor.py
+3 -3
View File
@@ -254,7 +254,7 @@ def execute_conversion(
audio_sink=audio_sink, audio_sink=audio_sink,
subtitle_mode=effective_subtitle_mode, subtitle_mode=effective_subtitle_mode,
max_subtitle_words=request.max_subtitle_words, max_subtitle_words=request.max_subtitle_words,
lang_code=request.language, language=request.language,
use_spacy_segmentation=use_spacy, use_spacy_segmentation=use_spacy,
) )
@@ -470,7 +470,7 @@ def execute_conversion(
subtitle_writer, subtitle_writer,
subtitle_mode=request.subtitle_mode, subtitle_mode=request.subtitle_mode,
max_subtitle_words=request.max_subtitle_words, max_subtitle_words=request.max_subtitle_words,
lang_code=request.language, language=request.language,
use_spacy_segmentation=use_spacy, use_spacy_segmentation=use_spacy,
fallback_end_time=stats.current_time, fallback_end_time=stats.current_time,
) )
@@ -480,7 +480,7 @@ def execute_conversion(
chapter_subtitle_writer, chapter_subtitle_writer,
subtitle_mode=request.subtitle_mode, subtitle_mode=request.subtitle_mode,
max_subtitle_words=request.max_subtitle_words, max_subtitle_words=request.max_subtitle_words,
lang_code=request.language, language=request.language,
use_spacy_segmentation=use_spacy, use_spacy_segmentation=use_spacy,
fallback_end_time=stats.current_time, fallback_end_time=stats.current_time,
) )
abogen/constants.py
+27 -14
View File
@@ -1,4 +1,5 @@
from abogen.utils import get_version from abogen.utils import get_version
from abogen.domain.enums import Language
# Program Information # Program Information
PROGRAM_NAME = "abogen" PROGRAM_NAME = "abogen"
@@ -16,8 +17,22 @@ SUBTITLE_FORMATS = [
("ass_centered_narrow", "ASS (centered narrow)"), ("ass_centered_narrow", "ASS (centered narrow)"),
] ]
# Language description mapping # Language description mapping (Language enum → human-readable label).
LANGUAGE_DESCRIPTIONS = { LANGUAGE_DESCRIPTIONS = {
Language.EN_US: "American English",
Language.EN_GB: "British English",
Language.ES: "Spanish",
Language.FR: "French",
Language.HI: "Hindi",
Language.IT: "Italian",
Language.JA: "Japanese",
Language.PT_BR: "Brazilian Portuguese",
Language.ZH: "Mandarin Chinese",
}
# Display-only mapping for kokoro codes → labels.
# Used by voice catalog and PyQt (legacy) where kokoro codes are still present.
KOKORO_CODE_LABELS = {
"a": "American English", "a": "American English",
"b": "British English", "b": "British English",
"e": "Spanish", "e": "Spanish",
@@ -56,24 +71,22 @@ SUPPORTED_INPUT_FORMATS = [
] ]
# Supported languages for subtitle generation # Supported languages for subtitle generation
# Currently, only 'a (American English)' and 'b (British English)' are supported for subtitle generation. # Currently, only English (EN_US, EN_GB) are supported for subtitle generation.
# This is because tokens that contain timestamps are not generated for other languages in the Kokoro pipeline. # This is because tokens that contain timestamps are not generated for other languages in the Kokoro pipeline.
# Please refer to: https://github.com/hexgrad/kokoro/blob/6d87f4ae7abc2d14dbc4b3ef2e5f19852e861ac2/kokoro/pipeline.py # Please refer to: https://github.com/hexgrad/kokoro/blob/6d87f4ae7abc2d14dbc4b3ef2e5f19852e861ac2/kokoro/pipeline.py
# 383 English processing (unchanged) SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION = [Language.EN_US, Language.EN_GB]
# 384 if self.lang_code in 'ab':
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION = list(LANGUAGE_DESCRIPTIONS.keys())
# Voice and sample text mapping # Voice and sample text mapping
SAMPLE_VOICE_TEXTS = { SAMPLE_VOICE_TEXTS = {
"a": "This is a sample of the selected voice.", Language.EN_US: "This is a sample of the selected voice.",
"b": "This is a sample of the selected voice.", Language.EN_GB: "This is a sample of the selected voice.",
"e": "Este es una muestra de la voz seleccionada.", Language.ES: "Este es una muestra de la voz seleccionada.",
"f": "Ceci est un exemple de la voix sélectionnée.", Language.FR: "Ceci est un exemple de la voix sélectionnée.",
"h": "यह चयनित आवाज़ का एक नमूना है।", Language.HI: "यह चयनित आवाज़ का एक नमूना है।",
"i": "Questo è un esempio della voce selezionata.", Language.IT: "Questo è un esempio della voce selezionata.",
"j": "これは選択した声のサンプルです。", Language.JA: "これは選択した声のサンプルです。",
"p": "Este é um exemplo da voz selecionada.", Language.PT_BR: "Este é um exemplo da voz selecionada.",
"z": "这是所选语音的示例。", Language.ZH: "这是所选语音的示例。",
} }
COLORS = { COLORS = {
abogen/domain/chunk_utils.py
+2 -1
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
from collections import defaultdict from collections import defaultdict
from typing import Any, Dict, Iterable, Mapping from typing import Any, Dict, Iterable, Mapping
from abogen.domain.enums import Language
from abogen.pronunciation_store import increment_usage from abogen.pronunciation_store import increment_usage
@@ -44,7 +45,7 @@ def record_override_usage(
if not usage_counter: if not usage_counter:
return return
language = getattr(job, "language", "") or "a" language = getattr(job, "language", Language.EN_US) or Language.EN_US
for normalized, amount in usage_counter.items(): for normalized, amount in usage_counter.items():
if amount <= 0: if amount <= 0:
continue continue
abogen/domain/conversion_engine.py
+4 -4
View File
@@ -23,7 +23,7 @@ from typing import Any, Callable, Optional, Protocol
from abogen.domain.audio_sink import AudioSink from abogen.domain.audio_sink import AudioSink
from abogen.domain.conversion_pipeline import tts_segments from abogen.domain.conversion_pipeline import tts_segments
from abogen.domain.enums import SubtitleMode from abogen.domain.enums import Language, SubtitleMode
from abogen.domain.normalization import TTSContext from abogen.domain.normalization import TTSContext
from abogen.domain.progress import calc_etr_str from abogen.domain.progress import calc_etr_str
from abogen.domain.subtitle_generation import process_subtitle_tokens from abogen.domain.subtitle_generation import process_subtitle_tokens
@@ -153,7 +153,7 @@ def process_and_write_subtitles(
*, *,
subtitle_mode: str, subtitle_mode: str,
max_subtitle_words: int, max_subtitle_words: int,
lang_code: str, language: Language,
use_spacy_segmentation: bool, use_spacy_segmentation: bool,
fallback_end_time: float, fallback_end_time: float,
) -> None: ) -> None:
@@ -169,7 +169,7 @@ def process_and_write_subtitles(
new_entries, new_entries,
max_subtitle_words, max_subtitle_words,
subtitle_mode, subtitle_mode,
lang_code, language,
use_spacy_segmentation=use_spacy_segmentation, use_spacy_segmentation=use_spacy_segmentation,
fallback_end_time=fallback_end_time, fallback_end_time=fallback_end_time,
) )
@@ -191,7 +191,7 @@ class SynthParams:
audio_sink: Optional[AudioSink] = None audio_sink: Optional[AudioSink] = None
subtitle_mode: str = "Disabled" subtitle_mode: str = "Disabled"
max_subtitle_words: int = 50 max_subtitle_words: int = 50
lang_code: str = "a" language: Language = Language.EN_US
use_spacy_segmentation: bool = False use_spacy_segmentation: bool = False
abogen/domain/conversion_pipeline.py
+1 -1
View File
@@ -282,7 +282,7 @@ def emit_text_to_sinks(
# subtitle # subtitle
subtitle_writer: Any = None, subtitle_writer: Any = None,
subtitle_mode: str = "Disabled", subtitle_mode: str = "Disabled",
subtitle_lang: str = "a", subtitle_lang: Language = Language.EN_US,
max_subtitle_words: int = 50, max_subtitle_words: int = 50,
use_spacy_segmentation: bool = True, use_spacy_segmentation: bool = True,
# normalization # normalization
abogen/domain/normalization.py
+6 -7
View File
@@ -15,6 +15,7 @@ from __future__ import annotations
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Mapping, Optional from typing import Any, Callable, Dict, List, Mapping, Optional
from abogen.domain.enums import Language
from abogen.kokoro_text_normalization import ( from abogen.kokoro_text_normalization import (
ApostropheConfig, ApostropheConfig,
normalize_for_pipeline as _normalize_for_pipeline, normalize_for_pipeline as _normalize_for_pipeline,
@@ -127,7 +128,7 @@ def prepare_text_for_tts(
def build_tts_context( def build_tts_context(
*, *,
language: str = "a", language: Language,
subtitle_mode: str = "Disabled", subtitle_mode: str = "Disabled",
pronunciation_overrides: Optional[List[Dict[str, Any]]] = None, pronunciation_overrides: Optional[List[Dict[str, Any]]] = None,
manual_overrides: Optional[List[Dict[str, Any]]] = None, manual_overrides: Optional[List[Dict[str, Any]]] = None,
@@ -143,7 +144,7 @@ def build_tts_context(
merges pronunciation overrides, and compiles all rules. merges pronunciation overrides, and compiles all rules.
Args: Args:
language: Language code (a, b, e, f, etc.). language: Language enum value.
subtitle_mode: Subtitle mode string. subtitle_mode: Subtitle mode string.
pronunciation_overrides: List of pronunciation override dicts. pronunciation_overrides: List of pronunciation override dicts.
manual_overrides: List of manual override dicts. manual_overrides: List of manual override dicts.
@@ -199,15 +200,13 @@ def build_tts_context(
) )
# Compute split pattern # Compute split pattern
try: if not isinstance(language, Language):
lang = Language.from_str(language) if not isinstance(language, Language) else language raise TypeError(f"language must be Language enum, got {type(language).__name__}: {language!r}")
except ValueError:
lang = Language.EN_US
try: try:
mode = SubtitleMode.from_str(subtitle_mode) if not isinstance(subtitle_mode, SubtitleMode) else subtitle_mode mode = SubtitleMode.from_str(subtitle_mode) if not isinstance(subtitle_mode, SubtitleMode) else subtitle_mode
except ValueError: except ValueError:
mode = SubtitleMode.DISABLED mode = SubtitleMode.DISABLED
split_pattern = get_split_pattern(lang, mode) split_pattern = get_split_pattern(language, mode)
# Merge pronunciation overrides (accepts dict or object) # Merge pronunciation overrides (accepts dict or object)
source = { source = {
abogen/domain/pipeline_factory.py
+13 -26
View File
@@ -2,6 +2,10 @@
Provides a unified interface for creating and managing TTS pipelines Provides a unified interface for creating and managing TTS pipelines
across all UI layers (WebUI, PyQt, CLI). across all UI layers (WebUI, PyQt, CLI).
Language handling: the engine owns the mapping between Language enum
and its internal format. Callers pass Language enum; the engine
converts internally. No engine-specific codes leak outside the engine.
""" """
from __future__ import annotations from __future__ import annotations
@@ -13,19 +17,6 @@ from abogen.domain.enums import Language
from abogen.domain.voice_resolution import initialize_voice_cache from abogen.domain.voice_resolution import initialize_voice_cache
from abogen.tts_plugin.utils import create_pipeline, is_plugin_registered from abogen.tts_plugin.utils import create_pipeline, is_plugin_registered
# Kokoro-specific language mapping (engine's responsibility)
_KOKORO_LANG_MAP = {
Language.EN_US: "a",
Language.EN_GB: "b",
Language.ES: "e",
Language.FR: "f",
Language.HI: "h",
Language.IT: "i",
Language.JA: "j",
Language.PT_BR: "p",
Language.ZH: "z",
}
def resolve_device(use_gpu: bool) -> str: def resolve_device(use_gpu: bool) -> str:
"""Determine compute device from job and global config flags.""" """Determine compute device from job and global config flags."""
@@ -39,29 +30,25 @@ def resolve_device(use_gpu: bool) -> str:
def create_pipeline_for_job( def create_pipeline_for_job(
provider: str, provider: str,
language: str, language: Language,
use_gpu: bool, use_gpu: bool,
) -> Any: ) -> Any:
"""Create a TTS pipeline with proper device selection. """Create a TTS pipeline with proper device selection.
Handles provider validation, GPU decision, and plugin checks. Args:
provider: TTS provider name ("kokoro" or "supertonic").
language: Language enum (app-layer type, not engine-specific).
use_gpu: Whether GPU acceleration is requested.
""" """
provider = str(provider or "kokoro").strip().lower() or "kokoro" provider = str(provider or "kokoro").strip().lower() or "kokoro"
if not is_plugin_registered(provider): if not is_plugin_registered(provider):
provider = "kokoro" provider = "kokoro"
# Convert Language enum to Kokoro single-letter code
try:
lang = Language.from_str(language) if not isinstance(language, Language) else language
except ValueError:
lang = Language.EN_US # fallback for unknown languages
kokoro_code = _KOKORO_LANG_MAP.get(lang, "a")
if provider == "supertonic": if provider == "supertonic":
return create_pipeline("supertonic") return create_pipeline("supertonic")
device = resolve_device(use_gpu) device = resolve_device(use_gpu)
return create_pipeline("kokoro", lang_code=kokoro_code, device=device) return create_pipeline("kokoro", language=language, device=device)
def dispose_pipelines(pipelines: Dict[str, Any]) -> None: def dispose_pipelines(pipelines: Dict[str, Any]) -> None:
@@ -80,7 +67,7 @@ class PipelinePool:
Usage:: Usage::
pool = PipelinePool() pool = PipelinePool()
backend = pool.get("kokoro", "en", use_gpu=True) backend = pool.get("kokoro", Language.EN_US, use_gpu=True)
# ... use backend ... # ... use backend ...
pool.dispose_all() pool.dispose_all()
""" """
@@ -92,7 +79,7 @@ class PipelinePool:
def get( def get(
self, self,
provider: str, provider: str,
language: str, language: Language,
use_gpu: bool, use_gpu: bool,
*, *,
request: Any = None, request: Any = None,
@@ -102,7 +89,7 @@ class PipelinePool:
Args: Args:
provider: TTS provider name ("kokoro" or "supertonic"). provider: TTS provider name ("kokoro" or "supertonic").
language: Language code (for kokoro). language: Language enum (app-layer type).
use_gpu: Whether GPU acceleration is requested. use_gpu: Whether GPU acceleration is requested.
request: ConversionRequest for voice cache initialization. request: ConversionRequest for voice cache initialization.
events: ConversionEvents for logging during cache init. events: ConversionEvents for logging during cache init.
abogen/domain/settings_core.py
+3 -3
View File
@@ -13,7 +13,7 @@ from dataclasses import dataclass
from typing import Any, Callable, Dict, Mapping from typing import Any, Callable, Dict, Mapping
from abogen.constants import ( from abogen.constants import (
LANGUAGE_DESCRIPTIONS, KOKORO_CODE_LABELS,
SUBTITLE_FORMATS, SUBTITLE_FORMATS,
SUPPORTED_SOUND_FORMATS, SUPPORTED_SOUND_FORMATS,
) )
@@ -135,10 +135,10 @@ def _norm_speaker_spec(value: Any, default: str) -> str:
def _norm_language_list(value: Any, default: list) -> list: def _norm_language_list(value: Any, default: list) -> list:
if isinstance(value, (list, tuple, set)): if isinstance(value, (list, tuple, set)):
return [code for code in value if isinstance(code, str) and code in LANGUAGE_DESCRIPTIONS] return [code for code in value if isinstance(code, str) and code in KOKORO_CODE_LABELS]
if isinstance(value, str): if isinstance(value, str):
parts = [item.strip().lower() for item in value.split(",") if item.strip()] parts = [item.strip().lower() for item in value.split(",") if item.strip()]
return [code for code in parts if code in LANGUAGE_DESCRIPTIONS] return [code for code in parts if code in KOKORO_CODE_LABELS]
return default return default
abogen/domain/split_pattern.py
+5 -9
View File
@@ -12,35 +12,31 @@ PUNCTUATION_SENTENCE_COMMA = r".!?,？。！？،，、।"
PUNCTUATION_COMMAS = ",，、" PUNCTUATION_COMMAS = ",，、"
def get_split_pattern(language: str, subtitle_mode: str) -> str: def get_split_pattern(language: Language, subtitle_mode: str) -> str:
"""Get the appropriate split pattern based on language and subtitle mode. """Get the appropriate split pattern based on language and subtitle mode.
Args: Args:
language: Language code (a, b, e, f, etc.) language: Language enum value.
subtitle_mode: Subtitle mode ("Sentence", "Sentence + Comma", "Line", etc.) subtitle_mode: Subtitle mode ("Sentence", "Sentence + Comma", "Line", etc.)
Returns: Returns:
Split pattern string Split pattern string
""" """
try:
lang = Language.from_str(language) if not isinstance(language, Language) else language
except ValueError:
lang = None # unknown language — treat as non-English, non-CJK
try: try:
mode = SubtitleMode.from_str(subtitle_mode) if not isinstance(subtitle_mode, SubtitleMode) else subtitle_mode mode = SubtitleMode.from_str(subtitle_mode) if not isinstance(subtitle_mode, SubtitleMode) else subtitle_mode
except ValueError: except ValueError:
mode = SubtitleMode.DISABLED mode = SubtitleMode.DISABLED
# For English, always use newline splitting only # For English, always use newline splitting only
if lang in (Language.EN_US, Language.EN_GB): if language in (Language.EN_US, Language.EN_GB):
return "\n" return "\n"
# Determine spacing pattern based on language # Determine spacing pattern based on language
spacing = r"\s*" if lang and lang.is_cjk else r"\s+" spacing = r"\s*" if language.is_cjk else r"\s+"
# For CJK languages, when subtitle mode is Disabled or Line, prefer # For CJK languages, when subtitle mode is Disabled or Line, prefer
# punctuation-based splitting instead of plain newline splitting. # punctuation-based splitting instead of plain newline splitting.
if mode in (SubtitleMode.DISABLED, SubtitleMode.LINE) and lang and lang.is_cjk: if mode in (SubtitleMode.DISABLED, SubtitleMode.LINE) and language.is_cjk:
return rf"(?<=[{PUNCTUATION_SENTENCE}]){spacing}|\n+" return rf"(?<=[{PUNCTUATION_SENTENCE}]){spacing}|\n+"
if mode == SubtitleMode.LINE: if mode == SubtitleMode.LINE:
abogen/domain/subtitle_generation.py
+6 -6
View File
@@ -19,7 +19,7 @@ def process_subtitle_tokens(
subtitle_entries: List[Tuple[float, float, str]], subtitle_entries: List[Tuple[float, float, str]],
max_subtitle_words: int, max_subtitle_words: int,
subtitle_mode: str, subtitle_mode: str,
lang_code: str, language: Language,
use_spacy_segmentation: bool = False, use_spacy_segmentation: bool = False,
fallback_end_time: Optional[float] = None, fallback_end_time: Optional[float] = None,
) -> None: ) -> None:
@@ -35,7 +35,7 @@ def process_subtitle_tokens(
max_subtitle_words: Maximum number of words per subtitle entry. max_subtitle_words: Maximum number of words per subtitle entry.
subtitle_mode: One of "Disabled", "Line", "Sentence", "Sentence + Comma", subtitle_mode: One of "Disabled", "Line", "Sentence", "Sentence + Comma",
"Sentence + Highlighting", or a string like "5" for word-count mode. "Sentence + Highlighting", or a string like "5" for word-count mode.
lang_code: Language code for spaCy processing (e.g., "a" for English). language: Language enum value for spaCy processing.
use_spacy_segmentation: Whether to use spaCy for sentence boundary detection. use_spacy_segmentation: Whether to use spaCy for sentence boundary detection.
fallback_end_time: Fallback end time for the last entry if none is available. fallback_end_time: Fallback end time for the last entry if none is available.
""" """
@@ -49,7 +49,7 @@ def process_subtitle_tokens(
use_spacy_for_english = ( use_spacy_for_english = (
use_spacy_segmentation use_spacy_segmentation
and subtitle_mode not in [SubtitleMode.DISABLED, SubtitleMode.LINE] and subtitle_mode not in [SubtitleMode.DISABLED, SubtitleMode.LINE]
and lang_code in [Language.EN_US, Language.EN_GB] and language in [Language.EN_US, Language.EN_GB]
and subtitle_mode in [SubtitleMode.SENTENCE, SubtitleMode.SENTENCE_COMMA] and subtitle_mode in [SubtitleMode.SENTENCE, SubtitleMode.SENTENCE_COMMA]
) )
@@ -61,7 +61,7 @@ def process_subtitle_tokens(
if use_spacy_for_english and subtitle_mode != SubtitleMode.LINE: if use_spacy_for_english and subtitle_mode != SubtitleMode.LINE:
_process_spacy_sentences( _process_spacy_sentences(
processed_tokens, subtitle_entries, max_subtitle_words, processed_tokens, subtitle_entries, max_subtitle_words,
subtitle_mode, lang_code, fallback_end_time subtitle_mode, language, fallback_end_time
) )
else: else:
_process_regex_sentences( _process_regex_sentences(
@@ -141,7 +141,7 @@ def _process_spacy_sentences(
subtitle_entries: List[Tuple[float, float, str]], subtitle_entries: List[Tuple[float, float, str]],
max_subtitle_words: int, max_subtitle_words: int,
subtitle_mode: str, subtitle_mode: str,
lang_code: str, language: Language,
fallback_end_time: Optional[float], fallback_end_time: Optional[float],
) -> None: ) -> None:
"""Process tokens using spaCy for sentence boundary detection.""" """Process tokens using spaCy for sentence boundary detection."""
@@ -155,7 +155,7 @@ def _process_spacy_sentences(
) )
return return
nlp = get_spacy_model(lang_code) nlp = get_spacy_model(language)
if not nlp: if not nlp:
_process_regex_sentences( _process_regex_sentences(
tokens, subtitle_entries, max_subtitle_words, tokens, subtitle_entries, max_subtitle_words,
abogen/spacy_utils.py
+14 -40
View File
@@ -21,20 +21,6 @@ SPACY_MODELS = {
Language.HI: "xx_sent_ud_sm", Language.HI: "xx_sent_ud_sm",
} }
# Kokoro single-letter codes -> Language enum (inverse of pipeline_factory._KOKORO_LANG_MAP)
_KOKORO_TO_LANGUAGE = {
"a": Language.EN_US,
"b": Language.EN_GB,
"e": Language.ES,
"f": Language.FR,
"h": Language.HI,
"i": Language.IT,
"j": Language.JA,
"p": Language.PT_BR,
"z": Language.ZH,
}
def _load_spacy(): def _load_spacy():
"""Lazy load spaCy module.""" """Lazy load spaCy module."""
global _spacy global _spacy
@@ -48,12 +34,12 @@ def _load_spacy():
return _spacy return _spacy
def get_spacy_model(lang_code, log_callback=None): def get_spacy_model(language: Language, log_callback=None):
""" """
Get or load a spaCy model for the given language code. Get or load a spaCy model for the given language.
Args: Args:
lang_code: Language code or Language enum (e.g., "a", "en-US", Language.EN_US) language: Language enum value.
log_callback: Optional function to log messages log_callback: Optional function to log messages
Returns: Returns:
@@ -61,36 +47,24 @@ def get_spacy_model(lang_code, log_callback=None):
""" """
def log(msg, is_error=False): def log(msg, is_error=False):
# Prefer GUI log callback when provided to avoid spamming stdout.
if log_callback: if log_callback:
color = "red" if is_error else "grey" color = "red" if is_error else "grey"
try: try:
log_callback((msg, color)) log_callback((msg, color))
except Exception: except Exception:
# Fallback to printing if callback misbehaves
print(msg) print(msg)
else: else:
print(msg) print(msg)
# Normalize to Language enum if not isinstance(language, Language):
if not isinstance(lang_code, Language): raise TypeError(f"language must be Language enum, got {type(language).__name__}: {language!r}")
if isinstance(lang_code, str) and lang_code in _KOKORO_TO_LANGUAGE:
lang_code = _KOKORO_TO_LANGUAGE[lang_code]
else:
try:
lang_code = Language.from_str(lang_code)
except ValueError:
log(f"\nspaCy: Unknown language '{lang_code}'...")
return None
# Check if model is cached if language in _nlp_cache:
if lang_code in _nlp_cache: return _nlp_cache[language]
return _nlp_cache[lang_code]
# Check if language is supported model_name = SPACY_MODELS.get(language)
model_name = SPACY_MODELS.get(lang_code)
if not model_name: if not model_name:
log(f"\nspaCy: No model mapping for language '{lang_code}'...") log(f"\nspaCy: No model mapping for language '{language}'...")
return None return None
# Lazy load spaCy # Lazy load spaCy
@@ -114,7 +88,7 @@ def get_spacy_model(lang_code, log_callback=None):
if "parser" not in nlp.pipe_names and "sentencizer" not in nlp.pipe_names: if "parser" not in nlp.pipe_names and "sentencizer" not in nlp.pipe_names:
nlp.add_pipe("sentencizer") nlp.add_pipe("sentencizer")
_nlp_cache[lang_code] = nlp _nlp_cache[language] = nlp
return nlp return nlp
except OSError: except OSError:
# Model not found, attempt download # Model not found, attempt download
@@ -131,7 +105,7 @@ def get_spacy_model(lang_code, log_callback=None):
if "parser" not in nlp.pipe_names and "sentencizer" not in nlp.pipe_names: if "parser" not in nlp.pipe_names and "sentencizer" not in nlp.pipe_names:
nlp.add_pipe("sentencizer") nlp.add_pipe("sentencizer")
_nlp_cache[lang_code] = nlp _nlp_cache[language] = nlp
log(f"spaCy model '{model_name}' downloaded and loaded") log(f"spaCy model '{model_name}' downloaded and loaded")
return nlp return nlp
except Exception as e: except Exception as e:
@@ -145,19 +119,19 @@ def get_spacy_model(lang_code, log_callback=None):
return None return None
def segment_sentences(text, lang_code, log_callback=None): def segment_sentences(text, language: Language, log_callback=None):
""" """
Segment text into sentences using spaCy. Segment text into sentences using spaCy.
Args: Args:
text: Text to segment text: Text to segment
lang_code: Language code language: Language enum value
log_callback: Optional function to log messages log_callback: Optional function to log messages
Returns: Returns:
List of sentence strings, or None if spaCy unavailable List of sentence strings, or None if spaCy unavailable
""" """
nlp = get_spacy_model(lang_code, log_callback) nlp = get_spacy_model(language, log_callback)
if nlp is None: if nlp is None:
return None return None
abogen/speaker_configs.py
+2 -2
View File
@@ -4,7 +4,7 @@ import json
import os import os
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
from abogen.constants import LANGUAGE_DESCRIPTIONS from abogen.constants import KOKORO_CODE_LABELS
from abogen.utils import get_user_config_path from abogen.utils import get_user_config_path
_CONFIG_WRAPPER_KEY = "abogen_speaker_configs" _CONFIG_WRAPPER_KEY = "abogen_speaker_configs"
@@ -163,4 +163,4 @@ def list_configs() -> List[Dict[str, Any]]:
def describe_language(code: str) -> str: def describe_language(code: str) -> str:
code = (code or "a").lower() code = (code or "a").lower()
return LANGUAGE_DESCRIPTIONS.get(code, code.upper()) return KOKORO_CODE_LABELS.get(code, code.upper())
abogen/subtitle_utils.py
+13 -2
View File
@@ -1,6 +1,7 @@
import re import re
from abogen.utils import detect_encoding, load_config from abogen.utils import detect_encoding, load_config
from abogen.constants import SAMPLE_VOICE_TEXTS from abogen.constants import SAMPLE_VOICE_TEXTS
from abogen.domain.enums import Language
# Pre-compile frequently used regex patterns for better performance # Pre-compile frequently used regex patterns for better performance
_METADATA_TAG_PATTERN = re.compile(r"<<METADATA_[^:]+:[^>]*>>") _METADATA_TAG_PATTERN = re.compile(r"<<METADATA_[^:]+:[^>]*>>")
@@ -388,8 +389,18 @@ def parse_ass_file(file_path):
return subtitles return subtitles
def get_sample_voice_text(lang_code): def get_sample_voice_text(language):
return SAMPLE_VOICE_TEXTS.get(lang_code, SAMPLE_VOICE_TEXTS["a"]) """Get sample voice text for a language.
Args:
language: Language enum value or string (for backward compatibility).
"""
if isinstance(language, str):
try:
language = Language.from_str(language)
except (ValueError, AttributeError):
language = Language.EN_US
return SAMPLE_VOICE_TEXTS.get(language, SAMPLE_VOICE_TEXTS[Language.EN_US])
# Backward-compatible re-exports — canonical location is domain/output_paths.py # Backward-compatible re-exports — canonical location is domain/output_paths.py
abogen/tts_plugin/plugin_manager.py
+1 -1
View File
@@ -8,7 +8,7 @@ Usage:
from abogen.tts_plugin.plugin_manager import get_plugin_manager from abogen.tts_plugin.plugin_manager import get_plugin_manager
manager = get_plugin_manager() manager = get_plugin_manager()
engine = manager.create_engine("kokoro", lang_code="a", device="cpu") engine = manager.create_engine("kokoro", language=Language.EN_US, device="cpu")
session = engine.create_session() session = engine.create_session()
try: try:
result = session.synthesize("Hello world") result = session.synthesize("Hello world")
abogen/tts_plugin/types.py
+5 -3
View File
@@ -9,6 +9,8 @@ from __future__ import annotations
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any, Mapping from typing import Any, Mapping
from abogen.domain.enums import Language
@dataclass(frozen=True) @dataclass(frozen=True)
class AudioFormat: class AudioFormat:
@@ -103,9 +105,9 @@ class EngineConfig:
Attributes: Attributes:
device: Device to use (e.g., "cpu", "cuda:0"). device: Device to use (e.g., "cpu", "cuda:0").
lang_code: Language code for the engine (e.g., "a" for Kokoro English). language: Language enum value. The engine converts to its internal
Plugins that do not require a language code ignore this field. format internally — callers never see engine-specific codes.
""" """
device: str = "cpu" device: str = "cpu"
lang_code: str = "a" language: Language = Language.EN_US
abogen/tts_plugin/utils.py
+5 -4
View File
@@ -10,6 +10,7 @@ from typing import Any, Iterator
import numpy as np import numpy as np
from abogen.domain.enums import Language
from abogen.tts_plugin.plugin_manager import get_plugin_manager from abogen.tts_plugin.plugin_manager import get_plugin_manager
@@ -123,7 +124,7 @@ class Pipeline:
Presents the same interface that old callers expect:: Presents the same interface that old callers expect::
pipeline = create_pipeline("kokoro", lang_code="a", device="cpu") pipeline = create_pipeline("kokoro", language=Language.EN_US, device="cpu")
for segment in pipeline(text, voice="af_nova", speed=1.0): for segment in pipeline(text, voice="af_nova", speed=1.0):
audio = segment.audio audio = segment.audio
""" """
@@ -200,7 +201,7 @@ class Pipeline:
def create_pipeline( def create_pipeline(
plugin_id: str, plugin_id: str,
*, *,
lang_code: str = "a", language: Language = Language.EN_US,
device: str = "cpu", device: str = "cpu",
) -> Pipeline: ) -> Pipeline:
"""Create a callable TTS pipeline via the Plugin Architecture. """Create a callable TTS pipeline via the Plugin Architecture.
@@ -211,7 +212,7 @@ def create_pipeline(
Args: Args:
plugin_id: Plugin identifier (e.g., "kokoro", "supertonic"). plugin_id: Plugin identifier (e.g., "kokoro", "supertonic").
lang_code: Language code for the engine. language: Language enum value (app-layer type, not engine-specific).
device: Device to use (e.g., "cpu", "cuda:0"). device: Device to use (e.g., "cpu", "cuda:0").
Returns: Returns:
@@ -235,7 +236,7 @@ def create_pipeline(
})(), })(),
) )
config = EngineConfig(device=device, lang_code=lang_code) config = EngineConfig(device=device, language=language)
engine = manager.create_engine(plugin_id, context=ctx, model_path=None, config=config) engine = manager.create_engine(plugin_id, context=ctx, model_path=None, config=config)
return Pipeline(engine) return Pipeline(engine)
abogen/webui/debug_tts_runner.py
+16 -28
View File
@@ -14,9 +14,12 @@ from abogen.kokoro_text_normalization import normalize_for_pipeline
from abogen.normalization_settings import build_apostrophe_config from abogen.normalization_settings import build_apostrophe_config
from abogen.text_extractor import extract_from_path from abogen.text_extractor import extract_from_path
from abogen.voice_cache import ensure_voice_assets from abogen.voice_cache import ensure_voice_assets
from abogen.webui.conversion_runner import SAMPLE_RATE, _select_device, _to_float32, _spec_to_voice_ids from abogen.domain.device import select_device as _select_device
from abogen.domain.audio_helpers import to_float32 as _to_float32, SAMPLE_RATE
from abogen.domain.voice_resolution import spec_to_voice_ids as _spec_to_voice_ids
from abogen.domain.voice_loader import resolve_voice from abogen.domain.voice_loader import resolve_voice
from abogen.domain.split_pattern import get_split_pattern from abogen.domain.split_pattern import get_split_pattern
from abogen.domain.enums import Language
from abogen.tts_plugin.utils import create_pipeline from abogen.tts_plugin.utils import create_pipeline
@@ -43,11 +46,11 @@ def _resolve_voice_setting(value: str) -> tuple[str, Optional[str], Optional[str
return resolve_voice_setting(value) return resolve_voice_setting(value)
def _load_pipeline(language: str, use_gpu: bool) -> Any: def _load_pipeline(language: Language, use_gpu: bool) -> Any:
device = "cpu" device = "cpu"
if use_gpu: if use_gpu:
device = _select_device() device = _select_device()
return create_pipeline("kokoro", lang_code=language, device=device) return create_pipeline("kokoro", language=language, device=device)
def _extract_cases_from_text(text: str) -> List[Tuple[str, str]]: def _extract_cases_from_text(text: str) -> List[Tuple[str, str]]:
@@ -127,32 +130,14 @@ def run_debug_tts_wavs(
if missing: if missing:
raise RuntimeError(f"Debug EPUB missing expected codes: {', '.join(missing)}") raise RuntimeError(f"Debug EPUB missing expected codes: {', '.join(missing)}")
language = str(settings.get("language") or "a").strip() or "a" raw_language = str(settings.get("language") or "en-US").strip() or "en-US"
# Kokoro's KPipeline expects short language codes like "a" (American English), try:
# but older settings may store ISO-like values such as "en". language = Language.from_str(raw_language)
language_aliases = { except ValueError:
"en": "a", language = Language.EN_US
"en-us": "a",
"en_us": "a",
"en-gb": "b",
"en_gb": "b",
"es": "e",
"es-es": "e",
"fr": "f",
"fr-fr": "f",
"hi": "h",
"it": "i",
"pt": "p",
"pt-br": "p",
"ja": "j",
"jp": "j",
"zh": "z",
"zh-cn": "z",
}
language = language_aliases.get(language.lower(), language)
voice_spec = str(settings.get("default_voice") or "").strip() voice_spec = str(settings.get("default_voice") or "").strip()
use_gpu = bool(settings.get("use_gpu", False)) use_gpu = bool(settings.get("use_gpu", False))
speed = float(settings.get("default_speed", 1.0) or 1.0) speed = float(settings.get("default_speed") or 1.0)
# Settings may store "profile:<name>" which is not a Kokoro voice ID. # Settings may store "profile:<name>" which is not a Kokoro voice ID.
# Resolve it to a concrete voice formula (e.g. "af_heart*0.5+...") so Kokoro # Resolve it to a concrete voice formula (e.g. "af_heart*0.5+...") so Kokoro
@@ -162,7 +147,10 @@ def run_debug_tts_wavs(
if resolved_voice: if resolved_voice:
voice_spec = resolved_voice voice_spec = resolved_voice
if profile_language: if profile_language:
language = str(profile_language).strip() or language try:
language = Language.from_str(str(profile_language).strip()) or language
except (ValueError, AttributeError):
pass
except Exception: except Exception:
# Voice profile resolution is best-effort; fall back to raw voice_spec. # Voice profile resolution is best-effort; fall back to raw voice_spec.
pass pass
abogen/webui/routes/api.py
+20 -4
View File
@@ -6,6 +6,7 @@ from pathlib import Path
from flask import Blueprint, request, jsonify, send_file, url_for, current_app from flask import Blueprint, request, jsonify, send_file, url_for, current_app
from flask.typing import ResponseReturnValue from flask.typing import ResponseReturnValue
from abogen.domain.enums import Language
from abogen.webui.routes.utils.settings import ( from abogen.webui.routes.utils.settings import (
load_settings, load_settings,
load_integration_settings, load_integration_settings,
@@ -47,6 +48,21 @@ from werkzeug.utils import secure_filename
api_bp = Blueprint("api", __name__) api_bp = Blueprint("api", __name__)
def _parse_language(value: Any) -> Language:
"""Parse a frontend language value to Language enum.
This is the API boundary — frontend sends strings, backend parses
to Language enum. No engine-specific codes leak outside the engine.
"""
if isinstance(value, Language):
return value
try:
return Language.from_str(str(value or "").strip())
except (ValueError, AttributeError):
return Language.EN_US
# --- Voice Profile Routes --- # --- Voice Profile Routes ---
@api_bp.get("/voice-profiles") @api_bp.get("/voice-profiles")
@@ -152,7 +168,7 @@ def api_export_voice_profiles() -> ResponseReturnValue:
def api_voice_profiles_preview() -> ResponseReturnValue: def api_voice_profiles_preview() -> ResponseReturnValue:
payload = request.get_json(force=True, silent=True) or {} payload = request.get_json(force=True, silent=True) or {}
text = str(payload.get("text") or "").strip() or "Hello world" text = str(payload.get("text") or "").strip() or "Hello world"
language = str(payload.get("language") or "a").strip().lower() or "a" language = _parse_language(payload.get("language"))
speed = coerce_float(payload.get("speed"), 1.0) speed = coerce_float(payload.get("speed"), 1.0)
max_seconds = coerce_float(payload.get("max_seconds"), 8.0) max_seconds = coerce_float(payload.get("max_seconds"), 8.0)
@@ -186,7 +202,7 @@ def api_voice_profiles_preview() -> ResponseReturnValue:
speed = float(normalized_entry.get("speed") or speed) speed = float(normalized_entry.get("speed") or speed)
else: else:
voice_spec = formula_from_profile(normalized_entry) or "" voice_spec = formula_from_profile(normalized_entry) or ""
language = str(normalized_entry.get("language") or language) language = _parse_language(normalized_entry.get("language") or language)
elif formula: elif formula:
voice_spec = formula voice_spec = formula
resolved_provider = "kokoro" resolved_provider = "kokoro"
@@ -221,7 +237,7 @@ def api_speaker_preview() -> ResponseReturnValue:
pending_id = str(payload.get("pending_id") or "").strip() pending_id = str(payload.get("pending_id") or "").strip()
text = payload.get("text", "Hello world") text = payload.get("text", "Hello world")
voice = payload.get("voice", "af_heart") voice = payload.get("voice", "af_heart")
language = payload.get("language", "a") language = _parse_language(payload.get("language"))
speed_value = payload.get("speed") speed_value = payload.get("speed")
speed = coerce_float(speed_value, 1.0) speed = coerce_float(speed_value, 1.0)
tts_provider = str(payload.get("tts_provider") or "").strip().lower() tts_provider = str(payload.get("tts_provider") or "").strip().lower()
@@ -576,7 +592,7 @@ def api_entity_pronunciation_preview() -> ResponseReturnValue:
token = payload.get("token", "").strip() token = payload.get("token", "").strip()
pronunciation = payload.get("pronunciation", "").strip() pronunciation = payload.get("pronunciation", "").strip()
voice = payload.get("voice", "").strip() voice = payload.get("voice", "").strip()
language = payload.get("language", "a").strip() language = _parse_language(payload.get("language"))
if not token and not pronunciation: if not token and not pronunciation:
return jsonify({"error": "Token or pronunciation required"}), 400 return jsonify({"error": "Token or pronunciation required"}), 400
abogen/webui/routes/utils/synthesize.py
+7 -27
View File
@@ -15,19 +15,6 @@ from abogen.domain.pronunciation import (
apply_pronunciation_rules, apply_pronunciation_rules,
) )
# Kokoro-specific language mapping (engine's responsibility)
_KOKORO_LANG_MAP = {
Language.EN_US: "a",
Language.EN_GB: "b",
Language.ES: "e",
Language.FR: "f",
Language.HI: "h",
Language.IT: "i",
Language.JA: "j",
Language.PT_BR: "p",
Language.ZH: "z",
}
SAMPLE_RATE = 24000 SAMPLE_RATE = 24000
@@ -46,7 +33,7 @@ def clear_preview_pipelines() -> None:
_preview_pipelines.clear() _preview_pipelines.clear()
def _resolve_pipeline(language: str, use_gpu: bool) -> Tuple[Any, bool]: def _resolve_pipeline(language: Language, use_gpu: bool) -> Tuple[Any, bool]:
devices: List[str] = ["cpu"] devices: List[str] = ["cpu"]
if use_gpu: if use_gpu:
preferred = _select_device() preferred = _select_device()
@@ -63,29 +50,22 @@ def _resolve_pipeline(language: str, use_gpu: bool) -> Tuple[Any, bool]:
raise RuntimeError("Preview pipeline is unavailable") from last_error raise RuntimeError("Preview pipeline is unavailable") from last_error
def get_preview_pipeline(language: str, device: str) -> Any: def get_preview_pipeline(language: Language, device: str) -> Any:
# Convert Language enum to Kokoro single-letter code key = (language, device)
try:
lang = Language.from_str(language) if not isinstance(language, Language) else language
except ValueError:
lang = Language.EN_US
kokoro_code = _KOKORO_LANG_MAP.get(lang, "a")
key = (kokoro_code, device)
with _preview_pipeline_lock: with _preview_pipeline_lock:
pipeline = _preview_pipelines.get(key) pipeline = _preview_pipelines.get(key)
if pipeline is not None: if pipeline is not None:
return pipeline return pipeline
from abogen.tts_plugin.utils import create_pipeline from abogen.tts_plugin.utils import create_pipeline
pipeline = create_pipeline("kokoro", lang_code=kokoro_code, device=device) pipeline = create_pipeline("kokoro", language=lang, device=device)
_preview_pipelines[key] = pipeline _preview_pipelines[key] = pipeline
return pipeline return pipeline
def generate_preview_audio( def generate_preview_audio(
text: str, text: str,
voice_spec: str, voice_spec: str,
language: str, language: Language,
speed: float, speed: float,
use_gpu: bool, use_gpu: bool,
tts_provider: str = "kokoro", tts_provider: str = "kokoro",
@@ -131,7 +111,7 @@ def generate_preview_audio(
current_app.logger.exception("Preview normalization failed; using raw text") current_app.logger.exception("Preview normalization failed; using raw text")
normalized_text = source_text normalized_text = source_text
preview_split = get_split_pattern(str(language or "a"), "Disabled") preview_split = get_split_pattern(language, "Disabled")
if provider == "supertonic": if provider == "supertonic":
from abogen.tts_plugin.utils import create_pipeline from abogen.tts_plugin.utils import create_pipeline
@@ -194,7 +174,7 @@ def generate_preview_audio(
def synthesize_preview( def synthesize_preview(
text: str, text: str,
voice_spec: str, voice_spec: str,
language: str, language: Language,
speed: float, speed: float,
use_gpu: bool, use_gpu: bool,
tts_provider: str = "kokoro", tts_provider: str = "kokoro",
abogen/webui/routes/utils/voice.py
+5 -3
View File
@@ -277,17 +277,19 @@ def filter_voice_catalog(
def build_voice_catalog() -> List[Dict[str, str]]: def build_voice_catalog() -> List[Dict[str, str]]:
from plugins.kokoro.engine import language_for_voice_id
catalog: List[Dict[str, str]] = [] catalog: List[Dict[str, str]] = []
gender_map = {"f": "Female", "m": "Male"} gender_map = {"f": "Female", "m": "Male"}
for voice_id in get_voices("kokoro"): for voice_id in get_voices("kokoro"):
prefix, _, rest = voice_id.partition("_") prefix, _, rest = voice_id.partition("_")
language_code = prefix[0] if prefix else "a"
gender_code = prefix[1] if len(prefix) > 1 else "" gender_code = prefix[1] if len(prefix) > 1 else ""
lang = language_for_voice_id(voice_id)
catalog.append( catalog.append(
{ {
"id": voice_id, "id": voice_id,
"language": language_code, "language": lang.value,
"language_label": LANGUAGE_DESCRIPTIONS.get(language_code, language_code.upper()), "language_label": LANGUAGE_DESCRIPTIONS.get(lang, lang.value.upper()),
"gender": gender_map.get(gender_code, "Unknown"), "gender": gender_map.get(gender_code, "Unknown"),
"gender_code": gender_code, "gender_code": gender_code,
"display_name": rest.replace("_", " ").title() if rest else voice_id, "display_name": rest.replace("_", " ").title() if rest else voice_id,
abogen/webui/routes/voices.py
+2 -1
View File
@@ -2,6 +2,7 @@ from typing import Any, Dict, List, Optional
from flask import Blueprint, render_template, request, jsonify, abort, flash, redirect, url_for from flask import Blueprint, render_template, request, jsonify, abort, flash, redirect, url_for
from flask.typing import ResponseReturnValue from flask.typing import ResponseReturnValue
from abogen.domain.enums import Language
from abogen.webui.routes.utils.voice import ( from abogen.webui.routes.utils.voice import (
template_options, template_options,
resolve_voice_setting, resolve_voice_setting,
@@ -39,7 +40,7 @@ def test_voice() -> ResponseReturnValue:
return synthesize_preview( return synthesize_preview(
text=text, text=text,
voice_spec=voice, voice_spec=voice,
language="a", # Default language language=Language.EN_US,
speed=speed, speed=speed,
use_gpu=use_gpu, use_gpu=use_gpu,
) )
abogen/webui/service.py
+14 -5
View File
@@ -14,6 +14,7 @@ from enum import Enum
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Optional, Mapping from typing import Any, Callable, Dict, Iterable, List, Optional, Mapping
from abogen.domain.enums import Language
from abogen.utils import get_internal_cache_path, get_user_settings_dir, load_config from abogen.utils import get_internal_cache_path, get_user_settings_dir, load_config
from abogen.voice_cache import bootstrap_voice_cache from abogen.voice_cache import bootstrap_voice_cache
from abogen.integrations.audiobookshelf import ( from abogen.integrations.audiobookshelf import (
@@ -105,7 +106,7 @@ class Job:
id: str id: str
original_filename: str original_filename: str
stored_path: Path stored_path: Path
language: str language: Language
voice: str voice: str
speed: float speed: float
use_gpu: bool use_gpu: bool
@@ -296,7 +297,7 @@ class PendingJob:
id: str id: str
original_filename: str original_filename: str
stored_path: Path stored_path: Path
language: str language: Language
voice: str voice: str
speed: float speed: float
use_gpu: bool use_gpu: bool
@@ -384,7 +385,7 @@ class ConversionService:
*, *,
original_filename: str, original_filename: str,
stored_path: Path, stored_path: Path,
language: str, language: Language,
voice: str, voice: str,
speed: float, speed: float,
tts_provider: str = "kokoro", tts_provider: str = "kokoro",
@@ -910,7 +911,7 @@ class ConversionService:
"id": job.id, "id": job.id,
"original_filename": job.original_filename, "original_filename": job.original_filename,
"stored_path": str(job.stored_path), "stored_path": str(job.stored_path),
"language": job.language, "language": job.language.value if isinstance(job.language, Language) else str(job.language),
"tts_provider": getattr(job, "tts_provider", "kokoro"), "tts_provider": getattr(job, "tts_provider", "kokoro"),
"voice": job.voice, "voice": job.voice,
"speed": job.speed, "speed": job.speed,
@@ -1026,11 +1027,19 @@ class ConversionService:
stored_path = Path(payload["stored_path"]) stored_path = Path(payload["stored_path"])
output_folder_raw = payload.get("output_folder") output_folder_raw = payload.get("output_folder")
output_folder = Path(output_folder_raw) if output_folder_raw else None output_folder = Path(output_folder_raw) if output_folder_raw else None
raw_lang = payload.get("language", "")
if isinstance(raw_lang, Language):
language = raw_lang
else:
try:
language = Language.from_str(str(raw_lang or "").strip())
except (ValueError, AttributeError):
language = Language.EN_US
job = Job( job = Job(
id=payload["id"], id=payload["id"],
original_filename=payload["original_filename"], original_filename=payload["original_filename"],
stored_path=stored_path, stored_path=stored_path,
language=payload.get("language", "a"), language=language,
tts_provider=str(payload.get("tts_provider") or "kokoro"), tts_provider=str(payload.get("tts_provider") or "kokoro"),
voice=payload.get("voice", ""), voice=payload.get("voice", ""),
speed=float(payload.get("speed", 1.0)), speed=float(payload.get("speed", 1.0)),
plugins/kokoro/__init__.py
+3 -1
View File
@@ -165,6 +165,7 @@ def create_engine(
""" """
try: try:
KPipeline = _load_kpipeline() KPipeline = _load_kpipeline()
from plugins.kokoro.engine import engine_language
# Determine repo_id from model_path or use default # Determine repo_id from model_path or use default
repo_id = "hexgrad/Kokoro-82M" repo_id = "hexgrad/Kokoro-82M"
@@ -172,8 +173,9 @@ def create_engine(
# If a specific model path is provided, use it as repo_id # If a specific model path is provided, use it as repo_id
repo_id = str(model_path) repo_id = str(model_path)
kokoro_code = engine_language(config.language)
pipeline = KPipeline( pipeline = KPipeline(
lang_code=config.lang_code, lang_code=kokoro_code,
repo_id=repo_id, repo_id=repo_id,
device=config.device, device=config.device,
) )
plugins/kokoro/engine.py
+58
View File
@@ -2,6 +2,10 @@
This module adapts the existing Kokoro backend to the new Engine/EngineSession This module adapts the existing Kokoro backend to the new Engine/EngineSession
protocol. It wraps the KokoroBackend without modifying it. protocol. It wraps the KokoroBackend without modifying it.
Language mapping: this is the engine's responsibility. The engine knows
which languages it supports and converts Language enum → internal format.
Callers outside this module never see engine-specific codes.
""" """
from __future__ import annotations from __future__ import annotations
@@ -11,6 +15,7 @@ from typing import Any
import numpy as np import numpy as np
from abogen.domain.enums import Language
from abogen.tts_plugin.capabilities import VoiceLister from abogen.tts_plugin.capabilities import VoiceLister
from abogen.tts_plugin.engine import Engine, EngineSession from abogen.tts_plugin.engine import Engine, EngineSession
from abogen.tts_plugin.errors import EngineError from abogen.tts_plugin.errors import EngineError
@@ -27,6 +32,59 @@ logger = logging.getLogger(__name__)
# Sample rate for Kokoro audio # Sample rate for Kokoro audio
_KOKORO_SAMPLE_RATE = 24000 _KOKORO_SAMPLE_RATE = 24000
# Engine-internal language mapping: Language enum → kokoro code.
# ONLY visible inside this module — callers never see kokoro codes.
_KOKORO_LANG_MAP: dict[Language, str] = {
Language.EN_US: "a",
Language.EN_GB: "b",
Language.ES: "e",
Language.FR: "f",
Language.HI: "h",
Language.IT: "i",
Language.JA: "j",
Language.PT_BR: "p",
Language.ZH: "z",
}
# Reverse mapping: engine-internal code → Language enum.
# Used by voice catalog and other places that need to convert
# engine codes back to Language enum (e.g. voice ID prefix extraction).
_CODE_TO_LANGUAGE: dict[str, Language] = {v: k for k, v in _KOKORO_LANG_MAP.items()}
def supported_languages() -> list[Language]:
"""Return the list of Language enum values this engine supports.
This is the engine's responsibility — the engine knows which
languages it supports and exposes them as Language enum values.
UI layers query this to populate language selectors.
"""
return list(_KOKORO_LANG_MAP.keys())
def engine_language(lang: Language) -> str:
"""Map a Language enum to the engine's internal code.
This is the engine's responsibility — the engine owns the mapping
between Language enum and its internal format. Callers pass Language
enum; the engine converts internally. The returned string is ONLY
used inside the engine implementation.
"""
return _KOKORO_LANG_MAP.get(lang, "a")
def language_for_voice_id(voice_id: str) -> Language:
"""Determine which Language a voice belongs to from its voice ID.
Kokoro voice IDs encode language as a prefix (e.g. "af_heart" → "a" → EN_US).
This is kokoro-specific knowledge that stays inside the engine.
Callers pass a voice ID string; the engine returns a Language enum.
"""
prefix = str(voice_id or "").strip()[:1].lower()
if prefix in _CODE_TO_LANGUAGE:
return _CODE_TO_LANGUAGE[prefix]
return Language.EN_US
class KokoroSession: class KokoroSession:
"""EngineSession implementation for Kokoro. """EngineSession implementation for Kokoro.
tests/contracts/test_integration.py
+3 -1
View File
@@ -14,6 +14,8 @@ from unittest.mock import MagicMock, patch
import numpy as np import numpy as np
from abogen.domain.enums import Language
from abogen.tts_plugin.engine import Engine, EngineSession from abogen.tts_plugin.engine import Engine, EngineSession
from abogen.tts_plugin.errors import EngineError from abogen.tts_plugin.errors import EngineError
from abogen.tts_plugin.plugin_manager import PluginManager, get_plugin_manager, reset_plugin_manager from abogen.tts_plugin.plugin_manager import PluginManager, get_plugin_manager, reset_plugin_manager
@@ -328,7 +330,7 @@ class TestRegression:
manager._loaded = True manager._loaded = True
with patch("abogen.tts_plugin.utils.get_plugin_manager", return_value=manager): with patch("abogen.tts_plugin.utils.get_plugin_manager", return_value=manager):
backend = create_pipeline("mock_tts", lang_code="a", device="cpu") backend = create_pipeline("mock_tts", language=Language.EN_US, device="cpu")
# Old interface: pipeline(text, voice=..., speed=..., split_pattern=...) # Old interface: pipeline(text, voice=..., speed=..., split_pattern=...)
segments = list(backend( segments = list(backend(
tests/contracts/test_plugin_manager_contract.py
+3 -2
View File
@@ -3,6 +3,7 @@
import pytest import pytest
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from abogen.domain.enums import Language
from abogen.tts_plugin.plugin_manager import PluginManager, get_plugin_manager, reset_plugin_manager from abogen.tts_plugin.plugin_manager import PluginManager, get_plugin_manager, reset_plugin_manager
from abogen.tts_plugin.utils import Pipeline, create_pipeline from abogen.tts_plugin.utils import Pipeline, create_pipeline
from abogen.tts_plugin.engine import Engine, EngineSession from abogen.tts_plugin.engine import Engine, EngineSession
@@ -175,7 +176,7 @@ class TestCreatePipelineCompat:
mock_engine = FakeEngine() mock_engine = FakeEngine()
mock_manager.create_engine.return_value = mock_engine mock_manager.create_engine.return_value = mock_engine
backend = create_pipeline("kokoro", lang_code="a", device="cpu") backend = create_pipeline("kokoro", language=Language.EN_US, device="cpu")
assert callable(backend) assert callable(backend)
mock_manager.create_engine.assert_called_once() mock_manager.create_engine.assert_called_once()
@@ -185,7 +186,7 @@ class TestCreatePipelineCompat:
assert call_args.kwargs["model_path"] is None assert call_args.kwargs["model_path"] is None
assert isinstance(call_args.kwargs["config"], EngineConfig) assert isinstance(call_args.kwargs["config"], EngineConfig)
assert call_args.kwargs["config"].device == "cpu" assert call_args.kwargs["config"].device == "cpu"
assert call_args.kwargs["config"].lang_code == "a" assert call_args.kwargs["config"].language == Language.EN_US
def test_create_pipeline_raises_for_unknown_plugin(self): def test_create_pipeline_raises_for_unknown_plugin(self):
"""create_pipeline raises KeyError for unknown plugins.""" """create_pipeline raises KeyError for unknown plugins."""
tests/contracts/test_types_contract.py
+13 -12
View File
@@ -8,6 +8,7 @@ These tests verify that value objects satisfy the architectural requirements:
import pytest import pytest
from abogen.domain.enums import Language
from abogen.tts_plugin.types import ( from abogen.tts_plugin.types import (
AudioFormat, AudioFormat,
Duration, Duration,
@@ -192,23 +193,23 @@ class TestEngineConfigContract:
config = EngineConfig(device="cuda:0") config = EngineConfig(device="cuda:0")
assert config.device == "cuda:0" assert config.device == "cuda:0"
def test_default_lang_code(self) -> None: def test_default_language(self) -> None:
config = EngineConfig() config = EngineConfig()
assert config.lang_code == "a" assert config.language == Language.EN_US
def test_custom_lang_code(self) -> None: def test_custom_language(self) -> None:
config = EngineConfig(lang_code="j") config = EngineConfig(language=Language.JA)
assert config.lang_code == "j" assert config.language == Language.JA
def test_immutability(self) -> None: def test_immutability(self) -> None:
config = EngineConfig() config = EngineConfig()
with pytest.raises(AttributeError): with pytest.raises(AttributeError):
config.device = "cuda:0" # type: ignore[misc] config.device = "cuda:0" # type: ignore[misc]
def test_immutability_lang_code(self) -> None: def test_immutability_language(self) -> None:
config = EngineConfig() config = EngineConfig()
with pytest.raises(AttributeError): with pytest.raises(AttributeError):
config.lang_code = "j" # type: ignore[misc] config.language = Language.JA # type: ignore[misc]
def test_unknown_keys_ignored_per_spec(self) -> None: def test_unknown_keys_ignored_per_spec(self) -> None:
"""Architecture spec: Unknown keys are ignored (no error). """Architecture spec: Unknown keys are ignored (no error).
@@ -225,11 +226,11 @@ class TestEngineConfigContract:
EngineConfig may contain fields that are not relevant to every plugin. EngineConfig may contain fields that are not relevant to every plugin.
Plugins MUST ignore fields they do not need, not raise on them. Plugins MUST ignore fields they do not need, not raise on them.
""" """
config = EngineConfig(device="cuda:0", lang_code="j") config = EngineConfig(device="cuda:0", language=Language.JA)
assert config.device == "cuda:0" assert config.device == "cuda:0"
assert config.lang_code == "j" assert config.language == Language.JA
# A plugin that only needs device simply reads config.device # A plugin that only needs device simply reads config.device
# and ignores config.lang_code — this must not raise. # and ignores config.language — this must not raise.
def test_engine_config_contains_engine_instance_configuration(self) -> None: def test_engine_config_contains_engine_instance_configuration(self) -> None:
"""Architecture Amendment #1: EngineConfig definition. """Architecture Amendment #1: EngineConfig definition.
@@ -238,7 +239,7 @@ class TestEngineConfigContract:
Engine instance is created and that remain constant throughout Engine instance is created and that remain constant throughout
the lifetime of that Engine. the lifetime of that Engine.
""" """
config = EngineConfig(device="cpu", lang_code="a") config = EngineConfig(device="cpu", language=Language.EN_US)
# Both fields are init-time, immutable, engine-scoped. # Both fields are init-time, immutable, engine-scoped.
assert config.device == "cpu" assert config.device == "cpu"
assert config.lang_code == "a" assert config.language == Language.EN_US
tests/test_behavioral_regression.py
+2 -1
View File
@@ -928,10 +928,11 @@ class TestValueObjectsBehavioral:
def test_engine_config_defaults(self) -> None: def test_engine_config_defaults(self) -> None:
from abogen.tts_plugin.types import EngineConfig from abogen.tts_plugin.types import EngineConfig
from abogen.domain.enums import Language
config = EngineConfig() config = EngineConfig()
assert config.device == "cpu" assert config.device == "cpu"
assert config.lang_code == "a" assert config.language == Language.EN_US
def test_parameter_values_defaults(self) -> None: def test_parameter_values_defaults(self) -> None:
pv = ParameterValues() pv = ParameterValues()
tests/test_conversion_executor.py
+7 -6
View File
@@ -14,6 +14,7 @@ from unittest.mock import MagicMock, patch
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any from typing import Any
from abogen.domain.enums import Language
from abogen.domain.conversion_engine import ( from abogen.domain.conversion_engine import (
synthesize_text, synthesize_text,
SynthParams, SynthParams,
@@ -254,7 +255,7 @@ class TestProcessAndWriteSubtitles:
writer, writer,
subtitle_mode="Sentence", subtitle_mode="Sentence",
max_subtitle_words=5, max_subtitle_words=5,
lang_code="a", language=Language.EN_US,
use_spacy_segmentation=False, use_spacy_segmentation=False,
fallback_end_time=10.0, fallback_end_time=10.0,
) )
@@ -271,7 +272,7 @@ class TestProcessAndWriteSubtitles:
writer, writer,
subtitle_mode="Sentence", subtitle_mode="Sentence",
max_subtitle_words=5, max_subtitle_words=5,
lang_code="a", language=Language.EN_US,
use_spacy_segmentation=False, use_spacy_segmentation=False,
fallback_end_time=2.0, fallback_end_time=2.0,
) )
@@ -293,7 +294,7 @@ class TestProcessAndWriteSubtitles:
writer, writer,
subtitle_mode="Line", subtitle_mode="Line",
max_subtitle_words=5, max_subtitle_words=5,
lang_code="a", language=Language.EN_US,
use_spacy_segmentation=False, use_spacy_segmentation=False,
fallback_end_time=3.0, fallback_end_time=3.0,
) )
@@ -312,7 +313,7 @@ class TestProcessAndWriteSubtitles:
writer, writer,
subtitle_mode="Disabled", subtitle_mode="Disabled",
max_subtitle_words=5, max_subtitle_words=5,
lang_code="a", language=Language.EN_US,
use_spacy_segmentation=False, use_spacy_segmentation=False,
fallback_end_time=2.0, fallback_end_time=2.0,
) )
@@ -348,7 +349,7 @@ class TestFullPipeline:
audio_sink=merged_sink, audio_sink=merged_sink,
subtitle_mode="Sentence", subtitle_mode="Sentence",
max_subtitle_words=5, max_subtitle_words=5,
lang_code="a", language=Language.EN_US,
use_spacy_segmentation=False, use_spacy_segmentation=False,
) )
@@ -367,7 +368,7 @@ class TestFullPipeline:
subtitle_writer, subtitle_writer,
subtitle_mode="Sentence", subtitle_mode="Sentence",
max_subtitle_words=5, max_subtitle_words=5,
lang_code="a", language=Language.EN_US,
use_spacy_segmentation=False, use_spacy_segmentation=False,
fallback_end_time=stats.current_time, fallback_end_time=stats.current_time,
) )
tests/test_conversion_request.py
+3 -3
View File
@@ -37,13 +37,13 @@ class TestConversionRequestBasics:
assert "merge_chapters_at_end" in defaults assert "merge_chapters_at_end" in defaults
def test_split_pattern_computation(self): def test_split_pattern_computation(self):
pattern = get_split_pattern("a", "Disabled") pattern = get_split_pattern(Language.EN_US, "Disabled")
assert isinstance(pattern, str) assert isinstance(pattern, str)
assert len(pattern) > 0 assert len(pattern) > 0
def test_split_pattern_varies_by_subtitle_mode(self): def test_split_pattern_varies_by_subtitle_mode(self):
pattern_disabled = get_split_pattern("a", "Disabled") pattern_disabled = get_split_pattern(Language.EN_US, "Disabled")
pattern_sentence = get_split_pattern("a", "Sentence") pattern_sentence = get_split_pattern(Language.EN_US, "Sentence")
# Different modes should produce different patterns # Different modes should produce different patterns
assert isinstance(pattern_disabled, str) assert isinstance(pattern_disabled, str)
assert isinstance(pattern_sentence, str) assert isinstance(pattern_sentence, str)
tests/test_domain_normalization_pipeline.py
+16 -11
View File
@@ -2,6 +2,7 @@
import pytest import pytest
from unittest.mock import patch, MagicMock from unittest.mock import patch, MagicMock
from abogen.domain.enums import Language
from abogen.domain.normalization import prepare_text_for_tts, normalize_text_for_pipeline, build_tts_context, TTSContext from abogen.domain.normalization import prepare_text_for_tts, normalize_text_for_pipeline, build_tts_context, TTSContext
@@ -151,21 +152,21 @@ class TestBuildTtsContext:
"""Tests for the build_tts_context factory.""" """Tests for the build_tts_context factory."""
def test_returns_tts_context(self): def test_returns_tts_context(self):
ctx = build_tts_context() ctx = build_tts_context(language=Language.EN_US)
assert isinstance(ctx, TTSContext) assert isinstance(ctx, TTSContext)
def test_default_split_pattern(self): def test_default_split_pattern(self):
ctx = build_tts_context(language="a", subtitle_mode="Disabled") ctx = build_tts_context(language=Language.EN_US, subtitle_mode="Disabled")
assert isinstance(ctx.split_pattern, str) assert isinstance(ctx.split_pattern, str)
assert len(ctx.split_pattern) > 0 assert len(ctx.split_pattern) > 0
def test_english_uses_newline_split(self): def test_english_uses_newline_split(self):
ctx = build_tts_context(language="a", subtitle_mode="Disabled") ctx = build_tts_context(language=Language.EN_US, subtitle_mode="Disabled")
assert ctx.split_pattern == "\n" assert ctx.split_pattern == "\n"
def test_cjk_uses_punctuation_split(self): def test_cjk_uses_punctuation_split(self):
ctx = build_tts_context(language="j", subtitle_mode="Disabled") ctx = build_tts_context(language=Language.JA, subtitle_mode="Disabled")
assert "[.??.?!]" in ctx.split_pattern or "\\n" not in ctx.split_pattern assert r"\n" in ctx.split_pattern
def test_pronunciation_overrides_compiled(self): def test_pronunciation_overrides_compiled(self):
overrides = [ overrides = [
@@ -176,6 +177,7 @@ class TestBuildTtsContext:
} }
] ]
ctx = build_tts_context( ctx = build_tts_context(
language=Language.EN_US,
pronunciation_overrides=overrides, pronunciation_overrides=overrides,
) )
assert ctx.pronunciation_rules is not None assert ctx.pronunciation_rules is not None
@@ -190,6 +192,7 @@ class TestBuildTtsContext:
} }
] ]
ctx = build_tts_context( ctx = build_tts_context(
language=Language.EN_US,
manual_overrides=overrides, manual_overrides=overrides,
) )
assert ctx.pronunciation_rules is not None assert ctx.pronunciation_rules is not None
@@ -203,6 +206,7 @@ class TestBuildTtsContext:
{"token": "x", "pronunciation": "RIGHT", "normalized": "x"} {"token": "x", "pronunciation": "RIGHT", "normalized": "x"}
] ]
ctx = build_tts_context( ctx = build_tts_context(
language=Language.EN_US,
pronunciation_overrides=pronunciation, pronunciation_overrides=pronunciation,
manual_overrides=manual, manual_overrides=manual,
) )
@@ -224,22 +228,23 @@ class TestBuildTtsContext:
} }
] ]
ctx = build_tts_context( ctx = build_tts_context(
language=Language.EN_US,
heteronym_overrides=overrides, heteronym_overrides=overrides,
) )
assert ctx.heteronym_rules is not None assert ctx.heteronym_rules is not None
def test_usage_counter_passed_through(self): def test_usage_counter_passed_through(self):
counter = {} counter = {}
ctx = build_tts_context(usage_counter=counter) ctx = build_tts_context(language=Language.EN_US, usage_counter=counter)
assert ctx.usage_counter is counter assert ctx.usage_counter is counter
def test_usage_counter_default_empty(self): def test_usage_counter_default_empty(self):
ctx = build_tts_context() ctx = build_tts_context(language=Language.EN_US)
assert ctx.usage_counter == {} assert ctx.usage_counter == {}
def test_normalization_overrides_stored(self): def test_normalization_overrides_stored(self):
overrides = {"normalization_numbers": False} overrides = {"normalization_numbers": False}
ctx = build_tts_context(normalization_overrides=overrides) ctx = build_tts_context(language=Language.EN_US, normalization_overrides=overrides)
assert ctx.normalization_overrides is overrides assert ctx.normalization_overrides is overrides
def test_speakers_used_for_pronunciation(self): def test_speakers_used_for_pronunciation(self):
@@ -250,7 +255,7 @@ class TestBuildTtsContext:
"resolved_voice": "M1", "resolved_voice": "M1",
} }
} }
ctx = build_tts_context(speakers=speakers) ctx = build_tts_context(language=Language.EN_US, speakers=speakers)
assert ctx.pronunciation_rules is not None assert ctx.pronunciation_rules is not None
assert len(ctx.pronunciation_rules) >= 1 assert len(ctx.pronunciation_rules) >= 1
@@ -265,7 +270,7 @@ class TestBuildTtsContext:
mock_cfg.return_value = MagicMock(convert_numbers=True) mock_cfg.return_value = MagicMock(convert_numbers=True)
with patch("builtins.__import__", side_effect=ImportError): with patch("builtins.__import__", side_effect=ImportError):
try: try:
build_tts_context(log_callback=lambda lvl, msg: logs.append((lvl, msg))) build_tts_context(language=Language.EN_US, log_callback=lambda lvl, msg: logs.append((lvl, msg)))
except ImportError: except ImportError:
pass pass
# If num2words is missing and convert_numbers is True, a warning should be logged # If num2words is missing and convert_numbers is True, a warning should be logged
@@ -276,7 +281,7 @@ class TestBuildTtsContext:
"normalization_apostrophe_mode": "llm", "normalization_apostrophe_mode": "llm",
}): }):
with pytest.raises(RuntimeError, match="LLM"): with pytest.raises(RuntimeError, match="LLM"):
build_tts_context() build_tts_context(language=Language.EN_US)
def test_dict_source_accepted(self): def test_dict_source_accepted(self):
"""merge_pronunciation_overrides should accept a dict.""" """merge_pronunciation_overrides should accept a dict."""
tests/test_domain_pipeline_factory.py
+24 -25
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from abogen.domain.enums import Language
from abogen.domain.pipeline_factory import ( from abogen.domain.pipeline_factory import (
PipelinePool, PipelinePool,
create_pipeline_for_job, create_pipeline_for_job,
@@ -31,7 +32,7 @@ class TestCreatePipelineForJob:
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True) @patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True)
def test_supertonic_provider(self, _reg, mock_create): def test_supertonic_provider(self, _reg, mock_create):
mock_create.return_value = MagicMock() mock_create.return_value = MagicMock()
result = create_pipeline_for_job("supertonic", "en", use_gpu=True) result = create_pipeline_for_job("supertonic", Language.EN_US, use_gpu=True)
mock_create.assert_called_once_with("supertonic") mock_create.assert_called_once_with("supertonic")
assert result is mock_create.return_value assert result is mock_create.return_value
@@ -40,43 +41,41 @@ class TestCreatePipelineForJob:
@patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu") @patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu")
def test_kokoro_provider(self, _dev, _reg, mock_create): def test_kokoro_provider(self, _dev, _reg, mock_create):
mock_create.return_value = MagicMock() mock_create.return_value = MagicMock()
result = create_pipeline_for_job("kokoro", "en", use_gpu=False) result = create_pipeline_for_job("kokoro", Language.EN_US, use_gpu=False)
# "en" → fallback to EN_US → kokoro code "a" mock_create.assert_called_once_with("kokoro", language=Language.EN_US, device="cpu")
mock_create.assert_called_once_with("kokoro", lang_code="a", device="cpu")
assert result is mock_create.return_value assert result is mock_create.return_value
@patch("abogen.domain.pipeline_factory.create_pipeline") @patch("abogen.domain.pipeline_factory.create_pipeline")
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True) @patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True)
@patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu") @patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu")
def test_kokoro_provider_iso_code(self, _dev, _reg, mock_create): def test_kokoro_provider_en_gb(self, _dev, _reg, mock_create):
mock_create.return_value = MagicMock() mock_create.return_value = MagicMock()
result = create_pipeline_for_job("kokoro", "en-GB", use_gpu=False) result = create_pipeline_for_job("kokoro", Language.EN_GB, use_gpu=False)
# "en-GB" → EN_GB → kokoro code "b" mock_create.assert_called_once_with("kokoro", language=Language.EN_GB, device="cpu")
mock_create.assert_called_once_with("kokoro", lang_code="b", device="cpu")
@patch("abogen.domain.pipeline_factory.create_pipeline") @patch("abogen.domain.pipeline_factory.create_pipeline")
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=False) @patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=False)
@patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu") @patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu")
def test_unknown_provider_falls_back_to_kokoro(self, _dev, _reg, mock_create): def test_unknown_provider_falls_back_to_kokoro(self, _dev, _reg, mock_create):
mock_create.return_value = MagicMock() mock_create.return_value = MagicMock()
result = create_pipeline_for_job("unknown_provider", "en", use_gpu=False) result = create_pipeline_for_job("unknown_provider", Language.EN_US, use_gpu=False)
mock_create.assert_called_once_with("kokoro", lang_code="a", device="cpu") mock_create.assert_called_once_with("kokoro", language=Language.EN_US, device="cpu")
@patch("abogen.domain.pipeline_factory.create_pipeline") @patch("abogen.domain.pipeline_factory.create_pipeline")
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True) @patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True)
@patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu") @patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu")
def test_empty_provider_defaults_to_kokoro(self, _dev, _reg, mock_create): def test_empty_provider_defaults_to_kokoro(self, _dev, _reg, mock_create):
mock_create.return_value = MagicMock() mock_create.return_value = MagicMock()
result = create_pipeline_for_job("", "en", use_gpu=False) result = create_pipeline_for_job("", Language.EN_US, use_gpu=False)
mock_create.assert_called_once_with("kokoro", lang_code="a", device="cpu") mock_create.assert_called_once_with("kokoro", language=Language.EN_US, device="cpu")
@patch("abogen.domain.pipeline_factory.create_pipeline") @patch("abogen.domain.pipeline_factory.create_pipeline")
@patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True) @patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True)
@patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu") @patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu")
def test_none_provider_defaults_to_kokoro(self, _dev, _reg, mock_create): def test_none_provider_defaults_to_kokoro(self, _dev, _reg, mock_create):
mock_create.return_value = MagicMock() mock_create.return_value = MagicMock()
result = create_pipeline_for_job(None, "en", use_gpu=False) result = create_pipeline_for_job(None, Language.EN_US, use_gpu=False)
mock_create.assert_called_once_with("kokoro", lang_code="a", device="cpu") mock_create.assert_called_once_with("kokoro", language=Language.EN_US, device="cpu")
class TestDisposePipelines: class TestDisposePipelines:
@@ -110,11 +109,11 @@ class TestPipelinePool:
mock_create.return_value = mock_pipeline mock_create.return_value = mock_pipeline
pool = PipelinePool() pool = PipelinePool()
result = pool.get("kokoro", "en", use_gpu=True) result = pool.get("kokoro", Language.EN_US, use_gpu=True)
assert result is mock_pipeline assert result is mock_pipeline
mock_create.assert_called_once() mock_create.assert_called_once()
result2 = pool.get("kokoro", "en", use_gpu=True) result2 = pool.get("kokoro", Language.EN_US, use_gpu=True)
assert result2 is mock_pipeline assert result2 is mock_pipeline
assert mock_create.call_count == 1 assert mock_create.call_count == 1
@@ -125,10 +124,10 @@ class TestPipelinePool:
pool = PipelinePool() pool = PipelinePool()
request = MagicMock() request = MagicMock()
pool.get("kokoro", "en", use_gpu=True, request=request) pool.get("kokoro", Language.EN_US, use_gpu=True, request=request)
assert mock_cache.call_count == 1 assert mock_cache.call_count == 1
pool.get("kokoro", "en", use_gpu=True, request=request) pool.get("kokoro", Language.EN_US, use_gpu=True, request=request)
assert mock_cache.call_count == 1 assert mock_cache.call_count == 1
@patch("abogen.domain.pipeline_factory.initialize_voice_cache") @patch("abogen.domain.pipeline_factory.initialize_voice_cache")
@@ -136,7 +135,7 @@ class TestPipelinePool:
def test_get_no_job_skips_voice_cache(self, mock_create, mock_cache): def test_get_no_job_skips_voice_cache(self, mock_create, mock_cache):
mock_create.return_value = MagicMock() mock_create.return_value = MagicMock()
pool = PipelinePool() pool = PipelinePool()
pool.get("kokoro", "en", use_gpu=True) pool.get("kokoro", Language.EN_US, use_gpu=True)
mock_cache.assert_not_called() mock_cache.assert_not_called()
@patch("abogen.domain.pipeline_factory.create_pipeline_for_job") @patch("abogen.domain.pipeline_factory.create_pipeline_for_job")
@@ -146,8 +145,8 @@ class TestPipelinePool:
mock_create.side_effect = [p1, p2] mock_create.side_effect = [p1, p2]
pool = PipelinePool() pool = PipelinePool()
r1 = pool.get("kokoro", "en", use_gpu=True) r1 = pool.get("kokoro", Language.EN_US, use_gpu=True)
r2 = pool.get("supertonic", "en", use_gpu=True) r2 = pool.get("supertonic", Language.EN_US, use_gpu=True)
assert r1 is p1 assert r1 is p1
assert r2 is p2 assert r2 is p2
assert mock_create.call_count == 2 assert mock_create.call_count == 2
@@ -160,8 +159,8 @@ class TestPipelinePool:
mock_create.side_effect = [p1, p2] mock_create.side_effect = [p1, p2]
pool = PipelinePool() pool = PipelinePool()
pool.get("kokoro", "en", use_gpu=True) pool.get("kokoro", Language.EN_US, use_gpu=True)
pool.get("supertonic", "en", use_gpu=True) pool.get("supertonic", Language.EN_US, use_gpu=True)
pool.dispose_all() pool.dispose_all()
p1.dispose.assert_called_once() p1.dispose.assert_called_once()
@@ -181,5 +180,5 @@ class TestPipelinePool:
def test_unknown_provider_falls_back(self, _reg, _cache, mock_create): def test_unknown_provider_falls_back(self, _reg, _cache, mock_create):
mock_create.return_value = MagicMock() mock_create.return_value = MagicMock()
pool = PipelinePool() pool = PipelinePool()
pool.get("bogus_provider", "en", use_gpu=True) pool.get("bogus_provider", Language.EN_US, use_gpu=True)
mock_create.assert_called_once_with("kokoro", "en", True) mock_create.assert_called_once_with("kokoro", Language.EN_US, True)
tests/test_preview_applies_manual_overrides.py
+2 -1
View File
@@ -1,3 +1,4 @@
from abogen.domain.enums import Language
from abogen.webui.routes.utils import synthesize from abogen.webui.routes.utils import synthesize
@@ -45,7 +46,7 @@ def test_preview_applies_manual_override_before_normalization(monkeypatch):
synthesize.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=Language.EN_US,
speed=1.0, speed=1.0,
use_gpu=False, use_gpu=False,
tts_provider="supertonic", tts_provider="supertonic",
tests/test_split_pattern.py
+21 -25
View File
@@ -1,10 +1,11 @@
"""Tests for split pattern logic (3 identical copies in codebase).""" """Tests for split pattern logic."""
import os import os
import sys import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
import pytest import pytest
from abogen.domain.enums import Language
from abogen.domain.split_pattern import get_split_pattern from abogen.domain.split_pattern import get_split_pattern
@@ -12,49 +13,49 @@ from abogen.domain.split_pattern import get_split_pattern
class TestEnglish: class TestEnglish:
def test_english_sentence(self): def test_english_sentence(self):
assert get_split_pattern("en-US", "Sentence") == "\n" assert get_split_pattern(Language.EN_US, "Sentence") == "\n"
def test_english_sentence_comma(self): def test_english_sentence_comma(self):
assert get_split_pattern("en-US", "Sentence + Comma") == "\n" assert get_split_pattern(Language.EN_US, "Sentence + Comma") == "\n"
def test_english_line(self): def test_english_line(self):
assert get_split_pattern("en-US", "Line") == "\n" assert get_split_pattern(Language.EN_US, "Line") == "\n"
def test_english_disabled(self): def test_english_disabled(self):
assert get_split_pattern("en-US", "Disabled") == "\n" assert get_split_pattern(Language.EN_US, "Disabled") == "\n"
def test_english_gb(self): def test_english_gb(self):
assert get_split_pattern("en-GB", "Sentence") == "\n" assert get_split_pattern(Language.EN_GB, "Sentence") == "\n"
# --- CJK languages --- # --- CJK languages ---
class TestCJK: class TestCJK:
def test_chinese_disabled(self): def test_chinese_disabled(self):
pattern = get_split_pattern("zh", "Disabled") pattern = get_split_pattern(Language.ZH, "Disabled")
assert pattern != "\n" assert pattern != "\n"
assert r"\n+" in pattern assert r"\n+" in pattern
def test_chinese_line(self): def test_chinese_line(self):
pattern = get_split_pattern("zh", "Line") pattern = get_split_pattern(Language.ZH, "Line")
assert pattern != "\n" assert pattern != "\n"
assert r"\n+" in pattern assert r"\n+" in pattern
def test_chinese_sentence(self): def test_chinese_sentence(self):
pattern = get_split_pattern("zh", "Sentence") pattern = get_split_pattern(Language.ZH, "Sentence")
assert r"\n+" in pattern assert r"\n+" in pattern
def test_chinese_sentence_comma(self): def test_chinese_sentence_comma(self):
pattern = get_split_pattern("zh", "Sentence + Comma") pattern = get_split_pattern(Language.ZH, "Sentence + Comma")
assert r"\n+" in pattern assert r"\n+" in pattern
def test_japanese_disabled(self): def test_japanese_disabled(self):
pattern = get_split_pattern("ja", "Disabled") pattern = get_split_pattern(Language.JA, "Disabled")
assert pattern != "\n" assert pattern != "\n"
assert r"\n+" in pattern assert r"\n+" in pattern
def test_japanese_sentence(self): def test_japanese_sentence(self):
pattern = get_split_pattern("ja", "Sentence") pattern = get_split_pattern(Language.JA, "Sentence")
assert r"\n+" in pattern assert r"\n+" in pattern
@@ -62,22 +63,17 @@ class TestCJK:
class TestOtherLanguages: class TestOtherLanguages:
def test_spanish_sentence(self): def test_spanish_sentence(self):
pattern = get_split_pattern("es", "Sentence") pattern = get_split_pattern(Language.ES, "Sentence")
assert r"\n+" in pattern assert r"\n+" in pattern
def test_spanish_line(self): def test_spanish_line(self):
assert get_split_pattern("es", "Line") == "\n" assert get_split_pattern(Language.ES, "Line") == "\n"
def test_spanish_disabled(self): def test_spanish_disabled(self):
# canonical: \n+ for non-CJK Disabled assert get_split_pattern(Language.ES, "Disabled") == r"\n+"
assert get_split_pattern("es", "Disabled") == r"\n+"
def test_french_sentence_comma(self): def test_french_sentence_comma(self):
pattern = get_split_pattern("fr", "Sentence + Comma") pattern = get_split_pattern(Language.FR, "Sentence + Comma")
assert r"\n+" in pattern
def test_unknown_lang(self):
pattern = get_split_pattern("x", "Sentence")
assert r"\n+" in pattern assert r"\n+" in pattern
@@ -85,17 +81,17 @@ class TestOtherLanguages:
class TestPatternStructure: class TestPatternStructure:
def test_sentence_has_lookbehind(self): def test_sentence_has_lookbehind(self):
pattern = get_split_pattern("es", "Sentence") pattern = get_split_pattern(Language.ES, "Sentence")
assert r"(?<=" in pattern assert r"(?<=" in pattern
def test_sentence_comma_has_comma_chars(self): def test_sentence_comma_has_comma_chars(self):
pattern = get_split_pattern("es", "Sentence + Comma") pattern = get_split_pattern(Language.ES, "Sentence + Comma")
assert "," in pattern assert "," in pattern
def test_cjk_spacing_uses_star(self): def test_cjk_spacing_uses_star(self):
pattern = get_split_pattern("zh", "Sentence") pattern = get_split_pattern(Language.ZH, "Sentence")
assert r"\s*" in pattern assert r"\s*" in pattern
def test_non_cjk_spacing_uses_plus(self): def test_non_cjk_spacing_uses_plus(self):
pattern = get_split_pattern("es", "Sentence") pattern = get_split_pattern(Language.ES, "Sentence")
assert r"\s+" in pattern assert r"\s+" in pattern
tests/test_subtitle_generation.py
+10 -9
View File
@@ -2,6 +2,7 @@
import pytest import pytest
from abogen.domain.enums import Language
from abogen.domain.subtitle_generation import ( from abogen.domain.subtitle_generation import (
process_subtitle_tokens, process_subtitle_tokens,
PUNCTUATION_SENTENCE, PUNCTUATION_SENTENCE,
@@ -20,7 +21,7 @@ class TestProcessSubtitleTokens:
subtitle_entries=entries, subtitle_entries=entries,
max_subtitle_words=50, max_subtitle_words=50,
subtitle_mode="Sentence", subtitle_mode="Sentence",
lang_code="a", language=Language.EN_US,
) )
assert entries == [] assert entries == []
@@ -41,7 +42,7 @@ class TestProcessSubtitleTokens:
subtitle_entries=entries, subtitle_entries=entries,
max_subtitle_words=50, max_subtitle_words=50,
subtitle_mode="Disabled", subtitle_mode="Disabled",
lang_code="a", language=Language.EN_US,
) )
# Disabled mode doesn't have special handling in current implementation # Disabled mode doesn't have special handling in current implementation
# It processes tokens normally # It processes tokens normally
@@ -59,7 +60,7 @@ class TestProcessSubtitleTokens:
subtitle_entries=entries, subtitle_entries=entries,
max_subtitle_words=50, max_subtitle_words=50,
subtitle_mode="Line", subtitle_mode="Line",
lang_code="a", language=Language.EN_US,
) )
# Line mode processes all tokens into entries # Line mode processes all tokens into entries
assert len(entries) >= 1 assert len(entries) >= 1
@@ -82,7 +83,7 @@ class TestProcessSubtitleTokens:
subtitle_entries=entries, subtitle_entries=entries,
max_subtitle_words=50, max_subtitle_words=50,
subtitle_mode="Sentence", subtitle_mode="Sentence",
lang_code="a", language=Language.EN_US,
) )
assert len(entries) >= 1 assert len(entries) >= 1
# Should have at least one entry with both sentences or split # Should have at least one entry with both sentences or split
@@ -106,7 +107,7 @@ class TestProcessSubtitleTokens:
subtitle_entries=entries, subtitle_entries=entries,
max_subtitle_words=50, max_subtitle_words=50,
subtitle_mode="2", # 2 words per entry subtitle_mode="2", # 2 words per entry
lang_code="a", language=Language.EN_US,
) )
assert len(entries) >= 2 assert len(entries) >= 2
# Check that entries are split roughly by word count # Check that entries are split roughly by word count
@@ -125,7 +126,7 @@ class TestProcessSubtitleTokens:
subtitle_entries=entries, subtitle_entries=entries,
max_subtitle_words=50, max_subtitle_words=50,
subtitle_mode="Line", subtitle_mode="Line",
lang_code="a", language=Language.EN_US,
fallback_end_time=10.0, fallback_end_time=10.0,
) )
assert len(entries) == 1 assert len(entries) == 1
@@ -143,7 +144,7 @@ class TestProcessSubtitleTokens:
subtitle_entries=entries, subtitle_entries=entries,
max_subtitle_words=50, max_subtitle_words=50,
subtitle_mode="Sentence + Highlighting", subtitle_mode="Sentence + Highlighting",
lang_code="a", language=Language.EN_US,
) )
assert len(entries) >= 1 assert len(entries) >= 1
# Should contain karaoke tags # Should contain karaoke tags
@@ -162,7 +163,7 @@ class TestProcessSubtitleTokens:
subtitle_entries=entries, subtitle_entries=entries,
max_subtitle_words=3, max_subtitle_words=3,
subtitle_mode="Line", subtitle_mode="Line",
lang_code="a", language=Language.EN_US,
) )
# Should have more than 1 entry due to word limit # Should have more than 1 entry due to word limit
assert len(entries) > 1 assert len(entries) > 1
@@ -179,7 +180,7 @@ class TestProcessSubtitleTokens:
subtitle_entries=entries, subtitle_entries=entries,
max_subtitle_words=50, max_subtitle_words=50,
subtitle_mode="Sentence", subtitle_mode="Sentence",
lang_code="a", language=Language.EN_US,
) )
assert len(entries) >= 1 assert len(entries) >= 1
# Check that timing is preserved # Check that timing is preserved