feat: Add debug TTS functionality with EPUB generation, WAV artifact creation, and settings integration

2026-07-18 05:40:26 +02:00 · 2025-12-15 13:46:13 -08:00
parent daf4d78766
commit aa71783e5a
6 changed files with 551 additions and 1 deletions
@@ -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] = [
        "<?xml version=\"1.0\" encoding=\"utf-8\"?>",
        "<!DOCTYPE html>",
        "<html xmlns=\"http://www.w3.org/1999/xhtml\">",
        "<head>",
        f"  <title>{title}</title>",
        "  <meta charset=\"utf-8\" />",
        "</head>",
        "<body>",
        f"  <h1>{title}</h1>",
        "  <p>Each paragraph begins with a stable debug code marker.</p>",
    ]
    for sample in DEBUG_TTS_SAMPLES:
        safe_label = sample.label.replace("&", "and")
        chapter_lines.append(f"  <h2>{safe_label}</h2>")
        chapter_lines.append(
            "  <p><strong>" + marker_for(sample.code) + "</strong> " + sample.text + "</p>"
        )
    chapter_lines += ["</body>", "</html>"]
    chapter_xhtml = "\n".join(chapter_lines)
    container_xml = """<?xml version="1.0" encoding="UTF-8"?>
 <container version="1.0" xmlns="urn:oasis:names:tc:opendocument:xmlns:container">
  <rootfiles>
    <rootfile full-path="OEBPS/content.opf" media-type="application/oebps-package+xml"/>
  </rootfiles>
 </container>
 """
    content_opf = """<?xml version="1.0" encoding="utf-8"?>
 <package xmlns="http://www.idpf.org/2007/opf" unique-identifier="bookid" version="3.0">
  <metadata xmlns:dc="http://purl.org/dc/elements/1.1/">
    <dc:identifier id="bookid">abogen-debug-samples</dc:identifier>
    <dc:title>abogen debug samples</dc:title>
    <dc:language>en</dc:language>
  </metadata>
  <manifest>
    <item id="chapter" href="chapter.xhtml" media-type="application/xhtml+xml" />
    <item id="nav" href="nav.xhtml" media-type="application/xhtml+xml" properties="nav" />
  </manifest>
  <spine>
    <itemref idref="chapter" />
  </spine>
 </package>
 """
    nav_xhtml = """<?xml version="1.0" encoding="utf-8"?>
 <!DOCTYPE html>
 <html xmlns="http://www.w3.org/1999/xhtml">
 <head>
  <title>Navigation</title>
  <meta charset="utf-8" />
 </head>
 <body>
  <nav epub:type="toc" id="toc">
    <h2>Table of Contents</h2>
    <ol>
      <li><a href="chapter.xhtml">Debug samples</a></li>
    </ol>
  </nav>
 </body>
 </html>
 """
    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
@@ -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<code>[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_<CODE>.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
@@ -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/<run_id>/<filename>")
 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)
@@ -20,6 +20,7 @@
      <button type="button" class="settings-nav__item{% if not llm_ready %} is-disabled{% endif %}" data-section="llm">LLM</button>
      <button type="button" class="settings-nav__item" data-section="normalization">Text Normalization</button>
      <button type="button" class="settings-nav__item" data-section="integrations">Integrations</button>
      <button type="button" class="settings-nav__item" data-section="debug">Debug</button>
    </nav>
    <form action="{{ url_for('settings.update_settings') }}" method="post" class="settings__form">
      <div class="settings-panels">
@@ -472,6 +473,42 @@
            </div>
          </fieldset>
        </section>
        <section class="settings-panel" data-section="debug">
          <fieldset class="settings__section">
            <legend>Debug · TTS transformations</legend>
            <p class="hint">Generate a set of WAV files from a purpose-built EPUB containing code-tagged examples. When something sounds wrong, report the code (e.g. <code>NUM_001</code>) to pinpoint the failing transformation.</p>
            <div class="field field--stack">
              <button type="submit" class="button" formaction="{{ url_for('settings.run_debug_wavs') }}" formmethod="post">Generate debug WAVs</button>
              <p class="hint">Uses your current Settings defaults (voice, language, speed, GPU).</p>
            </div>
            {% if debug_manifest and debug_manifest.artifacts %}
            <div class="field field--wide">
              <label>Latest debug WAVs</label>
              <ul>
                {% for item in debug_manifest.artifacts %}
                <li>
                  <a href="{{ url_for('settings.download_debug_wav', run_id=debug_manifest.run_id, filename=item.filename) }}">{{ item.label }} · {{ item.filename }}</a>
                </li>
                {% endfor %}
              </ul>
            </div>
            {% endif %}
            {% if debug_samples %}
            <div class="field field--wide">
              <label>Included examples</label>
              <ul>
                {% for sample in debug_samples %}
                <li><strong>{{ sample.code }}</strong> — {{ sample.label }}: <span class="muted">{{ sample.text }}</span></li>
                {% endfor %}
              </ul>
            </div>
            {% endif %}
          </fieldset>
        </section>
      </div>
      <div class="modal" data-role="contraction-modal" hidden>
