diff --git a/abogen/debug_tts_samples.py b/abogen/debug_tts_samples.py new file mode 100644 index 0000000..8c0d44c --- /dev/null +++ b/abogen/debug_tts_samples.py @@ -0,0 +1,179 @@ +from __future__ import annotations + +import tempfile +import zipfile +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable, List, Sequence + + +MARKER_PREFIX = "[[ABOGEN-DBG:" +MARKER_SUFFIX = "]]" + + +@dataclass(frozen=True) +class DebugTTSSample: + code: str + label: str + text: str + + +DEBUG_TTS_SAMPLES: Sequence[DebugTTSSample] = ( + DebugTTSSample( + code="APOS_001", + label="Apostrophes & contractions", + text="It's a beautiful day, isn't it? Let's see what we'll do.", + ), + DebugTTSSample( + code="POS_001", + label="Plural possessives", + text="The dogs' bowls were empty, but the boss's office was quiet.", + ), + DebugTTSSample( + code="NUM_001", + label="Grouped numbers", + text="There are 1,234 apples, 56 oranges, and 7.89 liters of juice.", + ), + DebugTTSSample( + code="YEAR_001", + label="Years and decades", + text="In 1999, people said the '90s were over.", + ), + DebugTTSSample( + code="DATE_001", + label="ISO dates", + text="On 2023-01-01, we celebrated the new year.", + ), + DebugTTSSample( + code="CUR_001", + label="Currency symbols", + text="The price is $10.50, but it was £8.00 yesterday.", + ), + DebugTTSSample( + code="TITLE_001", + label="Titles and abbreviations", + text="Dr. Smith lives on Elm St. near the U.S. border.", + ), + DebugTTSSample( + code="PUNC_001", + label="Terminal punctuation", + text="This sentence ends without punctuation", + ), + DebugTTSSample( + code="QUOTE_001", + label="ALL CAPS inside quotes", + text='He shouted, "THIS IS IMPORTANT!" and then whispered, "ok."', + ), + DebugTTSSample( + code="FOOT_001", + label="Footnote indicators", + text="This is a sentence with a footnote[1] and another[12].", + ), +) + + +def marker_for(code: str) -> str: + return f"{MARKER_PREFIX}{code}{MARKER_SUFFIX}" + + +def build_debug_epub(dest_path: Path, *, title: str = "abogen debug samples") -> Path: + """Create a tiny EPUB containing all debug samples. + + The text includes stable marker codes so developers can report failures + precisely. + """ + + dest_path = Path(dest_path) + dest_path.parent.mkdir(parents=True, exist_ok=True) + + chapter_lines: List[str] = [ + "", + "", + "", + "", + f" {title}", + " ", + "", + "", + f"

{title}

", + "

Each paragraph begins with a stable debug code marker.

", + ] + + for sample in DEBUG_TTS_SAMPLES: + safe_label = sample.label.replace("&", "and") + chapter_lines.append(f"

{safe_label}

") + chapter_lines.append( + "

" + marker_for(sample.code) + " " + sample.text + "

" + ) + + chapter_lines += ["", ""] + chapter_xhtml = "\n".join(chapter_lines) + + container_xml = """ + + + + + +""" + + content_opf = """ + + + abogen-debug-samples + abogen debug samples + en + + + + + + + + + +""" + + nav_xhtml = """ + + + + Navigation + + + + + + +""" + + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp) + (tmp_path / "mimetype").write_text("application/epub+zip", encoding="utf-8") + meta_inf = tmp_path / "META-INF" + meta_inf.mkdir(parents=True, exist_ok=True) + (meta_inf / "container.xml").write_text(container_xml, encoding="utf-8") + oebps = tmp_path / "OEBPS" + oebps.mkdir(parents=True, exist_ok=True) + (oebps / "content.opf").write_text(content_opf, encoding="utf-8") + (oebps / "chapter.xhtml").write_text(chapter_xhtml, encoding="utf-8") + (oebps / "nav.xhtml").write_text(nav_xhtml, encoding="utf-8") + + # Per EPUB spec: mimetype must be the first entry and stored (no compression). + with zipfile.ZipFile(dest_path, "w") as zf: + zf.write(tmp_path / "mimetype", "mimetype", compress_type=zipfile.ZIP_STORED) + for source in (meta_inf / "container.xml", oebps / "content.opf", oebps / "chapter.xhtml", oebps / "nav.xhtml"): + arcname = str(source.relative_to(tmp_path)).replace("\\", "/") + zf.write(source, arcname, compress_type=zipfile.ZIP_DEFLATED) + + return dest_path + + +def iter_expected_codes() -> Iterable[str]: + for sample in DEBUG_TTS_SAMPLES: + yield sample.code diff --git a/abogen/web/debug_tts_runner.py b/abogen/web/debug_tts_runner.py new file mode 100644 index 0000000..9a54591 --- /dev/null +++ b/abogen/web/debug_tts_runner.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import json +import re +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple + +import numpy as np + +from abogen.debug_tts_samples import MARKER_PREFIX, MARKER_SUFFIX, build_debug_epub, iter_expected_codes +from abogen.kokoro_text_normalization import normalize_for_pipeline +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.web.conversion_runner import SAMPLE_RATE, SPLIT_PATTERN, _select_device, _to_float32, _resolve_voice, _spec_to_voice_ids +from abogen.utils import load_numpy_kpipeline + + +_MARKER_RE = re.compile(re.escape(MARKER_PREFIX) + r"(?P[A-Z0-9_]+)" + re.escape(MARKER_SUFFIX)) + + +@dataclass(frozen=True) +class DebugWavArtifact: + label: str + filename: str + code: Optional[str] = None + + +def _load_pipeline(language: str, use_gpu: bool) -> Any: + device = "cpu" + if use_gpu: + device = _select_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]]: + raw = str(text or "") + matches = list(_MARKER_RE.finditer(raw)) + cases: List[Tuple[str, str]] = [] + if not matches: + return cases + for idx, match in enumerate(matches): + code = match.group("code") + start = match.end() + end = matches[idx + 1].start() if idx + 1 < len(matches) else len(raw) + snippet = raw[start:end] + # Keep it small and predictable: collapse whitespace. + snippet = " ".join(snippet.strip().split()) + cases.append((code, snippet)) + return cases + + +def run_debug_tts_wavs( + *, + output_root: Path, + settings: Mapping[str, Any], + epub_path: Optional[Path] = None, +) -> Dict[str, Any]: + """Generate WAV artifacts for the debug EPUB samples. + + Writes: + - overall.wav: concatenation of all samples + - case_.wav: each sample rendered separately + - manifest.json: metadata + file list + """ + + output_root = Path(output_root) + output_root.mkdir(parents=True, exist_ok=True) + + run_id = uuid.uuid4().hex + run_dir = output_root / "debug" / run_id + run_dir.mkdir(parents=True, exist_ok=True) + + if epub_path is None: + epub_path = run_dir / "abogen_debug_samples.epub" + build_debug_epub(epub_path) + else: + epub_path = Path(epub_path) + + extraction = extract_from_path(epub_path) + combined_text = extraction.combined_text or "\n\n".join((c.text or "") for c in extraction.chapters) + cases = _extract_cases_from_text(combined_text) + + expected = list(iter_expected_codes()) + found_codes = {code for code, _ in cases} + missing = [code for code in expected if code not in found_codes] + if missing: + raise RuntimeError(f"Debug EPUB missing expected codes: {', '.join(missing)}") + + language = str(settings.get("language") or "en").strip() or "en" + voice_spec = str(settings.get("default_voice") or "").strip() + use_gpu = bool(settings.get("use_gpu", False)) + speed = float(settings.get("default_speed", 1.0) or 1.0) + + # Best-effort voice caching (only for known Kokoro internal voices). + voice_ids = _spec_to_voice_ids(voice_spec) + if voice_ids: + try: + ensure_voice_assets(voice_ids) + except Exception: + # Network / optional dependency variance; debug runner can still proceed. + pass + + pipeline = _load_pipeline(language, use_gpu) + voice_choice = _resolve_voice(pipeline, voice_spec, use_gpu) + + apostrophe_config = build_apostrophe_config(settings=settings) + normalization_settings = dict(settings) + + artifacts: List[DebugWavArtifact] = [] + + overall_path = run_dir / "overall.wav" + overall_audio: List[np.ndarray] = [] + + def synth(text: str) -> np.ndarray: + normalized = normalize_for_pipeline( + text, + config=apostrophe_config, + settings=normalization_settings, + ) + parts: List[np.ndarray] = [] + for segment in pipeline( + normalized, + voice=voice_choice, + speed=speed, + split_pattern=SPLIT_PATTERN, + ): + audio = _to_float32(getattr(segment, "audio", None)) + if audio.size: + parts.append(audio) + if not parts: + return np.zeros(0, dtype="float32") + return np.concatenate(parts).astype("float32", copy=False) + + # Per sample + for code, snippet in cases: + if not snippet: + continue + audio = synth(snippet) + filename = f"case_{code}.wav" + path = run_dir / filename + # Write float32 PCM WAV. + import soundfile as sf + + sf.write(path, audio, SAMPLE_RATE, subtype="FLOAT") + artifacts.append(DebugWavArtifact(label=f"{code}", filename=filename, code=code)) + overall_audio.append(audio) + + # Overall + if overall_audio: + combined = np.concatenate(overall_audio).astype("float32", copy=False) + else: + combined = np.zeros(0, dtype="float32") + import soundfile as sf + + sf.write(overall_path, combined, SAMPLE_RATE, subtype="FLOAT") + artifacts.insert(0, DebugWavArtifact(label="Overall", filename="overall.wav", code=None)) + + manifest = { + "run_id": run_id, + "epub": str(epub_path), + "artifacts": [artifact.__dict__ for artifact in artifacts], + "sample_rate": SAMPLE_RATE, + } + (run_dir / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8") + return manifest diff --git a/abogen/web/routes/settings.py b/abogen/web/routes/settings.py index a8fff90..b3340b2 100644 --- a/abogen/web/routes/settings.py +++ b/abogen/web/routes/settings.py @@ -1,4 +1,6 @@ -from flask import Blueprint, render_template, request, redirect, url_for, flash +from pathlib import Path + +from flask import Blueprint, current_app, render_template, request, redirect, url_for, flash, send_file, abort from flask.typing import ResponseReturnValue from abogen.web.routes.utils.settings import ( @@ -7,11 +9,16 @@ from abogen.web.routes.utils.settings import ( save_settings, coerce_bool, coerce_int, + SAVE_MODE_LABELS, + llm_ready, _NORMALIZATION_BOOLEAN_KEYS, _NORMALIZATION_STRING_KEYS, _DEFAULT_ANALYSIS_THRESHOLD, ) from abogen.web.routes.utils.voice import template_options +from abogen.web.debug_tts_runner import run_debug_tts_wavs +from abogen.debug_tts_samples import DEBUG_TTS_SAMPLES +from abogen.utils import get_user_output_path settings_bp = Blueprint("settings", __name__) @@ -147,10 +154,71 @@ def settings_page() -> str | ResponseReturnValue: if request.method == "POST": return update_settings() + debug_run_id = (request.args.get("debug_run_id") or "").strip() + debug_manifest = None + if debug_run_id: + run_dir = Path(current_app.config.get("OUTPUT_FOLDER") or get_user_output_path("web")) / "debug" / debug_run_id + manifest_path = run_dir / "manifest.json" + if manifest_path.exists(): + try: + import json + + debug_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + except Exception: + debug_manifest = None + + save_locations = [{"value": key, "label": label} for key, label in SAVE_MODE_LABELS.items()] + default_output_dir = str(Path(get_user_output_path()).resolve()) + return render_template( "settings.html", settings=load_settings(), integrations=load_integration_settings(), options=template_options(), normalization_samples=_NORMALIZATION_SAMPLES, + save_locations=save_locations, + default_output_dir=default_output_dir, + llm_ready=llm_ready(load_settings()), + debug_samples=DEBUG_TTS_SAMPLES, + debug_manifest=debug_manifest, ) + + +@settings_bp.post("/debug/run") +def run_debug_wavs() -> ResponseReturnValue: + settings = load_settings() + output_root = Path(current_app.config.get("OUTPUT_FOLDER") or get_user_output_path("web")) + try: + manifest = run_debug_tts_wavs(output_root=output_root, settings=settings) + except Exception as exc: + flash(f"Debug WAV generation failed: {exc}", "error") + return redirect(url_for("settings.settings_page", _anchor="debug")) + + flash("Debug WAV generation completed.", "success") + return redirect( + url_for( + "settings.settings_page", + debug_run_id=str(manifest.get("run_id") or ""), + _anchor="debug", + ) + ) + + +@settings_bp.get("/debug//") +def download_debug_wav(run_id: str, filename: str) -> ResponseReturnValue: + safe_run = (run_id or "").strip() + safe_name = (filename or "").strip() + if not safe_run or not safe_name or "/" in safe_name or "\\" in safe_name: + abort(404) + if not safe_name.lower().endswith(".wav") and safe_name != "manifest.json": + abort(404) + + root = Path(current_app.config.get("OUTPUT_FOLDER") or get_user_output_path("web")) + path = (root / "debug" / safe_run / safe_name).resolve() + if not path.exists() or not path.is_file(): + abort(404) + # Ensure path is within root/debug/run_id + expected_dir = (root / "debug" / safe_run).resolve() + if expected_dir not in path.parents: + abort(404) + return send_file(path, as_attachment=True, download_name=path.name) diff --git a/abogen/web/templates/settings.html b/abogen/web/templates/settings.html index a5d1513..44fe35e 100644 --- a/abogen/web/templates/settings.html +++ b/abogen/web/templates/settings.html @@ -20,6 +20,7 @@ +
@@ -472,6 +473,42 @@
+ +
+
+ Debug · TTS transformations +

Generate a set of WAV files from a purpose-built EPUB containing code-tagged examples. When something sounds wrong, report the code (e.g. NUM_001) to pinpoint the failing transformation.

+ +
+ +

Uses your current Settings defaults (voice, language, speed, GPU).

+
+ + {% if debug_manifest and debug_manifest.artifacts %} +
+ + +
+ {% endif %} + + {% if debug_samples %} +
+ +
    + {% for sample in debug_samples %} +
  • {{ sample.code }} — {{ sample.label }}: {{ sample.text }}
  • + {% endfor %} +
+
+ {% endif %} +
+