177 lines
7.0 KiB
Python
177 lines
7.0 KiB
Python
"""Inline text tags for TTS, engine independent (idea only; no third-party code).
|
|
|
|
Tags: [pause 500ms] [pause 1.2s], [spell]USA[/spell], [slow]..[/slow], [fast]..[/fast].
|
|
parse() turns text into segments {text, speed, pause_ms}; render() synthesises and joins them.
|
|
Only a fixed set of tag names is ever stripped, so "[1]" or "[sic]" stay untouched.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Callable
|
|
|
|
MAX_PAUSE_MS = 10_000
|
|
MAX_TOTAL_PAUSE_MS = 60_000
|
|
SPEEDS = {"slow": 0.85, "fast": 1.15}
|
|
|
|
# Control-style tags from other tools that must never be spoken. Allowlist, not "any [word]".
|
|
STRIP_NAMES = frozenset({
|
|
"laugh", "laughs", "laughing", "chuckle", "sigh", "gasp", "cough", "breath", "breathe",
|
|
"inhale", "exhale", "whisper", "shout", "music", "sfx", "noise", "beat", "silence",
|
|
"emphasis", "excited", "sad", "angry", "happy", "calm", "sniff", "gulp", "clears_throat",
|
|
})
|
|
|
|
_TAG_RE = re.compile(r"\[(/?)([A-Za-z_]+)(?:[ \t]+([^\[\]\n]{0,24}))?\]")
|
|
_PAUSE_ARG_RE = re.compile(r"^(\d+(?:\.\d+)?)\s*(ms|s)$", re.I)
|
|
_PARA_RE = re.compile(r"(\n\s*\n)")
|
|
|
|
_DE_LETTERS = {
|
|
"a": "Ah", "b": "Be", "c": "Ze", "d": "De", "e": "Eh", "f": "Eff", "g": "Ge", "h": "Ha",
|
|
"i": "Ih", "j": "Jott", "k": "Ka", "l": "Ell", "m": "Emm", "n": "Enn", "o": "Oh", "p": "Pe",
|
|
"q": "Ku", "r": "Err", "s": "Ess", "t": "Te", "u": "Uh", "v": "Fau", "w": "We", "x": "Iks",
|
|
"y": "Üpsilon", "z": "Zett", "ä": "Ä", "ö": "Ö", "ü": "Ü", "ß": "Ess-Zett",
|
|
}
|
|
|
|
|
|
def _spell(content: str, language: str) -> str:
|
|
chars = [c for c in content if not c.isspace()]
|
|
if language == "de":
|
|
return " ".join(_DE_LETTERS.get(c.lower(), c) for c in chars)
|
|
from core.pronunciations import spell_letters
|
|
return spell_letters("".join(chars))
|
|
|
|
|
|
def _lang(text: str, language: str) -> str:
|
|
low = (language or "auto").strip().lower()
|
|
if low in {"de", "german", "deutsch"} or low.startswith("de-"):
|
|
return "de"
|
|
if low in {"", "auto"}:
|
|
from core.pronunciations import detect_text_language
|
|
return detect_text_language(text)
|
|
return "en"
|
|
|
|
|
|
def _pause_ms(arg: str | None) -> int | None:
|
|
m = _PAUSE_ARG_RE.match((arg or "").strip())
|
|
if not m:
|
|
return None
|
|
ms = float(m.group(1)) * (1000 if m.group(2).lower() == "s" else 1)
|
|
return min(int(ms), MAX_PAUSE_MS)
|
|
|
|
|
|
def has_tags(text: str) -> bool:
|
|
return any(m.group(2).lower() in {"pause", "spell", "slow", "fast"} | STRIP_NAMES
|
|
for m in _TAG_RE.finditer(text or ""))
|
|
|
|
|
|
def _parse_paragraph(text: str, lang: str) -> list[dict]:
|
|
out: list[dict] = []
|
|
speeds: list[float] = []
|
|
spell_from: int | None = None # index in `out` where an open [spell] started
|
|
pos = 0
|
|
|
|
def add(chunk: str) -> None:
|
|
if chunk:
|
|
out.append({"text": chunk, "speed": speeds[-1] if speeds else 1.0, "pause_ms": 0})
|
|
|
|
def close_spell() -> None:
|
|
nonlocal spell_from
|
|
if spell_from is not None:
|
|
joined = "".join(s["text"] for s in out[spell_from:])
|
|
del out[spell_from:]
|
|
add(_spell(joined, lang))
|
|
spell_from = None
|
|
|
|
for m in _TAG_RE.finditer(text):
|
|
name, closing = m.group(2).lower(), m.group(1) == "/"
|
|
known = name in {"pause", "spell", "slow", "fast"} or name in STRIP_NAMES
|
|
if not known:
|
|
continue # ordinary prose such as "[1]" or "[sic]": handled below
|
|
add(text[pos:m.start()])
|
|
pos = m.end()
|
|
if name == "pause" and not closing:
|
|
ms = _pause_ms(m.group(3))
|
|
if ms:
|
|
out.append({"text": "", "speed": 1.0, "pause_ms": ms})
|
|
elif name == "spell":
|
|
if closing:
|
|
close_spell()
|
|
elif spell_from is None:
|
|
spell_from = len(out)
|
|
elif name in SPEEDS:
|
|
if closing:
|
|
if speeds:
|
|
speeds.pop()
|
|
else:
|
|
speeds.append(SPEEDS[name])
|
|
# anything else in STRIP_NAMES (and a stray [/pause]) is dropped
|
|
add(text[pos:])
|
|
close_spell() # unclosed [spell] runs to the end of the paragraph
|
|
return out
|
|
|
|
|
|
def parse(text: str, language: str = "auto") -> list[dict]:
|
|
"""Split text into segments; unclosed tags end at the paragraph break; pauses are capped."""
|
|
lang = _lang(text, language)
|
|
segments: list[dict] = []
|
|
for part in _PARA_RE.split(text or ""):
|
|
if _PARA_RE.fullmatch(part):
|
|
if segments and not segments[-1]["pause_ms"]:
|
|
segments[-1]["text"] += part
|
|
continue
|
|
segments.extend(_parse_paragraph(part, lang))
|
|
budget = MAX_TOTAL_PAUSE_MS
|
|
merged: list[dict] = []
|
|
for seg in segments:
|
|
if seg["pause_ms"]:
|
|
seg["pause_ms"] = min(seg["pause_ms"], budget)
|
|
budget -= seg["pause_ms"]
|
|
if not seg["pause_ms"]:
|
|
continue
|
|
elif merged and not merged[-1]["pause_ms"] and merged[-1]["speed"] == seg["speed"]:
|
|
merged[-1]["text"] += seg["text"]
|
|
continue
|
|
merged.append(seg)
|
|
return [s for s in merged if s["pause_ms"] or s["text"].strip()]
|
|
|
|
|
|
def plain_text(text: str, language: str = "auto") -> str:
|
|
"""Text with all tags resolved and no timing (for paths that cannot render pauses/speed)."""
|
|
return re.sub(r"[ \t]+", " ", " ".join(s["text"].strip() for s in parse(text, language) if s["text"].strip()))
|
|
|
|
|
|
def render(text: str, synth: Callable[[str], tuple[bytes, str]], tempo: Callable[[bytes, float], bytes],
|
|
language: str = "auto") -> tuple[bytes, str]:
|
|
"""Synthesise every segment, stretch slow/fast ones, join with generated silence -> WAV bytes.
|
|
|
|
The result's duration already includes pauses and stretching, so timings measured on it stay right.
|
|
"""
|
|
import io
|
|
from pydub import AudioSegment
|
|
|
|
combined = AudioSegment.empty()
|
|
for seg in parse(text, language):
|
|
if seg["pause_ms"]:
|
|
rate = combined.frame_rate if len(combined) else 24000
|
|
combined += AudioSegment.silent(duration=seg["pause_ms"], frame_rate=rate)
|
|
continue
|
|
audio, media_type = synth(seg["text"].strip())
|
|
fmt = "wav" if audio[:4] == b"RIFF" or "wav" in media_type.lower() else None
|
|
piece = AudioSegment.from_file(io.BytesIO(audio), format=fmt)
|
|
if abs(seg["speed"] - 1.0) > 0.01:
|
|
buf = io.BytesIO()
|
|
piece.export(buf, format="wav")
|
|
piece = AudioSegment.from_file(io.BytesIO(tempo(buf.getvalue(), seg["speed"])), format="wav")
|
|
combined += piece
|
|
out = io.BytesIO()
|
|
combined.export(out, format="wav")
|
|
return out.getvalue(), "audio/wav"
|
|
|
|
|
|
def render_with_tags(text: str, synth: Callable[[str], tuple[bytes, str]], language: str = "auto",
|
|
enabled: bool = False) -> tuple[bytes, str]:
|
|
"""Shared entry for every path that can render tags: plain `synth(text)` unless enabled and tagged."""
|
|
if not (enabled and has_tags(text)):
|
|
return synth(text)
|
|
from core.audio import _change_tempo
|
|
return render(text, synth, _change_tempo, language)
|