Compare commits

...
Author SHA1 Message Date
Artem Akymenko 146000886d ci: add uv cache prune to optimize cache size 2026-07-08 21:14:15 +00:00
Artem Akymenko 31f95137dd ci: replace pip with uv for faster dependency installation 2026-07-08 18:34:35 +00:00
Artem Akymenko 6f02fda41c fix(ci): set QT_QPA_PLATFORM=offscreen for headless PyQt6 tests 2026-07-08 17:36:59 +00:00
Artem Akymenko a3c3462348 fix(ci): install libegl1 on Ubuntu and normalize line endings in epub test
- Add system dependency step for libegl1 to fix PyQt6 import on headless CI
- Normalize CRLF to LF in epub exporter whitespace test for Windows CI
2026-07-08 17:16:33 +00:00
Artem AkymenkoandGitHub 79332204d3 Merge pull request #189 from denizsafak/feat/registry-voice-resolution
refactor: Move backend resolution by voice spec into registry
2026-07-08 20:03:19 +03:00
Artem Akymenko 6deec3b9b6 refactor: move backend resolution by voice spec into registry
- Add resolve_backend_for_voice() to TTSBackendRegistry
- Add module-level wrapper resolve_backend_for_voice()
- Simplify _infer_provider_from_spec() to use registry API
- Simplify _supertonic_voice_from_spec() to only normalize
- Add 11 test cases for the new method

Resolution rules:
1. Empty spec -> fallback
2. Kokoro formula (* or +) -> kokoro
3. Exact voice ID match -> backend id
4. Unknown voice -> fallback
2026-07-08 17:02:33 +00:00
Artem Akymenko c4d14112d4 refactor: replace hardcoded backend ID sets with registry checks
Add TTSBackendRegistry.is_registered() and module-level
is_registered_backend() to validate backend IDs dynamically.
Replace all Category A hardcoded sets (validation-only) in
voice_profiles, api routes, conversion_runner, and form utils.
2026-07-08 16:33:16 +00:00
Artem Akymenko f4cb2c2329 ci: add pytest, use actions/cache@v6 2026-07-08 19:26:42 +03:00
Artem AkymenkoandGitHub 783738882f Merge pull request #188 from denizsafak/refactor/move-kokoro-voices-into-backend
refactor: move VOICES_INTERNAL into KokoroBackend module
2026-07-08 19:23:33 +03:00
Artem Akymenko e94ba5257e refactor: move VOICES_INTERNAL into KokoroBackend module
Make the Kokoro voice list an internal implementation detail of the
backend instead of a shared constant. The rest of the project already
accesses voices via get_metadata('kokoro').voices.

- Move VOICES_INTERNAL from constants.py to kokoro.py as _VOICES_INTERNAL
- Update tests to use get_metadata('kokoro').voices instead of importing
  the constant directly
