Compare commits

..
Author SHA1 Message Date
Deniz ŞafakandGitHub cbc05ead42 Merge pull request #173 from k0sm0naft/refactor/tts-backend-interface
refactor: introduce TTS backend abstraction
2026-07-03 21:04:47 +03:00
Artem Akymenko 50b4d6872a feat: Add minimal TTSBackend interface for future extensibility
- Create TTSBackend abstract base class with minimal contract
- Implement KokoroTTSBackend that maintains existing behavior
- Update conversion_runner.py to use new interface
- No behavioral changes, GUI unchanged, no new features
2026-07-03 01:25:41 +03:00
66 changed files with 299 additions and 624 deletions
+13 -18
View File
@@ -38,8 +38,6 @@ 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
@@ -67,9 +65,6 @@ 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.
@@ -178,7 +173,7 @@ Abogen offers **two interfaces**, but currently they have different feature sets
| Command | Interface | Features |
|---------|-----------|----------|
| `abogen` | PyQt6 Desktop GUI | Stable core features + **Supertonic TTS**|
| `abogen` | PyQt6 Desktop GUI | Stable core features |
| `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.
@@ -412,18 +407,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,13 +323,6 @@ 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.
)
@@ -355,13 +348,6 @@ 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.

After

Width:  |  Height:  |  Size: 571 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 992 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 786 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 877 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.5 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.0 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 832 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.0 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 521 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 868 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 808 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 372 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.1 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 883 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.6 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 465 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.5 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 885 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 381 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 855 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 917 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 441 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.5 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 856 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 837 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 875 B

Binary file not shown.

After

Width:  |  Height:  |  Size: 617 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 843 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 875 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 891 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.0 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.1 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.3 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.1 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 851 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.3 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.1 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 431 B

-51
View File
@@ -29,57 +29,6 @@ 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",
+90 -94
View File
@@ -20,7 +20,6 @@ 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
@@ -267,10 +266,7 @@ class ConversionThread(QThread):
use_gpu=True,
from_queue=False,
save_base_path=None,
tts_provider="kokoro",
supertonic_language="en",
supertonic_total_steps=8,
):
): # Add use_gpu parameter
super().__init__()
self._chapter_options_event = threading.Event()
self._timestamp_response_event = threading.Event()
@@ -280,10 +276,6 @@ class ConversionThread(QThread):
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
@@ -435,10 +427,6 @@ 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(
@@ -511,16 +499,9 @@ class ConversionThread(QThread):
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
)
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
@@ -766,7 +747,7 @@ class ConversionThread(QThread):
merged_out_path = f"{base_filepath_no_ext}.{self.output_format}"
subtitle_entries = []
current_time = 0.0
rate = self.sample_rate
rate = 24000
subtitle_mode = self.subtitle_mode
self.etr_start_time = time.time()
self.processed_char_count = 0
@@ -782,7 +763,7 @@ class ConversionThread(QThread):
merged_out_file = sf.SoundFile(
merged_out_path,
"w",
samplerate=self.sample_rate,
samplerate=24000,
channels=1,
format=self.output_format,
)
@@ -798,18 +779,62 @@ class ConversionThread(QThread):
# Prepare ffmpeg command for m4b output
cmd = [
"ffmpeg",
"-y",
"-thread_queue_size",
"32768",
"-f",
"f32le",
"-ar",
str(self.sample_rate),
"-ac",
"1",
"-i",
"pipe:0",
"-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",
]
)
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)
@@ -891,7 +916,7 @@ class ConversionThread(QThread):
merged_out_path = None
subtitle_entries = []
current_time = 0.0
rate = self.sample_rate
rate = 24000
subtitle_mode = self.subtitle_mode
self.etr_start_time = time.time()
self.processed_char_count = 0
@@ -944,7 +969,7 @@ class ConversionThread(QThread):
chapter_out_file = sf.SoundFile(
chapter_out_path,
"w",
samplerate=self.sample_rate,
samplerate=24000,
channels=1,
format=separate_chapters_format,
)
@@ -959,7 +984,7 @@ class ConversionThread(QThread):
"-f",
"f32le",
"-ar",
str(self.sample_rate),
"24000",
"-ac",
"1",
"-i",
@@ -1341,7 +1366,7 @@ 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 * self.sample_rate
self.silence_duration * 24000
) # Silence duration at 24,000 Hz
silence_audio = self.np.zeros(silence_samples, dtype="float32")
silence_bytes = silence_audio.tobytes()
@@ -1572,7 +1597,7 @@ class ConversionThread(QThread):
parent_dir, f"{sanitized_base_name}{suffix}"
)
merged_out_path = f"{base_filepath_no_ext}.{self.output_format}"
rate = self.sample_rate
rate = 24000
# Setup audio output
merged_out_file, ffmpeg_proc = None, None
@@ -2422,9 +2447,6 @@ class VoicePreviewThread(QThread):
speed,
use_gpu=False,
parent=None,
tts_provider="kokoro",
supertonic_language="en",
supertonic_total_steps=8,
):
super().__init__(parent)
self.np_module = np_module
@@ -2433,10 +2455,6 @@ class VoicePreviewThread(QThread):
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")
@@ -2446,11 +2464,6 @@ 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 = (
@@ -2465,56 +2478,39 @@ class VoicePreviewThread(QThread):
def run(self):
print(
f"\nVoice: {self.voice}\nLanguage: {self.lang_code}\nSpeed: {self.speed}\nGPU: {self.use_gpu}\nTTS Provider: {self.tts_provider}\n"
f"\nVoice: {self.voice}\nLanguage: {self.lang_code}\nSpeed: {self.speed}\nGPU: {self.use_gpu}\n"
)
# Generate the preview and save to cache
try:
if self.tts_provider == "supertonic":
from abogen.tts_supertonic import SupertonicPipeline
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)
# 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:
# 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)
device = "cpu"
tts = self.kpipeline_class(
lang_code=self.lang_code, repo_id="hexgrad/Kokoro-82M", device=device
)
# Enable voice formula support for preview
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 = self.np_module.concatenate(audio_segments)
sf.write(self.cache_path, audio, self.sample_rate)
# Save directly to the cache path
sf.write(self.cache_path, audio, 24000)
self.temp_wav = self.cache_path
self.finished.emit()
except Exception as e:
+21 -247
View File
@@ -9,7 +9,6 @@ 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,
@@ -84,8 +83,6 @@ from abogen.constants import (
PROGRAM_DESCRIPTION,
LANGUAGE_DESCRIPTIONS,
VOICES_INTERNAL,
KOKORO_LANG_TO_COUNTRY,
SUPERTONIC_LANG_TO_COUNTRY,
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
COLORS,
SUBTITLE_FORMATS,
@@ -973,9 +970,6 @@ 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
@@ -1024,16 +1018,6 @@ 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()
@@ -1123,53 +1107,6 @@ 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)
@@ -1827,12 +1764,6 @@ 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(
@@ -1892,48 +1823,6 @@ 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:"):
@@ -1942,26 +1831,10 @@ class abogen(QWidget):
from abogen.voice_profiles import load_profiles
entry = load_profiles().get(pname, {})
# set mixed voices and language
if isinstance(entry, dict):
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")
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
@@ -1974,12 +1847,7 @@ class abogen(QWidget):
else:
self.mixed_voice_state = None
self.selected_profile_name = None
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.selected_voice, self.selected_lang = data, data[0]
self.config["selected_voice"] = data
if "selected_profile_name" in self.config:
del self.config["selected_profile_name"]
@@ -1998,40 +1866,19 @@ 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 matching current provider
# re-add profiles
profile_icon = QIcon(get_resource_path("abogen.assets", "profile.png"))
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}")
for pname in load_profiles().keys():
self.voice_combo.addItem(profile_icon, pname, f"profile:{pname}")
# re-add voices
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)
for v in VOICES_INTERNAL:
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)
# restore selection
idx = -1
if self.selected_profile_name:
@@ -2222,9 +2069,6 @@ 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
@@ -2368,15 +2212,6 @@ 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:
@@ -2399,8 +2234,6 @@ 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
@@ -2410,8 +2243,6 @@ 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
@@ -2501,10 +2332,6 @@ 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,
@@ -2520,11 +2347,8 @@ class abogen(QWidget):
total_char_count=self.char_count,
use_gpu=self.gpu_ok,
from_queue=from_queue,
save_base_path=self.displayed_file_path,
tts_provider=tts_provider,
supertonic_language=supertonic_language,
supertonic_total_steps=supertonic_total_steps,
)
save_base_path=self.displayed_file_path, # Pass the save base path (original file for EPUB)
) # Use gpu_ok status
# 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
@@ -2601,15 +2425,9 @@ 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))
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:
self.update_log("Loading modules...")
load_thread = LoadPipelineThread(pipeline_loaded_callback)
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()
@@ -2922,32 +2740,9 @@ 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 = ""
@@ -3052,13 +2847,6 @@ 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
@@ -3117,28 +2905,14 @@ class abogen(QWidget):
else None
)
else:
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 ""
lang = self.selected_voice[0]
voice = self.selected_voice
# use same gpu/cpu logic as in conversion
gpu_msg, gpu_ok = get_gpu_acceleration(self.use_gpu)
self.preview_thread = VoicePreviewThread(
np_module, kpipeline_class, lang, voice, speed, gpu_ok,
tts_provider=tts_provider,
supertonic_language=supertonic_language,
supertonic_total_steps=supertonic_total_steps,
np_module, kpipeline_class, lang, voice, speed, gpu_ok
)
self.preview_thread.finished.connect(self._play_preview_audio)
self.preview_thread.error.connect(self._preview_error)
-4
View File
@@ -26,7 +26,3 @@ 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
+2 -5
View File
@@ -31,7 +31,6 @@ from abogen.constants import (
VOICES_INTERNAL,
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
LANGUAGE_DESCRIPTIONS,
KOKORO_LANG_TO_COUNTRY,
COLORS,
)
import re
@@ -190,9 +189,8 @@ 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"{country_code}.png"
"abogen.assets.flags", f"{language_code}.png"
)
gender_icon_path = get_resource_path(
"abogen.assets", "female.png" if is_female else "male.png"
@@ -514,8 +512,7 @@ class VoiceFormulaDialog(QDialog):
header_row.addWidget(QLabel("Language:"))
self.language_combo = QComboBox()
for code, desc in LANGUAGE_OPTIONS:
country_code = KOKORO_LANG_TO_COUNTRY.get(code, code)
flag = get_resource_path("abogen.assets.flags", f"{country_code}.png")
flag = get_resource_path("abogen.assets.flags", f"{code}.png")
if flag and os.path.exists(flag):
self.language_combo.addItem(QIcon(flag), desc, code)
else:
+117
View File
@@ -0,0 +1,117 @@
"""
Minimal TTS Backend Interface
This module defines a minimal interface for TTS backends to enable future
extensibility while maintaining backward compatibility with existing Kokoro
implementation.
"""
from abc import ABC, abstractmethod
from typing import Any, Iterator, Optional, Union
class TTSBackend(ABC):
"""
Minimal interface for TTS backends.
This interface is designed to be minimal and focused on the essential
operations needed for text-to-speech conversion.
"""
@abstractmethod
def __call__(
self,
text: str,
voice: Union[str, Any],
speed: float = 1.0,
**kwargs: Any
) -> Iterator[Any]:
"""
Generate speech segments from text.
Args:
text: Text to convert to speech
voice: Voice specification or object
speed: Speed multiplier for speech
**kwargs: Additional backend-specific parameters
Yields:
Speech segments (audio data, timing info, etc.)
"""
pass
class KokoroTTSBackend(TTSBackend):
"""
Implementation of TTSBackend using Kokoro.
This class provides the concrete implementation that maintains
the existing behavior while conforming to the TTSBackend interface.
"""
def __init__(self, lang_code: str, repo_id: str = "hexgrad/Kokoro-82M", device: str = "cpu"):
"""
Initialize Kokoro backend.
Args:
lang_code: Language code for the model
repo_id: Repository ID for the Kokoro model
device: Device to run the model on (cpu, cuda, etc.)
"""
self.lang_code = lang_code
self.repo_id = repo_id
self.device = device
self._pipeline = None
def _get_pipeline(self):
"""Lazy initialization of the Kokoro pipeline."""
if self._pipeline is None:
from abogen.utils import load_numpy_kpipeline
_, KPipeline = load_numpy_kpipeline()
try:
self._pipeline = KPipeline(
lang_code=self.lang_code,
repo_id=self.repo_id,
device=self.device
)
except RuntimeError as e:
if "CUDA" in str(e) and self.device != "cpu":
# Fall back to CPU if CUDA fails
self._pipeline = KPipeline(
lang_code=self.lang_code,
repo_id=self.repo_id,
device="cpu"
)
else:
raise
return self._pipeline
def __call__(
self,
text: str,
voice: Union[str, Any],
speed: float = 1.0,
split_pattern: str = r"\n+",
**kwargs: Any
) -> Iterator[Any]:
"""
Generate speech segments from text using Kokoro.
Args:
text: Text to convert to speech
voice: Voice specification or object
speed: Speed multiplier for speech
split_pattern: Pattern to split text into segments
**kwargs: Additional parameters passed to the pipeline
Yields:
Speech segments
"""
pipeline = self._get_pipeline()
return pipeline(
text,
voice=voice,
speed=speed,
split_pattern=split_pattern,
**kwargs
)
+6 -22
View File
@@ -4,7 +4,6 @@ import ast
from dataclasses import dataclass
import logging
import math
import os
import re
from typing import Any, Iterable, Iterator, Optional
@@ -16,13 +15,6 @@ logger = logging.getLogger(__name__)
DEFAULT_SUPERTONIC_VOICES = ("M1", "M2", "M3", "M4", "M5", "F1", "F2", "F3", "F4", "F5")
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
class SupertonicSegment:
@@ -97,7 +89,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
@@ -163,7 +155,6 @@ 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,14 +165,11 @@ class SupertonicPipeline:
sample_rate: int,
auto_download: bool = True,
total_steps: int = 5,
max_chunk_length: int = SUPERTONIC_MAX_CHUNK_LENGTH,
lang: str = "en",
intra_op_num_threads: Optional[int] = None,
max_chunk_length: int = 300,
) -> 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()
@@ -193,8 +181,7 @@ class SupertonicPipeline:
"Supertonic is not installed. Install it with `pip install supertonic`."
) from exc
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)
self._tts = TTS(auto_download=auto_download)
def __call__(
self,
@@ -204,14 +191,12 @@ 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(
@@ -222,13 +207,12 @@ 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,
@@ -254,14 +238,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:
+24 -27
View File
@@ -41,6 +41,7 @@ from abogen.utils import (
load_config,
load_numpy_kpipeline,
)
from abogen.tts_backend import KokoroTTSBackend
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
@@ -59,7 +60,7 @@ def _supertonic_voice_from_spec(spec: Any, fallback: str) -> str:
raw = str(spec or "").strip()
fallback_raw = str(fallback or "").strip()
# Supertonic voices are discrete IDs (M1/F3/...). If we see a Kokoro mix
# 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
@@ -1584,7 +1585,7 @@ def run_conversion_job(job: Job) -> None:
pipelines[provider_norm] = SupertonicPipeline(
sample_rate=SAMPLE_RATE,
auto_download=True,
total_steps=int(getattr(job, "supertonic_total_steps", 8) or 8),
total_steps=int(getattr(job, "supertonic_total_steps", 5) or 5),
)
return pipelines[provider_norm]
@@ -1594,16 +1595,12 @@ def run_conversion_job(job: Job) -> None:
device = "cpu"
if not disable_gpu:
device = _select_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
# Create KokoroTTSBackend instance instead of directly instantiating KPipeline
pipelines[provider_norm] = KokoroTTSBackend(
lang_code=job.language,
repo_id="hexgrad/Kokoro-82M",
device=device
)
if not kokoro_cache_ready:
_initialize_voice_cache(job)
kokoro_cache_ready = True
@@ -1618,7 +1615,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", 8) or 8)
steps = int(entry.get("total_steps") or getattr(job, "supertonic_total_steps", 5) or 5)
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)
@@ -1634,7 +1631,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)
@@ -1644,8 +1641,8 @@ def run_conversion_job(job: Job) -> None:
return provider, resolved, cached, speed, steps
if provider == "kokoro":
kokoro_pipeline = get_pipeline("kokoro")
choice = _resolve_voice(kokoro_pipeline, resolved, job.use_gpu)
kokoro_backend = get_pipeline("kokoro")
choice = _resolve_voice(kokoro_backend, resolved, job.use_gpu)
else:
choice = resolved
@@ -1774,8 +1771,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_pipeline = get_pipeline("kokoro")
voice_cache[f"kokoro:{base_voice_resolved}"] = _resolve_voice(kokoro_pipeline, base_voice_resolved, job.use_gpu)
kokoro_backend = get_pipeline("kokoro")
voice_cache[f"kokoro:{base_voice_resolved}"] = _resolve_voice(kokoro_backend, base_voice_resolved, job.use_gpu)
processed_chars = 0
subtitle_index = 1
current_time = 0.0
@@ -1857,11 +1854,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", 8)),
total_steps=int(supertonic_steps_override if supertonic_steps_override is not None else getattr(job, "supertonic_total_steps", 5)),
)
else:
kokoro_pipeline = get_pipeline("kokoro")
segment_iter = kokoro_pipeline(
kokoro_backend = get_pipeline("kokoro")
segment_iter = kokoro_backend(
normalized,
voice=voice_choice,
speed=float(speed_override if speed_override is not None else job.speed),
@@ -1950,8 +1947,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_pipeline = get_pipeline("kokoro")
voice_choice = _resolve_voice(kokoro_pipeline, chapter_voice_resolved, job.use_gpu)
kokoro_backend = get_pipeline("kokoro")
voice_choice = _resolve_voice(kokoro_backend, chapter_voice_resolved, job.use_gpu)
voice_cache[chapter_cache_key] = voice_choice
else:
voice_choice = chapter_voice_resolved
@@ -2095,9 +2092,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_pipeline = get_pipeline("kokoro")
kokoro_backend = get_pipeline("kokoro")
chunk_voice_choice = _resolve_voice(
kokoro_pipeline,
kokoro_backend,
chunk_voice_resolved,
job.use_gpu,
)
@@ -2448,7 +2445,7 @@ def _load_pipeline(job: Job):
return SupertonicPipeline(
sample_rate=SAMPLE_RATE,
auto_download=True,
total_steps=int(getattr(job, "supertonic_total_steps", 8) or 8),
total_steps=int(getattr(job, "supertonic_total_steps", 5) or 5),
)
device = "cpu"
@@ -2610,7 +2607,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 -3
View File
@@ -162,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 8)
supertonic_total_steps = int(payload.get("supertonic_total_steps") or payload.get("total_steps") or settings.get("supertonic_total_steps") or 5)
voice_spec = ""
resolved_provider = provider or "kokoro"
@@ -224,7 +224,7 @@ 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 8)
supertonic_total_steps = int(payload.get("supertonic_total_steps") or 5)
settings = load_settings()
use_gpu = settings.get("use_gpu", False)
@@ -269,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 8),
supertonic_total_steps=supertonic_total_steps or int(settings.get("supertonic_total_steps") or 5),
pronunciation_overrides=pronunciation_overrides,
manual_overrides=manual_overrides,
speakers=speakers,
+1 -5
View File
@@ -43,12 +43,8 @@ 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:
total_steps = int(form.get("supertonic_total_steps", current.get("supertonic_total_steps", 8)))
current["supertonic_total_steps"] = max(2, min(15, total_steps))
current["supertonic_total_steps"] = max(2, min(15, int(form.get("supertonic_total_steps", current.get("supertonic_total_steps", 5)))))
except (TypeError, ValueError):
pass
try:
+1 -12
View File
@@ -577,7 +577,7 @@ 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 provider_value in {"kokoro", "supertonic"}:
pending.tts_provider = provider_value
@@ -913,15 +913,6 @@ 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,
@@ -937,8 +928,6 @@ 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,
+2 -2
View File
@@ -92,7 +92,7 @@ def generate_preview_audio(
speed: float,
use_gpu: bool,
tts_provider: str = "kokoro",
supertonic_total_steps: int = 8,
supertonic_total_steps: int = 5,
max_seconds: float = 8.0,
pronunciation_overrides: Optional[Iterable[Mapping[str, Any]]] = None,
manual_overrides: Optional[Iterable[Mapping[str, Any]]] = None,
@@ -201,7 +201,7 @@ def synthesize_preview(
speed: float,
use_gpu: bool,
tts_provider: str = "kokoro",
supertonic_total_steps: int = 8,
supertonic_total_steps: int = 5,
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", 8),
supertonic_total_steps=getattr(pending, "supertonic_total_steps", 5),
use_gpu=pending.use_gpu,
subtitle_mode=pending.subtitle_mode,
output_format=pending.output_format,
+1 -2
View File
@@ -175,8 +175,7 @@ def settings_defaults() -> Dict[str, Any]:
"save_mode": "default_output" if has_output_override() else "save_next_to_input",
"default_speaker": "",
"default_voice": VOICES_INTERNAL[0] if VOICES_INTERNAL else "",
"tts_provider": "kokoro",
"supertonic_total_steps": 8,
"supertonic_total_steps": 5,
"supertonic_speed": 1.0,
"replace_single_newlines": False,
"use_gpu": True,
+1 -1
View File
@@ -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":
+7 -7
View File
@@ -111,7 +111,7 @@ class Job:
subtitle_format: str
created_at: float
tts_provider: str = "kokoro"
supertonic_total_steps: int = 8
supertonic_total_steps: int = 5
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", 8),
"supertonic_total_steps": getattr(self, "supertonic_total_steps", 5),
"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 = 8
supertonic_total_steps: int = 5
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 = 8,
supertonic_total_steps: int = 5,
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 8),
supertonic_total_steps=int(supertonic_total_steps or 5),
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", 8),
"supertonic_total_steps": getattr(job, "supertonic_total_steps", 5),
"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", 8)),
supertonic_total_steps=int(payload.get("supertonic_total_steps", 5)),
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,26 +26,6 @@
{% 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 %}
@@ -293,13 +273,6 @@
<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>
@@ -307,13 +280,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 }}" 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>
<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>
{% endfor %}
</optgroup>
{% endif %}
</select>
</div>
<div class="field" data-role="voice-field" data-provider="kokoro" {% if profile_value != '__standard' or is_supertonic %}hidden aria-hidden="true"{% endif %}>
<div class="field" data-role="voice-field" {% if profile_value != '__standard' %}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 %}
@@ -321,23 +294,10 @@
{% endfor %}
</select>
</div>
<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 %}>
<div class="field" data-conditional="formula" data-role="formula-field" {% if profile_value != '__formula' %}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">
@@ -463,54 +423,3 @@
</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,15 +61,6 @@
<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>=1.3.1",
"supertonic>=0.1.0",
"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", "onnxruntime-gpu>=1.26.0"]
cuda126 = ["torch"]
# NVIDIA GPU (Windows) (CUDA 12.8) # uv tool install abogen[cuda]
cuda = ["torch", "onnxruntime-gpu>=1.26.0"]
cuda = ["torch"]
# NVIDIA GPU (Windows) (CUDA 13.0) # uv tool install abogen[cuda130]
cuda130 = ["torch", "onnxruntime-gpu>=1.26.0"]
cuda130 = ["torch"]
# 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
@@ -6,13 +6,13 @@ 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