Compare commits

..
90 changed files with 1161 additions and 2078 deletions
+11 -32
View File
@@ -1,9 +1,7 @@
name: CI
run-name: CI
name: pip install
run-name: pip install
on:
push:
branches: [main]
paths:
- '**.py'
- 'pyproject.toml'
@@ -13,42 +11,23 @@ on:
- 'pyproject.toml'
- '.github/workflows/**'
workflow_dispatch:
jobs:
test:
install-and-run:
strategy:
matrix:
os: [ubuntu-latest, macos-14, windows-latest]
os: [ubuntu-latest, macos-latest, windows-latest]
python-version: ['3.12']
fail-fast: false
continue-on-error: true
runs-on: ${{ matrix.os }}
steps:
- name: Checkout repository
uses: actions/checkout@v7
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v6
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- 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
- name: Install from repository
run: python -m pip install .
#- name: Run abogen
# run: abogen
+1 -1
View File
@@ -18,7 +18,7 @@ jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v7
- uses: actions/checkout@v4
- name: Login to Github Container Registry
# Only if we need to push an image
+18 -13
View File
@@ -38,6 +38,8 @@ This method handles everything automatically - installing all dependencies inclu
#### <b>OPTION 2: Install using uv</b>
First, [install uv](https://docs.astral.sh/uv/getting-started/installation/) if you haven't already.
The CUDA extras install both GPU-accelerated Kokoro (via PyTorch) and Supertonic (via onnxruntime-gpu).
```bash
# For NVIDIA GPUs (CUDA 12.8) - Recommended
uv tool install --python 3.12 abogen[cuda] --extra-index-url https://download.pytorch.org/whl/cu128 --index-strategy unsafe-best-match
@@ -65,6 +67,9 @@ venv\Scripts\activate
# We need to use an older version of PyTorch (2.8.0) until this issue is fixed: https://github.com/pytorch/pytorch/issues/166628
pip install torch==2.8.0+cu128 torchvision==0.23.0+cu128 torchaudio==2.8.0 --index-url https://download.pytorch.org/whl/cu128
# Also install onnxruntime-gpu for Supertonic GPU acceleration:
pip install onnxruntime-gpu
# For AMD GPUs:
# Not supported yet, because ROCm is not available on Windows. Use Linux if you have AMD GPU.
@@ -173,7 +178,7 @@ Abogen offers **two interfaces**, but currently they have different feature sets
| Command | Interface | Features |
|---------|-----------|----------|
| `abogen` | PyQt6 Desktop GUI | Stable core features |
| `abogen` | PyQt6 Desktop GUI | Stable core features + **Supertonic TTS**|
| `abogen-web` | Flask Web UI | Core features + **Supertonic TTS**, **LLM Normalization**, **Audiobookshelf Integration** and more! |
> **Note:** The Web UI is under active development. We are working to integrate these new features into the PyQt desktop app. until then, the Web UI provides the most feature-rich experience.
@@ -407,18 +412,18 @@ When Audiobookshelf sits behind Nginx Proxy Manager (NPM), make sure the API pat
1. Create a **Proxy Host** that points to your ABS container or host (default forward port `13378`).
2. Under the **SSL** tab, enable your certificate and tick **Force SSL** if you want HTTPS only.
3. In the **Advanced** tab, append the snippet below so bearer tokens, client IPs, and large uploads survive the proxy hop:
```nginx
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header X-Forwarded-Host $host;
proxy_set_header X-Forwarded-Port $server_port;
proxy_set_header Authorization $http_authorization;
client_max_body_size 5g;
proxy_read_timeout 300s;
proxy_connect_timeout 300s;
```
```nginx
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header X-Forwarded-Host $host;
proxy_set_header X-Forwarded-Port $server_port;
proxy_set_header Authorization $http_authorization;
client_max_body_size 5g;
proxy_read_timeout 300s;
proxy_connect_timeout 300s;
```
4. Disable **Block Common Exploits** (it strips Authorization headers in some NPM builds).
5. Enable **Websockets Support** on the main proxy screen (Audiobookshelf uses it for the web UI, and it keeps the reverse proxy configuration consistent).
6. If you publish Audiobookshelf under a path prefix (for example `/abs`), add a **Custom Location** with `Location: /abs/` and set the **Forward Path** to `/`. That rewrite strips the `/abs` prefix before traffic reaches Audiobookshelf so `/abs/api/...` on the internet becomes `/api/...` on the backend. Use the same prefixed URL in Abogens “Base URL” field.
+14
View File
@@ -323,6 +323,13 @@ if /I "%IS_NVIDIA%"=="true" (
pause
exit /b
)
echo Installing onnxruntime-gpu for Supertonic GPU acceleration...
%PYTHON_CONSOLE_PATH% -m uv pip install --system onnxruntime-gpu
if errorlevel 1 (
echo Failed to install onnxruntime-gpu.
pause
exit /b
)
) else (
echo CUDA is available on NVIDIA GPU.
)
@@ -348,6 +355,13 @@ if /I "%IS_NVIDIA%"=="true" (
pause
exit /b
)
echo Installing onnxruntime-gpu for Supertonic GPU acceleration...
%PYTHON_CONSOLE_PATH% -m uv pip install --system onnxruntime-gpu
if errorlevel 1 (
echo Failed to install onnxruntime-gpu.
pause
exit /b
)
)
)
Binary file not shown.

Before

Width:  |  Height:  |  Size: 571 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 992 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 786 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 877 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.5 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.0 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 832 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.0 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 521 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 868 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 808 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 372 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.1 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 883 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.6 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 465 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.5 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 885 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 381 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 855 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 917 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 441 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.5 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 856 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 837 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 875 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 617 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 843 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 875 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 891 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.0 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.1 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.3 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.1 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 851 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.3 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.1 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 431 B

+109
View File
@@ -29,6 +29,57 @@ LANGUAGE_DESCRIPTIONS = {
"z": "Mandarin Chinese",
}
# Mapping from Kokoro single-letter language codes to ISO 3166-1 alpha-2 country codes
# Used for loading flag icons
KOKORO_LANG_TO_COUNTRY = {
"a": "us", # American English -> United States
"b": "gb", # British English -> United Kingdom
"e": "es", # Spanish -> Spain
"f": "fr", # French -> France
"h": "in", # Hindi -> India
"i": "it", # Italian -> Italy
"j": "jp", # Japanese -> Japan
"p": "br", # Brazilian Portuguese -> Brazil
"z": "cn", # Mandarin Chinese -> China
}
# Mapping from Supertonic ISO 639-1 language codes to ISO 3166-1 alpha-2 country codes
# Used for loading flag icons in the Supertonic language picker
SUPERTONIC_LANG_TO_COUNTRY = {
"en": "gb",
"ko": "kr",
"ja": "jp",
"ar": "ae",
"bg": "bg",
"cs": "cz",
"da": "dk",
"de": "de",
"el": "gr",
"es": "es",
"et": "ee",
"fi": "fi",
"fr": "fr",
"hi": "in",
"hr": "hr",
"hu": "hu",
"id": "id",
"it": "it",
"lt": "lt",
"lv": "lv",
"nl": "nl",
"pl": "pl",
"pt": "pt",
"ro": "ro",
"ru": "ru",
"sk": "sk",
"sl": "si",
"sv": "se",
"tr": "tr",
"uk": "ua",
"vi": "vn",
"na": "na",
}
# Supported sound formats
SUPPORTED_SOUND_FORMATS = [
"wav",
@@ -63,6 +114,64 @@ 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.",
+4 -5
View File
@@ -21,8 +21,7 @@ from PyQt6.QtWidgets import (
)
from PyQt6.QtCore import QThread, pyqtSignal
from abogen.constants import COLORS
from abogen.tts_backend_registry import get_metadata
from abogen.constants import COLORS, VOICES_INTERNAL
from abogen.spacy_utils import SPACY_MODELS
import abogen.hf_tracker
@@ -115,7 +114,7 @@ class PreDownloadWorker(QThread):
self._voices_success = False
return
voice_list = get_metadata("kokoro").voices
voice_list = VOICES_INTERNAL
for idx, voice in enumerate(voice_list, start=1):
if self._cancelled:
self._voices_success = False
@@ -463,14 +462,14 @@ class PreDownloadDialog(QDialog):
try:
from huggingface_hub import try_to_load_from_cache
for voice in get_metadata("kokoro").voices:
for voice in VOICES_INTERNAL:
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(get_metadata("kokoro").voices)
return False, list(VOICES_INTERNAL)
return (len(missing) == 0), missing
def _check_kokoro_model(self) -> bool:
+134 -102
View File
@@ -5,7 +5,6 @@ import hashlib # For generating unique cache filenames
from platformdirs import user_desktop_dir
from PyQt6.QtCore import QThread, pyqtSignal, Qt, QTimer
from PyQt6.QtWidgets import QCheckBox, QVBoxLayout, QDialog, QLabel, QDialogButtonBox
import numpy as np
import soundfile as sf
from abogen.utils import (
create_process,
@@ -21,6 +20,7 @@ from abogen.constants import (
SUPPORTED_SOUND_FORMATS,
SUPPORTED_SUBTITLE_FORMATS,
)
from abogen.tts_supertonic import SupertonicPipeline, SUPERTONIC_AVAILABLE_LANGS, DEFAULT_SUPERTONIC_VOICES
from abogen.voice_formulas import get_new_voice
import abogen.hf_tracker as hf_tracker
import static_ffmpeg
@@ -260,21 +260,30 @@ class ConversionThread(QThread):
output_folder,
subtitle_mode,
output_format,
backend,
np_module,
kpipeline_class,
start_time,
total_char_count,
use_gpu=True,
from_queue=False,
save_base_path=None,
): # Add use_gpu parameter
tts_provider="kokoro",
supertonic_language="en",
supertonic_total_steps=8,
):
super().__init__()
self._chapter_options_event = threading.Event()
self._timestamp_response_event = threading.Event()
self.backend = backend
self.np = np_module
self.KPipeline = kpipeline_class
self.file_name = file_name
self.lang_code = lang_code
self.speed = speed
self.voice = voice
self.tts_provider = tts_provider
self.supertonic_language = supertonic_language
self.supertonic_total_steps = supertonic_total_steps
self.sample_rate = 44100 if tts_provider == "supertonic" else 24000
self.save_option = save_option
self.output_folder = output_folder
self.subtitle_mode = subtitle_mode
@@ -426,6 +435,10 @@ class ConversionThread(QThread):
)
self.log_updated.emit(f"- Voice: {self.voice}")
self.log_updated.emit(f"- Speed: {self.speed}")
tts_provider_label = self.tts_provider.capitalize()
if self.tts_provider == "supertonic":
tts_provider_label += f" (lang={self.supertonic_language}, steps={self.supertonic_total_steps})"
self.log_updated.emit(f"- TTS Engine: {tts_provider_label}")
self.log_updated.emit(f"- Subtitle mode: {self.subtitle_mode}")
self.log_updated.emit(f"- Output format: {self.output_format}")
self.log_updated.emit(
@@ -489,6 +502,26 @@ class ConversionThread(QThread):
self.log_updated.emit(("\nInitializing TTS pipeline...", "grey"))
# Set device based on use_gpu setting and platform
if self.use_gpu:
if platform.system() == "Darwin" and platform.processor() == "arm":
device = "mps" # Use MPS for Apple Silicon
else:
device = "cuda" # Use CUDA for other platforms
else:
device = "cpu"
if self.tts_provider == "supertonic":
tts = SupertonicPipeline(
sample_rate=self.sample_rate,
lang=self.supertonic_language,
total_steps=self.supertonic_total_steps,
)
else:
tts = self.KPipeline(
lang_code=self.lang_code, repo_id="hexgrad/Kokoro-82M", device=device
)
# Check if the input is a subtitle file or timestamp text file
is_subtitle_file = False
is_timestamp_text = False
@@ -524,7 +557,7 @@ class ConversionThread(QThread):
# Process subtitle files separately
if is_subtitle_file or is_timestamp_text:
self._process_subtitle_file(self.backend, base_path, is_timestamp_text)
self._process_subtitle_file(tts, base_path, is_timestamp_text)
return
if self.is_direct_text:
@@ -733,7 +766,7 @@ class ConversionThread(QThread):
merged_out_path = f"{base_filepath_no_ext}.{self.output_format}"
subtitle_entries = []
current_time = 0.0
rate = 24000
rate = self.sample_rate
subtitle_mode = self.subtitle_mode
self.etr_start_time = time.time()
self.processed_char_count = 0
@@ -749,7 +782,7 @@ class ConversionThread(QThread):
merged_out_file = sf.SoundFile(
merged_out_path,
"w",
samplerate=24000,
samplerate=self.sample_rate,
channels=1,
format=self.output_format,
)
@@ -765,62 +798,18 @@ class ConversionThread(QThread):
# Prepare ffmpeg command for m4b output
cmd = [
"ffmpeg",
"-y",
"-thread_queue_size",
"32768",
"-f",
"f32le",
"-ar",
"24000",
"-ac",
"1",
"-i",
"pipe:0",
]
if cover_path and os.path.exists(cover_path):
cmd.extend(
[
"-i",
cover_path,
"-map",
"0:a",
"-map",
"1",
"-c:v",
"copy",
"-disposition:v",
"attached_pic",
]
)
cmd.extend(
[
"-c:a",
"aac",
"-q:a",
"2",
"-movflags",
"+faststart+use_metadata_tags",
"-y",
"-thread_queue_size",
"32768",
"-f",
"f32le",
"-ar",
str(self.sample_rate),
"-ac",
"1",
"-i",
"pipe:0",
]
)
cmd += metadata_options
cmd.append(merged_out_path)
ffmpeg_proc = create_process(cmd, stdin=subprocess.PIPE, text=False)
elif self.output_format == "opus":
static_ffmpeg.add_paths()
cmd = [
"ffmpeg",
"-y",
"-thread_queue_size",
"32768",
"-f",
"f32le",
"-ar",
"24000",
"-ac",
"1",
"-i",
"pipe:0",
]
cmd.extend(["-c:a", "libopus", "-b:a", "24000"])
cmd.append(merged_out_path)
ffmpeg_proc = create_process(cmd, stdin=subprocess.PIPE, text=False)
@@ -902,7 +891,7 @@ class ConversionThread(QThread):
merged_out_path = None
subtitle_entries = []
current_time = 0.0
rate = 24000
rate = self.sample_rate
subtitle_mode = self.subtitle_mode
self.etr_start_time = time.time()
self.processed_char_count = 0
@@ -955,7 +944,7 @@ class ConversionThread(QThread):
chapter_out_file = sf.SoundFile(
chapter_out_path,
"w",
samplerate=24000,
samplerate=self.sample_rate,
channels=1,
format=separate_chapters_format,
)
@@ -970,7 +959,7 @@ class ConversionThread(QThread):
"-f",
"f32le",
"-ar",
"24000",
str(self.sample_rate),
"-ac",
"1",
"-i",
@@ -1057,7 +1046,7 @@ class ConversionThread(QThread):
for segment_idx, (voice_name, segment_text) in enumerate(voice_segments):
# Load voice for this segment (with caching)
try:
loaded_voice = self.load_voice_cached(voice_name, self.backend)
loaded_voice = self.load_voice_cached(voice_name, tts)
if segment_idx > 0:
voice_display = voice_name if len(voice_name) < 50 else voice_name[:47] + "..."
self.log_updated.emit((f" → Voice: {voice_display}", "grey"))
@@ -1066,7 +1055,7 @@ class ConversionThread(QThread):
(f"⚠ Voice loading error for '{voice_name}', continuing with previous", "orange")
)
if segment_idx == 0:
loaded_voice = self.load_voice_cached(self.voice, self.backend)
loaded_voice = self.load_voice_cached(self.voice, tts)
# Determine if spaCy segmentation should be used for PRE-TTS segmentation
# Only non-English languages use spaCy for pre-segmentation
@@ -1152,7 +1141,7 @@ class ConversionThread(QThread):
print("Using split pattern: (unprintable)")
for text_segment in text_segments:
for result in self.backend(
for result in tts(
text_segment,
voice=loaded_voice,
speed=self.speed,
@@ -1352,9 +1341,9 @@ class ConversionThread(QThread):
# Add silence between chapters for merged output (except after the last chapter)
if merge_chapters_at_end and chapter_idx < total_chapters:
silence_samples = int(
self.silence_duration * 24000
self.silence_duration * self.sample_rate
) # Silence duration at 24,000 Hz
silence_audio = np.zeros(silence_samples, dtype="float32")
silence_audio = self.np.zeros(silence_samples, dtype="float32")
silence_bytes = silence_audio.tobytes()
if merged_out_file:
@@ -1583,7 +1572,7 @@ class ConversionThread(QThread):
parent_dir, f"{sanitized_base_name}{suffix}"
)
merged_out_path = f"{base_filepath_no_ext}.{self.output_format}"
rate = 24000
rate = self.sample_rate
# Setup audio output
merged_out_file, ffmpeg_proc = None, None
@@ -1693,7 +1682,7 @@ class ConversionThread(QThread):
max_end_time = max(
(end for _, end, _ in subtitles if end is not None), default=0
)
audio_buffer = np.zeros(
audio_buffer = self.np.zeros(
int(max_end_time * rate) + rate, dtype="float32"
)
@@ -1757,7 +1746,7 @@ class ConversionThread(QThread):
# Generate TTS audio
tts_results = [
r
for r in self.backend(
for r in tts(
processed_text,
voice=loaded_voice,
speed=self.speed,
@@ -1775,11 +1764,11 @@ class ConversionThread(QThread):
# Concatenate audio and determine duration
full_audio = (
np.concatenate(
self.np.concatenate(
[a.numpy() if hasattr(a, "numpy") else a for a in audio_chunks]
)
if audio_chunks
else np.zeros(
else self.np.zeros(
int((subtitle_duration or 0) * rate), dtype="float32"
)
)
@@ -1813,8 +1802,8 @@ class ConversionThread(QThread):
num_stages = max(
1,
int(
np.ceil(
np.log(speed_factor) / np.log(2.0)
self.np.ceil(
self.np.log(speed_factor) / self.np.log(2.0)
)
),
)
@@ -1847,7 +1836,7 @@ class ConversionThread(QThread):
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
full_audio = np.frombuffer(
full_audio = self.np.frombuffer(
speed_proc.communicate(input=full_audio.tobytes())[0],
dtype="float32",
)
@@ -1861,7 +1850,7 @@ class ConversionThread(QThread):
tts_results = [
r
for r in self.backend(
for r in tts(
processed_text,
voice=loaded_voice,
speed=new_speed,
@@ -1872,14 +1861,14 @@ class ConversionThread(QThread):
audio_chunks = [r.audio for r in tts_results]
full_audio = (
np.concatenate(
self.np.concatenate(
[
a.numpy() if hasattr(a, "numpy") else a
for a in audio_chunks
]
)
if audio_chunks
else np.zeros(
else self.np.zeros(
int(subtitle_duration * rate), dtype="float32"
)
)
@@ -1896,10 +1885,10 @@ class ConversionThread(QThread):
# Pad or trim to subtitle duration
target_samples = int(subtitle_duration * rate)
if len(full_audio) < target_samples:
full_audio = np.concatenate(
full_audio = self.np.concatenate(
[
full_audio,
np.zeros(
self.np.zeros(
target_samples - len(full_audio), dtype="float32"
),
]
@@ -1912,10 +1901,10 @@ class ConversionThread(QThread):
end_sample = start_sample + len(full_audio)
if end_sample > len(audio_buffer):
# Extend buffer if needed
audio_buffer = np.concatenate(
audio_buffer = self.np.concatenate(
[
audio_buffer,
np.zeros(
self.np.zeros(
end_sample - len(audio_buffer), dtype="float32"
),
]
@@ -1957,7 +1946,7 @@ class ConversionThread(QThread):
self.progress_updated.emit(percent, etr_str)
# Normalize audio buffer to prevent clipping from mixed overlaps
max_amplitude = np.abs(audio_buffer).max()
max_amplitude = self.np.abs(audio_buffer).max()
if max_amplitude > 1.0:
self.log_updated.emit(
f"\n -> Normalizing audio (peak: {max_amplitude:.2f})"
@@ -2426,19 +2415,28 @@ class VoicePreviewThread(QThread):
def __init__(
self,
backend,
np_module,
kpipeline_class,
lang_code,
voice,
speed,
use_gpu=False,
parent=None,
tts_provider="kokoro",
supertonic_language="en",
supertonic_total_steps=8,
):
super().__init__(parent)
self.backend = backend
self.np_module = np_module
self.kpipeline_class = kpipeline_class
self.lang_code = lang_code
self.voice = voice
self.speed = speed
self.use_gpu = use_gpu
self.tts_provider = tts_provider
self.supertonic_language = supertonic_language
self.supertonic_total_steps = supertonic_total_steps
self.sample_rate = 44100 if tts_provider == "supertonic" else 24000
# Cache location for preview audio
self.cache_dir = get_user_cache_path("preview_cache")
@@ -2448,6 +2446,11 @@ class VoicePreviewThread(QThread):
def _get_cache_path(self):
"""Generate a unique filename for the voice with its parameters"""
if self.tts_provider == "supertonic":
voice_id = self.voice or "M1"
filename = f"st_{voice_id}_{self.supertonic_language}_steps{self.supertonic_total_steps}_{self.speed:.2f}.wav"
return os.path.join(self.cache_dir, filename)
# For a voice formula, use a hash of the formula
if "*" in self.voice:
voice_id = (
@@ -2462,27 +2465,56 @@ class VoicePreviewThread(QThread):
def run(self):
print(
f"\nVoice: {self.voice}\nLanguage: {self.lang_code}\nSpeed: {self.speed}\nGPU: {self.use_gpu}\n"
f"\nVoice: {self.voice}\nLanguage: {self.lang_code}\nSpeed: {self.speed}\nGPU: {self.use_gpu}\nTTS Provider: {self.tts_provider}\n"
)
# Generate the preview and save to cache
try:
if self.tts_provider == "supertonic":
from abogen.tts_supertonic import SupertonicPipeline
# Enable voice formula support for preview
if "*" in self.voice:
loaded_voice = get_new_voice(self.backend, self.voice, self.use_gpu)
tts = SupertonicPipeline(
sample_rate=self.sample_rate,
lang=self.supertonic_language,
total_steps=self.supertonic_total_steps,
)
loaded_voice = self.voice or "M1"
sample_text = "Hello, this is a sample of the selected voice."
audio_segments = []
for result in tts(
sample_text,
voice=loaded_voice,
speed=self.speed,
split_pattern=None,
):
audio_segments.append(result.audio)
else:
loaded_voice = self.voice
sample_text = get_sample_voice_text(self.lang_code)
audio_segments = []
for result in self.backend(
sample_text, voice=loaded_voice, speed=self.speed, split_pattern=None
):
audio_segments.append(result.audio)
# Set device based on use_gpu setting and platform
if self.use_gpu:
if platform.system() == "Darwin" and platform.processor() == "arm":
device = "mps"
else:
device = "cuda"
else:
device = "cpu"
tts = self.kpipeline_class(
lang_code=self.lang_code, repo_id="hexgrad/Kokoro-82M", device=device
)
if "*" in self.voice:
loaded_voice = get_new_voice(tts, self.voice, self.use_gpu)
else:
loaded_voice = self.voice
sample_text = get_sample_voice_text(self.lang_code)
audio_segments = []
for result in tts(
sample_text, voice=loaded_voice, speed=self.speed, split_pattern=None
):
audio_segments.append(result.audio)
if audio_segments:
audio = np.concatenate(audio_segments)
# Save directly to the cache path
sf.write(self.cache_path, audio, 24000)
audio = self.np_module.concatenate(audio_segments)
sf.write(self.cache_path, audio, self.sample_rate)
self.temp_wav = self.cache_path
self.finished.emit()
except Exception as e:
+256 -51
View File
@@ -9,6 +9,7 @@ from abogen.pyqt.queue_manager_gui import QueueManager
from abogen.pyqt.queued_item import QueuedItem
import abogen.hf_tracker as hf_tracker
import hashlib # Added for cache path generation
from abogen.tts_supertonic import SUPERTONIC_AVAILABLE_LANGS, DEFAULT_SUPERTONIC_VOICES
from PyQt6.QtWidgets import (
QApplication,
QWidget,
@@ -82,11 +83,13 @@ from abogen.constants import (
GITHUB_URL,
PROGRAM_DESCRIPTION,
LANGUAGE_DESCRIPTIONS,
VOICES_INTERNAL,
KOKORO_LANG_TO_COUNTRY,
SUPERTONIC_LANG_TO_COUNTRY,
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
@@ -970,6 +973,9 @@ class abogen(QWidget):
self.fix_nonstandard_punctuation = self.config.get(
"fix_nonstandard_punctuation", False
)
self.tts_provider_config = self.config.get("tts_provider", "kokoro")
self.supertonic_language_config = self.config.get("supertonic_language", "en")
self.supertonic_total_steps_config = self.config.get("supertonic_total_steps", 8)
self._pending_close_event = None
self.gpu_ok = False # Initialize GPU availability status
@@ -1018,6 +1024,16 @@ class abogen(QWidget):
else:
self.mixed_voice_state = entry
self.selected_lang = entry[0][0] if entry and entry[0] else None
# Restore TTS provider and supertonic settings from config
provider_text = "Supertonic" if self.tts_provider_config == "supertonic" else "Kokoro"
idx_st = self.provider_combo.findText(provider_text)
if idx_st >= 0:
self.provider_combo.setCurrentIndex(idx_st)
self.st_lang_combo.setCurrentText(self.supertonic_language_config)
idx_steps = self.st_steps_combo.findData(self.supertonic_total_steps_config)
if idx_steps >= 0:
self.st_steps_combo.setCurrentIndex(idx_steps)
if self.save_option == "Choose output folder" and self.selected_output_folder:
self.save_path_label.setText(self.selected_output_folder)
self.save_path_row_widget.show()
@@ -1107,6 +1123,53 @@ class abogen(QWidget):
speed_layout.addWidget(self.speed_label)
controls_layout.addLayout(speed_layout)
self.speed_slider.valueChanged.connect(self.update_speed_label)
# TTS Provider selection
provider_layout = QHBoxLayout()
provider_layout.setSpacing(7)
provider_label = QLabel("TTS Engine:", self)
provider_layout.addWidget(provider_label)
self.provider_combo = QComboBox(self)
self.provider_combo.addItem("Kokoro", "kokoro")
self.provider_combo.addItem("Supertonic", "supertonic")
self.provider_combo.setStyleSheet("QComboBox { min-height: 20px; padding: 6px 12px; }")
self.provider_combo.currentIndexChanged.connect(self.on_provider_changed)
provider_layout.addWidget(self.provider_combo)
controls_layout.addLayout(provider_layout)
# Supertonic-specific controls (language + steps), hidden by default
self.supertonic_row = QWidget()
supertonic_row_layout = QHBoxLayout(self.supertonic_row)
supertonic_row_layout.setContentsMargins(0, 0, 0, 0)
supertonic_row_layout.setSpacing(7)
st_lang_label = QLabel("Language:", self)
supertonic_row_layout.addWidget(st_lang_label)
self.st_lang_combo = QComboBox(self)
for code in SUPERTONIC_AVAILABLE_LANGS:
country_code = SUPERTONIC_LANG_TO_COUNTRY.get(code, code)
flag = get_resource_path("abogen.assets.flags", f"{country_code}.png")
icon_st = QIcon(flag) if flag and os.path.exists(flag) else QIcon()
self.st_lang_combo.addItem(icon_st, code, code)
self.st_lang_combo.setCurrentText("en")
self.st_lang_combo.setStyleSheet("QComboBox { min-height: 20px; padding: 6px 12px; }")
self.st_lang_combo.currentTextChanged.connect(self._on_st_lang_changed)
supertonic_row_layout.addWidget(self.st_lang_combo)
st_steps_label = QLabel("Steps:", self)
supertonic_row_layout.addWidget(st_steps_label)
self.st_steps_combo = QComboBox(self)
for val in range(2, 16):
self.st_steps_combo.addItem(str(val), val)
self.st_steps_combo.setCurrentIndex(self.st_steps_combo.findData(8))
self.st_steps_combo.setStyleSheet("QComboBox { min-height: 20px; padding: 6px 12px; }")
self.st_steps_combo.currentIndexChanged.connect(self._on_st_steps_changed)
supertonic_row_layout.addWidget(self.st_steps_combo)
supertonic_row_layout.addStretch()
self.supertonic_row.hide()
controls_layout.addWidget(self.supertonic_row)
# Voice selection
voice_layout = QHBoxLayout()
voice_layout.setSpacing(7)
@@ -1764,6 +1827,12 @@ class abogen(QWidget):
Update the enabled state of subtitle options based on the selected language.
For non-English languages, only sentence-based and line-based modes are supported.
"""
provider = self.provider_combo.currentData()
if provider == "supertonic":
self.subtitle_combo.setEnabled(False)
self.subtitle_format_combo.setEnabled(False)
return
# Check if current file is a subtitle file
is_subtitle_input = False
if self.selected_file and self.selected_file.lower().endswith(
@@ -1823,6 +1892,48 @@ class abogen(QWidget):
# Enable/disable subtitle options based on language
self.update_subtitle_options_availability()
def on_provider_changed(self, index):
provider = self.provider_combo.itemData(index)
self.config["tts_provider"] = provider
save_config(self.config)
is_supertonic = provider == "supertonic"
# Show/hide Supertonic controls
self.supertonic_row.setVisible(is_supertonic)
# Update subtitles availability
self.update_subtitle_options_availability()
# Repopulate voice list
self.populate_profiles_in_voice_combo()
# Clear/reset mixed voice state when switching provider
if is_supertonic:
self.mixed_voice_state = None
self.btn_voice_formula_mixer.setEnabled(False)
self.voice_combo.setToolTip(
"Supertonic voices:\n"
"M1-M5 = Male voices\n"
"F1-F5 = Female voices"
)
else:
self.btn_voice_formula_mixer.setEnabled(True)
self.voice_combo.setToolTip(
"The first character represents the language:\n"
'"a" => American English\n"b" => British English\n"e" => Spanish\n"f" => French\n"h" => Hindi\n"i" => Italian\n"j" => Japanese\n"p" => Brazilian Portuguese\n"z" => Mandarin Chinese\nThe second character represents the gender:\n"m" => Male\n"f" => Female'
)
def _on_st_lang_changed(self, lang):
self.config["supertonic_language"] = lang
save_config(self.config)
if self.provider_combo.currentData() == "supertonic":
self.selected_lang = lang
self.update_subtitle_options_availability()
def _on_st_steps_changed(self):
self.config["supertonic_total_steps"] = self.st_steps_combo.currentData()
save_config(self.config)
def on_voice_combo_changed(self, index):
data = self.voice_combo.itemData(index)
if isinstance(data, str) and data.startswith("profile:"):
@@ -1831,10 +1942,26 @@ class abogen(QWidget):
from abogen.voice_profiles import load_profiles
entry = load_profiles().get(pname, {})
# set mixed voices and language
if isinstance(entry, dict):
self.mixed_voice_state = entry.get("voices", [])
self.selected_lang = entry.get("language")
entry_provider = str(entry.get("provider", "")).strip().lower()
if entry_provider == "supertonic":
# Switch provider to Supertonic if not already
if self.provider_combo.currentData() != "supertonic":
self.provider_combo.setCurrentIndex(1)
self.mixed_voice_state = None
self.selected_lang = entry.get("language", self.st_lang_combo.currentText())
# Sync supertonic controls from profile
profile_steps = entry.get("total_steps")
if profile_steps is not None:
idx_steps = self.st_steps_combo.findData(int(profile_steps))
if idx_steps >= 0:
self.st_steps_combo.setCurrentIndex(idx_steps)
profile_lang = entry.get("language")
if profile_lang and profile_lang in SUPERTONIC_AVAILABLE_LANGS:
self.st_lang_combo.setCurrentText(profile_lang)
else:
self.mixed_voice_state = entry.get("voices", [])
self.selected_lang = entry.get("language")
else:
self.mixed_voice_state = entry
self.selected_lang = entry[0][0] if entry and entry[0] else None
@@ -1847,7 +1974,12 @@ class abogen(QWidget):
else:
self.mixed_voice_state = None
self.selected_profile_name = None
self.selected_voice, self.selected_lang = data, data[0]
self.selected_voice = data
provider = self.provider_combo.currentData()
if provider == "supertonic":
self.selected_lang = self.st_lang_combo.currentText()
else:
self.selected_lang = data[0] if data else ""
self.config["selected_voice"] = data
if "selected_profile_name" in self.config:
del self.config["selected_profile_name"]
@@ -1866,19 +1998,40 @@ class abogen(QWidget):
def populate_profiles_in_voice_combo(self):
# preserve current voice or profile
current = self.voice_combo.currentData()
provider = self.provider_combo.currentData()
self.voice_combo.blockSignals(True)
self.voice_combo.clear()
# re-add profiles
# re-add profiles matching current provider
profile_icon = QIcon(get_resource_path("abogen.assets", "profile.png"))
for pname in load_profiles().keys():
self.voice_combo.addItem(profile_icon, pname, f"profile:{pname}")
for pname, entry in load_profiles().items():
entry_provider = ""
if isinstance(entry, dict):
entry_provider = str(entry.get("provider", "")).strip().lower()
if provider == "supertonic":
if entry_provider == "supertonic":
self.voice_combo.addItem(profile_icon, pname, f"profile:{pname}")
else:
if entry_provider != "supertonic":
self.voice_combo.addItem(profile_icon, pname, f"profile:{pname}")
# re-add voices
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):
icon = QIcon(flag_path)
self.voice_combo.addItem(icon, f"{v}", v)
if provider == "supertonic":
for v in DEFAULT_SUPERTONIC_VOICES:
icon = QIcon()
if v.startswith("F"):
icon_path = get_resource_path("abogen.assets", "female.png")
else:
icon_path = get_resource_path("abogen.assets", "male.png")
if icon_path and os.path.exists(icon_path):
icon = QIcon(icon_path)
self.voice_combo.addItem(icon, f"{v}", v)
else:
for v in VOICES_INTERNAL:
icon = QIcon()
country_code = KOKORO_LANG_TO_COUNTRY.get(v[0], v[0])
flag_path = get_resource_path("abogen.assets.flags", f"{country_code}.png")
if flag_path and os.path.exists(flag_path):
icon = QIcon(flag_path)
self.voice_combo.addItem(icon, f"{v}", v)
# restore selection
idx = -1
if self.selected_profile_name:
@@ -2069,6 +2222,9 @@ class abogen(QWidget):
save_base_path=save_base_path,
save_chapters_separately=getattr(self, "save_chapters_separately", None),
merge_chapters_at_end=getattr(self, "merge_chapters_at_end", None),
tts_provider=self.provider_combo.currentData(),
supertonic_language=self.st_lang_combo.currentText(),
supertonic_total_steps=self.st_steps_combo.currentData(),
)
# Prevent adding duplicate items to the queue
@@ -2212,6 +2368,15 @@ class abogen(QWidget):
self.config["replace_numerals"] = self.replace_numerals
self.config["fix_nonstandard_punctuation"] = self.fix_nonstandard_punctuation
# TTS provider settings
tts_provider = getattr(queued_item, "tts_provider", "kokoro")
self.provider_combo.setCurrentText("Supertonic" if tts_provider == "supertonic" else "Kokoro")
self.st_lang_combo.setCurrentText(getattr(queued_item, "supertonic_language", "en"))
steps_val = getattr(queued_item, "supertonic_total_steps", 8)
idx_steps = self.st_steps_combo.findData(steps_val)
if idx_steps >= 0:
self.st_steps_combo.setCurrentIndex(idx_steps)
# Sync Voice/Profile in config
self.config["selected_voice"] = self.selected_voice
if "selected_profile_name" in self.config:
@@ -2234,6 +2399,8 @@ class abogen(QWidget):
self.current_queue_index = 0 # Reset for next time
def get_voice_formula(self) -> str:
if self.provider_combo.currentData() == "supertonic":
return self._get_supertonic_voice()
if self.mixed_voice_state:
formula_components = [
f"{name}*{weight}" for name, weight in self.mixed_voice_state
@@ -2243,6 +2410,8 @@ class abogen(QWidget):
return self.selected_voice
def get_selected_lang(self, voice_formula) -> str:
if self.provider_combo.currentData() == "supertonic":
return self.st_lang_combo.currentText()
if self.selected_profile_name:
from abogen.voice_profiles import load_profiles
@@ -2316,9 +2485,9 @@ class abogen(QWidget):
file_size_str = "Unknown"
# pipeline_loaded_callback remains unchanged
def pipeline_loaded_callback(backend, error):
def pipeline_loaded_callback(np_module, kpipeline_class, error):
if error:
self.update_log((f"Error loading TTS backend: {error}", "red"))
self.update_log((f"Error loading numpy or KPipeline: {error}", "red"))
prevent_sleep_end()
return
@@ -2332,6 +2501,10 @@ class abogen(QWidget):
# determine selected language: use profile setting if profile selected, else voice code
selected_lang = self.get_selected_lang(voice_formula)
tts_provider = self.provider_combo.currentData()
supertonic_language = self.st_lang_combo.currentText()
supertonic_total_steps = self.st_steps_combo.currentData()
self.conversion_thread = ConversionThread(
self.selected_file,
selected_lang,
@@ -2341,13 +2514,17 @@ class abogen(QWidget):
self.selected_output_folder,
subtitle_mode=actual_subtitle_mode,
output_format=self.selected_format,
backend=backend,
np_module=np_module,
kpipeline_class=kpipeline_class,
start_time=self.start_time,
total_char_count=self.char_count,
use_gpu=self.gpu_ok,
from_queue=from_queue,
save_base_path=self.displayed_file_path, # Pass the save base path (original file for EPUB)
) # Use gpu_ok status
save_base_path=self.displayed_file_path,
tts_provider=tts_provider,
supertonic_language=supertonic_language,
supertonic_total_steps=supertonic_total_steps,
)
# Pass the displayed file path to the log_updated signal handler in ConversionThread
self.conversion_thread.display_path = display_path
# Pass the file size string
@@ -2424,22 +2601,15 @@ class abogen(QWidget):
# Store gpu_ok status to use when creating the conversion thread
self.gpu_ok = gpu_ok
self.update_log((gpu_msg, gpu_ok))
self.update_log("Loading modules...")
# Determine device based on GPU availability
if gpu_ok:
if platform.system() == "Darwin" and platform.processor() == "arm":
device = "mps"
else:
device = "cuda"
tts_provider = self.provider_combo.currentData()
if tts_provider == "supertonic":
# Supertonic doesn't need KPipeline, call callback directly
import numpy as np
pipeline_loaded_callback(np, None, None)
else:
device = "cpu"
lang_code = self.selected_lang or "a"
load_thread = LoadPipelineThread(
pipeline_loaded_callback, lang_code=lang_code, device=device
)
load_thread.start()
self.update_log("Loading modules...")
load_thread = LoadPipelineThread(pipeline_loaded_callback)
load_thread.start()
threading.Thread(target=gpu_and_load, daemon=True).start()
@@ -2752,9 +2922,32 @@ class abogen(QWidget):
"Open File Error", f"Could not open file:\n{e}"
)
def _get_supertonic_voice(self) -> str:
"""Resolve the effective Supertonic voice from the current combo selection."""
if self.selected_profile_name:
from abogen.voice_profiles import load_profiles
entry = load_profiles().get(self.selected_profile_name, {})
if isinstance(entry, dict):
return str(entry.get("voice", "M1"))
return "M1"
current_data = self.voice_combo.currentData()
if current_data and isinstance(current_data, str) and not current_data.startswith("profile:"):
return current_data
return "M1"
def _get_preview_cache_path(self):
"""Generate the expected cache path for the current voice settings."""
speed = self.speed_slider.value() / 100.0
provider = self.provider_combo.currentData()
if provider == "supertonic":
voice_to_cache = self._get_supertonic_voice()
lang_to_cache = self.st_lang_combo.currentText()
steps = self.st_steps_combo.currentData()
cache_dir = get_user_cache_path("preview_cache")
filename = f"st_{voice_to_cache}_{lang_to_cache}_steps{steps}_{speed:.2f}.wav"
return os.path.join(cache_dir, filename)
voice_to_cache = ""
lang_to_cache = ""
@@ -2859,6 +3052,13 @@ class abogen(QWidget):
self.btn_voice_formula_mixer.setEnabled(False) # Disable mixer button
self.btn_start.setEnabled(False) # Disable start button during preview
# For Supertonic, skip KPipeline loading and use SupertonicPipeline directly
if self.provider_combo.currentData() == "supertonic":
import numpy as np
self.loading_movie.start()
self._on_pipeline_loaded_for_preview(np, None, None)
return
# Start loading animation - ensure signal connection is always active
if hasattr(self, "loading_movie"):
# Disconnect previous connections to avoid multiple connections
@@ -2875,27 +3075,18 @@ class abogen(QWidget):
)
self.loading_movie.start()
# Determine device based on GPU availability
if self.gpu_ok:
if platform.system() == "Darwin" and platform.processor() == "arm":
device = "mps"
else:
device = "cuda"
else:
device = "cpu"
def pipeline_loaded_callback(np_module, kpipeline_class, error):
self._on_pipeline_loaded_for_preview(np_module, kpipeline_class, error)
lang = self.selected_lang or "a"
load_thread = LoadPipelineThread(
self._on_pipeline_loaded_for_preview, lang_code=lang, device=device
)
load_thread = LoadPipelineThread(pipeline_loaded_callback)
load_thread.start()
def _on_pipeline_loaded_for_preview(self, backend, error):
def _on_pipeline_loaded_for_preview(self, np_module, kpipeline_class, error):
# stop loading animation and restore icon on error
if error:
self.loading_movie.stop()
self._show_error_message_box(
"Loading Error", f"Error loading TTS backend: {error}"
"Loading Error", f"Error loading numpy or KPipeline: {error}"
)
self.btn_preview.setIcon(self.play_icon)
self.btn_preview.setEnabled(True)
@@ -2926,14 +3117,28 @@ class abogen(QWidget):
else None
)
else:
lang = self.selected_voice[0]
voice = self.selected_voice
if self.provider_combo.currentData() == "supertonic":
voice = self._get_supertonic_voice()
else:
voice = self.selected_voice or ""
tts_provider = self.provider_combo.currentData()
supertonic_language = self.st_lang_combo.currentText()
supertonic_total_steps = self.st_steps_combo.currentData()
if tts_provider == "supertonic":
lang = supertonic_language
else:
lang = self.selected_voice[0] if self.selected_voice else ""
# use same gpu/cpu logic as in conversion
gpu_msg, gpu_ok = get_gpu_acceleration(self.use_gpu)
self.preview_thread = VoicePreviewThread(
backend, lang, voice, speed, gpu_ok
np_module, kpipeline_class, lang, voice, speed, gpu_ok,
tts_provider=tts_provider,
supertonic_language=supertonic_language,
supertonic_total_steps=supertonic_total_steps,
)
self.preview_thread.finished.connect(self._play_preview_audio)
self.preview_thread.error.connect(self._preview_error)
+4 -5
View File
@@ -21,8 +21,7 @@ from PyQt6.QtWidgets import (
)
from PyQt6.QtCore import QThread, pyqtSignal
from abogen.constants import COLORS
from abogen.tts_backend_registry import get_metadata
from abogen.constants import COLORS, VOICES_INTERNAL
from abogen.spacy_utils import SPACY_MODELS
import abogen.hf_tracker
@@ -115,7 +114,7 @@ class PreDownloadWorker(QThread):
self._voices_success = False
return
voice_list = get_metadata("kokoro").voices
voice_list = VOICES_INTERNAL
for idx, voice in enumerate(voice_list, start=1):
if self._cancelled:
self._voices_success = False
@@ -463,14 +462,14 @@ class PreDownloadDialog(QDialog):
try:
from huggingface_hub import try_to_load_from_cache
for voice in get_metadata("kokoro").voices:
for voice in VOICES_INTERNAL:
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(get_metadata("kokoro").voices)
return False, list(VOICES_INTERNAL)
return (len(missing) == 0), missing
def _check_kokoro_model(self) -> bool:
+4
View File
@@ -26,3 +26,7 @@ class QueuedItem:
replace_all_caps: bool = False
replace_numerals: bool = False
fix_nonstandard_punctuation: bool = False
# TTS Provider fields
tts_provider: str = "kokoro"
supertonic_language: str = "en"
supertonic_total_steps: int = 8
+8 -5
View File
@@ -28,11 +28,12 @@ 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,
KOKORO_LANG_TO_COUNTRY,
COLORS,
)
from abogen.tts_backend_registry import get_metadata
import re
import platform
from abogen.utils import get_resource_path
@@ -179,7 +180,7 @@ class VoiceMixer(QWidget):
layout.addWidget(QLabel(name), alignment=Qt.AlignmentFlag.AlignCenter)
# Voice name label with gender icon
is_female = self.voice_name in get_metadata("kokoro").voices and self.voice_name[1] == "f"
is_female = self.voice_name in VOICES_INTERNAL and self.voice_name[1] == "f"
# Icons layout (flag and gender)
icons_layout = QHBoxLayout()
@@ -189,8 +190,9 @@ class VoiceMixer(QWidget):
) # Center the icons horizontally
# Flag icon
country_code = KOKORO_LANG_TO_COUNTRY.get(language_code, language_code)
flag_icon_path = get_resource_path(
"abogen.assets.flags", f"{language_code}.png"
"abogen.assets.flags", f"{country_code}.png"
)
gender_icon_path = get_resource_path(
"abogen.assets", "female.png" if is_female else "male.png"
@@ -512,7 +514,8 @@ class VoiceFormulaDialog(QDialog):
header_row.addWidget(QLabel("Language:"))
self.language_combo = QComboBox()
for code, desc in LANGUAGE_OPTIONS:
flag = get_resource_path("abogen.assets.flags", f"{code}.png")
country_code = KOKORO_LANG_TO_COUNTRY.get(code, code)
flag = get_resource_path("abogen.assets.flags", f"{country_code}.png")
if flag and os.path.exists(flag):
self.language_combo.addItem(QIcon(flag), desc, code)
else:
@@ -772,7 +775,7 @@ class VoiceFormulaDialog(QDialog):
def add_voices(self, initial_state):
first_enabled_voice = None
for voice in get_metadata("kokoro").voices:
for voice in VOICES_INTERNAL:
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 available voices (case-insensitive).
"""Validate voice name against VOICES_INTERNAL list (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.tts_backend_registry import get_metadata
from abogen.constants import VOICES_INTERNAL
# Create case-insensitive lookup set (done once per call)
voice_lookup_lower = {v.lower() for v in get_metadata("kokoro").voices}
voice_lookup_lower = {v.lower() for v in VOICES_INTERNAL}
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 canonical voice names.
Voice names are normalized to lowercase to match VOICES_INTERNAL.
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.tts_backend_registry import get_metadata
from abogen.constants import VOICES_INTERNAL
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 get_metadata("kokoro").voices if v.lower() == voice_part_lower),
(v for v in VOICES_INTERNAL 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 get_metadata("kokoro").voices if v.lower() == voice_name_lower),
(v for v in VOICES_INTERNAL if v.lower() == voice_name_lower),
voice_name
)
valid_markers += 1
-89
View File
@@ -1,89 +0,0 @@
"""
TTS Backend Interface
This module defines the protocol for TTS backends and the
metadata model that describes a backend implementation.
"""
from dataclasses import dataclass
from typing import Protocol, List, Dict, Any
@dataclass(frozen=True)
class TTSBackendMetadata:
"""
Immutable metadata describing a TTS backend implementation.
Attributes:
id: Unique backend identifier (e.g. ``"kokoro"``, ``"supertonic"``).
name: Human-readable display name.
description: Short description of the backend.
voices: Tuple of supported voice identifiers.
"""
id: str
name: str
description: str
voices: tuple[str, ...] = ()
class TTSBackend(Protocol):
"""
Protocol for TTS backends.
All TTS backends must implement this interface to be compatible
with the application.
"""
@property
def metadata(self) -> TTSBackendMetadata:
...
def __init__(self, **kwargs) -> None:
"""
Initialize the TTS backend.
Args:
**kwargs: Backend-specific configuration parameters
"""
...
def synthesize(self, text: str, **kwargs) -> bytes:
"""
Synthesize speech from text.
Args:
text: Text to synthesize
**kwargs: Additional parameters for synthesis
Returns:
Audio data as bytes
"""
...
def get_available_voices(self) -> List[str]:
"""
Get list of available voices.
Returns:
List of voice identifiers
"""
...
def get_supported_formats(self) -> List[str]:
"""
Get list of supported audio formats.
Returns:
List of supported audio formats
"""
...
def get_info(self) -> Dict[str, Any]:
"""
Get backend information.
Returns:
Dictionary with backend information
"""
...
-146
View File
@@ -1,146 +0,0 @@
"""
TTS Backend Registry
Provides a global registry for TTS backend factories.
Backends register themselves with metadata and a factory callable.
The registry is universal and does not know about backend constructors.
"""
from typing import Callable, Any
from abogen.tts_backend import TTSBackend, TTSBackendMetadata
class TTSBackendRegistry:
"""Registry of TTS backend factories.
Stores metadata and factory callables for registered backends.
"""
def __init__(self) -> None:
self._backends: dict[str, TTSBackendMetadata] = {}
self._factories: dict[str, Callable[..., TTSBackend]] = {}
def register(
self,
metadata: TTSBackendMetadata,
factory: Callable[..., TTSBackend],
) -> None:
"""Register a backend with its metadata and factory callable."""
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())
def get_metadata(self, backend_id: str) -> TTSBackendMetadata:
"""Get metadata for a specific backend.
Raises:
KeyError: If backend with given id is not registered.
"""
if backend_id not in self._backends:
raise KeyError(f"Unknown backend: {backend_id}")
return self._backends[backend_id]
def create_backend(self, backend_id: str, **kwargs: Any) -> TTSBackend:
"""Create a backend instance by id.
Raises:
KeyError: If backend with given id is not registered.
"""
if backend_id not in self._factories:
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()
def register_backend(
metadata: TTSBackendMetadata,
factory: Callable[..., TTSBackend],
) -> None:
"""Register a TTS backend in the global registry."""
_registry.register(metadata, factory)
def get_metadata(backend_id: str) -> TTSBackendMetadata:
"""Get metadata for a specific backend by id.
Ensures all backends are registered by importing the tts_backends
package on first access.
Raises:
KeyError: If backend with given id is not registered.
"""
import abogen.tts_backends # noqa: F401 — triggers backend registration
return _registry.get_metadata(backend_id)
def get_default_voice(backend_id: str, fallback: str = "") -> str:
"""Return the first voice of a backend, or *fallback* if none."""
voices = get_metadata(backend_id).voices
return voices[0] if voices else fallback
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)
-20
View File
@@ -1,20 +0,0 @@
"""TTS backends package.
Backend modules are auto-discovered and imported here.
Each backend module registers itself with the global registry
when imported.
"""
import importlib
import pkgutil
def _discover_backends():
"""Import all modules in this package to trigger their registration."""
package = __name__
for _importer, modname, _ispkg in pkgutil.iter_modules(path=__path__):
importlib.import_module(f"{package}.{modname}")
_discover_backends()
-179
View File
@@ -1,179 +0,0 @@
"""
Kokoro TTS Backend
Encapsulates the Kokoro KPipeline as a TTSBackend implementation.
"""
from __future__ import annotations
from typing import Any, Dict, Iterator, List, Optional
import numpy as np
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),
)
def _load_kpipeline():
"""Lazy-load Kokoro dependencies."""
from kokoro import KPipeline # type: ignore[import-not-found]
return KPipeline
class KokoroBackend:
"""TTSBackend implementation wrapping the Kokoro KPipeline.
All interaction with KPipeline is encapsulated here.
The rest of the project depends only on this class.
"""
def __init__(self, **kwargs: Any) -> None:
lang_code = kwargs["lang_code"]
repo_id = kwargs.get("repo_id", "hexgrad/Kokoro-82M")
device = kwargs.get("device", "cpu")
KPipeline = _load_kpipeline()
self._pipeline = KPipeline(
lang_code=lang_code,
repo_id=repo_id,
device=device,
)
self._lang_code = lang_code
@property
def metadata(self) -> TTSBackendMetadata:
return _KOKORO_METADATA
def __call__(
self,
text: str,
*,
voice: Any,
speed: float = 1.0,
split_pattern: Optional[str] = None,
) -> Iterator[Any]:
"""Delegate to KPipeline's __call__."""
return self._pipeline(
text,
voice=voice,
speed=speed,
split_pattern=split_pattern,
)
def load_single_voice(self, voice_name: str) -> Any:
"""Load a single voice tensor. Used by voice formula system."""
return self._pipeline.load_single_voice(voice_name)
def synthesize(self, text: str, **kwargs: Any) -> bytes:
"""Synthesize speech from text. Returns raw audio bytes."""
voice = kwargs.get("voice", "")
speed = kwargs.get("speed", 1.0)
split_pattern = kwargs.get("split_pattern", None)
audio_parts: list[np.ndarray] = []
for segment in self(text, voice=voice, speed=speed, split_pattern=split_pattern):
audio = segment.audio
if hasattr(audio, "numpy"):
audio = audio.numpy()
audio_parts.append(np.asarray(audio, dtype="float32"))
if not audio_parts:
return b""
combined = np.concatenate(audio_parts).astype("float32", copy=False)
return combined.tobytes()
def get_available_voices(self) -> List[str]:
"""Return known Kokoro voice identifiers."""
return list(self.metadata.voices)
def get_supported_formats(self) -> List[str]:
"""Kokoro outputs raw PCM float32 audio."""
return ["pcm_float32"]
def get_info(self) -> Dict[str, Any]:
return {
"id": "kokoro",
"name": "Kokoro",
"lang_code": self._lang_code,
}
def create_kokoro_backend(**kwargs: Any) -> KokoroBackend:
"""Factory callable registered with TTSBackendRegistry."""
return KokoroBackend(**kwargs)
# --- Registration ---
from abogen.tts_backend_registry import register_backend # noqa: E402
register_backend(
metadata=_KOKORO_METADATA,
factory=create_kokoro_backend,
)
@@ -4,8 +4,9 @@ import ast
from dataclasses import dataclass
import logging
import math
import os
import re
from typing import Any, Dict, Iterable, Iterator, List, Optional
from typing import Any, Iterable, Iterator, Optional
import numpy as np
@@ -15,14 +16,12 @@ logger = logging.getLogger(__name__)
DEFAULT_SUPERTONIC_VOICES = ("M1", "M2", "M3", "M4", "M5", "F1", "F2", "F3", "F4", "F5")
from abogen.tts_backend import TTSBackendMetadata
_SUPERTONIC_METADATA = TTSBackendMetadata(
id="supertonic",
name="SuperTonic",
description="SuperTonic TTS engine",
voices=DEFAULT_SUPERTONIC_VOICES,
)
SUPERTONIC_AVAILABLE_LANGS = [
"en", "ko", "ja", "ar", "bg", "cs", "da", "de", "el",
"es", "et", "fi", "fr", "hi", "hr", "hu", "id", "it",
"lt", "lv", "nl", "pl", "pt", "ro", "ru", "sk", "sl",
"sv", "tr", "uk", "vi", "na",
]
@dataclass
@@ -98,7 +97,7 @@ _UNSUPPORTED_CHARS_RE = re.compile(
def _parse_unsupported_characters(error: BaseException) -> list[str]:
"""Best-effort extraction of unsupported characters from SuperTonic errors."""
"""Best-effort extraction of unsupported characters from Supertonic errors."""
message = " ".join(
str(part) for part in getattr(error, "args", ()) if part is not None
@@ -164,6 +163,7 @@ def _configure_supertonic_gpu() -> None:
except Exception as exc:
logger.warning("Could not configure supertonic GPU providers: %s", exc)
SUPERTONIC_MAX_CHUNK_LENGTH = 500
class SupertonicPipeline:
"""Minimal adapter that mimics Kokoro's pipeline iteration interface."""
@@ -174,11 +174,14 @@ class SupertonicPipeline:
sample_rate: int,
auto_download: bool = True,
total_steps: int = 5,
max_chunk_length: int = 300,
max_chunk_length: int = SUPERTONIC_MAX_CHUNK_LENGTH,
lang: str = "en",
intra_op_num_threads: Optional[int] = None,
) -> None:
self.sample_rate = int(sample_rate)
self.total_steps = int(total_steps)
self.max_chunk_length = int(max_chunk_length)
self.lang = str(lang or "en")
# Configure GPU providers before importing TTS
_configure_supertonic_gpu()
@@ -190,7 +193,8 @@ class SupertonicPipeline:
"Supertonic is not installed. Install it with `pip install supertonic`."
) from exc
self._tts = TTS(auto_download=auto_download)
threads = intra_op_num_threads if intra_op_num_threads is not None else os.cpu_count()
self._tts = TTS(auto_download=auto_download, intra_op_num_threads=threads)
def __call__(
self,
@@ -200,12 +204,14 @@ class SupertonicPipeline:
speed: float,
split_pattern: Optional[str] = None,
total_steps: Optional[int] = None,
lang: Optional[str] = None,
) -> Iterator[SupertonicSegment]:
voice_name = (voice or "").strip() or "M1"
steps = int(total_steps) if total_steps is not None else self.total_steps
steps = max(2, min(15, steps))
speed_value = float(speed) if speed is not None else 1.0
speed_value = max(0.7, min(2.0, speed_value))
language = str(lang or self.lang or "en")
style = self._tts.get_voice_style(voice_name=voice_name)
chunks = _split_text(
@@ -216,12 +222,13 @@ class SupertonicPipeline:
removed: set[str] = set()
last_exc: Exception | None = None
# SuperTonic can raise ValueError for unsupported characters; strip and retry.
# Supertonic can raise ValueError for unsupported characters; strip and retry.
for attempt in range(3):
try:
wav, duration = self._tts.synthesize(
text=chunk_to_speak,
voice_style=style,
lang=language,
total_steps=steps,
speed=speed_value,
max_chunk_length=self.max_chunk_length,
@@ -247,14 +254,14 @@ class SupertonicPipeline:
chunk_to_speak = sanitized
if not chunk_to_speak:
logger.warning(
"SuperTonic: dropped a chunk after removing unsupported characters: %s",
"Supertonic: dropped a chunk after removing unsupported characters: %s",
sorted(removed),
)
break
if attempt == 0:
logger.warning(
"SuperTonic: removed unsupported characters %s and retried.",
"Supertonic: removed unsupported characters %s and retried.",
sorted(removed),
)
else:
@@ -282,111 +289,3 @@ class SupertonicPipeline:
audio = _resample_linear(audio, src_rate, self.sample_rate)
yield SupertonicSegment(graphemes=chunk_to_speak, audio=audio)
class SupertonicBackend:
"""Supertonic TTS backend implementing the TTSBackend protocol.
Encapsulates ``SupertonicPipeline`` as an internal implementation detail.
"""
@property
def metadata(self) -> TTSBackendMetadata:
return _SUPERTONIC_METADATA
def __init__(self, **kwargs: Any) -> None:
self._pipeline = SupertonicPipeline(
sample_rate=kwargs.get("sample_rate", 24000),
auto_download=kwargs.get("auto_download", True),
total_steps=kwargs.get("total_steps", 5),
)
def synthesize(self, text: str, **kwargs: Any) -> bytes:
"""Synthesize speech and return raw audio bytes (WAV).
Delegates to the internal :class:`SupertonicPipeline` and concatenates
all produced segments into a single byte buffer.
"""
import io
import soundfile as sf
voice = kwargs.get("voice", "M1")
speed = float(kwargs.get("speed", 1.0))
split_pattern = kwargs.get("split_pattern")
total_steps = kwargs.get("total_steps")
segments = self._pipeline(
text,
voice=voice,
speed=speed,
split_pattern=split_pattern,
total_steps=total_steps,
)
audio_parts: list[np.ndarray] = []
for seg in segments:
audio_parts.append(seg.audio)
if not audio_parts:
return b""
combined = np.concatenate(audio_parts)
buf = io.BytesIO()
sf.write(buf, combined, self._pipeline.sample_rate, format="WAV")
return buf.getvalue()
def get_available_voices(self) -> List[str]:
"""Return the list of built-in SuperTonic voice identifiers."""
return list(self.metadata.voices)
def get_supported_formats(self) -> List[str]:
return ["wav"]
def get_info(self) -> Dict[str, Any]:
return {
"sample_rate": self._pipeline.sample_rate,
"total_steps": self._pipeline.total_steps,
"max_chunk_length": self._pipeline.max_chunk_length,
"voices": list(DEFAULT_SUPERTONIC_VOICES),
}
def __call__(
self,
text: str,
*,
voice: str,
speed: float,
split_pattern: Optional[str] = None,
total_steps: Optional[int] = None,
) -> Iterator[SupertonicSegment]:
"""Backward-compatible call interface, delegates to the pipeline."""
return self._pipeline(
text,
voice=voice,
speed=speed,
split_pattern=split_pattern,
total_steps=total_steps,
)
def create_supertonic_backend(**kwargs: Any) -> SupertonicBackend:
"""Create a SuperTonic TTS backend instance.
Args:
sample_rate: Audio sample rate. Defaults to 24000.
auto_download: Auto-download models. Defaults to True.
total_steps: Inference steps. Defaults to 5.
Returns:
SupertonicBackend instance.
"""
return SupertonicBackend(**kwargs)
from abogen.tts_backend_registry import register_backend # noqa: E402
register_backend(
metadata=_SUPERTONIC_METADATA,
factory=create_supertonic_backend,
)
+11 -10
View File
@@ -529,20 +529,21 @@ def prevent_sleep_end():
_sleep_procs[system] = None
def load_numpy_kpipeline():
import numpy as np
from kokoro import KPipeline # type: ignore[import-not-found]
return np, KPipeline
class LoadPipelineThread(Thread):
def __init__(self, callback, lang_code="a", device="cpu"):
def __init__(self, callback):
super().__init__()
self.callback = callback
self.lang_code = lang_code
self.device = device
def run(self):
try:
from abogen.tts_backend_registry import create_backend
backend = create_backend(
"kokoro", lang_code=self.lang_code, device=self.device
)
self.callback(backend, None)
np_module, kpipeline_class = load_numpy_kpipeline()
self.callback(np_module, kpipeline_class, None)
except Exception as e:
self.callback(None, str(e))
self.callback(None, None, str(e))
+3 -4
View File
@@ -17,7 +17,7 @@ if LocalEntryNotFoundError is None: # pragma: no cover - fallback for tests
pass
from abogen.tts_backend_registry import get_metadata
from abogen.constants import VOICES_INTERNAL
_CACHE_LOCK = threading.Lock()
_CACHED_VOICES: Set[str] = set()
@@ -26,9 +26,8 @@ _BOOTSTRAPPED = False
def _normalize_targets(voices: Optional[Iterable[str]]) -> Set[str]:
kokoro_voices = get_metadata("kokoro").voices
if not voices:
return set(kokoro_voices)
return set(VOICES_INTERNAL)
normalized: Set[str] = set()
for voice in voices:
if not voice:
@@ -36,7 +35,7 @@ def _normalize_targets(voices: Optional[Iterable[str]]) -> Set[str]:
voice_id = str(voice).strip()
if not voice_id:
continue
if voice_id in kokoro_voices:
if voice_id in VOICES_INTERNAL:
normalized.add(voice_id)
return normalized
+2 -3
View File
@@ -1,7 +1,7 @@
import re
from typing import List, Tuple
from abogen.tts_backend_registry import get_metadata
from abogen.constants import VOICES_INTERNAL
# Calls parsing and loads the voice to gpu or cpu
@@ -22,7 +22,6 @@ def parse_formula_terms(formula: str) -> List[Tuple[str, float]]:
raise ValueError("Empty voice formula")
terms: List[Tuple[str, float]] = []
kokoro_voices = get_metadata("kokoro").voices
for segment in formula.split("+"):
part = segment.strip()
if not part:
@@ -31,7 +30,7 @@ def parse_formula_terms(formula: str) -> List[Tuple[str, float]]:
raise ValueError("Each component must be in the form voice*weight")
voice_name, raw_weight = part.split("*", 1)
voice_name = voice_name.strip()
if voice_name not in kokoro_voices:
if voice_name not in VOICES_INTERNAL:
raise ValueError(f"Unknown voice: {voice_name}")
try:
weight = float(raw_weight.strip())
-33
View File
@@ -1,33 +0,0 @@
from dataclasses import dataclass
@dataclass(frozen=True)
class VoiceMetadata:
"""
Immutable metadata describing a voice from a TTS backend.
This model describes a voice independently of any backend implementation.
Backends populate these objects; the application consumes them.
The ``backend_id`` field is set by the backend itself (via
``self.metadata.id``) the application never hardcodes it.
This ensures renaming a backend does not require touching voice definitions.
"""
id: str
"""Unique voice identifier within the backend (e.g. ``"af_alloy"``, ``"M1"``)."""
display_name: str
"""Human-readable display name (e.g. ``"Alloy"``, ``"Male 1"``)."""
language: str
"""Language code — backend-specific format is acceptable (e.g. ``"a"``, ``"en"``)."""
gender: str
"""Gender category: ``"female"``, ``"male"``, or ``"unknown"``."""
backend_id: str
"""Identifier of the backend that owns this voice (e.g. ``"kokoro"``).
Set automatically by the backend never hardcoded in voice definitions.
"""
+5 -6
View File
@@ -2,7 +2,8 @@ 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.constants import VOICES_INTERNAL
from abogen.tts_supertonic import DEFAULT_SUPERTONIC_VOICES
from abogen.utils import get_user_config_path
@@ -69,8 +70,7 @@ def serialize_profiles() -> Dict[str, Dict[str, Iterable[Tuple[str, float]]]]:
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"
return raw if raw in DEFAULT_SUPERTONIC_VOICES else "M1"
def _coerce_supertonic_steps(value: Any) -> int:
@@ -101,7 +101,7 @@ def normalize_profile_entry(entry: Any) -> Dict[str, Any]:
return {}
provider = str(entry.get("provider") or "kokoro").strip().lower()
if not is_registered_backend(provider):
if provider not in {"kokoro", "supertonic"}:
provider = "kokoro"
language = str(entry.get("language") or "a").strip().lower() or "a"
@@ -135,7 +135,6 @@ def normalize_profile_entry(entry: Any) -> Dict[str, Any]:
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")
@@ -144,7 +143,7 @@ def _normalize_voice_entries(entries: Iterable) -> List[Tuple[str, float]]:
voice, weight = item[0], item[1]
else:
continue
if voice not in kokoro_voices:
if voice not in VOICES_INTERNAL:
continue
if weight is None:
continue
+12 -11
View File
@@ -2,6 +2,7 @@ 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
@@ -26,22 +27,22 @@ RUN python3 -m venv "$VIRTUAL_ENV"
WORKDIR /app
COPY pyproject.toml README.md ./
RUN pip install uv \
&& if [ -n "$TORCH_VERSION" ]; then \
uv pip install --system torch=="$TORCH_VERSION" torchvision=="$TORCH_VERSION" torchaudio=="$TORCH_VERSION" --index-url "$TORCH_INDEX_URL"; \
else \
uv pip install --system torch torchvision torchaudio --index-url "$TORCH_INDEX_URL"; \
fi \
&& 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 \
&& uv pip install --system "mutagen>=1.47.0"
COPY abogen ./abogen
RUN pip install --upgrade pip \
&& if [ -n "$TORCH_VERSION" ]; then \
pip install 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"; \
fi \
&& pip install --no-cache-dir . \
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"
# 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 \
uv pip install --system onnxruntime-gpu; \
pip install --no-cache-dir onnxruntime-gpu; \
fi
ENV ABOGEN_HOST=0.0.0.0 \
+62 -53
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, is_registered_backend, resolve_backend_for_voice
from abogen.constants import VOICES_INTERNAL
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 (
@@ -39,15 +39,14 @@ from abogen.utils import (
get_user_cache_path,
get_user_output_path,
load_config,
load_numpy_kpipeline,
)
from abogen.tts_backend_registry import create_backend
from abogen.tts_backend import TTSBackend
from abogen.voice_cache import ensure_voice_assets
from abogen.voice_formulas import extract_voice_ids, get_new_voice
from abogen.voice_profiles import load_profiles, normalize_profile_entry
from abogen.pronunciation_store import increment_usage
from abogen.llm_client import LLMClientError
from abogen.tts_supertonic import DEFAULT_SUPERTONIC_VOICES, SupertonicPipeline
from .service import Job, JobStatus
@@ -57,26 +56,25 @@ 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()
# Normalize to uppercase for Supertonic voice IDs
upper = raw.upper() if raw else ""
# 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"
# 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 ""
upper = raw.upper()
if upper in DEFAULT_SUPERTONIC_VOICES:
return upper
# If still empty, use default Supertonic voice
if not upper or "*" in upper or "+" in upper:
upper = "M1"
fallback_upper = fallback_raw.upper() if fallback_raw else ""
if fallback_upper in DEFAULT_SUPERTONIC_VOICES:
return fallback_upper
return upper
return "M1"
def _split_speaker_reference(value: Any) -> tuple[Optional[str], str]:
@@ -120,7 +118,15 @@ def _formula_from_kokoro_entry(entry: Mapping[str, Any]) -> str:
def _infer_provider_from_spec(value: Any, fallback: str = "kokoro") -> str:
return resolve_backend_for_voice(str(value or ""), fallback=fallback)
raw = str(value or "").strip()
if not raw:
return fallback
upper = raw.upper()
if upper in DEFAULT_SUPERTONIC_VOICES:
return "supertonic"
if "*" in raw or "+" in raw:
return "kokoro"
return fallback
class _JobCancelled(Exception):
@@ -569,7 +575,7 @@ def _spec_to_voice_ids(spec: Any) -> Set[str]:
return set(extract_voice_ids(text))
except ValueError:
return set()
if text in get_metadata("kokoro").voices:
if text in VOICES_INTERNAL:
return {text}
return set()
@@ -633,7 +639,7 @@ def _collect_required_voice_ids(job: Job) -> Set[str]:
for key in ("resolved_voice", "voice_formula", "voice"):
voices.update(_spec_to_voice_ids(payload.get(key)))
voices.update(get_metadata("kokoro").voices)
voices.update(VOICES_INTERNAL)
return voices
@@ -1567,7 +1573,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 not is_registered_backend(provider_norm):
if provider_norm not in {"kokoro", "supertonic"}:
provider_norm = "kokoro"
existing = pipelines.get(provider_norm)
@@ -1575,11 +1581,10 @@ def run_conversion_job(job: Job) -> None:
return existing
if provider_norm == "supertonic":
pipelines[provider_norm] = create_backend(
"supertonic",
pipelines[provider_norm] = SupertonicPipeline(
sample_rate=SAMPLE_RATE,
auto_download=True,
total_steps=int(getattr(job, "supertonic_total_steps", 5) or 5),
total_steps=int(getattr(job, "supertonic_total_steps", 8) or 8),
)
return pipelines[provider_norm]
@@ -1589,12 +1594,16 @@ def run_conversion_job(job: Job) -> None:
device = "cpu"
if not disable_gpu:
device = _select_device()
# Create KPipeline instance directly (conforms to TTSBackend protocol)
pipelines[provider_norm] = create_backend(
"kokoro",
lang_code=job.language,
device=device
)
_np, KPipeline = load_numpy_kpipeline()
# Try to initialize with the selected device; fall back to CPU if CUDA fails
try:
pipelines[provider_norm] = KPipeline(lang_code=job.language, repo_id="hexgrad/Kokoro-82M", device=device)
except RuntimeError as e:
if "CUDA" in str(e) and device != "cpu":
job.add_log(f"CUDA initialization failed, falling back to CPU: {e}", level="warning")
pipelines[provider_norm] = KPipeline(lang_code=job.language, repo_id="hexgrad/Kokoro-82M", device="cpu")
else:
raise
if not kokoro_cache_ready:
_initialize_voice_cache(job)
kokoro_cache_ready = True
@@ -1609,7 +1618,7 @@ def run_conversion_job(job: Job) -> None:
provider = str(entry.get("provider") or "kokoro").strip().lower() or "kokoro"
if provider == "supertonic":
voice = str(entry.get("voice") or getattr(job, "voice", "M1") or "M1").strip() or "M1"
steps = int(entry.get("total_steps") or getattr(job, "supertonic_total_steps", 5) or 5)
steps = int(entry.get("total_steps") or getattr(job, "supertonic_total_steps", 8) or 8)
speed = float(entry.get("speed") or getattr(job, "speed", 1.0) or 1.0)
return "supertonic", _supertonic_voice_from_spec(voice, getattr(job, "voice", "M1")), speed, steps
formula = _formula_from_kokoro_entry(entry)
@@ -1625,7 +1634,7 @@ def run_conversion_job(job: Job) -> None:
"""Resolve a raw voice spec into (provider, resolved_spec, choice, speed, steps).
For Kokoro formulas, `choice` will be a resolved voice tensor (via `voice_formulas`).
For SuperTonic, `choice` will be a valid SuperTonic voice id.
For Supertonic, `choice` will be a valid Supertonic voice id.
"""
provider, resolved, speed, steps = resolve_voice_target(raw_spec)
@@ -1635,8 +1644,8 @@ def run_conversion_job(job: Job) -> None:
return provider, resolved, cached, speed, steps
if provider == "kokoro":
kokoro_backend = get_pipeline("kokoro")
choice = _resolve_voice(kokoro_backend, resolved, job.use_gpu)
kokoro_pipeline = get_pipeline("kokoro")
choice = _resolve_voice(kokoro_pipeline, resolved, job.use_gpu)
else:
choice = resolved
@@ -1765,8 +1774,8 @@ def run_conversion_job(job: Job) -> None:
voice_cache: Dict[str, Any] = {}
base_provider, base_voice_resolved, _, _ = resolve_voice_target(base_voice_spec)
if base_provider == "kokoro" and base_voice_resolved and "*" not in base_voice_resolved:
kokoro_backend = get_pipeline("kokoro")
voice_cache[f"kokoro:{base_voice_resolved}"] = _resolve_voice(kokoro_backend, base_voice_resolved, job.use_gpu)
kokoro_pipeline = get_pipeline("kokoro")
voice_cache[f"kokoro:{base_voice_resolved}"] = _resolve_voice(kokoro_pipeline, base_voice_resolved, job.use_gpu)
processed_chars = 0
subtitle_index = 1
current_time = 0.0
@@ -1796,8 +1805,8 @@ def run_conversion_job(job: Job) -> None:
fallback_key = next(iter(voice_cache.keys()), "")
if fallback_key and fallback_key != "__custom_mix":
intro_voice_spec = fallback_key.split(":", 1)[-1]
if not intro_voice_spec:
intro_voice_spec = get_default_voice("kokoro")
if not intro_voice_spec and VOICES_INTERNAL:
intro_voice_spec = VOICES_INTERNAL[0]
if intro_voice_spec:
intro_provider, _, intro_voice_choice, intro_speed, intro_steps = resolve_voice_choice(
@@ -1848,11 +1857,11 @@ def run_conversion_job(job: Job) -> None:
voice=voice_name,
speed=float(speed_override if speed_override is not None else job.speed),
split_pattern=split_pattern,
total_steps=int(supertonic_steps_override if supertonic_steps_override is not None else getattr(job, "supertonic_total_steps", 5)),
total_steps=int(supertonic_steps_override if supertonic_steps_override is not None else getattr(job, "supertonic_total_steps", 8)),
)
else:
kokoro_backend = get_pipeline("kokoro")
segment_iter = kokoro_backend(
kokoro_pipeline = get_pipeline("kokoro")
segment_iter = kokoro_pipeline(
normalized,
voice=voice_choice,
speed=float(speed_override if speed_override is not None else job.speed),
@@ -1941,8 +1950,8 @@ def run_conversion_job(job: Job) -> None:
if chapter_provider == "kokoro":
voice_choice = voice_cache.get(chapter_cache_key)
if voice_choice is None:
kokoro_backend = get_pipeline("kokoro")
voice_choice = _resolve_voice(kokoro_backend, chapter_voice_resolved, job.use_gpu)
kokoro_pipeline = get_pipeline("kokoro")
voice_choice = _resolve_voice(kokoro_pipeline, chapter_voice_resolved, job.use_gpu)
voice_cache[chapter_cache_key] = voice_choice
else:
voice_choice = chapter_voice_resolved
@@ -2086,9 +2095,9 @@ def run_conversion_job(job: Job) -> None:
if chunk_provider == "kokoro":
chunk_voice_choice = voice_cache.get(chunk_cache_key)
if chunk_voice_choice is None:
kokoro_backend = get_pipeline("kokoro")
kokoro_pipeline = get_pipeline("kokoro")
chunk_voice_choice = _resolve_voice(
kokoro_backend,
kokoro_pipeline,
chunk_voice_resolved,
job.use_gpu,
)
@@ -2230,8 +2239,8 @@ def run_conversion_job(job: Job) -> None:
if fallback_key and fallback_key != "__custom_mix":
# `voice_cache` keys are internal and include provider prefixes.
outro_voice_spec = fallback_key.split(":", 1)[-1]
if not outro_voice_spec:
outro_voice_spec = get_default_voice("kokoro")
if not outro_voice_spec and VOICES_INTERNAL:
outro_voice_spec = VOICES_INTERNAL[0]
if outro_text and outro_voice_spec:
outro_start_time = current_time
@@ -2436,17 +2445,17 @@ def _load_pipeline(job: Job):
disable_gpu = not job.use_gpu or not cfg.get("use_gpu", True)
provider = str(getattr(job, "tts_provider", "kokoro") or "kokoro").strip().lower()
if provider == "supertonic":
return create_backend(
"supertonic",
return SupertonicPipeline(
sample_rate=SAMPLE_RATE,
auto_download=True,
total_steps=int(getattr(job, "supertonic_total_steps", 5) or 5),
total_steps=int(getattr(job, "supertonic_total_steps", 8) or 8),
)
device = "cpu"
if not disable_gpu:
device = _select_device()
return create_backend("kokoro", lang_code=job.language, device=device)
_np, KPipeline = load_numpy_kpipeline()
return KPipeline(lang_code=job.language, repo_id="hexgrad/Kokoro-82M", device=device)
def _select_device() -> str:
@@ -2601,7 +2610,7 @@ def _build_ffmpeg_command(path: Path, fmt: str, metadata: Optional[Dict[str, str
def _resolve_voice(pipeline, voice_spec: str, use_gpu: bool):
if "*" in voice_spec:
# Voice formulas are a Kokoro-only feature (they require a pipeline that can
# load individual Kokoro voices). When running with SuperTonic (or when the
# load individual Kokoro voices). When running with Supertonic (or when the
# pipeline is otherwise unavailable), treat the spec as a plain string and
# allow downstream provider-specific resolution to choose a safe fallback.
if pipeline is None or not hasattr(pipeline, "load_single_voice"):
+3 -2
View File
@@ -15,7 +15,7 @@ from abogen.normalization_settings import build_apostrophe_config
from abogen.text_extractor import extract_from_path
from abogen.voice_cache import ensure_voice_assets
from abogen.webui.conversion_runner import SAMPLE_RATE, SPLIT_PATTERN, _select_device, _to_float32, _resolve_voice, _spec_to_voice_ids
from abogen.tts_backend_registry import create_backend
from abogen.utils import load_numpy_kpipeline
_MARKER_RE = re.compile(re.escape(MARKER_PREFIX) + r"(?P<code>[A-Z0-9_]+)" + re.escape(MARKER_SUFFIX))
@@ -45,7 +45,8 @@ def _load_pipeline(language: str, use_gpu: bool) -> Any:
device = "cpu"
if use_gpu:
device = _select_device()
return create_backend("kokoro", lang_code=language, device=device)
_np, KPipeline = load_numpy_kpipeline()
return KPipeline(lang_code=language, repo_id="hexgrad/Kokoro-82M", device=device)
def _extract_cases_from_text(text: str) -> List[Tuple[str, str]]:
+5 -6
View File
@@ -34,7 +34,6 @@ 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,
@@ -64,7 +63,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 not is_registered_backend(provider):
if provider not in {"kokoro", "supertonic"}:
provider = "kokoro"
if provider == "supertonic":
profile = {
@@ -163,7 +162,7 @@ def api_voice_profiles_preview() -> ResponseReturnValue:
formula = str(payload.get("formula") or "").strip()
profile_name = str(payload.get("profile") or "").strip()
provider = str(payload.get("tts_provider") or payload.get("provider") or "").strip().lower() or None
supertonic_total_steps = int(payload.get("supertonic_total_steps") or payload.get("total_steps") or settings.get("supertonic_total_steps") or 5)
supertonic_total_steps = int(payload.get("supertonic_total_steps") or payload.get("total_steps") or settings.get("supertonic_total_steps") or 8)
voice_spec = ""
resolved_provider = provider or "kokoro"
@@ -225,13 +224,13 @@ def api_speaker_preview() -> ResponseReturnValue:
speed_value = payload.get("speed")
speed = coerce_float(speed_value, 1.0)
tts_provider = str(payload.get("tts_provider") or "").strip().lower()
supertonic_total_steps = int(payload.get("supertonic_total_steps") or 5)
supertonic_total_steps = int(payload.get("supertonic_total_steps") or 8)
settings = load_settings()
use_gpu = settings.get("use_gpu", False)
base_spec, speaker_name = split_profile_spec(voice)
resolved_provider = tts_provider if is_registered_backend(tts_provider) else ""
resolved_provider = tts_provider if tts_provider in {"kokoro", "supertonic"} else ""
if speaker_name:
entry = normalize_profile_entry(load_profiles().get(speaker_name))
@@ -270,7 +269,7 @@ def api_speaker_preview() -> ResponseReturnValue:
use_gpu=use_gpu
,
tts_provider=resolved_provider,
supertonic_total_steps=supertonic_total_steps or int(settings.get("supertonic_total_steps") or 5),
supertonic_total_steps=supertonic_total_steps or int(settings.get("supertonic_total_steps") or 8),
pronunciation_overrides=pronunciation_overrides,
manual_overrides=manual_overrides,
speakers=speakers,
+5 -1
View File
@@ -43,8 +43,12 @@ def update_settings() -> ResponseReturnValue:
current["language"] = (form.get("language") or "en").strip()
current["default_speaker"] = (form.get("default_speaker") or "").strip()
current["default_voice"] = (form.get("default_voice") or "").strip()
provider = str(form.get("tts_provider") or "kokoro").strip().lower()
if provider in {"kokoro", "supertonic"}:
current["tts_provider"] = provider
try:
current["supertonic_total_steps"] = max(2, min(15, int(form.get("supertonic_total_steps", current.get("supertonic_total_steps", 5)))))
total_steps = int(form.get("supertonic_total_steps", current.get("supertonic_total_steps", 8)))
current["supertonic_total_steps"] = max(2, min(15, total_steps))
except (TypeError, ValueError):
pass
try:
+18 -8
View File
@@ -7,7 +7,6 @@ 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,
@@ -33,7 +32,7 @@ from abogen.webui.routes.utils.common import split_profile_spec
from abogen.utils import calculate_text_length
from abogen.voice_profiles import serialize_profiles, normalize_profile_entry
from abogen.chunking import ChunkLevel, build_chunks_for_chapters
from abogen.tts_backend_registry import get_default_voice
from abogen.constants import VOICES_INTERNAL
from abogen.speaker_configs import get_config
from abogen.kokoro_text_normalization import normalize_roman_numeral_titles
from dataclasses import dataclass
@@ -578,9 +577,9 @@ def apply_book_step_form(
# NOTE: Do not auto-set a global TTS provider at the book level based on the
# narrator defaults. Provider is resolved per-speaker/per-chunk from the voice
# spec (e.g. "speaker:Name" for saved speakers, or a Kokoro mix formula).
# This enables mixed-provider conversions (e.g. narrator=SuperTonic, characters=Kokoro).
# This enables mixed-provider conversions (e.g. narrator=Supertonic, characters=Kokoro).
provider_value = str(form.get("tts_provider") or "").strip().lower()
if is_registered_backend(provider_value):
if provider_value in {"kokoro", "supertonic"}:
pending.tts_provider = provider_value
# Determine the base speaker selection (saved speaker ref or raw voice).
@@ -617,8 +616,8 @@ def apply_book_step_form(
custom_formula = ""
base_voice_spec = resolved_default_voice or narrator_voice_raw
if not base_voice_spec:
base_voice_spec = get_default_voice("kokoro")
if not base_voice_spec and VOICES_INTERNAL:
base_voice_spec = VOICES_INTERNAL[0]
voice_choice, resolved_language, selected_profile = resolve_voice_choice(
pending.language,
@@ -797,8 +796,8 @@ def build_pending_job_from_extraction(
profile_selection = inferred_profile
base_voice = base_voice_input or resolved_default_voice or str(default_voice_setting).strip()
if not base_voice:
base_voice = get_default_voice("kokoro")
if not base_voice and VOICES_INTERNAL:
base_voice = VOICES_INTERNAL[0]
selected_speaker_config = (form.get("speaker_config") or "").strip()
speaker_config_payload = get_config(selected_speaker_config) if selected_speaker_config else None
@@ -914,6 +913,15 @@ def build_pending_job_from_extraction(
else:
normalization_overrides[key] = default_val
provider_value = str(form.get("tts_provider") or "").strip().lower()
if provider_value not in {"kokoro", "supertonic"}:
provider_value = settings.get("tts_provider", "kokoro")
try:
total_steps = int(form.get("supertonic_total_steps", settings.get("supertonic_total_steps", 8)))
supertonic_steps = max(2, min(15, total_steps))
except (TypeError, ValueError):
supertonic_steps = int(settings.get("supertonic_total_steps", 8))
pending = PendingJob(
id=uuid.uuid4().hex,
original_filename=original_name,
@@ -929,6 +937,8 @@ def build_pending_job_from_extraction(
replace_single_newlines=replace_single_newlines,
subtitle_format=subtitle_format,
total_characters=total_chars,
tts_provider=provider_value,
supertonic_total_steps=supertonic_steps,
save_chapters_separately=save_chapters_separately,
merge_chapters_at_end=merge_chapters_at_end,
separate_chapters_format=separate_chapters_format,
+7 -6
View File
@@ -78,9 +78,10 @@ def get_preview_pipeline(language: str, device: str) -> Any:
pipeline = _preview_pipelines.get(key)
if pipeline is not None:
return pipeline
from abogen.tts_backend_registry import create_backend
from abogen.utils import load_numpy_kpipeline
pipeline = create_backend("kokoro", lang_code=language, device=device)
_, KPipeline = load_numpy_kpipeline()
pipeline = KPipeline(lang_code=language, repo_id="hexgrad/Kokoro-82M", device=device)
_preview_pipelines[key] = pipeline
return pipeline
@@ -91,7 +92,7 @@ def generate_preview_audio(
speed: float,
use_gpu: bool,
tts_provider: str = "kokoro",
supertonic_total_steps: int = 5,
supertonic_total_steps: int = 8,
max_seconds: float = 8.0,
pronunciation_overrides: Optional[Iterable[Mapping[str, Any]]] = None,
manual_overrides: Optional[Iterable[Mapping[str, Any]]] = None,
@@ -136,9 +137,9 @@ def generate_preview_audio(
normalized_text = source_text
if provider == "supertonic":
from abogen.tts_backend_registry import create_backend
from abogen.tts_supertonic import SupertonicPipeline
pipeline = create_backend("supertonic", sample_rate=SAMPLE_RATE, auto_download=True, total_steps=supertonic_total_steps)
pipeline = SupertonicPipeline(sample_rate=SAMPLE_RATE, auto_download=True, total_steps=supertonic_total_steps)
segments = pipeline(
normalized_text,
voice=voice_spec,
@@ -200,7 +201,7 @@ def synthesize_preview(
speed: float,
use_gpu: bool,
tts_provider: str = "kokoro",
supertonic_total_steps: int = 5,
supertonic_total_steps: int = 8,
max_seconds: float = 8.0,
pronunciation_overrides: Optional[Iterable[Mapping[str, Any]]] = None,
manual_overrides: Optional[Iterable[Mapping[str, Any]]] = None,
+1 -1
View File
@@ -25,7 +25,7 @@ def submit_job(pending: PendingJob) -> str:
tts_provider=getattr(pending, "tts_provider", "kokoro"),
voice=pending.voice,
speed=pending.speed,
supertonic_total_steps=getattr(pending, "supertonic_total_steps", 5),
supertonic_total_steps=getattr(pending, "supertonic_total_steps", 8),
use_gpu=pending.use_gpu,
subtitle_mode=pending.subtitle_mode,
output_format=pending.output_format,
+4 -3
View File
@@ -6,8 +6,8 @@ from abogen.constants import (
LANGUAGE_DESCRIPTIONS,
SUBTITLE_FORMATS,
SUPPORTED_SOUND_FORMATS,
VOICES_INTERNAL,
)
from abogen.tts_backend_registry import get_default_voice
from abogen.normalization_settings import (
DEFAULT_LLM_PROMPT,
environment_llm_defaults,
@@ -174,8 +174,9 @@ def settings_defaults() -> Dict[str, Any]:
"subtitle_format": "srt",
"save_mode": "default_output" if has_output_override() else "save_next_to_input",
"default_speaker": "",
"default_voice": get_default_voice("kokoro"),
"supertonic_total_steps": 5,
"default_voice": VOICES_INTERNAL[0] if VOICES_INTERNAL else "",
"tts_provider": "kokoro",
"supertonic_total_steps": 8,
"supertonic_speed": 1.0,
"replace_single_newlines": False,
"use_gpu": True,
+7 -6
View File
@@ -17,10 +17,10 @@ from abogen.constants import (
SUPPORTED_SOUND_FORMATS,
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
SAMPLE_VOICE_TEXTS,
VOICES_INTERNAL,
)
from abogen.tts_backend_registry import get_metadata
from abogen.speaker_configs import list_configs
from abogen.tts_backend_registry import create_backend
from abogen.utils import load_numpy_kpipeline
from abogen.webui.conversion_runner import _select_device, _to_float32, SAMPLE_RATE, SPLIT_PATTERN
_preview_pipeline_lock = threading.RLock()
@@ -285,7 +285,7 @@ def filter_voice_catalog(
def build_voice_catalog() -> List[Dict[str, str]]:
catalog: List[Dict[str, str]] = []
gender_map = {"f": "Female", "m": "Male"}
for voice_id in get_metadata("kokoro").voices:
for voice_id in VOICES_INTERNAL:
prefix, _, rest = voice_id.partition("_")
language_code = prefix[0] if prefix else "a"
gender_code = prefix[1] if len(prefix) > 1 else ""
@@ -590,7 +590,7 @@ def template_options() -> Dict[str, Any]:
voice_catalog = build_voice_catalog()
return {
"languages": LANGUAGE_DESCRIPTIONS,
"voices": get_metadata("kokoro").voices,
"voices": VOICES_INTERNAL,
"subtitle_formats": SUBTITLE_FORMATS,
"supported_langs_for_subs": SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
"output_formats": SUPPORTED_SOUND_FORMATS,
@@ -666,7 +666,7 @@ def resolve_voice_choice(
# Provider-aware behavior:
# - Kokoro profiles typically represent mixes (formula strings).
# - SuperTonic profiles represent a discrete voice id + settings.
# - Supertonic profiles represent a discrete voice id + settings.
# In that case, we return a speaker reference so downstream can
# resolve provider per-speaker and allow mixed-provider casting.
if provider == "supertonic":
@@ -741,7 +741,8 @@ def get_preview_pipeline(language: str, device: str):
pipeline = _preview_pipelines.get(key)
if pipeline is not None:
return pipeline
pipeline = create_backend("kokoro", lang_code=language, device=device)
_, KPipeline = load_numpy_kpipeline()
pipeline = KPipeline(lang_code=language, repo_id="hexgrad/Kokoro-82M", device=device)
_preview_pipelines[key] = pipeline
return pipeline
+1 -1
View File
@@ -17,7 +17,7 @@ from abogen.speaker_configs import (
save_configs,
delete_config,
)
from abogen.constants import VOICES_INTERNAL
voices_bp = Blueprint("voices", __name__)
+7 -7
View File
@@ -111,7 +111,7 @@ class Job:
subtitle_format: str
created_at: float
tts_provider: str = "kokoro"
supertonic_total_steps: int = 5
supertonic_total_steps: int = 8
save_chapters_separately: bool = False
merge_chapters_at_end: bool = True
separate_chapters_format: str = "wav"
@@ -204,7 +204,7 @@ class Job:
"queue_position": self.queue_position,
"options": {
"tts_provider": getattr(self, "tts_provider", "kokoro"),
"supertonic_total_steps": getattr(self, "supertonic_total_steps", 5),
"supertonic_total_steps": getattr(self, "supertonic_total_steps", 8),
"save_chapters_separately": self.save_chapters_separately,
"merge_chapters_at_end": self.merge_chapters_at_end,
"separate_chapters_format": self.separate_chapters_format,
@@ -552,7 +552,7 @@ class PendingJob:
normalization_overrides: Dict[str, Any]
created_at: float
tts_provider: str = "kokoro"
supertonic_total_steps: int = 5
supertonic_total_steps: int = 8
cover_image_path: Optional[Path] = None
cover_image_mime: Optional[str] = None
chapter_intro_delay: float = 0.5
@@ -621,7 +621,7 @@ class ConversionService:
voice: str,
speed: float,
tts_provider: str = "kokoro",
supertonic_total_steps: int = 5,
supertonic_total_steps: int = 8,
use_gpu: bool,
subtitle_mode: str,
output_format: str,
@@ -674,7 +674,7 @@ class ConversionService:
voice=voice,
speed=speed,
tts_provider=tts_provider,
supertonic_total_steps=int(supertonic_total_steps or 5),
supertonic_total_steps=int(supertonic_total_steps or 8),
use_gpu=use_gpu,
subtitle_mode=subtitle_mode,
output_format=output_format,
@@ -1147,7 +1147,7 @@ class ConversionService:
"tts_provider": getattr(job, "tts_provider", "kokoro"),
"voice": job.voice,
"speed": job.speed,
"supertonic_total_steps": getattr(job, "supertonic_total_steps", 5),
"supertonic_total_steps": getattr(job, "supertonic_total_steps", 8),
"use_gpu": job.use_gpu,
"subtitle_mode": job.subtitle_mode,
"output_format": job.output_format,
@@ -1275,7 +1275,7 @@ class ConversionService:
replace_single_newlines=bool(payload.get("replace_single_newlines", False)),
subtitle_format=payload.get("subtitle_format", "srt"),
created_at=float(payload.get("created_at", time.time())),
supertonic_total_steps=int(payload.get("supertonic_total_steps", 5)),
supertonic_total_steps=int(payload.get("supertonic_total_steps", 8)),
save_chapters_separately=bool(payload.get("save_chapters_separately", False)),
merge_chapters_at_end=bool(payload.get("merge_chapters_at_end", True)),
separate_chapters_format=payload.get("separate_chapters_format", "wav"),
@@ -26,6 +26,26 @@
{% set subtitle_value = settings_dict.get('subtitle_mode', 'Disabled') %}
{% endif %}
{% endif %}
{% set tts_provider_value = form_values.get('tts_provider') if form_values else None %}
{% if not tts_provider_value %}
{% if pending and pending.tts_provider %}
{% set tts_provider_value = pending.tts_provider %}
{% else %}
{% set tts_provider_value = settings_dict.get('tts_provider', 'kokoro') %}
{% endif %}
{% endif %}
{% set supertonic_steps_value = form_values.get('supertonic_total_steps') if form_values else None %}
{% if supertonic_steps_value is none %}
{% if pending and pending.supertonic_total_steps is not none %}
{% set supertonic_steps_value = pending.supertonic_total_steps %}
{% else %}
{% set supertonic_steps_value = settings_dict.get('supertonic_total_steps', 8) %}
{% endif %}
{% endif %}
{% if supertonic_steps_value is not none and supertonic_steps_value is string %}
{% set supertonic_steps_value = supertonic_steps_value|int %}
{% endif %}
{% set is_supertonic = tts_provider_value == 'supertonic' %}
{% set generate_flag = form_values.get('generate_epub3') if form_values else None %}
{% if generate_flag is not none %}
{% set generate_epub3 = True %}
@@ -273,6 +293,13 @@
<div class="form-section__layout form-section__layout--split">
<div class="form-section__group">
<div class="field">
<label for="tts_provider">TTS Engine</label>
<select id="tts_provider" name="tts_provider" data-role="tts-provider" {{ 'disabled' if readonly else '' }}>
<option value="kokoro" {% if tts_provider_value == 'kokoro' %}selected{% endif %}>Kokoro</option>
<option value="supertonic" {% if tts_provider_value == 'supertonic' %}selected{% endif %}>Supertonic</option>
</select>
</div>
<div class="field" data-role="voice-profile-field">
<label for="voice_profile">Voice profile</label>
<select id="voice_profile" name="voice_profile" data-role="voice-profile" {{ 'disabled' if readonly else '' }}>
<option value="__standard" {% if profile_value == '__standard' %}selected{% endif %}>Standard voice</option>
@@ -280,13 +307,13 @@
{% if options.voice_profile_options %}
<optgroup label="Saved mixes">
{% for profile in options.voice_profile_options %}
<option value="{{ profile.name }}" data-language="{{ profile.language }}" data-formula="{{ profile.formula|e }}" {% if profile_value == profile.name %}selected{% endif %}>{{ profile.name }}{% if profile.language %} ({{ profile.language|upper }}){% endif %}</option>
<option value="{{ profile.name }}" data-language="{{ profile.language }}" data-formula="{{ profile.formula|e }}" data-provider="{{ profile.provider|default('kokoro')|lower }}" {% if profile_value == profile.name %}selected{% endif %}>{{ profile.name }}{% if profile.language %} ({{ profile.language|upper }}){% endif %}{% if profile.provider and profile.provider|lower != 'kokoro' %} · {{ profile.provider|capitalize }}{% endif %}</option>
{% endfor %}
</optgroup>
{% endif %}
</select>
</div>
<div class="field" data-role="voice-field" {% if profile_value != '__standard' %}hidden aria-hidden="true"{% endif %}>
<div class="field" data-role="voice-field" data-provider="kokoro" {% if profile_value != '__standard' or is_supertonic %}hidden aria-hidden="true"{% endif %}>
<label for="voice">Voice</label>
<select id="voice" name="voice" data-role="voice-select" data-default="{{ narrator_voice or settings_dict.get('default_voice', '') }}" {{ 'disabled' if readonly else '' }}>
{% for voice in options.voices %}
@@ -294,10 +321,23 @@
{% endfor %}
</select>
</div>
<div class="field" data-conditional="formula" data-role="formula-field" {% if profile_value != '__formula' %}hidden aria-hidden="true"{% endif %}>
<div class="field" data-role="voice-field" data-provider="supertonic" {% if profile_value != '__standard' or not is_supertonic %}hidden aria-hidden="true"{% endif %}>
<label for="voice_st">Supertonic voice</label>
<select id="voice_st" name="voice" data-role="voice-select" data-default="{{ narrator_voice or 'M1' }}" {{ 'disabled' if readonly else '' }}>
{% for voice in ['M1','M2','M3','M4','M5','F1','F2','F3','F4','F5'] %}
<option value="{{ voice }}" {% if narrator_voice == voice and profile_value == '__standard' %}selected{% endif %}>{{ voice }}</option>
{% endfor %}
</select>
</div>
<div class="field" data-conditional="formula" data-role="formula-field" data-provider="kokoro" {% if profile_value != '__formula' or is_supertonic %}hidden aria-hidden="true"{% endif %}>
<label for="voice_formula">Custom voice formula</label>
<input type="text" id="voice_formula" name="voice_formula" placeholder="af_nova*0.4+am_liam*0.6" data-role="voice-formula" value="{{ voice_formula_value }}" {{ 'disabled' if readonly else '' }}>
</div>
<div class="field" data-role="supertonic-steps-field" {% if not is_supertonic %}hidden aria-hidden="true"{% endif %}>
<label for="supertonic_total_steps">Supertonic quality (total steps)</label>
<input type="number" id="supertonic_total_steps" name="supertonic_total_steps" min="2" max="15" value="{{ supertonic_steps_value }}" {{ 'disabled' if readonly else '' }}>
<p class="hint">2 = fastest/lowest quality, 15 = slowest/highest quality.</p>
</div>
</div>
<div class="form-section__group">
<div class="field field--slider">
@@ -423,3 +463,54 @@
</div>
</footer>
</form>
<script nonce="{{ csp_nonce() if csp_nonce else '' }}">
(function() {
const form = document.querySelector('[data-wizard-form="true"][data-step="book"]');
if (!form) return;
const providerSelect = form.querySelector('[data-role="tts-provider"]');
if (!providerSelect) return;
function filterProfilesByProvider(provider) {
const profileSelect = form.querySelector('[data-role="voice-profile"]');
if (!profileSelect) return;
const options = profileSelect.querySelectorAll('option[data-provider]');
options.forEach(function(opt) {
const matches = !opt.dataset.provider || opt.dataset.provider === provider;
opt.hidden = !matches;
if (opt.selected && opt.hidden) {
opt.selected = false;
}
});
if (!profileSelect.value || profileSelect.selectedOptions[0]?.hidden) {
const firstVisible = profileSelect.querySelector('option:not([hidden])');
if (firstVisible) profileSelect.value = firstVisible.value;
}
profileSelect.dispatchEvent(new Event('change', { bubbles: true }));
}
function syncProviderUI(provider) {
var isSupertonic = provider === 'supertonic';
form.querySelectorAll('[data-role="voice-field"]').forEach(function(el) {
el.hidden = el.dataset.provider !== provider;
el.setAttribute('aria-hidden', el.hidden ? 'true' : 'false');
});
var formulaField = form.querySelector('[data-role="formula-field"]');
if (formulaField) {
formulaField.hidden = isSupertonic;
formulaField.setAttribute('aria-hidden', isSupertonic ? 'true' : 'false');
}
var stepsField = form.querySelector('[data-role="supertonic-steps-field"]');
if (stepsField) {
stepsField.hidden = !isSupertonic;
stepsField.setAttribute('aria-hidden', isSupertonic ? 'false' : 'true');
}
filterProfilesByProvider(provider);
}
providerSelect.addEventListener('change', function() {
syncProviderUI(providerSelect.value);
});
syncProviderUI(providerSelect.value);
})();
</script>
+9
View File
@@ -61,6 +61,15 @@
<p class="hint">Pick a saved speaker from Speaker Studio to use by default for new jobs.</p>
</div>
<div class="field">
<label for="tts_provider">Default TTS Engine</label>
<select id="tts_provider" name="tts_provider">
<option value="kokoro" {% if settings.tts_provider == 'kokoro' %}selected{% endif %}>Kokoro</option>
<option value="supertonic" {% if settings.tts_provider == 'supertonic' %}selected{% endif %}>Supertonic</option>
</select>
<p class="hint">Select the default TTS engine for new jobs.</p>
</div>
<div class="field field--wide">
<p class="tag">Kokoro settings</p>
</div>
+4 -4
View File
@@ -30,7 +30,7 @@ dependencies = [
"pip",
"kokoro>=0.9.4",
"misaki[zh]>=0.9.4",
"supertonic>=0.1.0",
"supertonic>=1.3.1",
"ebooklib>=0.19",
"beautifulsoup4>=4.13.4",
"spacy>=3.8.7,<4.0",
@@ -111,11 +111,11 @@ filterwarnings = [
[project.optional-dependencies]
# NVIDIA GPU (Windows) (CUDA 12.6) # uv tool install abogen[cuda126]
cuda126 = ["torch"]
cuda126 = ["torch", "onnxruntime-gpu>=1.26.0"]
# NVIDIA GPU (Windows) (CUDA 12.8) # uv tool install abogen[cuda]
cuda = ["torch"]
cuda = ["torch", "onnxruntime-gpu>=1.26.0"]
# NVIDIA GPU (Windows) (CUDA 13.0) # uv tool install abogen[cuda130]
cuda130 = ["torch"]
cuda130 = ["torch", "onnxruntime-gpu>=1.26.0"]
# AMD GPU (Linux) (ROCm 6.4) # uv tool install abogen[rocm]
rocm = ["torch", "pytorch-triton-rocm"]
# Development dependencies # uv tool install abogen[dev]
+2 -2
View File
@@ -1,7 +1,7 @@
from types import SimpleNamespace
from typing import cast
from abogen.tts_backend_registry import get_metadata
from abogen.constants import VOICES_INTERNAL
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(get_metadata("kokoro").voices)
assert voices.issuperset(VOICES_INTERNAL)
+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.replace("\r\n", "\n")
assert "Second line\n\nThird paragraph." in original_text
def test_epub3_sentence_chunks_render_as_paragraphs(tmp_path) -> None:
-216
View File
@@ -1,216 +0,0 @@
"""Tests for KokoroBackend class."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Iterator, List
from unittest.mock import MagicMock, patch
import numpy as np
import pytest
from abogen.tts_backend import TTSBackendMetadata
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
@dataclass
class _FakeSegment:
graphemes: str
audio: Any # np.ndarray or torch-like tensor
class _FakePipeline:
"""Minimal mock for kokoro.KPipeline."""
def __init__(self, *, lang_code: str, repo_id: str, device: str):
self.lang_code = lang_code
self.repo_id = repo_id
self.device = device
self._voices: dict[str, np.ndarray] = {}
def __call__(
self,
text: str,
*,
voice: Any = "",
speed: float = 1.0,
split_pattern: str | None = None,
) -> Iterator[_FakeSegment]:
yield _FakeSegment(graphemes=text, audio=np.zeros(100, dtype="float32"))
def load_single_voice(self, name: str) -> np.ndarray:
if name not in self._voices:
self._voices[name] = np.ones((1, 256), dtype="float32") * 0.5
return self._voices[name]
def _make_backend(**kwargs: Any):
"""Create KokoroBackend with mocked KPipeline."""
with patch("abogen.tts_backends.kokoro._load_kpipeline") as load:
load.return_value = _FakePipeline
from abogen.tts_backends.kokoro import KokoroBackend
return KokoroBackend(**kwargs)
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestKokoroBackendMetadata:
def test_metadata_returns_tts_backend_metadata(self):
backend = _make_backend(lang_code="a")
meta = backend.metadata
assert isinstance(meta, TTSBackendMetadata)
def test_metadata_fields(self):
backend = _make_backend(lang_code="a")
meta = backend.metadata
assert meta.id == "kokoro"
assert meta.name == "Kokoro"
assert "Kokoro" in meta.description
class TestKokoroBackendInit:
def test_stores_lang_code(self):
backend = _make_backend(lang_code="b")
assert backend._lang_code == "b"
def test_default_repo_id(self):
with patch("abogen.tts_backends.kokoro._load_kpipeline") as load:
load.return_value = _FakePipeline
from abogen.tts_backends.kokoro import KokoroBackend
b = KokoroBackend(lang_code="a")
assert b._pipeline.repo_id == "hexgrad/Kokoro-82M"
def test_custom_repo_id(self):
backend = _make_backend(lang_code="a", repo_id="custom/repo")
assert backend._pipeline.repo_id == "custom/repo"
def test_default_device(self):
backend = _make_backend(lang_code="a")
assert backend._pipeline.device == "cpu"
def test_custom_device(self):
backend = _make_backend(lang_code="a", device="cuda")
assert backend._pipeline.device == "cuda"
class TestKokoroBackendCall:
def test_call_delegates_to_pipeline(self):
backend = _make_backend(lang_code="a")
results = list(backend("hello", voice="af_heart", speed=1.2, split_pattern=r"\n"))
assert len(results) == 1
assert results[0].graphemes == "hello"
def test_call_returns_iterator(self):
backend = _make_backend(lang_code="a")
result = backend("test", voice="af_heart")
assert hasattr(result, "__iter__")
def test_call_with_voice_tensor(self):
backend = _make_backend(lang_code="a")
voice_tensor = np.ones((1, 256), dtype="float32")
results = list(backend("test", voice=voice_tensor))
assert len(results) == 1
def test_call_default_speed(self):
backend = _make_backend(lang_code="a")
# Should not raise with default speed
list(backend("text", voice="af_heart"))
def test_call_default_split_pattern_is_none(self):
backend = _make_backend(lang_code="a")
# split_pattern defaults to None
list(backend("text", voice="af_heart"))
class TestLoadSingleVoice:
def test_load_single_voice_delegates(self):
backend = _make_backend(lang_code="a")
tensor = backend.load_single_voice("af_heart")
assert isinstance(tensor, np.ndarray)
assert tensor.shape == (1, 256)
def test_load_single_voice_caches(self):
backend = _make_backend(lang_code="a")
t1 = backend.load_single_voice("af_heart")
t2 = backend.load_single_voice("af_heart")
assert t1 is t2 # same object
class TestSynthesize:
def test_synthesize_returns_bytes(self):
backend = _make_backend(lang_code="a")
result = backend.synthesize("hello", voice="af_heart")
assert isinstance(result, bytes)
def test_synthesize_nonempty(self):
backend = _make_backend(lang_code="a")
result = backend.synthesize("hello", voice="af_heart")
assert len(result) > 0
def test_synthesize_with_speed(self):
backend = _make_backend(lang_code="a")
result = backend.synthesize("hello", voice="af_heart", speed=1.5)
assert isinstance(result, bytes)
def test_synthesize_empty_text(self):
backend = _make_backend(lang_code="a")
# Empty text produces no segments
result = backend.synthesize("", voice="af_heart")
assert isinstance(result, bytes)
class TestProtocolMethods:
def test_get_available_voices(self):
backend = _make_backend(lang_code="a")
voices = backend.get_available_voices()
assert isinstance(voices, list)
assert len(voices) > 0
assert all(isinstance(v, str) for v in voices)
def test_get_supported_formats(self):
backend = _make_backend(lang_code="a")
formats = backend.get_supported_formats()
assert "pcm_float32" in formats
def test_get_info(self):
backend = _make_backend(lang_code="a")
info = backend.get_info()
assert info["id"] == "kokoro"
assert info["name"] == "Kokoro"
assert info["lang_code"] == "a"
class TestRegistration:
def test_factory_creates_kokoro_backend(self):
from abogen.tts_backends.kokoro import create_kokoro_backend, KokoroBackend
with patch("abogen.tts_backends.kokoro._load_kpipeline") as load:
load.return_value = _FakePipeline
backend = create_kokoro_backend(lang_code="a")
assert isinstance(backend, KokoroBackend)
def test_registry_has_kokoro(self):
import abogen.tts_backends # noqa: F401
from abogen.tts_backend_registry import _registry
meta = _registry.get_metadata("kokoro")
assert meta.id == "kokoro"
assert meta.name == "Kokoro"
def test_registry_factory_returns_kokoro_backend(self):
import abogen.tts_backends # noqa: F401
from abogen.tts_backend_registry import _registry
from abogen.tts_backends.kokoro import KokoroBackend
factory = _registry._factories["kokoro"]
with patch("abogen.tts_backends.kokoro._load_kpipeline") as load:
load.return_value = _FakePipeline
backend = factory(lang_code="a")
assert isinstance(backend, KokoroBackend)
+6 -11
View File
@@ -19,7 +19,7 @@ def test_preview_applies_manual_override_before_normalization(monkeypatch):
# And stub the kokoro pipeline path so generate_preview_audio won't proceed.
# We'll instead validate by calling the override logic through generate_preview_audio
# with provider=supertonic and stub create_backend to capture input.
# with provider=supertonic and stub SupertonicPipeline to capture input.
captured = {}
class DummyPipeline:
@@ -30,16 +30,11 @@ def test_preview_applies_manual_override_before_normalization(monkeypatch):
captured["text"] = text
return iter(())
from abogen import tts_backend_registry
original_create_backend = tts_backend_registry.create_backend
def _mock_create_backend(backend_id, **kwargs):
if backend_id == "supertonic":
return DummyPipeline(**kwargs)
return original_create_backend(backend_id, **kwargs)
monkeypatch.setattr(tts_backend_registry, "create_backend", _mock_create_backend)
monkeypatch.setitem(
__import__("sys").modules,
"abogen.tts_supertonic",
type("M", (), {"SupertonicPipeline": DummyPipeline}),
)
try:
preview.generate_preview_audio(
-314
View File
@@ -1,314 +0,0 @@
from dataclasses import dataclass
from abogen.tts_backend import TTSBackendMetadata
from abogen.tts_backend_registry import TTSBackendRegistry
class TestTTSBackendMetadata:
def test_is_frozen_dataclass(self):
assert dataclass(TTSBackendMetadata)
def test_fields_are_present(self):
meta = TTSBackendMetadata(
id="test",
name="Test Backend",
description="A test backend",
)
assert meta.id == "test"
assert meta.name == "Test Backend"
assert meta.description == "A test backend"
def test_voices_field_default_empty(self):
meta = TTSBackendMetadata(
id="test",
name="Test",
description="Test backend",
)
assert meta.voices == ()
def test_voices_field_stored(self):
meta = TTSBackendMetadata(
id="test",
name="Test",
description="Test backend",
voices=("v1", "v2"),
)
assert meta.voices == ("v1", "v2")
def test_is_immutable(self):
import pytest
meta = TTSBackendMetadata(
id="kokoro",
name="Kokoro",
description="Test",
)
with pytest.raises(Exception):
meta.id = "changed"
class TestTTSBackendRegistry:
def test_register_and_list(self):
registry = TTSBackendRegistry()
meta = TTSBackendMetadata(id="a", name="A", description="Backend A")
registry.register(metadata=meta, factory=lambda: None)
backends = registry.list_backends()
assert len(backends) == 1
assert backends[0].id == "a"
def test_list_multiple(self):
registry = TTSBackendRegistry()
meta_a = TTSBackendMetadata(id="a", name="A", description="A")
meta_b = TTSBackendMetadata(id="b", name="B", description="B")
registry.register(metadata=meta_a, factory=lambda: None)
registry.register(metadata=meta_b, factory=lambda: None)
backends = registry.list_backends()
ids = [b.id for b in backends]
assert "a" in ids
assert "b" in ids
def test_get_metadata(self):
registry = TTSBackendRegistry()
meta = TTSBackendMetadata(id="x", name="X", description="X backend")
registry.register(metadata=meta, factory=lambda: None)
result = registry.get_metadata("x")
assert result.id == "x"
assert result.name == "X"
def test_get_metadata_unknown_raises(self):
import pytest
registry = TTSBackendRegistry()
with pytest.raises(KeyError, match="Unknown backend: nope"):
registry.get_metadata("nope")
def test_create_backend(self):
registry = TTSBackendRegistry()
meta = TTSBackendMetadata(id="test", name="Test", description="Test backend")
def factory(**kwargs):
return {"created": True, "kwargs": kwargs}
registry.register(metadata=meta, factory=factory)
result = registry.create_backend("test", foo="bar")
assert result == {"created": True, "kwargs": {"foo": "bar"}}
def test_create_backend_unknown_raises(self):
import pytest
registry = TTSBackendRegistry()
with pytest.raises(KeyError, match="Unknown backend: missing"):
registry.create_backend("missing")
def test_register_overwrites(self):
registry = TTSBackendRegistry()
meta1 = TTSBackendMetadata(id="x", name="V1", description="First")
meta2 = TTSBackendMetadata(id="x", name="V2", description="Second")
registry.register(metadata=meta1, factory=lambda: "v1")
registry.register(metadata=meta2, factory=lambda: "v2")
result = registry.get_metadata("x")
assert result.name == "V2"
assert registry.create_backend("x") == "v2"
class TestBackendRegistration:
"""Tests that existing backends are auto-registered."""
def test_import_triggers_registration(self):
import abogen.tts_backends # noqa: F401
from abogen.tts_backend_registry import _registry
backends = _registry.list_backends()
ids = [b.id for b in backends]
assert "kokoro" in ids
assert "supertonic" in ids
def test_kokoro_metadata(self):
import abogen.tts_backends # noqa: F401
from abogen.tts_backend_registry import _registry
meta = _registry.get_metadata("kokoro")
assert meta.id == "kokoro"
assert meta.name == "Kokoro"
assert "Kokoro" in meta.description
def test_supertonic_metadata(self):
import abogen.tts_backends # noqa: F401
from abogen.tts_backend_registry import _registry
meta = _registry.get_metadata("supertonic")
assert meta.id == "supertonic"
assert meta.name == "SuperTonic"
assert "SuperTonic" in meta.description
def test_kokoro_metadata_has_voices(self):
import abogen.tts_backends # noqa: F401
from abogen.tts_backend_registry import _registry
meta = _registry.get_metadata("kokoro")
assert isinstance(meta.voices, tuple)
assert len(meta.voices) > 0
assert all(isinstance(v, str) for v in meta.voices)
def test_supertonic_metadata_has_voices(self):
import abogen.tts_backends # noqa: F401
from abogen.tts_backend_registry import _registry
meta = _registry.get_metadata("supertonic")
assert isinstance(meta.voices, tuple)
assert len(meta.voices) == 10
assert meta.voices == ("M1", "M2", "M3", "M4", "M5", "F1", "F2", "F3", "F4", "F5")
def test_kokoro_factory_callable(self):
import abogen.tts_backends # noqa: F401
from abogen.tts_backend_registry import _registry
factory = _registry._factories["kokoro"]
assert callable(factory)
def test_supertonic_factory_callable(self):
import abogen.tts_backends # noqa: F401
from abogen.tts_backend_registry import _registry
factory = _registry._factories["supertonic"]
assert callable(factory)
def test_kokoro_metadata_voices_match_registry(self):
"""Ensure the metadata property on the instance shares voices with registry."""
from abogen.tts_backends.kokoro import _KOKORO_METADATA
from abogen.tts_backend_registry import _registry
registry_meta = _registry.get_metadata("kokoro")
assert _KOKORO_METADATA is registry_meta
assert _KOKORO_METADATA.voices == registry_meta.voices
def test_supertonic_metadata_voices_match_registry(self):
"""Ensure the metadata property on the instance shares voices with registry."""
from abogen.tts_backends.supertonic import _SUPERTONIC_METADATA
from abogen.tts_backend_registry import _registry
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"
+8 -63
View File
@@ -1,6 +1,6 @@
import numpy as np
from abogen.tts_backends.supertonic import SupertonicBackend, SupertonicPipeline
from abogen.tts_supertonic import SupertonicPipeline
class _DummyTTS:
@@ -26,23 +26,13 @@ class _DummyTTS:
return audio, 0.05
def _make_pipeline() -> SupertonicPipeline:
def test_supertonic_pipeline_strips_unsupported_characters_and_retries():
# Avoid importing/initializing real supertonic by manually constructing the pipeline.
pipeline = SupertonicPipeline.__new__(SupertonicPipeline)
pipeline.sample_rate = 24000
pipeline.total_steps = 5
pipeline.max_chunk_length = 1000
pipeline._tts = _DummyTTS()
return pipeline
def _make_backend() -> SupertonicBackend:
backend = SupertonicBackend.__new__(SupertonicBackend)
backend._pipeline = _make_pipeline()
return backend
def test_supertonic_pipeline_strips_unsupported_characters_and_retries():
pipeline = _make_pipeline()
segs = list(pipeline("Hello • world", voice="M1", speed=1.0))
assert len(segs) == 1
@@ -53,56 +43,11 @@ def test_supertonic_pipeline_strips_unsupported_characters_and_retries():
def test_supertonic_pipeline_drops_chunk_if_only_unsupported_characters():
pipeline = _make_pipeline()
pipeline = SupertonicPipeline.__new__(SupertonicPipeline)
pipeline.sample_rate = 24000
pipeline.total_steps = 5
pipeline.max_chunk_length = 1000
pipeline._tts = _DummyTTS()
segs = list(pipeline("", voice="M1", speed=1.0))
assert segs == []
# --- SupertonicBackend tests ---
def test_backend_metadata():
backend = _make_backend()
meta = backend.metadata
assert meta.id == "supertonic"
assert meta.name == "SuperTonic"
assert "SuperTonic" in meta.description
def test_backend_get_available_voices():
backend = _make_backend()
voices = backend.get_available_voices()
assert isinstance(voices, list)
assert "M1" in voices
assert "F1" in voices
def test_backend_get_supported_formats():
backend = _make_backend()
formats = backend.get_supported_formats()
assert "wav" in formats
def test_backend_get_info():
backend = _make_backend()
info = backend.get_info()
assert info["sample_rate"] == 24000
assert info["total_steps"] == 5
assert isinstance(info["voices"], list)
def test_backend_call_delegates_to_pipeline():
backend = _make_backend()
segs = list(backend("Hello • world", voice="M1", speed=1.0))
assert len(segs) == 1
assert segs[0].audio.size > 0
def test_backend_synthesize_returns_wav_bytes():
backend = _make_backend()
wav_bytes = backend.synthesize("Hello world", voice="M1", speed=1.0)
assert isinstance(wav_bytes, bytes)
assert len(wav_bytes) > 0
# WAV magic number
assert wav_bytes[:4] == b"RIFF"
+2 -2
View File
@@ -3,7 +3,7 @@ from typing import cast
import pytest
from abogen.tts_backend_registry import get_metadata
from abogen.constants import VOICES_INTERNAL
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(get_metadata("kokoro").voices)
assert voices.issuperset(VOICES_INTERNAL)
+3 -3
View File
@@ -1,18 +1,18 @@
from __future__ import annotations
from abogen.webui.conversion_runner import _resolve_voice, _supertonic_voice_from_spec
from abogen.tts_backends.supertonic import DEFAULT_SUPERTONIC_VOICES
from abogen.tts_supertonic import DEFAULT_SUPERTONIC_VOICES
def test_resolve_voice_formula_without_pipeline_does_not_crash() -> None:
# This can happen when a previously-saved Kokoro mix formula is present
# but the active provider is SuperTonic (no Kokoro pipeline object).
# but the active provider is Supertonic (no Kokoro pipeline object).
formula = "af_heart*0.5+af_sky*0.5"
resolved = _resolve_voice(None, formula, use_gpu=False)
assert resolved == formula
def test_supertonic_voice_from_formula_falls_back_to_valid_voice() -> None:
# When a stale Kokoro mix formula is present, SuperTonic should not receive it.
# When a stale Kokoro mix formula is present, Supertonic should not receive it.
chosen = _supertonic_voice_from_spec("af_heart*0.5+af_sky*0.5", "af_heart*1.0")
assert chosen in DEFAULT_SUPERTONIC_VOICES
-233
View File
@@ -1,233 +0,0 @@
import pytest
from abogen.voice_metadata import VoiceMetadata
class TestVoiceMetadataCreation:
def test_create_with_all_fields(self):
voice = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
assert voice.id == "af_alloy"
assert voice.display_name == "Alloy"
assert voice.language == "a"
assert voice.gender == "female"
assert voice.backend_id == "kokoro"
def test_create_supertonic_voice(self):
voice = VoiceMetadata(
id="M1",
display_name="Male 1",
language="en",
gender="male",
backend_id="supertonic",
)
assert voice.id == "M1"
assert voice.backend_id == "supertonic"
def test_create_with_unknown_gender(self):
voice = VoiceMetadata(
id="custom_voice",
display_name="Custom",
language="en",
gender="unknown",
backend_id="custom_backend",
)
assert voice.gender == "unknown"
class TestVoiceMetadataImmutability:
def test_frozen_dataclass(self):
voice = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
with pytest.raises(AttributeError):
voice.id = "new_id"
def test_cannot_modify_display_name(self):
voice = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
with pytest.raises(AttributeError):
voice.display_name = "New Name"
def test_cannot_modify_backend_id(self):
voice = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
with pytest.raises(AttributeError):
voice.backend_id = "new_backend"
class TestVoiceMetadataEquality:
def test_equal_voices_are_equal(self):
voice1 = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
voice2 = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
assert voice1 == voice2
def test_different_voices_are_not_equal(self):
voice1 = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
voice2 = VoiceMetadata(
id="am_adam",
display_name="Adam",
language="a",
gender="male",
backend_id="kokoro",
)
assert voice1 != voice2
def test_different_backend_id_not_equal(self):
voice1 = VoiceMetadata(
id="custom",
display_name="Custom",
language="en",
gender="unknown",
backend_id="backend_a",
)
voice2 = VoiceMetadata(
id="custom",
display_name="Custom",
language="en",
gender="unknown",
backend_id="backend_b",
)
assert voice1 != voice2
class TestVoiceMetadataHashing:
def test_hashable(self):
voice = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
assert hash(voice) is not None
def test_equal_voices_same_hash(self):
voice1 = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
voice2 = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
assert hash(voice1) == hash(voice2)
def test_usable_in_set(self):
voice1 = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
voice2 = VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id="kokoro",
)
voice3 = VoiceMetadata(
id="am_adam",
display_name="Adam",
language="a",
gender="male",
backend_id="kokoro",
)
voice_set = {voice1, voice2, voice3}
assert len(voice_set) == 2
class TestVoiceMetadataUseCases:
def test_backend_populates_backend_id(self):
"""Simulate how a backend would populate backend_id automatically."""
class MockBackend:
def __init__(self):
self._backend_id = "kokoro"
def get_voices(self):
return [
VoiceMetadata(
id="af_alloy",
display_name="Alloy",
language="a",
gender="female",
backend_id=self._backend_id,
),
]
backend = MockBackend()
voices = backend.get_voices()
assert voices[0].backend_id == "kokoro"
def test_filter_by_language(self):
voices = [
VoiceMetadata(id="af_alloy", display_name="Alloy", language="a", gender="female", backend_id="kokoro"),
VoiceMetadata(id="jf_alpha", display_name="Alpha", language="j", gender="female", backend_id="kokoro"),
VoiceMetadata(id="am_adam", display_name="Adam", language="a", gender="male", backend_id="kokoro"),
]
english_voices = [v for v in voices if v.language == "a"]
assert len(english_voices) == 2
def test_filter_by_gender(self):
voices = [
VoiceMetadata(id="af_alloy", display_name="Alloy", language="a", gender="female", backend_id="kokoro"),
VoiceMetadata(id="am_adam", display_name="Adam", language="a", gender="male", backend_id="kokoro"),
VoiceMetadata(id="am_puck", display_name="Puck", language="a", gender="male", backend_id="kokoro"),
]
male_voices = [v for v in voices if v.gender == "male"]
assert len(male_voices) == 2
def test_filter_by_backend(self):
voices = [
VoiceMetadata(id="af_alloy", display_name="Alloy", language="a", gender="female", backend_id="kokoro"),
VoiceMetadata(id="M1", display_name="Male 1", language="en", gender="male", backend_id="supertonic"),
]
kokoro_voices = [v for v in voices if v.backend_id == "kokoro"]
assert len(kokoro_voices) == 1
assert kokoro_voices[0].id == "af_alloy"