2026-07-08 16:19:34 +00:00
Artem AkymenkoandGitHub 49d66839dc Merge pull request #186 from denizsafak/refactor/migrate-remaining-voice-metadata-consumers
refactor: migrate remaining consumers to get_metadata API
2026-07-08 19:01:20 +03:00
Artem AkymenkoandGitHub d0e316ea7b Merge pull request #187 from denizsafak/refactor/migrate-pyqt-to-backend-metadata
refactor(pyqt): migrate from VOICES_INTERNAL to get_metadata API
2026-07-08 19:01:02 +03:00
Artem Akymenko bb96ae502c refactor: migrate remaining consumers to get_metadata API
Replace direct VOICES_INTERNAL imports with get_metadata('kokoro').voices:
- abogen/predownload_gui.py
- abogen/subtitle_utils.py
2026-07-08 15:58:51 +00:00
Artem Akymenko a4d25accc1 refactor(pyqt): migrate from VOICES_INTERNAL to get_metadata API
Replace direct VOICES_INTERNAL imports with get_metadata('kokoro').voices
from tts_backend_registry in all PyQt modules:
- abogen/pyqt/gui.py
- abogen/pyqt/predownload_gui.py
- abogen/pyqt/voice_formula_gui.py
2026-07-08 15:57:18 +00:00
Artem AkymenkoandGitHub 66964bfd0b Merge pull request #185 from denizsafak/refactor/use-backend-metadata-in-webui
refactor(webui): replace direct VOICES_INTERNAL/DEFAULT_SUPERTONIC_VOICES with get_metadata API
2026-07-08 18:49:21 +03:00
18 changed files with 539 additions and 357 deletions
+28 -8
View File
@@ -1,5 +1,6 @@
name: pip install
run-name: pip install
name: CI
run-name: CI
on:
push:
branches: [main]
@@ -12,8 +13,9 @@ on:
- 'pyproject.toml'
- '.github/workflows/**'
workflow_dispatch:
jobs:
install-and-run:
test:
strategy:
matrix:
os: [ubuntu-latest, macos-14, windows-latest]
@@ -23,12 +25,30 @@ jobs:
steps:
- name: Checkout repository
uses: actions/checkout@v7
- name: Set up Python
uses: actions/setup-python@v6
with:
python-version: ${{ matrix.python-version }}
cache: pip
- name: Install from repository
run: python -m pip install .
#- name: Run abogen
# run: abogen
- name: Install uv
uses: astral-sh/setup-uv@v8.3.1
with:
enable-cache: true
cache-dependency-glob: pyproject.toml
- name: Install system dependencies (Ubuntu)
if: runner.os == 'Linux'
run: sudo apt-get update && sudo apt-get install -y libegl1
- name: Install dependencies
run: uv pip install --system .[dev]
- name: Run tests
env:
QT_QPA_PLATFORM: offscreen
run: pytest tests/ -v --tb=short
- name: Minimize uv cache
if: always()
run: uv cache prune --ci
-58
View File
@@ -63,64 +63,6 @@ SUPPORTED_INPUT_FORMATS = [
# 384 if self.lang_code in 'ab':
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION = list(LANGUAGE_DESCRIPTIONS.keys())
# Voice and sample text constants
VOICES_INTERNAL = [
"af_alloy",
"af_aoede",
"af_bella",
"af_heart",
"af_jessica",
"af_kore",
"af_nicole",
"af_nova",
"af_river",
"af_sarah",
"af_sky",
"am_adam",
"am_echo",
"am_eric",
"am_fenrir",
"am_liam",
"am_michael",
"am_onyx",
"am_puck",
"am_santa",
"bf_alice",
"bf_emma",
"bf_isabella",
"bf_lily",
"bm_daniel",
"bm_fable",
"bm_george",
"bm_lewis",
"ef_dora",
"em_alex",
"em_santa",
"ff_siwis",
"hf_alpha",
"hf_beta",
"hm_omega",
"hm_psi",
"if_sara",
"im_nicola",
"jf_alpha",
"jf_gongitsune",
"jf_nezumi",
"jf_tebukuro",
"jm_kumo",
"pf_dora",
"pm_alex",
"pm_santa",
"zf_xiaobei",
"zf_xiaoni",
"zf_xiaoxiao",
"zf_xiaoyi",
"zm_yunjian",
"zm_yunxi",
"zm_yunxia",
"zm_yunyang",
]
# Voice and sample text mapping
SAMPLE_VOICE_TEXTS = {
"a": "This is a sample of the selected voice.",
+5 -4
View File
@@ -21,7 +21,8 @@ from PyQt6.QtWidgets import (
)
from PyQt6.QtCore import QThread, pyqtSignal
from abogen.constants import COLORS, VOICES_INTERNAL
from abogen.constants import COLORS
from abogen.tts_backend_registry import get_metadata
from abogen.spacy_utils import SPACY_MODELS
import abogen.hf_tracker
@@ -114,7 +115,7 @@ class PreDownloadWorker(QThread):
self._voices_success = False
return
voice_list = VOICES_INTERNAL
voice_list = get_metadata("kokoro").voices
for idx, voice in enumerate(voice_list, start=1):
if self._cancelled:
self._voices_success = False
@@ -462,14 +463,14 @@ class PreDownloadDialog(QDialog):
try:
from huggingface_hub import try_to_load_from_cache
for voice in VOICES_INTERNAL:
for voice in get_metadata("kokoro").voices:
if not try_to_load_from_cache(
repo_id="hexgrad/Kokoro-82M", filename=f"voices/{voice}.pt"
):
missing.append(voice)
except Exception:
# If HF missing, report all as missing
return False, list(VOICES_INTERNAL)
return False, list(get_metadata("kokoro").voices)
return (len(missing) == 0), missing
def _check_kokoro_model(self) -> bool:
+2 -2
View File
@@ -82,11 +82,11 @@ from abogen.constants import (
GITHUB_URL,
PROGRAM_DESCRIPTION,
LANGUAGE_DESCRIPTIONS,
VOICES_INTERNAL,
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
COLORS,
SUBTITLE_FORMATS,
)
from abogen.tts_backend_registry import get_metadata
import threading
from abogen.pyqt.voice_formula_gui import VoiceFormulaDialog
from abogen.voice_profiles import load_profiles
@@ -1873,7 +1873,7 @@ class abogen(QWidget):
for pname in load_profiles().keys():
self.voice_combo.addItem(profile_icon, pname, f"profile:{pname}")
# re-add voices
for v in VOICES_INTERNAL:
for v in get_metadata("kokoro").voices:
icon = QIcon()
flag_path = get_resource_path("abogen.assets.flags", f"{v[0]}.png")
if flag_path and os.path.exists(flag_path):
+5 -4
View File
@@ -21,7 +21,8 @@ from PyQt6.QtWidgets import (
)
from PyQt6.QtCore import QThread, pyqtSignal
from abogen.constants import COLORS, VOICES_INTERNAL
from abogen.constants import COLORS
from abogen.tts_backend_registry import get_metadata
from abogen.spacy_utils import SPACY_MODELS
import abogen.hf_tracker
@@ -114,7 +115,7 @@ class PreDownloadWorker(QThread):
self._voices_success = False
return
voice_list = VOICES_INTERNAL
voice_list = get_metadata("kokoro").voices
for idx, voice in enumerate(voice_list, start=1):
if self._cancelled:
self._voices_success = False
@@ -462,14 +463,14 @@ class PreDownloadDialog(QDialog):
try:
from huggingface_hub import try_to_load_from_cache
for voice in VOICES_INTERNAL:
for voice in get_metadata("kokoro").voices:
if not try_to_load_from_cache(
repo_id="hexgrad/Kokoro-82M", filename=f"voices/{voice}.pt"
):
missing.append(voice)
except Exception:
# If HF missing, report all as missing
return False, list(VOICES_INTERNAL)
return False, list(get_metadata("kokoro").voices)
return (len(missing) == 0), missing
def _check_kokoro_model(self) -> bool:
+3 -3
View File
@@ -28,11 +28,11 @@ from PyQt6.QtWidgets import (
from PyQt6.QtCore import Qt, QTimer, QPoint, QRect, QSize
from PyQt6.QtGui import QPixmap, QIcon, QAction
from abogen.constants import (
VOICES_INTERNAL,
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
LANGUAGE_DESCRIPTIONS,
COLORS,
)
from abogen.tts_backend_registry import get_metadata
import re
import platform
from abogen.utils import get_resource_path
@@ -179,7 +179,7 @@ class VoiceMixer(QWidget):
layout.addWidget(QLabel(name), alignment=Qt.AlignmentFlag.AlignCenter)
# Voice name label with gender icon
is_female = self.voice_name in VOICES_INTERNAL and self.voice_name[1] == "f"
is_female = self.voice_name in get_metadata("kokoro").voices and self.voice_name[1] == "f"
# Icons layout (flag and gender)
icons_layout = QHBoxLayout()
@@ -772,7 +772,7 @@ class VoiceFormulaDialog(QDialog):
def add_voices(self, initial_state):
first_enabled_voice = None
for voice in VOICES_INTERNAL:
for voice in get_metadata("kokoro").voices:
language_code = voice[0] # First character is the language code
matching_voice = next(
(item for item in initial_state if item[0] == voice), None
+7 -7
View File
@@ -466,7 +466,7 @@ def sanitize_name_for_os(name, is_folder=True):
def validate_voice_name(voice_name):
"""Validate voice name against VOICES_INTERNAL list (case-insensitive).
"""Validate voice name against available voices (case-insensitive).
Handles both single voices and formulas like 'af_heart*0.5 + am_echo*0.5'.
Args:
@@ -477,10 +477,10 @@ def validate_voice_name(voice_name):
- is_valid: True if all voices in the name/formula are valid
- invalid_voice_name: The first invalid voice found, or None if all valid
"""
from abogen.constants import VOICES_INTERNAL
from abogen.tts_backend_registry import get_metadata
# Create case-insensitive lookup set (done once per call)
voice_lookup_lower = {v.lower() for v in VOICES_INTERNAL}
voice_lookup_lower = {v.lower() for v in get_metadata("kokoro").voices}
voice_name = voice_name.strip()
# Check if it's a formula (contains *)
@@ -505,7 +505,7 @@ def split_text_by_voice_markers(text, default_voice):
"""Split text by voice markers, returning list of (voice, text) tuples.
IMPORTANT: Returns the last voice used so it can persist across chapters.
Voice names are normalized to lowercase to match VOICES_INTERNAL.
Voice names are normalized to lowercase to match canonical voice names.
Args:
text: Text potentially containing <<VOICE:name>> markers
@@ -518,7 +518,7 @@ def split_text_by_voice_markers(text, default_voice):
- valid_count: Number of valid voice markers processed
- invalid_count: Number of invalid voice markers skipped
"""
from abogen.constants import VOICES_INTERNAL
from abogen.tts_backend_registry import get_metadata
voice_splits = list(_VOICE_MARKER_SEARCH_PATTERN.finditer(text))
@@ -560,7 +560,7 @@ def split_text_by_voice_markers(text, default_voice):
# Find the canonical (lowercase) voice name
voice_part_lower = voice_part.strip().lower()
canonical_voice = next(
(v for v in VOICES_INTERNAL if v.lower() == voice_part_lower),
(v for v in get_metadata("kokoro").voices if v.lower() == voice_part_lower),
voice_part.strip()
)
normalized_parts.append(f"{canonical_voice}*{weight.strip()}")
@@ -569,7 +569,7 @@ def split_text_by_voice_markers(text, default_voice):
# Find the canonical (lowercase) voice name
voice_name_lower = voice_name.lower()
current_voice = next(
(v for v in VOICES_INTERNAL if v.lower() == voice_name_lower),
(v for v in get_metadata("kokoro").voices if v.lower() == voice_name_lower),
voice_name
)
valid_markers += 1
+56
View File
@@ -30,6 +30,10 @@ class TTSBackendRegistry:
self._backends[metadata.id] = metadata
self._factories[metadata.id] = factory
def is_registered(self, backend_id: str) -> bool:
"""Return True if a backend with the given id is registered."""
return backend_id in self._backends
def list_backends(self) -> list[TTSBackendMetadata]:
"""Return metadata for all registered backends."""
return list(self._backends.values())
@@ -54,6 +58,33 @@ class TTSBackendRegistry:
raise KeyError(f"Unknown backend: {backend_id}")
return self._factories[backend_id](**kwargs)
def resolve_backend_for_voice(
self,
spec: str,
fallback: str = "kokoro",
) -> str:
"""Determine which backend owns the given voice specification.
Resolution rules:
1. Empty spec -> fallback
2. Kokoro formula (contains '*' or '+') -> "kokoro"
3. Exact voice ID match against registered backends -> backend id
4. Unknown voice -> fallback
"""
raw = str(spec or "").strip()
if not raw:
return fallback
if "*" in raw or "+" in raw:
return "kokoro"
upper = raw.upper()
for metadata in self._backends.values():
if upper in metadata.voices:
return metadata.id
return fallback
_registry = TTSBackendRegistry()
@@ -88,3 +119,28 @@ def get_default_voice(backend_id: str, fallback: str = "") -> str:
def create_backend(backend_id: str, **kwargs: Any) -> TTSBackend:
"""Create a TTS backend instance by provider id."""
return _registry.create_backend(backend_id, **kwargs)
def is_registered_backend(backend_id: str) -> bool:
"""Return True if *backend_id* is a registered TTS backend."""
import abogen.tts_backends # noqa: F401 — triggers backend registration
return _registry.is_registered(backend_id)
def resolve_backend_for_voice(
spec: str,
fallback: str = "kokoro",
) -> str:
"""Determine which backend owns the given voice specification.
Ensures all backends are registered by importing the tts_backends
package on first access.
Resolution rules:
1. Empty spec -> fallback
2. Kokoro formula (contains '*' or '+') -> "kokoro"
3. Exact voice ID match against registered backends -> backend id
4. Unknown voice -> fallback
"""
import abogen.tts_backends # noqa: F401 — triggers backend registration
return _registry.resolve_backend_for_voice(spec, fallback=fallback)
+60 -2
View File
@@ -10,14 +10,72 @@ from typing import Any, Dict, Iterator, List, Optional
import numpy as np
from abogen.constants import VOICES_INTERNAL
from abogen.tts_backend import TTSBackendMetadata
# Internal voice list — source of truth for Kokoro voices.
# The rest of the project accesses voices via get_metadata("kokoro").voices.
_VOICES_INTERNAL = [
"af_alloy",
"af_aoede",
"af_bella",
"af_heart",
"af_jessica",
"af_kore",
"af_nicole",
"af_nova",
"af_river",
"af_sarah",
"af_sky",
"am_adam",
"am_echo",
"am_eric",
"am_fenrir",
"am_liam",
"am_michael",
"am_onyx",
"am_puck",
"am_santa",
"bf_alice",
"bf_emma",
"bf_isabella",
"bf_lily",
"bm_daniel",
"bm_fable",
"bm_george",
"bm_lewis",
"ef_dora",
"em_alex",
"em_santa",
"ff_siwis",
"hf_alpha",
"hf_beta",
"hm_omega",
"hm_psi",
"if_sara",
"im_nicola",
"jf_alpha",
"jf_gongitsune",
"jf_nezumi",
"jf_tebukuro",
"jm_kumo",
"pf_dora",
"pm_alex",
"pm_santa",
"zf_xiaobei",
"zf_xiaoni",
"zf_xiaoxiao",
"zf_xiaoyi",
"zm_yunjian",
"zm_yunxi",
"zm_yunxia",
"zm_yunyang",
]
_KOKORO_METADATA = TTSBackendMetadata(
id="kokoro",
name="Kokoro",
description="Kokoro TTS engine",
voices=tuple(VOICES_INTERNAL),
voices=tuple(_VOICES_INTERNAL),
)
+230 -230
View File
@@ -1,230 +1,230 @@
import json
import os
from typing import Any, Dict, Iterable, List, Tuple
from abogen.tts_backend_registry import get_metadata
from abogen.utils import get_user_config_path
def _get_profiles_path():
config_path = get_user_config_path()
config_dir = os.path.dirname(config_path)
return os.path.join(config_dir, "voice_profiles.json")
def load_profiles():
"""Load all voice profiles from JSON file."""
path = _get_profiles_path()
if os.path.exists(path):
try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
# always expect abogen_voice_profiles wrapper
if isinstance(data, dict) and "abogen_voice_profiles" in data:
return data["abogen_voice_profiles"]
# fallback: treat as profiles dict
if isinstance(data, dict):
return data
except Exception:
return {}
return {}
def save_profiles(profiles):
"""Save all voice profiles to JSON file."""
path = _get_profiles_path()
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
# always save with abogen_voice_profiles wrapper
json.dump({"abogen_voice_profiles": profiles}, f, indent=2)
def delete_profile(name):
"""Remove a profile by name."""
profiles = load_profiles()
if name in profiles:
del profiles[name]
save_profiles(profiles)
def duplicate_profile(src, dest):
"""Duplicate an existing profile."""
profiles = load_profiles()
if src in profiles and dest:
profiles[dest] = profiles[src]
save_profiles(profiles)
def export_profiles(export_path):
"""Export all profiles to specified JSON file."""
profiles = load_profiles()
with open(export_path, "w", encoding="utf-8") as f:
json.dump({"abogen_voice_profiles": profiles}, f, indent=2)
def serialize_profiles() -> Dict[str, Dict[str, Iterable[Tuple[str, float]]]]:
"""Return profiles in canonical dictionary form."""
return load_profiles()
def _normalize_supertonic_voice(value: Any) -> str:
raw = str(value or "").strip().upper()
supertonic_voices = get_metadata("supertonic").voices
return raw if raw in supertonic_voices else "M1"
def _coerce_supertonic_steps(value: Any) -> int:
try:
steps = int(value)
except (TypeError, ValueError):
return 5
return max(2, min(15, steps))
def _coerce_supertonic_speed(value: Any) -> float:
try:
speed = float(value)
except (TypeError, ValueError):
return 1.0
return max(0.7, min(2.0, speed))
def normalize_profile_entry(entry: Any) -> Dict[str, Any]:
"""Normalize a stored profile entry.
Backwards compatible:
- Legacy Kokoro-only entries: {language, voices}
- New entries: include provider.
"""
if not isinstance(entry, dict):
return {}
provider = str(entry.get("provider") or "kokoro").strip().lower()
if provider not in {"kokoro", "supertonic"}:
provider = "kokoro"
language = str(entry.get("language") or "a").strip().lower() or "a"
if provider == "supertonic":
return {
"provider": "supertonic",
"language": language,
"voice": _normalize_supertonic_voice(
entry.get("voice") or entry.get("voice_name") or entry.get("name")
),
"total_steps": _coerce_supertonic_steps(
entry.get("total_steps")
or entry.get("supertonic_total_steps")
or entry.get("quality")
),
"speed": _coerce_supertonic_speed(
entry.get("speed") or entry.get("supertonic_speed")
),
}
voices = _normalize_voice_entries(entry.get("voices", []))
if not voices:
return {}
return {
"provider": "kokoro",
"language": language,
"voices": voices,
}
def _normalize_voice_entries(entries: Iterable) -> List[Tuple[str, float]]:
normalized: List[Tuple[str, float]] = []
kokoro_voices = get_metadata("kokoro").voices
for item in entries or []:
if isinstance(item, dict):
voice = item.get("id") or item.get("voice")
weight = item.get("weight")
elif isinstance(item, (list, tuple)) and len(item) >= 2:
voice, weight = item[0], item[1]
else:
continue
if voice not in kokoro_voices:
continue
if weight is None:
continue
try:
weight_val = float(weight)
except (TypeError, ValueError):
continue
if weight_val <= 0:
continue
normalized.append((voice, weight_val))
return normalized
def normalize_voice_entries(entries: Iterable) -> List[Tuple[str, float]]:
"""Public helper to normalize voice-weight pairs from arbitrary payloads."""
return _normalize_voice_entries(entries)
def save_profile(name: str, *, language: str, voices: Iterable) -> None:
"""Persist a single profile after validating its data."""
name = (name or "").strip()
if not name:
raise ValueError("Profile name is required")
normalized = _normalize_voice_entries(voices)
if not normalized:
raise ValueError("At least one voice with a weight above zero is required")
if not language:
language = "a"
profiles = load_profiles()
profiles[name] = {"provider": "kokoro", "language": language, "voices": normalized}
save_profiles(profiles)
def remove_profile(name: str) -> None:
delete_profile(name)
def import_profiles_data(data: Dict, *, replace_existing: bool = False) -> List[str]:
"""Merge profiles from a dictionary structure and persist them.
Returns the list of profile names that were added or updated.
"""
if not isinstance(data, dict):
raise ValueError("Invalid profile payload")
if "abogen_voice_profiles" in data:
data = data["abogen_voice_profiles"]
if not isinstance(data, dict):
raise ValueError("Invalid profile payload")
current = load_profiles()
updated: List[str] = []
for name, entry in data.items():
normalized = normalize_profile_entry(entry)
if not normalized:
continue
if name in current and not replace_existing:
# skip duplicates unless explicit replacement is requested
continue
current[name] = normalized
updated.append(name)
if updated:
save_profiles(current)
return updated
def export_profiles_payload(names: Iterable[str] | None = None) -> Dict[str, Dict]:
"""Return profiles limited to the provided names for download/export."""
profiles = load_profiles()
if names is None:
subset = profiles
else:
subset = {name: profiles[name] for name in names if name in profiles}
return {"abogen_voice_profiles": subset}
import json
import os
from typing import Any, Dict, Iterable, List, Tuple
from abogen.tts_backend_registry import get_metadata, is_registered_backend
from abogen.utils import get_user_config_path
def _get_profiles_path():
config_path = get_user_config_path()
config_dir = os.path.dirname(config_path)
return os.path.join(config_dir, "voice_profiles.json")
def load_profiles():
"""Load all voice profiles from JSON file."""
path = _get_profiles_path()
if os.path.exists(path):
try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
# always expect abogen_voice_profiles wrapper
if isinstance(data, dict) and "abogen_voice_profiles" in data:
return data["abogen_voice_profiles"]
# fallback: treat as profiles dict
if isinstance(data, dict):
return data
except Exception:
return {}
return {}
def save_profiles(profiles):
"""Save all voice profiles to JSON file."""
path = _get_profiles_path()
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
# always save with abogen_voice_profiles wrapper
json.dump({"abogen_voice_profiles": profiles}, f, indent=2)
def delete_profile(name):
"""Remove a profile by name."""
profiles = load_profiles()
if name in profiles:
del profiles[name]
save_profiles(profiles)
def duplicate_profile(src, dest):
"""Duplicate an existing profile."""
profiles = load_profiles()
if src in profiles and dest:
profiles[dest] = profiles[src]
save_profiles(profiles)
def export_profiles(export_path):
"""Export all profiles to specified JSON file."""
profiles = load_profiles()
with open(export_path, "w", encoding="utf-8") as f:
json.dump({"abogen_voice_profiles": profiles}, f, indent=2)
def serialize_profiles() -> Dict[str, Dict[str, Iterable[Tuple[str, float]]]]:
"""Return profiles in canonical dictionary form."""
return load_profiles()
def _normalize_supertonic_voice(value: Any) -> str:
raw = str(value or "").strip().upper()
supertonic_voices = get_metadata("supertonic").voices
return raw if raw in supertonic_voices else "M1"
def _coerce_supertonic_steps(value: Any) -> int:
try:
steps = int(value)
except (TypeError, ValueError):
return 5
return max(2, min(15, steps))
def _coerce_supertonic_speed(value: Any) -> float:
try:
speed = float(value)
except (TypeError, ValueError):
return 1.0
return max(0.7, min(2.0, speed))
def normalize_profile_entry(entry: Any) -> Dict[str, Any]:
"""Normalize a stored profile entry.
Backwards compatible:
- Legacy Kokoro-only entries: {language, voices}
- New entries: include provider.
"""
if not isinstance(entry, dict):
return {}
provider = str(entry.get("provider") or "kokoro").strip().lower()
if not is_registered_backend(provider):
provider = "kokoro"
language = str(entry.get("language") or "a").strip().lower() or "a"
if provider == "supertonic":
return {
"provider": "supertonic",
"language": language,
"voice": _normalize_supertonic_voice(
entry.get("voice") or entry.get("voice_name") or entry.get("name")
),
"total_steps": _coerce_supertonic_steps(
entry.get("total_steps")
or entry.get("supertonic_total_steps")
or entry.get("quality")
),
"speed": _coerce_supertonic_speed(
entry.get("speed") or entry.get("supertonic_speed")
),
}
voices = _normalize_voice_entries(entry.get("voices", []))
if not voices:
return {}
return {
"provider": "kokoro",
"language": language,
"voices": voices,
}
def _normalize_voice_entries(entries: Iterable) -> List[Tuple[str, float]]:
normalized: List[Tuple[str, float]] = []
kokoro_voices = get_metadata("kokoro").voices
for item in entries or []:
if isinstance(item, dict):
voice = item.get("id") or item.get("voice")
weight = item.get("weight")
elif isinstance(item, (list, tuple)) and len(item) >= 2:
voice, weight = item[0], item[1]
else:
continue
if voice not in kokoro_voices:
continue
if weight is None:
continue
try:
weight_val = float(weight)
except (TypeError, ValueError):
continue
if weight_val <= 0:
continue
normalized.append((voice, weight_val))
return normalized
def normalize_voice_entries(entries: Iterable) -> List[Tuple[str, float]]:
"""Public helper to normalize voice-weight pairs from arbitrary payloads."""
return _normalize_voice_entries(entries)
def save_profile(name: str, *, language: str, voices: Iterable) -> None:
"""Persist a single profile after validating its data."""
name = (name or "").strip()
if not name:
raise ValueError("Profile name is required")
normalized = _normalize_voice_entries(voices)
if not normalized:
raise ValueError("At least one voice with a weight above zero is required")
if not language:
language = "a"
profiles = load_profiles()
profiles[name] = {"provider": "kokoro", "language": language, "voices": normalized}
save_profiles(profiles)
def remove_profile(name: str) -> None:
delete_profile(name)
def import_profiles_data(data: Dict, *, replace_existing: bool = False) -> List[str]:
"""Merge profiles from a dictionary structure and persist them.
Returns the list of profile names that were added or updated.
"""
if not isinstance(data, dict):
raise ValueError("Invalid profile payload")
if "abogen_voice_profiles" in data:
data = data["abogen_voice_profiles"]
if not isinstance(data, dict):
raise ValueError("Invalid profile payload")
current = load_profiles()
updated: List[str] = []
for name, entry in data.items():
normalized = normalize_profile_entry(entry)
if not normalized:
continue
if name in current and not replace_existing:
# skip duplicates unless explicit replacement is requested
continue
current[name] = normalized
updated.append(name)
if updated:
save_profiles(current)
return updated
def export_profiles_payload(names: Iterable[str] | None = None) -> Dict[str, Dict]:
"""Return profiles limited to the provided names for download/export."""
profiles = load_profiles()
if names is None:
subset = profiles
else:
subset = {name: profiles[name] for name in names if name in profiles}
return {"abogen_voice_profiles": subset}
+6 -7
View File
@@ -2,7 +2,6 @@ FROM nvidia/cuda:12.6.3-cudnn-runtime-ubuntu22.04
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_NO_CACHE_DIR=1 \
VIRTUAL_ENV=/opt/venv \
PATH=/opt/venv/bin:$PATH
@@ -27,22 +26,22 @@ RUN python3 -m venv "$VIRTUAL_ENV"
WORKDIR /app
COPY pyproject.toml README.md ./
RUN pip install --upgrade pip \
RUN pip install uv \
&& if [ -n "$TORCH_VERSION" ]; then \
pip install torch=="$TORCH_VERSION" torchvision=="$TORCH_VERSION" torchaudio=="$TORCH_VERSION" --index-url "$TORCH_INDEX_URL"; \
uv pip install --system torch=="$TORCH_VERSION" torchvision=="$TORCH_VERSION" torchaudio=="$TORCH_VERSION" --index-url "$TORCH_INDEX_URL"; \
else \
pip install torch torchvision torchaudio --index-url "$TORCH_INDEX_URL"; \
uv pip install --system torch torchvision torchaudio --index-url "$TORCH_INDEX_URL"; \
fi \
&& pip install --no-cache-dir . \
&& uv pip install --system . \
https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-3.8.0/en_core_web_sm-3.8.0-py3-none-any.whl \
&& pip install --no-cache-dir "mutagen>=1.47.0"
&& uv pip install --system "mutagen>=1.47.0"
COPY abogen ./abogen
# Install onnxruntime-gpu for CUDA acceleration (supertonic uses ONNX Runtime)
# Set USE_GPU=false to skip this for CPU-only deployments
RUN if [ "$USE_GPU" = "true" ]; then \
pip install --no-cache-dir onnxruntime-gpu; \
uv pip install --system onnxruntime-gpu; \
fi
ENV ABOGEN_HOST=0.0.0.0 \
+17 -24
View File
@@ -20,7 +20,7 @@ import numpy as np
import soundfile as sf
import static_ffmpeg
from abogen.tts_backend_registry import get_metadata
from abogen.tts_backend_registry import get_metadata, is_registered_backend, resolve_backend_for_voice
from abogen.epub3.exporter import build_epub3_package
from abogen.kokoro_text_normalization import ApostropheConfig, normalize_for_pipeline, HAS_NUM2WORDS
from abogen.normalization_settings import (
@@ -57,25 +57,26 @@ SAMPLE_RATE = 24000
def _supertonic_voice_from_spec(spec: Any, fallback: str) -> str:
"""Normalize a voice specification for Supertonic.
This function only performs Supertonic-specific normalization (uppercase conversion
and fallback handling). Backend resolution is handled by the registry.
"""
raw = str(spec or "").strip()
fallback_raw = str(fallback or "").strip()
# SuperTonic voices are discrete IDs (M1/F3/...). If we see a Kokoro mix
# formula (contains '*' or '+'), ignore it and fall back to a safe voice.
if not raw or "*" in raw or "+" in raw:
raw = fallback_raw
if not raw or "*" in raw or "+" in raw:
raw = "M1"
# Normalize to uppercase for Supertonic voice IDs
upper = raw.upper() if raw else ""
upper = raw.upper()
if upper in get_metadata("supertonic").voices:
return upper
# If empty or contains formula characters, use fallback
if not upper or "*" in upper or "+" in upper:
upper = fallback_raw.upper() if fallback_raw else ""
fallback_upper = fallback_raw.upper() if fallback_raw else ""
if fallback_upper in get_metadata("supertonic").voices:
return fallback_upper
# If still empty, use default Supertonic voice
if not upper or "*" in upper or "+" in upper:
upper = "M1"
return "M1"
return upper
def _split_speaker_reference(value: Any) -> tuple[Optional[str], str]:
@@ -119,15 +120,7 @@ def _formula_from_kokoro_entry(entry: Mapping[str, Any]) -> str:
def _infer_provider_from_spec(value: Any, fallback: str = "kokoro") -> str:
raw = str(value or "").strip()
if not raw:
return fallback
upper = raw.upper()
if upper in get_metadata("supertonic").voices:
return "supertonic"
if "*" in raw or "+" in raw:
return "kokoro"
return fallback
return resolve_backend_for_voice(str(value or ""), fallback=fallback)
class _JobCancelled(Exception):
@@ -1574,7 +1567,7 @@ def run_conversion_job(job: Job) -> None:
def get_pipeline(provider: str) -> Any:
nonlocal kokoro_cache_ready
provider_norm = str(provider or "kokoro").strip().lower() or "kokoro"
if provider_norm not in {"kokoro", "supertonic"}:
if not is_registered_backend(provider_norm):
provider_norm = "kokoro"
existing = pipelines.get(provider_norm)
+3 -2
View File
@@ -34,6 +34,7 @@ from abogen.normalization_settings import (
)
from abogen.llm_client import list_models, LLMClientError
from abogen.kokoro_text_normalization import normalize_for_pipeline
from abogen.tts_backend_registry import is_registered_backend
from abogen.integrations.audiobookshelf import AudiobookshelfClient, AudiobookshelfConfig
from abogen.integrations.calibre_opds import (
CalibreOPDSClient,
@@ -63,7 +64,7 @@ def api_save_voice_profile() -> ResponseReturnValue:
if profile is None:
# Speaker Studio payload format
provider = str(payload.get("provider") or "kokoro").strip().lower()
if provider not in {"kokoro", "supertonic"}:
if not is_registered_backend(provider):
provider = "kokoro"
if provider == "supertonic":
profile = {
@@ -230,7 +231,7 @@ def api_speaker_preview() -> ResponseReturnValue:
use_gpu = settings.get("use_gpu", False)
base_spec, speaker_name = split_profile_spec(voice)
resolved_provider = tts_provider if tts_provider in {"kokoro", "supertonic"} else ""
resolved_provider = tts_provider if is_registered_backend(tts_provider) else ""
if speaker_name:
entry = normalize_profile_entry(load_profiles().get(speaker_name))
+2 -1
View File
@@ -7,6 +7,7 @@ from flask.typing import ResponseReturnValue
from abogen.webui.service import PendingJob, JobStatus
from abogen.webui.routes.utils.service import get_service
from abogen.tts_backend_registry import is_registered_backend
from abogen.webui.routes.utils.settings import (
load_settings,
coerce_bool,
@@ -579,7 +580,7 @@ def apply_book_step_form(
# spec (e.g. "speaker:Name" for saved speakers, or a Kokoro mix formula).
# This enables mixed-provider conversions (e.g. narrator=SuperTonic, characters=Kokoro).
provider_value = str(form.get("tts_provider") or "").strip().lower()
if provider_value in {"kokoro", "supertonic"}:
if is_registered_backend(provider_value):
pending.tts_provider = provider_value
# Determine the base speaker selection (saved speaker ref or raw voice).
+2 -2
View File
@@ -1,7 +1,7 @@
from types import SimpleNamespace
from typing import cast
from abogen.constants import VOICES_INTERNAL
from abogen.tts_backend_registry import get_metadata
from abogen.webui.conversion_runner import (
_chapter_voice_spec,
_chunk_voice_spec,
@@ -49,4 +49,4 @@ def test_voice_collection_includes_formula_components():
voices = _collect_required_voice_ids(job)
assert {"af_nova", "am_liam"}.issubset(voices)
assert voices.issuperset(VOICES_INTERNAL)
assert voices.issuperset(get_metadata("kokoro").voices)
+1 -1
View File
@@ -197,7 +197,7 @@ def test_epub3_preserves_original_whitespace(tmp_path) -> None:
)
assert match is not None
original_text = html.unescape(match.group(1))
assert "Second line\n\nThird paragraph." in original_text
assert "Second line\n\nThird paragraph." in original_text.replace("\r\n", "\n")
def test_epub3_sentence_chunks_render_as_paragraphs(tmp_path) -> None:
+110
View File
@@ -202,3 +202,113 @@ class TestBackendRegistration:
registry_meta = _registry.get_metadata("supertonic")
assert _SUPERTONIC_METADATA is registry_meta
assert _SUPERTONIC_METADATA.voices == registry_meta.voices
class TestResolveBackendForVoice:
"""Tests for the resolve_backend_for_voice method."""
def test_empty_spec_returns_fallback(self):
registry = TTSBackendRegistry()
assert registry.resolve_backend_for_voice("", fallback="kokoro") == "kokoro"
assert registry.resolve_backend_for_voice("", fallback="supertonic") == "supertonic"
def test_none_spec_returns_fallback(self):
registry = TTSBackendRegistry()
assert registry.resolve_backend_for_voice(None, fallback="kokoro") == "kokoro"
def test_kokoro_formula_with_star_returns_kokoro(self):
registry = TTSBackendRegistry()
assert registry.resolve_backend_for_voice("af_nova*0.7") == "kokoro"
def test_kokoro_formula_with_plus_returns_kokoro(self):
registry = TTSBackendRegistry()
assert registry.resolve_backend_for_voice("af_nova*0.7+am_liam*0.3") == "kokoro"
def test_kokoro_voice_id_resolves_to_kokoro(self):
registry = TTSBackendRegistry()
meta = TTSBackendMetadata(
id="kokoro",
name="Kokoro",
description="Kokoro TTS",
voices=("af_nova", "am_liam"),
)
registry.register(metadata=meta, factory=lambda: None)
assert registry.resolve_backend_for_voice("af_nova") == "kokoro"
assert registry.resolve_backend_for_voice("am_liam") == "kokoro"
def test_supertonic_voice_id_resolves_to_supertonic(self):
registry = TTSBackendRegistry()
meta = TTSBackendMetadata(
id="supertonic",
name="SuperTonic",
description="SuperTonic TTS",
voices=("M1", "M2", "F1", "F2"),
)
registry.register(metadata=meta, factory=lambda: None)
assert registry.resolve_backend_for_voice("M1") == "supertonic"
assert registry.resolve_backend_for_voice("F2") == "supertonic"
def test_unknown_voice_returns_fallback(self):
registry = TTSBackendRegistry()
meta = TTSBackendMetadata(
id="kokoro",
name="Kokoro",
description="Kokoro TTS",
voices=("af_nova",),
)
registry.register(metadata=meta, factory=lambda: None)
assert registry.resolve_backend_for_voice("unknown_voice") == "kokoro"
assert registry.resolve_backend_for_voice("unknown_voice", fallback="supertonic") == "supertonic"
def test_case_insensitive_matching(self):
registry = TTSBackendRegistry()
meta = TTSBackendMetadata(
id="supertonic",
name="SuperTonic",
description="SuperTonic TTS",
voices=("M1", "F1"),
)
registry.register(metadata=meta, factory=lambda: None)
assert registry.resolve_backend_for_voice("m1") == "supertonic"
assert registry.resolve_backend_for_voice("f1") == "supertonic"
def test_default_fallback_is_kokoro(self):
registry = TTSBackendRegistry()
assert registry.resolve_backend_for_voice("unknown") == "kokoro"
def test_multiple_backends_resolution(self):
registry = TTSBackendRegistry()
kokoro_meta = TTSBackendMetadata(
id="kokoro",
name="Kokoro",
description="Kokoro TTS",
voices=("af_nova",),
)
supertonic_meta = TTSBackendMetadata(
id="supertonic",
name="SuperTonic",
description="SuperTonic TTS",
voices=("M1",),
)
registry.register(metadata=kokoro_meta, factory=lambda: None)
registry.register(metadata=supertonic_meta, factory=lambda: None)
assert registry.resolve_backend_for_voice("af_nova") == "kokoro"
assert registry.resolve_backend_for_voice("M1") == "supertonic"
def test_global_wrapper_resolve_backend_for_voice(self):
from abogen.tts_backend_registry import resolve_backend_for_voice
# Test with empty spec
assert resolve_backend_for_voice("") == "kokoro"
# Test with formula
assert resolve_backend_for_voice("af_nova*0.7") == "kokoro"
# Test with a registered voice
assert resolve_backend_for_voice("af_nova") == "kokoro"
assert resolve_backend_for_voice("M1") == "supertonic"
+2 -2
View File
@@ -3,7 +3,7 @@ from typing import cast
import pytest
from abogen.constants import VOICES_INTERNAL
from abogen.tts_backend_registry import get_metadata
from abogen.voice_cache import (
LocalEntryNotFoundError,
_CACHED_VOICES,
@@ -66,4 +66,4 @@ def test_collect_required_voice_ids_includes_all():
voices = _collect_required_voice_ids(cast(Job, job))
assert {"af_nova", "am_liam", "am_michael"}.issubset(voices)
assert voices.issuperset(VOICES_INTERNAL)
assert voices.issuperset(get_metadata("kokoro").voices)