@@ -0,0 +1,97 @@
 import json
 from pathlib import Path
 import numpy as np
 import pytest
 from abogen.debug_tts_samples import DEBUG_TTS_SAMPLES, MARKER_PREFIX, MARKER_SUFFIX, iter_expected_codes
 from abogen.kokoro_text_normalization import HAS_NUM2WORDS, normalize_for_pipeline
 from abogen.normalization_settings import build_apostrophe_config
 from abogen.text_extractor import extract_from_path
 from abogen.web.app import create_app
 def test_debug_epub_contains_all_codes():
    epub_path = Path("tests/fixtures/abogen_debug_tts_samples.epub")
    assert epub_path.exists()
    extraction = extract_from_path(epub_path)
    combined = extraction.combined_text or "\n\n".join((c.text or "") for c in extraction.chapters)
    for code in iter_expected_codes():
        marker = f"{MARKER_PREFIX}{code}{MARKER_SUFFIX}"
        assert marker in combined
 def test_debug_samples_normalize_smoke():
    # Use the same defaults as the web UI.
    from abogen.web.routes.utils.settings import settings_defaults
    settings = settings_defaults()
    apostrophe = build_apostrophe_config(settings=settings)
    runtime = dict(settings)
    normalized = {
        sample.code: normalize_for_pipeline(sample.text, config=apostrophe, settings=runtime)
        for sample in DEBUG_TTS_SAMPLES
    }
    # Contractions should expand under defaults.
    assert "it is" in normalized["APOS_001"].lower()
    # Titles should expand.
    assert "doctor" in normalized["TITLE_001"].lower()
    # Footnotes should be removed.
    assert "[1]" not in normalized["FOOT_001"]
    # Terminal punctuation should be added.
    assert normalized["PUNC_001"].strip()[-1] in {".", "!", "?"}
    if HAS_NUM2WORDS:
        # Currency and numbers should expand to words when num2words is available.
        assert "dollar" in normalized["CUR_001"].lower()
        assert "thousand" in normalized["NUM_001"].lower()
 def test_settings_debug_route_writes_manifest(tmp_path, monkeypatch):
    # Avoid pulling Kokoro models in tests: stub the pipeline.
    from abogen.web import debug_tts_runner as runner
    class _Seg:
        def __init__(self, audio):
            self.audio = audio
    class DummyPipeline:
        def __call__(self, text, **kwargs):
            # 100ms of audio per call, deterministic.
            audio = np.zeros(int(0.1 * runner.SAMPLE_RATE), dtype="float32")
            audio[::100] = 0.1
            yield _Seg(audio)
    monkeypatch.setattr(runner, "_load_pipeline", lambda language, use_gpu: DummyPipeline())
    app = create_app(
        {
            "TESTING": True,
            "SECRET_KEY": "test",
            "OUTPUT_FOLDER": str(tmp_path),
            "UPLOAD_FOLDER": str(tmp_path / "uploads"),
        }
    )
    with app.test_client() as client:
        resp = client.post("/settings/debug/run")
        assert resp.status_code in {302, 303}
        location = resp.headers.get("Location", "")
        assert "debug_run_id=" in location
        # Extract run id from query string.
        run_id = location.split("debug_run_id=")[1].split("&")[0].split("#")[0]
        manifest_path = tmp_path / "debug" / run_id / "manifest.json"
        assert manifest_path.exists()
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
        filenames = {item["filename"] for item in manifest.get("artifacts", [])}
        assert "overall.wav" in filenames
        assert any(name.startswith("case_") and name.endswith(".wav") for name in filenames)