408 lines
14 KiB
Python
408 lines
14 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import time
|
|
import typing
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import torch
|
|
|
|
if typing.TYPE_CHECKING:
|
|
import numpy as np
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
EMOTION_OPTIONS: list[str] = [
|
|
"neutral",
|
|
"angry",
|
|
"sad",
|
|
"happy",
|
|
"fearful",
|
|
"surprised",
|
|
"whisper",
|
|
"pleading",
|
|
"sarcastic",
|
|
"authoritative",
|
|
"thoughtful",
|
|
"excited",
|
|
"anxious",
|
|
"tired",
|
|
"romantic",
|
|
"disgusted",
|
|
"amused",
|
|
"nostalgic",
|
|
"tense",
|
|
"urgent",
|
|
"horror",
|
|
"furious",
|
|
"joyful",
|
|
"terrified",
|
|
"griefstricken",
|
|
"shocked",
|
|
]
|
|
|
|
PACE_OPTIONS: list[str] = ["slow", "moderate", "fast"]
|
|
|
|
EMOTION_INSTRUCT_MAP: dict[str, str] = {
|
|
"angry": (
|
|
"Speak in hot, rising anger. Voice tight, teeth clenched, "
|
|
"clipped aggressive delivery."
|
|
),
|
|
"sad": (
|
|
"Speak in heavy, aching sorrow. Voice cracks and dips; "
|
|
"draw words out on a defeated breath."
|
|
),
|
|
"happy": (
|
|
"Speak with bright, unguarded joy. Smile in the voice; "
|
|
"light, bouncy, almost-laughing rhythm."
|
|
),
|
|
"fearful": (
|
|
"Speak with sharp, trembling fear. Quick shallow breaths, "
|
|
"an unsteady high voice that cuts words short."
|
|
),
|
|
"surprised": (
|
|
"Speak with genuine shock. Pitch jumps, words tumble out "
|
|
"in disbelief, then a stunned silence."
|
|
),
|
|
"whisper": (
|
|
"Speak in a close, hushed whisper. Barely audible, intimate "
|
|
"and secret, tense with withheld breath."
|
|
),
|
|
"pleading": (
|
|
"Speak with desperate, vulnerable pleading. Straining voice, "
|
|
"swaying pitch, reaching drawn-out vowels."
|
|
),
|
|
"sarcastic": (
|
|
"Speak with dragged-out, knowing sarcasm. Heavy irony, "
|
|
"mocking emphasis, an amused cruel edge."
|
|
),
|
|
"authoritative": (
|
|
"Speak with commanding authority. Deep, slow, immovable; "
|
|
"every line lands without argument."
|
|
),
|
|
"thoughtful": (
|
|
"Speak slowly, drifting into thought. Words hang and search "
|
|
"mid-air; quiet revelations between pauses."
|
|
),
|
|
"excited": (
|
|
"Speak bursting with excitement. Rapid-fire, bright, rising; "
|
|
"barely pause for breath, energy spills over."
|
|
),
|
|
"anxious": (
|
|
"Speak with nervous, breathless anxiety. Quick flits between "
|
|
"words, swallowed endings, quiet panic beneath."
|
|
),
|
|
"tired": (
|
|
"Speak drained and exhausted. Heavy sagging words, slow and "
|
|
"quiet, barely any energy to finish a line."
|
|
),
|
|
"disgusted": (
|
|
"Speak curling with disgust. Bitten-off endings, a grimace "
|
|
"in every syllable, contempt in each word."
|
|
),
|
|
"amused": (
|
|
"Speak with barely suppressed amusement. The smile leaks "
|
|
"into the voice; light, teasing, warm laughter."
|
|
),
|
|
"nostalgic": (
|
|
"Speak with wistful, aching nostalgia. Warm and faraway, "
|
|
"lingering on memories, gentle sadness beneath."
|
|
),
|
|
"romantic": (
|
|
"Speak with tender intimate warmth. Low velvet delivery, "
|
|
"long caressed vowels, soft and close to the ear."
|
|
),
|
|
"tense": (
|
|
"Speak coiled with silent tension. Every syllable measured "
|
|
"and charged, shallow held breath, about to snap."
|
|
),
|
|
"urgent": (
|
|
"Speak at a racing, clipped emergency clip. Breath-driven, "
|
|
"decisive, tiny electric pauses, unignorable."
|
|
),
|
|
"horror": (
|
|
"Speak with dawning horror. Hushed, shaking dread, "
|
|
"flattening to a whisper as the terror settles in."
|
|
),
|
|
"furious": (
|
|
"Speak in barely controlled rage. A low growl: fast, violent, "
|
|
"almost shouting, one slip from destruction."
|
|
),
|
|
"joyful": (
|
|
"Speak with uncontainable exuberance. Laugh mid-sentence, "
|
|
"breathless and bright, delight tumbling words."
|
|
),
|
|
"terrified": (
|
|
"Speak in absolute terror. Strangled shaking whisper, gasping, "
|
|
"words cut short in panic, barely formed."
|
|
),
|
|
"griefstricken": (
|
|
"Speak devastated, newly grieving. Voice trembling and breaking, "
|
|
"long defeated pauses, tears barely held."
|
|
),
|
|
"shocked": (
|
|
"Speak stunned into stillness. Hollow and disbelieving; "
|
|
"each word lands like a blow, world just stopped."
|
|
),
|
|
"neutral": "",
|
|
}
|
|
|
|
PACE_INSTRUCT: dict[str, str] = {
|
|
"slow": (
|
|
"Deliver at a slow, deliberate tempo; drag the words out "
|
|
"and let the pauses linger."
|
|
),
|
|
"moderate": "",
|
|
"fast": (
|
|
"Deliver at a quick, clipped tempo; rapid syllables, "
|
|
"minimal pauses, constant momentum."
|
|
),
|
|
}
|
|
|
|
|
|
def compose_utterance_instruct(
|
|
instruct: str | None = None,
|
|
emotion: str | None = None,
|
|
pace: str | None = None,
|
|
) -> str | None:
|
|
"""Compose the TTS instruct from custom text + emotion + pace.
|
|
|
|
Only non-default choices contribute: custom instruct always, emotion
|
|
anything but neutral, pace anything but moderate. With no variation set the
|
|
result is None so a voice clone keeps its natural delivery.
|
|
"""
|
|
parts: list[str] = []
|
|
if instruct:
|
|
parts.append(instruct.strip())
|
|
if emotion and emotion not in ("neutral", ""):
|
|
emotion_instr = EMOTION_INSTRUCT_MAP.get(emotion)
|
|
if emotion_instr:
|
|
parts.append(emotion_instr)
|
|
if pace and pace not in ("moderate", ""):
|
|
pace_instr = PACE_INSTRUCT.get(pace)
|
|
if pace_instr:
|
|
parts.append(pace_instr)
|
|
return " ".join(parts) if parts else None
|
|
|
|
NARRATOR_COMPRESSION_INSTRUCT = (
|
|
"Read steadily and evenly. Low emphasis. Maintain calm, even pace. "
|
|
"Do not dramatize. Smooth, measured delivery for descriptive prose."
|
|
)
|
|
|
|
|
|
class ChapterGenerator:
|
|
def __init__(
|
|
self,
|
|
tts_engine: Any,
|
|
config: Any,
|
|
cache: Any | None = None,
|
|
normalizer: Any | None = None,
|
|
watermark_bits: np.ndarray | None = None,
|
|
author_name: str | None = None,
|
|
) -> None:
|
|
self.tts = tts_engine
|
|
self.config = config
|
|
self.cache = cache
|
|
self.normalizer = normalizer
|
|
self.watermark_bits = watermark_bits
|
|
self.author_name = author_name
|
|
self.max_retries = getattr(config.tts, "max_retries", 3) if hasattr(config, "tts") else 3
|
|
self.retry_delay = (
|
|
getattr(config.tts, "retry_delay", 2.0) if hasattr(config, "tts") else 2.0
|
|
)
|
|
|
|
def _build_utterance_instruct(self, utterance: dict[str, Any], is_narrator: bool) -> str | None:
|
|
if is_narrator:
|
|
return NARRATOR_COMPRESSION_INSTRUCT
|
|
|
|
speaker_profile = utterance.get("_speaker_profile")
|
|
base_instruct = self._build_speaker_base_instruct(speaker_profile)
|
|
varied = compose_utterance_instruct(
|
|
utterance.get("instruct"),
|
|
utterance.get("emotion"),
|
|
utterance.get("pace"),
|
|
)
|
|
return " ".join(p for p in (base_instruct, varied) if p) or None
|
|
|
|
ENERGY_INSTRUCT = {
|
|
"whispered": "Speak in a near-whisper, quiet and subdued.",
|
|
"subdued": "Speak softly, understated and reserved.",
|
|
"soft_spoken": "Speak gently, soft and warm delivery.",
|
|
"measured": "Speak with measured precision, calm and even.",
|
|
"assertive": "Speak with confidence and directness.",
|
|
"commanding": "Speak with authority, firm and commanding.",
|
|
"volatile": "Speak with unpredictable energy, shifting intensity.",
|
|
"explosive": "Speak with explosive force, intense and powerful.",
|
|
}
|
|
|
|
FORMALITY_INSTRUCT = {
|
|
"formal": "Use formal register, polished and precise diction.",
|
|
"casual": "Use casual register, relaxed and conversational.",
|
|
"profane": "Use rough, profane language delivery.",
|
|
"technical": "Use technical, precise terminology delivery.",
|
|
}
|
|
|
|
def _build_speaker_base_instruct(self, profile: dict[str, Any] | None) -> str | None:
|
|
if not profile:
|
|
return None
|
|
parts = []
|
|
energy = profile.get("energy", "")
|
|
if energy in self.ENERGY_INSTRUCT:
|
|
parts.append(self.ENERGY_INSTRUCT[energy])
|
|
formality = profile.get("formality", "")
|
|
if formality in self.FORMALITY_INSTRUCT:
|
|
parts.append(self.FORMALITY_INSTRUCT[formality])
|
|
baseline = profile.get("emotional_baseline", "")
|
|
if baseline and baseline not in ("neutral", ""):
|
|
emotion_instr = EMOTION_INSTRUCT_MAP.get(baseline)
|
|
if emotion_instr:
|
|
parts.append(f"Baseline temperament: {emotion_instr}")
|
|
texture = profile.get("texture", "")
|
|
if texture:
|
|
parts.append(f"Voice texture: {texture}.")
|
|
return " ".join(parts) if parts else None
|
|
|
|
def _compute_speaker_params(
|
|
self, speaker_name: str, profile: dict[str, Any] | None
|
|
) -> tuple[float, float]:
|
|
if profile:
|
|
from umbraaudio.voices.voice_designer import (
|
|
ATTRIBUTE_PARAM_MAP,
|
|
NARRATOR_CFG_WEIGHT,
|
|
NARRATOR_EXAGGERATION,
|
|
)
|
|
|
|
energy = profile.get("energy", "measured")
|
|
if speaker_name == "narrator":
|
|
return NARRATOR_EXAGGERATION, NARRATOR_CFG_WEIGHT
|
|
if energy in ATTRIBUTE_PARAM_MAP:
|
|
exag_range, cfg_range = ATTRIBUTE_PARAM_MAP[energy]
|
|
name_hash = int(hashlib.md5(speaker_name.encode()).hexdigest()[:8], 16)
|
|
exaggeration = exag_range[0] + ((name_hash % 100) / 100.0) * (
|
|
exag_range[1] - exag_range[0]
|
|
)
|
|
cfg_weight = cfg_range[0] + (((name_hash >> 8) % 100) / 100.0) * (
|
|
cfg_range[1] - cfg_range[0]
|
|
)
|
|
return exaggeration, cfg_weight
|
|
|
|
name_hash = int(hashlib.md5(speaker_name.encode()).hexdigest()[:8], 16)
|
|
return 0.2 + (name_hash % 100) / 200.0, 0.2 + ((name_hash >> 8) % 100) / 200.0
|
|
|
|
def generate_chapter(
|
|
self,
|
|
manifest: dict[str, Any],
|
|
output_dir: Path,
|
|
normalized_dir: Path | None = None,
|
|
speaker_profiles: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
if normalized_dir:
|
|
normalized_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
generated_files: list[Path] = []
|
|
failed: list[str] = []
|
|
skipped_cache = 0
|
|
ambiguity_flags: list[dict[str, Any]] = []
|
|
|
|
for utterance in manifest.get("utterances", []):
|
|
utt_id = utterance["id"]
|
|
out_path = output_dir / f"{utt_id}.wav"
|
|
|
|
if out_path.exists():
|
|
generated_files.append(out_path)
|
|
skipped_cache += 1
|
|
continue
|
|
|
|
if self.cache:
|
|
cached = self.cache.get(
|
|
utterance["text"],
|
|
utterance.get("voice_ref"),
|
|
utterance.get("instruct"),
|
|
)
|
|
if cached and cached.exists():
|
|
generated_files.append(cached)
|
|
skipped_cache += 1
|
|
continue
|
|
|
|
voice_ref = utterance.get("voice_ref")
|
|
speaker = utterance.get("speaker", "narrator")
|
|
is_narrator = speaker == "narrator"
|
|
|
|
instruct = self._build_utterance_instruct(utterance, is_narrator)
|
|
profile = (speaker_profiles or {}).get(speaker)
|
|
exaggeration, cfg_weight = self._compute_speaker_params(speaker, profile)
|
|
|
|
success = False
|
|
for attempt in range(self.max_retries):
|
|
try:
|
|
kwargs: dict[str, Any] = {}
|
|
if exaggeration is not None:
|
|
kwargs["exaggeration"] = exaggeration
|
|
if cfg_weight is not None:
|
|
kwargs["cfg_weight"] = cfg_weight
|
|
if self.watermark_bits is not None:
|
|
kwargs["watermark_bits"] = self.watermark_bits
|
|
|
|
with torch.no_grad():
|
|
self.tts.generate(
|
|
text=utterance["text"],
|
|
output_path=out_path,
|
|
voice_ref=Path(voice_ref) if voice_ref else None,
|
|
instruct=instruct,
|
|
**kwargs,
|
|
)
|
|
if torch.cuda.is_available():
|
|
torch.cuda.empty_cache()
|
|
success = True
|
|
break
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Retry %d/%d for %s: %s",
|
|
attempt + 1,
|
|
self.max_retries,
|
|
utt_id,
|
|
e,
|
|
)
|
|
if torch.cuda.is_available():
|
|
torch.cuda.empty_cache()
|
|
if attempt < self.max_retries - 1:
|
|
time.sleep(self.retry_delay)
|
|
|
|
if success:
|
|
if self.cache:
|
|
self.cache.put(
|
|
utterance["text"],
|
|
utterance.get("voice_ref"),
|
|
instruct,
|
|
out_path,
|
|
)
|
|
|
|
if self.normalizer and normalized_dir:
|
|
norm_path = normalized_dir / f"{utt_id}.wav"
|
|
try:
|
|
self.normalizer.normalize_file(out_path, norm_path)
|
|
generated_files.append(norm_path)
|
|
except Exception as e:
|
|
logger.warning("Normalization failed for %s: %s", utt_id, e)
|
|
generated_files.append(out_path)
|
|
else:
|
|
generated_files.append(out_path)
|
|
else:
|
|
failed.append(utt_id)
|
|
logger.error("Failed to generate %s after %d attempts", utt_id, self.max_retries)
|
|
|
|
return {
|
|
"chapter_id": manifest.get("chapter_id"),
|
|
"generated_files": generated_files,
|
|
"total": len(generated_files),
|
|
"failed": failed,
|
|
"skipped_cache": skipped_cache,
|
|
"ambiguity_flags": ambiguity_flags,
|
|
}
|