diff --git a/backend/database/__init__.py b/backend/database/__init__.py index fd1252bf4..5df700754 100644 --- a/backend/database/__init__.py +++ b/backend/database/__init__.py @@ -20,6 +20,7 @@ ProfileChannelMapping, ProfileSample, Project, + PronunciationEntry, Story, StoryItem, VoiceProfile, @@ -41,6 +42,7 @@ "MCPClientBinding", "ProfileChannelMapping", "ProfileSample", + "PronunciationEntry", "Project", "Story", "StoryItem", diff --git a/backend/database/migrations.py b/backend/database/migrations.py index d353b58c8..863442913 100644 --- a/backend/database/migrations.py +++ b/backend/database/migrations.py @@ -43,9 +43,90 @@ def run_migrations(engine) -> None: _migrate_generation_versions(engine, inspector, tables) _migrate_capture_settings(engine, inspector, tables) _migrate_mcp_bindings(engine, inspector, tables) + _migrate_pronunciation_entries(engine, inspector, tables) _normalize_storage_paths(engine, tables) +def _migrate_pronunciation_entries(engine, inspector, tables: set[str]) -> None: + """Add the scope-uniqueness index to an existing pronunciation_entries table. + + ``create_all`` builds the index with the table, so a fresh install needs + nothing here. A database created by an earlier build of this feature has the + table but not the index, and ``create_all`` will not add one to a table that + already exists. + + Duplicates that predate the index would make CREATE UNIQUE INDEX fail, so + they are collapsed first, keeping the oldest row in each scope. + + "Oldest" has to mean ``created_at``, not ``MIN(id)``: ids are random UUIDs, + so ordering by id would keep an arbitrary row and pick a different one on a + different machine. The concatenated key sorts by timestamp first and falls + back to id, which keeps the choice deterministic when two rows share a + timestamp -- ISO-8601 text sorts chronologically, so this needs no window + function and stays portable across SQLite builds. + + ``IF NOT EXISTS`` rather than relying on the inspector check alone: the + inspector reflects a snapshot, and a migration that raises takes startup + down with it. The check stays as the fast path that skips the dedup scan. + """ + if "pronunciation_entries" not in tables: + return + + # Strategy columns. Existing rows are all respellings, which is the + # default, so no backfill is needed. + columns = _get_columns(inspector, "pronunciation_entries") + for column, ddl in ( + ("strategy", "strategy VARCHAR NOT NULL DEFAULT 'respell'"), + ("spoken_language", "spoken_language VARCHAR"), + ("phonemes", "phonemes VARCHAR"), + ): + if column not in columns: + _add_column(engine, "pronunciation_entries", ddl, column) + + # sqlite_master rather than the inspector: SQLAlchemy cannot reflect an + # expression-based index and skips it with a warning, so the inspector + # never reports this one and the guard would never fire -- leaving the + # dedup scan to run on every startup for nothing. + with engine.connect() as conn: + already = conn.execute( + text( + "SELECT 1 FROM sqlite_master " + "WHERE type = 'index' AND name = 'uq_pronunciation_scope'" + ) + ).first() + if already: + return + + with engine.connect() as conn: + removed = conn.execute( + text( + """ + DELETE FROM pronunciation_entries + WHERE COALESCE(created_at, '') || '|' || id NOT IN ( + SELECT MIN(COALESCE(created_at, '') || '|' || id) + FROM pronunciation_entries + GROUP BY lower(term), COALESCE(language, ''), COALESCE(profile_id, '') + ) + """ + ) + ).rowcount + conn.execute( + text( + """ + CREATE UNIQUE INDEX IF NOT EXISTS uq_pronunciation_scope + ON pronunciation_entries ( + lower(term), COALESCE(language, ''), COALESCE(profile_id, '') + ) + """ + ) + ) + conn.commit() + + if removed: + logger.info("Collapsed %d duplicate pronunciation entries before indexing", removed) + logger.info("Added uq_pronunciation_scope index to pronunciation_entries") + + # -- helpers --------------------------------------------------------------- def _get_columns(inspector, table: str) -> set[str]: diff --git a/backend/database/models.py b/backend/database/models.py index b85a55b17..9ff928ab7 100644 --- a/backend/database/models.py +++ b/backend/database/models.py @@ -1,9 +1,21 @@ """ORM model definitions for the voicebox SQLite database.""" -from datetime import datetime import uuid +from datetime import datetime -from sqlalchemy import Column, String, Integer, Float, DateTime, Text, ForeignKey, Boolean, JSON +from sqlalchemy import ( + JSON, + Boolean, + Column, + DateTime, + Float, + ForeignKey, + Index, + Integer, + String, + Text, + func, +) from sqlalchemy.ext.declarative import declarative_base from ..utils.capture_chords import ( @@ -141,6 +153,67 @@ class GenerationVersion(Base): created_at = Column(DateTime, default=datetime.utcnow) +class PronunciationEntry(Base): + """A term the engine says wrong, and how to spell it so it says it right. + + Respelling rather than phonemes: every engine reads plain text, so + ``bandeja -> ban-DEH-ha`` works everywhere, where a phoneme string only + works on the engines that accept one. + """ + + __tablename__ = "pronunciation_entries" + + id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4())) + term = Column(String, nullable=False) + + # How to say it. Always populated, and always something every engine can + # read, so it doubles as the fallback when a richer strategy is not + # available on the target engine. For a `language` entry it is the term + # itself, since the text does not change -- only which language reads it. + replacement = Column(String, nullable=False) + + # Which realisation to prefer: "respell" | "language" | "phoneme". + # The compiler drops to `replacement` whenever the engine cannot do the + # preferred one, which is the same native-or-structural rule the + # directives follow. + strategy = Column(String, nullable=False, default="respell") + # For strategy="language": the language this term should be read in. + # Distinct from `language` below, which is about *when* the entry applies. + spoken_language = Column(String, nullable=True) + # For strategy="phoneme": engine-specific phoneme string. + phonemes = Column(String, nullable=True) + + # NULL applies in every generation language. A code restricts the entry to + # that language -- a Spanish word needs respelling when the engine is + # reading English, but not when it is already reading Spanish. + language = Column(String, nullable=True) + # NULL is a global entry; set to scope it to one voice. + profile_id = Column(String, ForeignKey("profiles.id"), nullable=True) + enabled = Column(Boolean, default=True, nullable=False) + notes = Column(Text, nullable=True) + created_at = Column(DateTime, default=datetime.utcnow) + updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + + # One entry per (term, language, profile) scope, enforced by the database + # rather than only by a pre-insert check, which two concurrent creates can + # both pass. + # + # A plain UniqueConstraint would not do it: SQL treats NULLs as distinct, + # so two global entries for the same term would both be accepted -- exactly + # the case worth catching. COALESCE maps the wildcard scopes onto a real + # value so they compare equal, and lower() makes the term case-insensitive + # to match how it is looked up. + __table_args__ = ( + Index( + "uq_pronunciation_scope", + func.lower(term), + func.coalesce(language, ""), + func.coalesce(profile_id, ""), + unique=True, + ), + ) + + class EffectPreset(Base): """Saved effect chain preset.""" diff --git a/backend/models.py b/backend/models.py index 7970ce41e..9fd3250f7 100644 --- a/backend/models.py +++ b/backend/models.py @@ -2,7 +2,8 @@ Pydantic models for request/response validation. """ -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, StringConstraints +from typing_extensions import Annotated from typing import Optional, List from datetime import datetime @@ -97,6 +98,16 @@ class GenerationRequest(BaseModel): default=50, ge=0, le=500, description="Crossfade duration in ms between chunks (0 for hard cut)" ) normalize: bool = Field(default=True, description="Normalize output audio volume") + prosody: bool = Field( + default=True, + description=( + "Resolve pronunciation entries and prosody markup (, , " + ", ) before synthesis. Set false to speak the text " + "literally, for a script that genuinely contains something shaped " + "like a tag. Text with no markup and no dictionary hits is " + "unaffected either way." + ), + ) effects_chain: Optional[List["EffectConfig"]] = Field( None, description="Effects chain to apply after generation (overrides profile default)" ) @@ -127,6 +138,191 @@ class Config: from_attributes = True +# A value of " " passes a raw min_length check and then stores as empty once +# the route strips it. Strip first, then length-check the result. +TrimmedTerm = Annotated[ + str, StringConstraints(strip_whitespace=True, min_length=1, max_length=200) +] +TrimmedReplacement = Annotated[ + str, StringConstraints(strip_whitespace=True, min_length=1, max_length=500) +] + + +PRONUNCIATION_STRATEGIES = "^(respell|language|phoneme)$" + + +class PronunciationEntryCreate(BaseModel): + """Request model for creating a pronunciation entry.""" + + term: TrimmedTerm + # Always required, and always plain text every engine can read: it is the + # fallback whenever the chosen strategy is unavailable on the target + # engine. For a `language` entry, repeat the term -- the text does not + # change, only which language reads it. + replacement: TrimmedReplacement + strategy: str = Field( + default="respell", + pattern=PRONUNCIATION_STRATEGIES, + description=( + "How to realise the term. 'respell' substitutes text and works " + "everywhere; 'language' renders it in another language; 'phoneme' " + "passes phonemes on engines that accept them." + ), + ) + spoken_language: Optional[str] = Field( + None, + pattern="^(zh|en|ja|ko|de|fr|ru|pt|es|it|he|ar|da|el|fi|hi|ms|nl|no|pl|sv|sw|tr)$", + description="For strategy='language': the language to read this term in.", + ) + phonemes: Optional[str] = Field( + None, max_length=500, description="For strategy='phoneme'." + ) + language: Optional[str] = Field( + None, + pattern="^(zh|en|ja|ko|de|fr|ru|pt|es|it|he|ar|da|el|fi|hi|ms|nl|no|pl|sv|sw|tr)$", + description="Apply only when generating in this language. Omit for all languages.", + ) + profile_id: Optional[str] = Field( + None, description="Scope to one voice. Omit for a global entry." + ) + enabled: bool = True + notes: Optional[str] = Field(None, max_length=1000) + + +class PronunciationEntryUpdate(BaseModel): + """Request model for updating a pronunciation entry. Omitted fields are left alone.""" + + term: Optional[TrimmedTerm] = None + replacement: Optional[TrimmedReplacement] = None + strategy: Optional[str] = Field(None, pattern=PRONUNCIATION_STRATEGIES) + spoken_language: Optional[str] = Field( + None, pattern="^(zh|en|ja|ko|de|fr|ru|pt|es|it|he|ar|da|el|fi|hi|ms|nl|no|pl|sv|sw|tr)$" + ) + phonemes: Optional[str] = Field(None, max_length=500) + language: Optional[str] = Field( + None, pattern="^(zh|en|ja|ko|de|fr|ru|pt|es|it|he|ar|da|el|fi|hi|ms|nl|no|pl|sv|sw|tr)$" + ) + profile_id: Optional[str] = None + enabled: Optional[bool] = None + notes: Optional[str] = Field(None, max_length=1000) + + +class PronunciationEntryResponse(BaseModel): + """Response model for a pronunciation entry.""" + + id: str + term: str + replacement: str + strategy: str = "respell" + spoken_language: Optional[str] = None + phonemes: Optional[str] = None + language: Optional[str] = None + profile_id: Optional[str] = None + enabled: bool = True + notes: Optional[str] = None + created_at: datetime + updated_at: datetime + + class Config: + from_attributes = True + + +class PronunciationPreviewRequest(BaseModel): + """Request to see what the dictionary would do to a piece of text.""" + + text: str = Field(..., min_length=1, max_length=50000) + language: Optional[str] = None + profile_id: Optional[str] = None + + +class PronunciationSubstitution(BaseModel): + """One replacement the dictionary made.""" + + term: str + replacement: str + entry_id: str + + +class PronunciationPreviewResponse(BaseModel): + """What the engine would actually be given, and why it differs.""" + + original: str + result: str + applied: List[PronunciationSubstitution] + + +class ProsodyPreviewRequest(BaseModel): + """Ask what a script compiles to, without generating anything.""" + + text: str = Field(..., min_length=1, max_length=50000) + engine: str = Field(default="qwen", max_length=50) + language: str = Field( + default="en", pattern="^(zh|en|ja|ko|de|fr|ru|pt|es|it|he|ar|da|el|fi|hi|ms|nl|no|pl|sv|sw|tr)$" + ) + profile_id: Optional[str] = None + instruct: Optional[str] = Field(None, max_length=500) + + +class ProsodyPlanNode(BaseModel): + """One step of the plan: a generation call, or a silence.""" + + kind: str + text: Optional[str] = None + language: Optional[str] = None + rate: Optional[float] = None + instruct: Optional[str] = None + # Set when a substitution changed what the engine hears, so a reviewer can + # see the difference rather than only the result. + source_text: Optional[str] = None + ms: Optional[int] = None + + +class ProsodyPlanWarning(BaseModel): + """Something the target engine cannot honour.""" + + code: str + detail: str + + +class ProsodyPreviewResponse(BaseModel): + """The compiled plan, before any audio exists.""" + + original: str + # The script with dictionary entries resolved into markup -- the same + # directives an author could have typed. + markup: str + dictionary_terms: List[str] + nodes: List[ProsodyPlanNode] + warnings: List[ProsodyPlanWarning] + run_count: int + # True when the script needs none of the harness and takes the ordinary + # single-shot generation path. + is_trivial: bool + + +class ProsodyAnnotateRequest(BaseModel): + """Ask the local LLM to draft markup for a script.""" + + text: str = Field(..., min_length=1, max_length=10000) + language: str = Field( + default="en", pattern="^(zh|en|ja|ko|de|fr|ru|pt|es|it|he|ar|da|el|fi|hi|ms|nl|no|pl|sv|sw|tr)$" + ) + model_size: Optional[str] = Field(default=None, pattern=r"^(0\.6B|1\.7B|4B)$") + + +class ProsodyAnnotateResponse(BaseModel): + """A suggestion for a human to review. Nothing is stored or generated.""" + + original: str + # Safe to use unconditionally: on rejection this is the original text. + markup: str + accepted: bool + changed: bool + rejected_reason: Optional[str] = None + model_size: Optional[str] = None + attempts: int = 0 + + class HistoryQuery(BaseModel): """Query model for generation history.""" diff --git a/backend/routes/__init__.py b/backend/routes/__init__.py index 42999d2d1..b4b4911bb 100644 --- a/backend/routes/__init__.py +++ b/backend/routes/__init__.py @@ -15,6 +15,8 @@ def register_routers(app: FastAPI) -> None: from .captures import router as captures_router from .stories import router as stories_router from .effects import router as effects_router + from .pronunciation import router as pronunciation_router + from .prosody import router as prosody_router from .audio import router as audio_router from .models import router as models_router from .settings import router as settings_router @@ -36,6 +38,8 @@ def register_routers(app: FastAPI) -> None: app.include_router(captures_router) app.include_router(stories_router) app.include_router(effects_router) + app.include_router(pronunciation_router) + app.include_router(prosody_router) app.include_router(audio_router) app.include_router(models_router) app.include_router(settings_router) diff --git a/backend/routes/generations.py b/backend/routes/generations.py index fbbeece67..4e13af936 100644 --- a/backend/routes/generations.py +++ b/backend/routes/generations.py @@ -139,6 +139,7 @@ async def generate_speech( mode="generate", max_chunk_chars=data.max_chunk_chars, crossfade_ms=data.crossfade_ms, + prosody=data.prosody, ) ) @@ -363,17 +364,34 @@ async def stream_speech( runaway_detector = has_tts_runaway - audio, sample_rate = await generate_chunked( - tts_model, + from ..services.prosody.pipeline import engine_capabilities, generate_with_prosody + + supports_instruct, engine_langs = engine_capabilities(engine) + + # The same transformer the persisted path uses, so a streamed preview + # matches what /generate would produce rather than approximating it. + audio, sample_rate = await generate_with_prosody( data.text, - voice_prompt, + engine=engine, language=data.language, + generate_chunked_fn=generate_chunked, + tts_model=tts_model, + voice_prompt=voice_prompt, + gen_kwargs=dict( + language=data.language, + seed=data.seed, + instruct=data.instruct, + max_chunk_chars=data.max_chunk_chars, + crossfade_ms=data.crossfade_ms, + trim_fn=trim_fn, + runaway_detector=runaway_detector, + ), + db=db, + profile_id=data.profile_id, + supports_instruct=supports_instruct, + engine_languages=engine_langs, seed=data.seed, - instruct=data.instruct, - max_chunk_chars=data.max_chunk_chars, - crossfade_ms=data.crossfade_ms, - trim_fn=trim_fn, - runaway_detector=runaway_detector, + enabled=data.prosody, ) effects_chain_config = None diff --git a/backend/routes/pronunciation.py b/backend/routes/pronunciation.py new file mode 100644 index 000000000..6d5dc52ed --- /dev/null +++ b/backend/routes/pronunciation.py @@ -0,0 +1,198 @@ +"""Pronunciation dictionary endpoints.""" + +import logging + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from .. import models +from ..database import PronunciationEntry, VoiceProfile as DBVoiceProfile, get_db +from ..services import pronunciation + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +_SCOPE_TAKEN = ( + "That scope already has an entry for this term. Update the existing entry instead." +) + + +def _commit_or_conflict(db: Session) -> None: + """Commit, turning a scope collision into a 409. + + ``find_duplicate`` runs first and gives a friendlier message naming the + existing row, but it is a check-then-act pair: two concurrent creates can + both pass it. ``uq_pronunciation_scope`` is what actually holds, so its + violation has to surface as a conflict rather than a 500. + """ + try: + db.commit() + except IntegrityError as exc: + db.rollback() + if "uq_pronunciation_scope" in str(exc.orig): + raise HTTPException(status_code=409, detail=_SCOPE_TAKEN) from exc + raise + + +def _validate_strategy(strategy: str, spoken_language: str | None, phonemes: str | None) -> None: + """A strategy has to carry what it needs to be realisable. + + Accepting `strategy="language"` with no language would store an entry that + silently degrades to its fallback -- the user would see the strategy they + picked and hear something else. + """ + if strategy == "language" and not spoken_language: + raise HTTPException( + status_code=400, + detail="strategy='language' needs spoken_language, e.g. 'es'.", + ) + if strategy == "phoneme" and not (phonemes or "").strip(): + raise HTTPException( + status_code=400, detail="strategy='phoneme' needs phonemes." + ) + + +def _validate_profile(profile_id: str | None, db: Session) -> None: + if profile_id is None: + return + if db.query(DBVoiceProfile).filter_by(id=profile_id).first() is None: + raise HTTPException(status_code=404, detail=f"Profile '{profile_id}' not found") + + +@router.get("/pronunciations", response_model=list[models.PronunciationEntryResponse]) +async def list_pronunciations( + language: str | None = Query(None, description="Filter to entries that apply to this language"), + profile_id: str | None = Query(None, description="Filter to entries that apply to this voice"), + include_disabled: bool = Query(True), + db: Session = Depends(get_db), +): + """List dictionary entries. + + With no filters this returns everything, which is what a management screen + wants. Passing ``language`` or ``profile_id`` narrows it to what would + actually apply to a generation with those settings. + """ + if language is None and profile_id is None: + q = db.query(PronunciationEntry) + if not include_disabled: + q = q.filter(PronunciationEntry.enabled.is_(True)) + return q.order_by(PronunciationEntry.term).all() + + return pronunciation.get_entries( + db, language=language, profile_id=profile_id, include_disabled=include_disabled + ) + + +@router.post("/pronunciations", response_model=models.PronunciationEntryResponse) +async def create_pronunciation( + data: models.PronunciationEntryCreate, + db: Session = Depends(get_db), +): + """Add a term and how to say it.""" + _validate_profile(data.profile_id, db) + _validate_strategy(data.strategy, data.spoken_language, data.phonemes) + + existing = pronunciation.find_duplicate(db, data.term, data.language, data.profile_id) + if existing is not None: + raise HTTPException( + status_code=409, + detail=( + f"An entry for '{data.term}' already exists in this scope " + f"(id {existing.id}). Update it instead." + ), + ) + + entry = PronunciationEntry( + term=data.term, + replacement=data.replacement, + strategy=data.strategy, + spoken_language=data.spoken_language, + phonemes=data.phonemes, + language=data.language, + profile_id=data.profile_id, + enabled=data.enabled, + notes=data.notes, + ) + db.add(entry) + _commit_or_conflict(db) + db.refresh(entry) + return entry + + +@router.put("/pronunciations/{entry_id}", response_model=models.PronunciationEntryResponse) +async def update_pronunciation( + entry_id: str, + data: models.PronunciationEntryUpdate, + db: Session = Depends(get_db), +): + """Update an entry. Omitted fields are left as they are.""" + entry = db.query(PronunciationEntry).filter_by(id=entry_id).first() + if entry is None: + raise HTTPException(status_code=404, detail="Pronunciation entry not found") + + fields = data.model_dump(exclude_unset=True) + if "profile_id" in fields: + _validate_profile(fields["profile_id"], db) + + # Validate against the row as it will be, not just what was sent -- a + # strategy change can rely on a field set by an earlier request. + if {"strategy", "spoken_language", "phonemes"} & fields.keys(): + _validate_strategy( + fields.get("strategy", entry.strategy), + fields.get("spoken_language", entry.spoken_language), + fields.get("phonemes", entry.phonemes), + ) + + # Re-check the scope only when something that defines it moved. + if {"term", "language", "profile_id"} & fields.keys(): + clash = pronunciation.find_duplicate( + db, + fields.get("term", entry.term), + fields.get("language", entry.language), + fields.get("profile_id", entry.profile_id), + exclude_id=entry_id, + ) + if clash is not None: + raise HTTPException( + status_code=409, + detail=f"That scope already has an entry for this term (id {clash.id}).", + ) + + for key, value in fields.items(): + setattr(entry, key, value) + + _commit_or_conflict(db) + db.refresh(entry) + return entry + + +@router.delete("/pronunciations/{entry_id}") +async def delete_pronunciation(entry_id: str, db: Session = Depends(get_db)): + """Delete an entry.""" + entry = db.query(PronunciationEntry).filter_by(id=entry_id).first() + if entry is None: + raise HTTPException(status_code=404, detail="Pronunciation entry not found") + db.delete(entry) + db.commit() + return {"message": "Pronunciation entry deleted"} + + +@router.post("/pronunciations/preview", response_model=models.PronunciationPreviewResponse) +async def preview_pronunciations( + data: models.PronunciationPreviewRequest, + db: Session = Depends(get_db), +): + """Show what the engine would be given for this text. + + The dictionary runs at generation time and the rewritten text is never + stored, so without this there is no way to see what a rule actually does + short of listening to the output. + """ + _validate_profile(data.profile_id, db) + result, applied = pronunciation.apply_pronunciations( + data.text, data.language, db, profile_id=data.profile_id + ) + return {"original": data.text, "result": result, "applied": applied} diff --git a/backend/routes/prosody.py b/backend/routes/prosody.py new file mode 100644 index 000000000..205909392 --- /dev/null +++ b/backend/routes/prosody.py @@ -0,0 +1,126 @@ +"""Prosody transformer endpoints. + +Two things a caller needs before generating: what the script will actually be +turned into, and optional help writing the markup in the first place. +""" + +import logging + +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy.orm import Session + +from .. import models +from ..database import get_db +from ..services import pronunciation +from ..services.prosody import annotate, compile_plan, rules_from_entries +from ..services.prosody.ir import Silence, Speech +from ..services.prosody.llm_annotate import ( + DEFAULT_MODEL_SIZE, + LLMUnavailableError, + annotate_with_llm, + is_llm_available, +) +from ..services.prosody.parser import ProsodyParseError +from ..services.prosody.pipeline import engine_capabilities + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +@router.post("/prosody/preview", response_model=models.ProsodyPreviewResponse) +async def preview_prosody( + data: models.ProsodyPreviewRequest, + db: Session = Depends(get_db), +): + """Show what a script compiles to, without generating anything. + + The dictionary and the markup both resolve into a plan long before any + audio exists, so this is the difference between trusting the pipeline and + inspecting it: every cut, every language, every silence, and everything the + chosen engine cannot honour. + """ + entries = pronunciation.get_entries( + db, language=data.language, profile_id=data.profile_id + ) + annotated, applied_terms = annotate(data.text, rules_from_entries(entries)) + + supports_instruct, languages = engine_capabilities(data.engine) + try: + plan = compile_plan( + annotated, + engine=data.engine, + default_language=data.language, + supports_instruct=supports_instruct, + engine_languages=languages, + base_instruct=data.instruct, + ) + except ProsodyParseError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + nodes = [ + models.ProsodyPlanNode( + kind="silence" if isinstance(n, Silence) else "speech", + text=getattr(n, "text", None), + language=getattr(n, "language", None), + rate=getattr(n, "rate", None), + instruct=getattr(n, "instruct", None), + source_text=getattr(n, "source_text", None), + ms=getattr(n, "ms", None), + ) + for n in plan.nodes + ] + + return models.ProsodyPreviewResponse( + original=data.text, + markup=annotated, + dictionary_terms=applied_terms, + nodes=nodes, + warnings=[ + models.ProsodyPlanWarning(code=w.code, detail=w.detail) for w in plan.warnings + ], + run_count=sum(1 for n in plan.nodes if isinstance(n, Speech)), + is_trivial=plan.is_trivial, + ) + + +@router.get("/prosody/annotate/availability") +async def annotation_availability(model_size: str = DEFAULT_MODEL_SIZE): + """Whether LLM annotation can run right now. + + Lets a client hide or disable the action instead of offering something that + will fail. Annotation is optional help -- everything else works without it. + """ + return {"available": is_llm_available(model_size), "model_size": model_size} + + +@router.post("/prosody/annotate", response_model=models.ProsodyAnnotateResponse) +async def annotate_prosody(data: models.ProsodyAnnotateRequest): + """Draft prosody markup for a script using the local LLM. + + The result is markup for a human to review, not audio. Nothing is stored + and nothing is generated -- accepting the suggestion means keeping the + returned text, which then runs through exactly the same pipeline as markup + typed by hand. + + A suggestion whose words differ from the input is rejected: the model can + fail to help, but it cannot rewrite the script. + """ + try: + result = await annotate_with_llm( + data.text, + language=data.language, + model_size=data.model_size or DEFAULT_MODEL_SIZE, + ) + except LLMUnavailableError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + + return models.ProsodyAnnotateResponse( + original=data.text, + markup=result.markup, + accepted=result.accepted, + changed=result.changed, + rejected_reason=result.rejected_reason, + model_size=result.model_size, + attempts=result.attempts, + ) diff --git a/backend/services/generation.py b/backend/services/generation.py index a4b2e8a3f..58e3c0717 100644 --- a/backend/services/generation.py +++ b/backend/services/generation.py @@ -42,6 +42,7 @@ async def run_generation( max_chunk_chars: Optional[int] = None, crossfade_ms: Optional[int] = None, version_id: Optional[str] = None, + prosody: bool = True, ) -> None: """Execute TTS inference and persist the result. @@ -56,6 +57,7 @@ async def run_generation( ) from ..utils.chunked_tts import generate_chunked from ..utils.audio import has_tts_runaway, normalize_audio, save_audio, trim_tts_output + from .prosody.pipeline import engine_capabilities, generate_with_prosody task_manager = get_task_manager() bg_db = next(get_db()) @@ -76,6 +78,7 @@ async def run_generation( ) await history.update_generation_status(generation_id, "generating", bg_db) + trim_fn = trim_tts_output if engine_needs_trim(engine) else None runaway_detector = has_tts_runaway if engine_retries_runaway(engine) else None @@ -91,7 +94,28 @@ async def run_generation( if crossfade_ms is not None: gen_kwargs["crossfade_ms"] = crossfade_ms - audio, sample_rate = await generate_chunked(tts_model, text, voice_prompt, **gen_kwargs) + # The transformer resolves dictionary entries and prosody markup into a + # plan. Unmarked text with no dictionary hits compiles to one plain run + # and takes the same single-shot call as before -- the row in + # `generations` keeps what the author wrote either way, so History + # stays readable and editing an entry changes future audio without + # rewriting the past. + supports_instruct, engine_langs = engine_capabilities(engine) + audio, sample_rate = await generate_with_prosody( + text, + engine=engine, + language=language, + generate_chunked_fn=generate_chunked, + tts_model=tts_model, + voice_prompt=voice_prompt, + gen_kwargs=gen_kwargs, + db=bg_db, + profile_id=profile_id, + supports_instruct=supports_instruct, + engine_languages=engine_langs, + seed=seed, + enabled=prosody, + ) # --- Normalize (generate and regenerate always; retry skips) ----- if normalize or mode == "regenerate": diff --git a/backend/services/pronunciation.py b/backend/services/pronunciation.py new file mode 100644 index 000000000..8def983ee --- /dev/null +++ b/backend/services/pronunciation.py @@ -0,0 +1,191 @@ +"""Pronunciation dictionary — reusable fixes for terms the engine says wrong. + +Names, acronyms, brands and loanwords come out wrong and there is no way to +correct them short of editing the text every time (#827). This maps a term to a +respelling applied just before TTS. + +Respelling rather than phonemes on purpose: every engine reads plain text, so +``bandeja -> ban-DEH-ha`` works on all of them, where a phoneme string only +works on the engines that accept one. It is cruder and it is portable. + +Applied at generation time, not when the text is saved, so ``generations.text`` +keeps what the author wrote. Editing an entry then changes future audio without +rewriting history, and the History tab never shows a reader ``ban-DEH-ha``. +""" + +import logging +import re + +from sqlalchemy import func, or_ +from sqlalchemy.orm import Session + +from ..database import PronunciationEntry + +logger = logging.getLogger(__name__) + +# Paralinguistic tags like [laugh] are engine syntax, not speech. A term that +# happens to appear inside one must not be rewritten. +_TAG_RE = re.compile(r"\[[^\]]*\]") + + +def get_entries( + db: Session, + language: str | None = None, + profile_id: str | None = None, + include_disabled: bool = False, +) -> list[PronunciationEntry]: + """Entries that apply to a given language and profile. + + An entry with a NULL language or profile is a wildcard, so the filters ask + for "matches, or unset" rather than equality. + """ + q = db.query(PronunciationEntry) + if not include_disabled: + q = q.filter(PronunciationEntry.enabled.is_(True)) + if language is not None: + q = q.filter( + or_(PronunciationEntry.language.is_(None), PronunciationEntry.language == language) + ) + if profile_id is not None: + q = q.filter( + or_(PronunciationEntry.profile_id.is_(None), PronunciationEntry.profile_id == profile_id) + ) + else: + q = q.filter(PronunciationEntry.profile_id.is_(None)) + return q.all() + + +def _match_case(source: str, replacement: str) -> str: + """Carry the matched text's capitalisation onto the replacement. + + A term at the start of a sentence is capitalised there and nowhere else, so + storing one lowercase entry has to cover both. + + "All caps" counts *cased* characters, not ``str.isupper()``: that returns + True for ``C++``, which has one cased letter, and shouting the replacement + would turn ``C plus plus`` into ``C PLUS PLUS``. An acronym like ``WCAG`` + has four and is genuinely all caps. + """ + cased = [c for c in source if c.isalpha()] + if len(cased) > 1 and all(c.isupper() for c in cased): + return replacement.upper() + if source[:1].isupper(): + return replacement[:1].upper() + replacement[1:] + return replacement + + +def _tag_spans(text: str) -> list[tuple[int, int]]: + return [(m.start(), m.end()) for m in _TAG_RE.finditer(text)] + + +def build_pattern(terms: list[str]) -> re.Pattern | None: + """One alternation over every term, longest first. + + Longest-first matters twice. It lets a multi-word entry beat the + single-word entry inside it, and because this is a single pass, a + replacement can never be re-matched by another rule — so + ``bandeja -> ban-DEH-ha`` and ``ha -> hah`` cannot compound into + ``ban-DEH-hah``, which is what a loop of per-term substitutions would do. + + Lookarounds rather than ``\\b`` so terms that start or end with punctuation + still anchor on a word boundary. + """ + usable = [t for t in terms if t and t.strip()] + if not usable: + return None + ordered = sorted(set(usable), key=len, reverse=True) + alternation = "|".join(re.escape(t) for t in ordered) + return re.compile(rf"(? tuple[str, list[dict]]: + """Rewrite *text* using the dictionary. + + Returns the rewritten text and a record of what was replaced, so a caller + can log or surface it — a silent rewrite of someone's script is worse than + no rewrite at all. + """ + if not text or not text.strip(): + return text, [] + + entries = get_entries(db, language=language, profile_id=profile_id) + if not entries: + return text, [] + + # A profile-scoped entry beats a global one for the same term; a + # language-specific entry beats a wildcard. Sorting the losers first lets + # the later assignment win. + by_term: dict = {} + for e in sorted( + entries, + key=lambda e: ((e.profile_id is not None), (e.language is not None)), + ): + by_term[e.term.lower()] = e + + pattern = build_pattern([e.term for e in by_term.values()]) + if pattern is None: + return text, [] + + skip = _tag_spans(text) + applied: list[dict] = [] + + def substitute(m: re.Match) -> str: + if any(start <= m.start() < end for start, end in skip): + return m.group(0) + entry = by_term.get(m.group(0).lower()) + if entry is None: + return m.group(0) + out = _match_case(m.group(0), entry.replacement) + applied.append({"term": m.group(0), "replacement": out, "entry_id": entry.id}) + return out + + result = pattern.sub(substitute, text) + if applied: + # Terms are user-supplied and are often names, so the values stay at + # DEBUG; INFO carries only how many were rewritten. + logger.info("Pronunciation dictionary rewrote %d term(s)", len(applied)) + logger.debug( + "Pronunciation substitutions: %s", + ", ".join(f"{a['term']}->{a['replacement']}" for a in applied), + ) + return result, applied + + +def find_duplicate( + db: Session, + term: str, + language: str | None, + profile_id: str | None, + exclude_id: str | None = None, +) -> PronunciationEntry | None: + """An existing entry for the same term in the same scope. + + An early check so the caller can return a useful 409 naming the existing + entry. The database enforces the same rule via ``uq_pronunciation_scope``, + which is what actually holds under concurrent creates. + + Compares ``lower(term)`` rather than ``ilike``: a term is a literal, and + ``ilike`` would read ``%`` and ``_`` in it as wildcards, so ``band_ja`` + would collide with ``bandeja``. Trimmed first, because the route stores the + trimmed value and an untrimmed lookup would miss its own duplicate. + """ + normalized = term.strip().lower() + q = db.query(PronunciationEntry).filter(func.lower(PronunciationEntry.term) == normalized) + q = ( + q.filter(PronunciationEntry.language.is_(None)) + if language is None + else q.filter(PronunciationEntry.language == language) + ) + q = ( + q.filter(PronunciationEntry.profile_id.is_(None)) + if profile_id is None + else q.filter(PronunciationEntry.profile_id == profile_id) + ) + if exclude_id: + q = q.filter(PronunciationEntry.id != exclude_id) + return q.first() diff --git a/backend/services/prosody/__init__.py b/backend/services/prosody/__init__.py new file mode 100644 index 000000000..959c6e978 --- /dev/null +++ b/backend/services/prosody/__init__.py @@ -0,0 +1,38 @@ +"""Prosody transformer: a harness around audio segment production. + +It never synthesises anything. It decides where to cut a script, what settings +each cut carries, and how the pieces are reassembled -- so pauses, per-span +language and rate work on every engine, including the ones that accept no +directives at all. + + markup ──▶ parse ──▶ compile(engine) ──▶ RenderPlan ──▶ render ──▶ audio + +The plan is the seam. It is plain data with no model behind it, so it can be +built, asserted on and previewed for free. +""" + +from .annotate import LANGUAGE, PHONEME, RESPELL, TermRule, annotate, rules_from_entries +from .compiler import compile_plan +from .ir import Attrs, Break, PlanWarning, RenderPlan, Silence, Speech, Text +from .parser import ProsodyParseError, has_markup, parse, strip_markup + +__all__ = [ + "LANGUAGE", + "PHONEME", + "RESPELL", + "Attrs", + "Break", + "PlanWarning", + "ProsodyParseError", + "RenderPlan", + "Silence", + "Speech", + "TermRule", + "Text", + "annotate", + "compile_plan", + "has_markup", + "parse", + "rules_from_entries", + "strip_markup", +] diff --git a/backend/services/prosody/annotate.py b/backend/services/prosody/annotate.py new file mode 100644 index 000000000..b1e5a149c --- /dev/null +++ b/backend/services/prosody/annotate.py @@ -0,0 +1,185 @@ +"""Turn dictionary entries into markup. + +The dictionary does not get its own execution path. It writes the same +directives an author would have written by hand, and the ordinary +parse-compile-render pipeline takes it from there. + +That is worth more than it first looks: + +* a dictionary term and a hand-written span compose, because by the time the + compiler sees them they are the same thing; +* every rule the compiler already enforces -- coalescing, orphaned punctuation, + engine capability -- applies to dictionary output for free; +* the result is *showable*. A preview can hand back the annotated markup, so a + rule that fires unexpectedly is visible rather than being an invisible + difference between what was typed and what was spoken. + +The forthcoming LLM annotator emits into exactly the same slot, which is what +keeps "with an LLM" and "without an LLM" the same code path downstream. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from .parser import _TAG_RE + +# Strategies an entry can ask for, best-first within each engine's abilities. +RESPELL = "respell" +LANGUAGE = "language" +PHONEME = "phoneme" +STRATEGIES = (RESPELL, LANGUAGE, PHONEME) + + +@dataclass(frozen=True) +class TermRule: + """One dictionary entry, reduced to what annotation needs. + + Decoupled from the ORM row so the annotator stays pure and testable, and so + the LLM path can synthesise rules without inventing database objects. + """ + + term: str + replacement: str + strategy: str = RESPELL + spoken_language: str | None = None + phonemes: str | None = None + + def realise(self, matched: str, *, supports_phonemes: bool) -> str: + """The markup this rule becomes on an engine with these abilities. + + Falls back to ``replacement`` whenever the preferred strategy is not + available -- which is why ``replacement`` is always populated, even for + rules that do not primarily respell. + """ + if self.strategy == LANGUAGE and self.spoken_language: + return f'{matched}' + if self.strategy == PHONEME and self.phonemes and supports_phonemes: + return f'{matched}' + if self.replacement.lower() == matched.lower(): + # Nothing to change and no strategy available: leave the text alone + # rather than wrapping it in a tag that does nothing. + return matched + # The alias is what gets spoken, so it has to carry the matched casing: + # one lowercase entry must cover the term mid-sentence and at the start + # of one. + alias = _match_case(matched, self.replacement) + return f'{matched}' + + +def _match_case(source: str, replacement: str) -> str: + """Carry the matched text's capitalisation onto the replacement. + + "All caps" counts *cased* characters rather than using ``str.isupper()``, + which returns True for ``C++`` -- shouting the replacement would turn + ``C plus plus`` into ``C PLUS PLUS``. + """ + cased = [c for c in source if c.isalpha()] + if len(cased) > 1 and all(c.isupper() for c in cased): + return replacement.upper() + if source[:1].isupper(): + return replacement[:1].upper() + replacement[1:] + return replacement + + +def _escape(value: str) -> str: + """Attribute values are quoted, so a quote inside one would end it early.""" + return value.replace('"', """) + + +def _build_pattern(terms: list[str]) -> re.Pattern | None: + """One alternation over every term, longest first. + + Same construction as the dictionary's own matcher, and for the same reason: + a single pass means an inserted respelling can never be re-matched by + another rule, and the longest term wins over one contained inside it. + """ + usable = sorted({t for t in terms if t and t.strip()}, key=len, reverse=True) + if not usable: + return None + alternation = "|".join(re.escape(t) for t in usable) + return re.compile(rf"(? list[tuple[int, int]]: + """Spans of *markup* that are outside any tag and outside any tagged span. + + A term already wrapped by hand must be left alone: the author has said what + they want, and re-wrapping it would nest a rule inside their decision. This + also keeps the annotator from rewriting the inside of an attribute value. + """ + regions: list[tuple[int, int]] = [] + cursor = 0 + depth = 0 + for match in _TAG_RE.finditer(markup): + if depth == 0 and match.start() > cursor: + regions.append((cursor, match.start())) + name = match.group("name").lower() + closing = bool(match.group("closing")) + self_closing = bool(match.group("void")) + if name != "break" and not self_closing: + depth += -1 if closing else 1 + depth = max(depth, 0) + cursor = match.end() + if depth == 0 and cursor < len(markup): + regions.append((cursor, len(markup))) + return regions + + +def annotate( + markup: str, + rules: list[TermRule], + *, + supports_phonemes: bool = False, +) -> tuple[str, list[str]]: + """Wrap dictionary terms in *markup* with the directives they imply. + + Returns the annotated markup and the terms that were matched, so a caller + can report what fired rather than silently changing what gets spoken. + """ + if not markup or not rules: + return markup, [] + + by_term = {r.term.lower(): r for r in rules} + pattern = _build_pattern([r.term for r in rules]) + if pattern is None: + return markup, [] + + applied: list[str] = [] + out: list[str] = [] + cursor = 0 + + for start, end in _plain_regions(markup): + out.append(markup[cursor:start]) + segment = markup[start:end] + + def substitute(m: re.Match) -> str: + rule = by_term.get(m.group(0).lower()) + if rule is None: + return m.group(0) + realised = rule.realise(m.group(0), supports_phonemes=supports_phonemes) + if realised == m.group(0): + return m.group(0) + applied.append(m.group(0)) + return realised + + out.append(pattern.sub(substitute, segment)) + cursor = end + + out.append(markup[cursor:]) + return "".join(out), applied + + +def rules_from_entries(entries) -> list[TermRule]: + """Adapt ORM rows to :class:`TermRule`.""" + return [ + TermRule( + term=e.term, + replacement=e.replacement, + strategy=getattr(e, "strategy", None) or RESPELL, + spoken_language=getattr(e, "spoken_language", None), + phonemes=getattr(e, "phonemes", None), + ) + for e in entries + ] diff --git a/backend/services/prosody/compiler.py b/backend/services/prosody/compiler.py new file mode 100644 index 000000000..34bbf245c --- /dev/null +++ b/backend/services/prosody/compiler.py @@ -0,0 +1,253 @@ +"""Flatten a parsed script into a RenderPlan for one specific engine. + +This is where the harness earns its name. Each directive has up to two +realisations: + +*native* + tell the engine — ``instruct=`` on the engines that honour it + +*structural* + cut the text, render the run with its own settings, reassemble — which + needs no engine support whatsoever + +Pauses, language spans and rate are all structural, so they work identically on +all eight engines. Delivery cues are the only ones that depend on capability, +and where an engine cannot take them the plan says so out loud instead of +dropping them silently. + +Nothing here loads a model or touches audio. A plan can be built, asserted on +and shown to a user for free, which is what makes the pipeline previewable. +""" + +from __future__ import annotations + +from dataclasses import replace + +from .ir import Attrs, Break, Node, PlanWarning, RenderPlan, Silence, Span, Speech, Text +from .parser import parse + + +def _is_over_articulated(alias: str) -> bool: + """Whether a respelling is shaped like the one that tested worst. + + ``ban-DEH-ha`` -- syllable hyphens plus capitals for stress -- was judged + exaggerated on every voice tried, and measurably so: it ran about 30% + longer than the same sentence unmarked. ``ban-deh-ha`` was acceptable and + plain ``bandeha`` sounded most natural, so it is the *combination* that + misfires, not either alone. + + Acronym expansions like ``W C A G`` are capitals without hyphens and are + deliberately not flagged. + """ + return "-" in alias and any(c.isupper() for c in alias[1:]) + + +def _flatten(nodes: list[Node], inherited: Attrs, out: list[tuple[str, Attrs] | int]) -> None: + """Walk the tree into a flat run of (text, attrs) pairs and break markers.""" + for node in nodes: + if isinstance(node, Text): + if node.value: + out.append((node.value, inherited)) + elif isinstance(node, Break): + out.append(node.ms) + elif isinstance(node, Span): + _flatten(node.children, inherited.merged_with(node.attrs), out) + + +def _resolve(items: list[tuple[str, Attrs] | int]) -> list[tuple[str, str, Attrs] | int]: + """Apply substitutions, producing (spoken, written, attrs). + + Once ````/```` has been materialised into the spoken text it + stops being a setting, so it must not keep the run apart from its + neighbours afterwards. ``spoken_as`` is cleared here for exactly that + reason -- it governs inheritance, not rendering. + """ + out: list[tuple[str, str, Attrs] | int] = [] + for item in items: + if isinstance(item, int): + out.append(item) + continue + written, attrs = item + spoken = written if attrs.spoken_as is None else attrs.spoken_as + out.append((spoken, written, replace(attrs, spoken_as=None))) + return out + + +def _coalesce(items: list[tuple[str, str, Attrs] | int]) -> list[tuple[str, str, Attrs] | int]: + """Merge neighbouring runs that render identically. + + Every avoided cut is one less place the model restarts its prosody, which + is the entire cost of this approach. It matters most for ````: a + respelling changes the characters, not the settings, so the sentence around + it should stay in one piece. Cutting there would buy the seams that + respelling exists to avoid. + """ + merged: list[tuple[str, str, Attrs] | int] = [] + for item in items: + if ( + isinstance(item, tuple) + and merged + and isinstance(merged[-1], tuple) + and merged[-1][2] == item[2] + ): + prev = merged[-1] + merged[-1] = (prev[0] + item[0], prev[1] + item[1], prev[2]) + else: + merged.append(item) + return merged + + +def compile_plan( + markup: str, + *, + engine: str, + default_language: str, + supports_instruct: bool = False, + engine_languages: list[str] | None = None, + base_instruct: str | None = None, +) -> RenderPlan: + """Compile *markup* into a plan for *engine*. + + Args: + markup: The script, with or without directives. + engine: Target engine id, recorded on the plan. + default_language: Language for text outside any ````. + supports_instruct: From the model registry, not guessed here. + engine_languages: What the engine can generate. A ```` outside + this set is a warning rather than an error -- the run still + renders, in the engine's own language, which is what it would have + done anyway. + base_instruct: The request's own delivery instruction, which + ```` composes with rather than replaces. + """ + nodes = parse(markup) + flat = _coalesce(_resolve(_walk(nodes))) + + plan_nodes: list[Speech | Silence] = [] + warnings: list[PlanWarning] = [] + seen_unsupported_language: set[str] = set() + emphasis_dropped = False + + for item in flat: + if isinstance(item, int): + if item > 0: + plan_nodes.append(Silence(item)) + continue + + text, raw_text, attrs = item + if not text.strip(): + # Whitespace between tags is not a run of its own, but it must not + # be lost either -- glue it onto the previous run. + if plan_nodes and isinstance(plan_nodes[-1], Speech): + prev = plan_nodes[-1] + plan_nodes[-1] = Speech( + text=prev.text + raw_text, + language=prev.language, + rate=prev.rate, + instruct=prev.instruct, + seed=prev.seed, + source_text=prev.source_text, + ) + continue + + language = attrs.language or default_language + if ( + engine_languages + and language not in engine_languages + and language not in seen_unsupported_language + ): + seen_unsupported_language.add(language) + warnings.append( + PlanWarning( + code="language_unsupported", + detail=( + f"Engine {engine!r} cannot generate {language!r}; that run will be " + f"read as {default_language!r}." + ), + ) + ) + language = default_language + + instruct = base_instruct + if attrs.emphasis: + if supports_instruct: + cue = f"Say this with {attrs.emphasis} emphasis." + instruct = f"{base_instruct} {cue}".strip() if base_instruct else cue + elif not emphasis_dropped: + emphasis_dropped = True + warnings.append( + PlanWarning( + code="emphasis_unsupported", + detail=( + f"Engine {engine!r} does not honour delivery instructions, so " + f" has no effect. Try qwen_custom_voice." + ), + ) + ) + + if text != raw_text and _is_over_articulated(text): + warnings.append( + PlanWarning( + code="over_articulated_respelling", + detail=( + f"{text!r} combines hyphens with capitals, which makes the engine " + f"over-articulate: in testing that read as exaggerated and ran ~30% " + f"longer than the same sentence. A plain letter substitution " + f"(bandeja -> bandeha) sounded most natural." + ), + ) + ) + + plan_nodes.append( + Speech( + text=text, + language=language, + rate=attrs.rate or 1.0, + instruct=instruct, + source_text=raw_text if text != raw_text else None, + ) + ) + + return RenderPlan(nodes=_absorb_unspeakable(plan_nodes), warnings=warnings, engine=engine) + + +def _has_speech(text: str) -> bool: + """Whether a run contains anything a model could pronounce.""" + return any(ch.isalnum() for ch in text) + + +def _absorb_unspeakable(nodes: list[Speech | Silence]) -> list[Speech | Silence]: + """Fold runs with no pronounceable content into a neighbour. + + A span boundary almost always orphans its trailing punctuation -- + ``.`` leaves a run holding just ``"."``. Generating that is a wasted + call that returns noise or silence, so it is appended to the run before it + (or prepended to the one after, when it comes first). + + Substituted runs are never merged: a ```` applies to specific words, + and absorbing text into it would put words through a substitution the + author never wrapped. + """ + out: list[Speech | Silence] = [] + for node in nodes: + if isinstance(node, Speech) and not _has_speech(node.text): + prev = out[-1] if out else None + if isinstance(prev, Speech) and prev.source_text is None: + out[-1] = replace(prev, text=prev.text + node.text) + continue + out.append(node) + + # A leading orphan has no predecessor; give it to the following run. + if len(out) > 1 and isinstance(out[0], Speech) and not _has_speech(out[0].text): + first, second = out[0], out[1] + if isinstance(second, Speech) and second.source_text is None: + out = [replace(second, text=first.text + second.text), *out[2:]] + + # Everything may have been punctuation; keep it rather than return nothing. + return out or nodes + + +def _walk(nodes: list[Node]) -> list[tuple[str, Attrs] | int]: + out: list[tuple[str, Attrs] | int] = [] + _flatten(nodes, Attrs(), out) + return out diff --git a/backend/services/prosody/ir.py b/backend/services/prosody/ir.py new file mode 100644 index 000000000..41af5c2dc --- /dev/null +++ b/backend/services/prosody/ir.py @@ -0,0 +1,184 @@ +"""Intermediate representation for the prosody transformer. + +The transformer is a harness around segment production: it never synthesises +anything, it decides where to cut, what settings each cut carries, and how the +pieces are reassembled. This module is the vocabulary those decisions are +expressed in. + +Two layers: + +``Directive``/``Span`` + The parse tree. Mirrors the markup, so spans nest and attributes inherit. + +``RenderPlan`` + The flattened, resolved result: a list of speech runs and silences with + every attribute already decided for one specific engine. Nothing after this + point has to know the markup existed. + +The plan is deliberately a plain data object with no model behind it. It can be +built, asserted on, diffed and shown to a user without loading 3.5 GB of +weights, which is what makes the whole pipeline testable and previewable. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field, replace + +# ── Parse tree ─────────────────────────────────────────────────────── + + +@dataclass(frozen=True) +class Attrs: + """Settings that flow down the span tree. + + ``None`` means "inherit from the enclosing span", which is what lets + ```` wrap a ```` without either having to know + about the other. + """ + + language: str | None = None + rate: float | None = None + instruct: str | None = None + emphasis: str | None = None + # Set by and : the text handed to the + # engine differs from the text the author wrote. + spoken_as: str | None = None + + def merged_with(self, child: Attrs) -> Attrs: + """Child wins where it says anything; parent shows through elsewhere.""" + return Attrs( + language=child.language if child.language is not None else self.language, + rate=child.rate if child.rate is not None else self.rate, + instruct=child.instruct if child.instruct is not None else self.instruct, + emphasis=child.emphasis if child.emphasis is not None else self.emphasis, + # Deliberately not inherited: a substitution applies to the text it + # wraps, not to everything nested inside a parent that had one. + spoken_as=child.spoken_as, + ) + + +@dataclass(frozen=True) +class Text: + """Literal text to be spoken.""" + + value: str + + +@dataclass(frozen=True) +class Break: + """A silence, in milliseconds. + + No engine accepts this, so it is always realised structurally -- the + renderer emits silence and the model never sees it. That is why pauses work + identically on all eight engines. + """ + + ms: int + + +@dataclass(frozen=True) +class Span: + """A run of nodes sharing a set of attribute overrides.""" + + attrs: Attrs + children: list[Node] = field(default_factory=list) + # The tag this came from, kept for error messages and round-tripping. + tag: str = "" + + +Node = Text | Break | Span + + +# ── Render plan ────────────────────────────────────────────────────── + + +@dataclass(frozen=True) +class Speech: + """One generation call: text plus every setting already resolved.""" + + text: str + language: str + rate: float = 1.0 + instruct: str | None = None + seed: int | None = None + # What the author wrote, when `text` was substituted by / or + # a dictionary entry. Kept so a preview can show the change rather than + # silently handing back something the author never typed. + source_text: str | None = None + + +@dataclass(frozen=True) +class Silence: + """A gap, produced by assembly rather than by the engine.""" + + ms: int + + +PlanNode = Speech | Silence + + +@dataclass(frozen=True) +class PlanWarning: + """Something the target engine cannot honour. + + Carried on the plan rather than logged and forgotten: an instruction that + silently does nothing reads as the model refusing to follow it, which is + the complaint behind #579. + """ + + code: str + detail: str + + +@dataclass(frozen=True) +class RenderPlan: + """Everything the renderer needs, for one engine, with nothing left to decide.""" + + nodes: list[PlanNode] = field(default_factory=list) + warnings: list[PlanWarning] = field(default_factory=list) + engine: str = "" + + @property + def speech_nodes(self) -> list[Speech]: + return [n for n in self.nodes if isinstance(n, Speech)] + + @property + def is_trivial(self) -> bool: + """Whether this needs none of the harness: one run, no assembly. + + The overwhelmingly common case, and the caller takes the existing + single-shot path for it, so unmarked text costs nothing. + + A substitution does *not* make a plan non-trivial. ``source_text`` is + provenance for display -- by this point the respelling is already in + ``text``, and one run with no silences and no rate change has nothing + for the renderer to assemble. Treating it as non-trivial would send + every respelled sentence through the renderer for no benefit, and would + contradict the property that makes respelling the preferred strategy: + that it does not cut the sentence. + """ + return ( + len(self.nodes) == 1 + and isinstance(self.nodes[0], Speech) + and self.nodes[0].rate == 1.0 + ) + + def with_seeds(self, base_seed: int | None) -> RenderPlan: + """Assign a deterministic per-run seed. + + Varied per run so neighbouring runs do not share RNG artefacts, but + derived from ``base_seed`` so the same plan always renders the same + audio. ``None`` stays ``None`` -- an unseeded plan should still vary + between takes. + """ + if base_seed is None: + return self + out: list[PlanNode] = [] + speech_index = 0 + for node in self.nodes: + if isinstance(node, Speech): + out.append(replace(node, seed=base_seed + speech_index)) + speech_index += 1 + else: + out.append(node) + return replace(self, nodes=out) diff --git a/backend/services/prosody/llm_annotate.py b/backend/services/prosody/llm_annotate.py new file mode 100644 index 000000000..41954d844 --- /dev/null +++ b/backend/services/prosody/llm_annotate.py @@ -0,0 +1,249 @@ +"""Draft prosody markup with the local LLM. + +The model is an *authoring aid*, never part of rendering. It reads a script and +returns the same script with directives inserted; from there the deterministic +pipeline runs exactly as it does for markup typed by hand. That is what keeps +"with an LLM" and "without an LLM" the same code path downstream, and it is why +generation stays reproducible -- a model in the render path would make the same +script produce different audio on every run. + +The invariant that makes this safe to accept +-------------------------------------------- +Strip the tags from the model's output and compare to the input. If a single +word moved, the model rewrote the script instead of annotating it, and the +result is thrown away. The model can fail to help; it cannot mangle. Without +that check, an LLM quietly rephrasing a line would be discovered only by +listening to the audio. + +Availability +------------ +The Qwen LLM ships with most installs but is not guaranteed present, and this +must never be the reason a feature stops working. A missing model is reported, +not downloaded on demand -- the caller falls back to the dictionary and +hand-written markup, which is the whole feature minus the typing assistance. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass + +from .parser import ProsodyParseError, has_markup, parse, strip_markup + +logger = logging.getLogger(__name__) + +DEFAULT_MODEL_SIZE = "1.7B" +# Low, because this is a structural edit rather than a creative one: the same +# script should get the same annotation. +TEMPERATURE = 0.2 + +SYSTEM_PROMPT = """You annotate a script for a text-to-speech engine. + +Insert tags. Never change, add, remove or reorder any word. + +Tags you may use: +word a word or phrase in another language + a pause + +Rules: +- Wrap foreign words in with the correct language code. +- Use only where the script clearly wants a beat. +- Output the annotated script and nothing else. No explanation, no quotes. +- If nothing needs annotating, output the script unchanged.""" + +_EXAMPLES: list[tuple[str, str]] = [ + ( + "He plays a bandeja, not a smash.", + 'He plays a bandeja, not a smash.', + ), + ( + "The tempo was allegro throughout.", + 'The tempo was allegro throughout.', + ), + ("Nothing unusual in this line.", "Nothing unusual in this line."), +] + + +@dataclass(frozen=True) +class AnnotationResult: + """What the annotator produced, and whether it was trustworthy. + + ``markup`` is always safe to use: on rejection it is the original text, so + a caller can use the result unconditionally and read ``rejected_reason`` + only to explain why nothing changed. + """ + + markup: str + accepted: bool + rejected_reason: str | None = None + model_size: str | None = None + attempts: int = 0 + + @property + def changed(self) -> bool: + return self.accepted and has_markup(self.markup) + + +class LLMUnavailableError(RuntimeError): + """The local LLM is not downloaded, so annotation cannot run.""" + + +def is_llm_available(model_size: str = DEFAULT_MODEL_SIZE) -> bool: + """Whether the LLM can run without downloading anything first. + + Deliberately does not trigger a download: annotation is optional help, and + a feature that silently pulls gigabytes when first used is not optional. + """ + try: + from ..llm import get_llm_model + + backend = get_llm_model() + if backend.is_loaded(): + return True + return bool(backend._is_model_cached(model_size)) + except Exception: + logger.debug("LLM availability check failed", exc_info=True) + return False + + +def validate_annotation(original: str, candidate: str) -> str | None: + """Why *candidate* is not an acceptable annotation of *original*. + + Returns ``None`` when it is acceptable. Two things have to hold: it must + parse, and stripping it must reproduce the input word for word. + """ + if not candidate or not candidate.strip(): + return "the model returned nothing" + + try: + parse(candidate) + except ProsodyParseError as exc: + return f"the markup is malformed ({exc})" + + if strip_markup(candidate) != strip_markup(original): + return "the model changed the words instead of only annotating them" + + return None + + +async def annotate_with_llm( + text: str, + *, + language: str = "en", + model_size: str = DEFAULT_MODEL_SIZE, + max_attempts: int = 2, +) -> AnnotationResult: + """Ask the LLM to mark up *text*. + + Retries once on a rejected candidate, because the usual failure is a model + wrapping its answer in prose rather than misunderstanding the task, and a + second attempt with the complaint fed back usually lands. + + Raises: + LLMUnavailableError: if the model is not downloaded. + """ + if not text or not text.strip(): + return AnnotationResult(markup=text, accepted=True, model_size=model_size) + + if not is_llm_available(model_size): + raise LLMUnavailableError( + f"The {model_size} LLM is not downloaded. Annotation is optional -- " + "dictionary entries and hand-written markup work without it." + ) + + from ..llm import get_llm_model + + backend = get_llm_model() + prompt = f"Script language: {language}\n\n{text}" + last_reason = "the model produced no usable annotation" + + for attempt in range(1, max_attempts + 1): + try: + raw = await backend.generate( + prompt=prompt, + system=SYSTEM_PROMPT, + max_tokens=min(2048, len(text) * 2 + 256), + temperature=TEMPERATURE, + model_size=model_size, + examples=_EXAMPLES, + ) + except Exception: + logger.exception("LLM annotation call failed") + return AnnotationResult( + markup=text, + accepted=False, + rejected_reason="the LLM call failed", + model_size=model_size, + attempts=attempt, + ) + + reason = "the model returned nothing" + for candidate in _candidates(raw): + reason = validate_annotation(text, candidate) + if reason is None: + return AnnotationResult( + markup=candidate, + accepted=True, + model_size=model_size, + attempts=attempt, + ) + + last_reason = reason + logger.info("Rejected LLM annotation (attempt %d): %s", attempt, reason) + # Feed the complaint back rather than re-asking identically. + prompt = ( + f"Script language: {language}\n\n{text}\n\n" + f"Your previous answer was rejected: {reason}. " + "Return the script with tags inserted and every word unchanged." + ) + + return AnnotationResult( + markup=text, + accepted=False, + rejected_reason=last_reason, + model_size=model_size, + attempts=max_attempts, + ) + + +def _unfence(text: str) -> str: + """Drop a surrounding ``` block, with or without a language hint.""" + if not text.startswith("```"): + return text + lines = text.splitlines() + if len(lines) < 2: + return text + lines = lines[1:] + if lines and lines[-1].strip().startswith("```"): + lines = lines[:-1] + return "\n".join(lines).strip() + + +def _unquote(text: str) -> str: + """Drop one pair of wrapping quotes, which small models add habitually. + + No cleverness about whether that quote also appears inside: markup is full + of quoted attributes, so any such test would refuse the very case this + exists for. Whether the unwrapping was right is settled by validating the + result, not by guessing here. + """ + if len(text) >= 2 and text[0] == text[-1] and text[0] in {chr(34), chr(39)}: + return text[1:-1].strip() + return text + + +def _candidates(raw: str) -> list[str]: + """Plausible readings of a model answer, most conservative first. + + Small models wrap their output in fences, quotes, or both. Rather than + guess which, every unwrapping is offered and the invariant picks the first + that is a faithful annotation -- so a wrapper is only removed when doing so + produces something demonstrably correct, and no unwrapping can launder a + changed word into acceptance. + """ + text = (raw or "").strip() + out: list[str] = [] + for candidate in (text, _unfence(text), _unquote(text), _unquote(_unfence(text))): + if candidate and candidate not in out: + out.append(candidate) + return out diff --git a/backend/services/prosody/parser.py b/backend/services/prosody/parser.py new file mode 100644 index 000000000..739a349bd --- /dev/null +++ b/backend/services/prosody/parser.py @@ -0,0 +1,223 @@ +"""Parse the SSML subset into the prosody IR. + +Not an XML parser, deliberately. A script is prose: it contains ``&``, it +contains ``5 < 6``, and an XML parser rejects both. Instead this recognises a +*closed set* of known tags and treats everything else as literal text, which +makes it strictly more forgiving than XML on the input it actually gets -- +nothing needs escaping unless it happens to spell one of our tags. + +The subset is chosen so that the two pronunciation strategies are SSML's own +(```` and ````) rather than something invented alongside them. + +Supported:: + + silence, also 0.7s; bare is 700ms + … render this run in another language + … speaking rate + … + bandeja + bandeja + +Angle brackets rather than square ones because square ones are taken: +``[laugh]`` is a Chatterbox Turbo paralinguistic tag, which is passed *to* the +engine, where a directive is intercepted *before* it. Same delimiter, opposite +behaviour -- see ``_PARA_TAG_RE`` in ``utils/chunked_tts.py``. +""" + +from __future__ import annotations + +import re + +from .ir import Attrs, Break, Node, Span, Text + +# Every tag the parser knows. Anything else stays literal text. +_VOID_TAGS = {"break"} +_SPAN_TAGS = {"lang", "prosody", "emphasis", "sub", "phoneme"} +_ALL_TAGS = _VOID_TAGS | _SPAN_TAGS + +_TAG_RE = re.compile( + r"<\s*(?P/)?\s*(?P" + "|".join(sorted(_ALL_TAGS)) + r")" + r"(?P[^<>]*?)(?P/)?\s*>", + re.IGNORECASE, +) + +_ATTR_RE = re.compile(r"""(?P[\w:.-]+)\s*=\s*(?P["'])(?P.*?)(?P=quote)""") + +# "700ms", "0.7s", "700" (bare numbers read as milliseconds). +_DURATION_RE = re.compile(r"^\s*(?P\d+(?:\.\d+)?)\s*(?Pms|s)?\s*$", re.IGNORECASE) + +MAX_BREAK_MS = 60_000 + +# What a bare means. 700ms was judged a good pause on every voice +# tried; 1500ms read as too long unless the script genuinely wants a beat to +# stop and think. The natural gap after a full stop is already 210-440ms +# depending on the voice, so this adds to a pause rather than creating one. +DEFAULT_BREAK_MS = 700 + + +class ProsodyParseError(ValueError): + """The markup is malformed in a way that would change what gets spoken.""" + + +def parse_duration(raw: str) -> int: + """``"700ms"``/``"0.7s"``/``"700"`` -> milliseconds.""" + m = _DURATION_RE.match(raw or "") + if not m: + raise ProsodyParseError(f"Not a duration: {raw!r}. Use e.g. \"700ms\" or \"0.7s\".") + value = float(m.group("value")) + ms = round(value * 1000) if (m.group("unit") or "ms").lower() == "s" else round(value) + if ms < 0 or ms > MAX_BREAK_MS: + raise ProsodyParseError(f"Break of {ms}ms is outside 0..{MAX_BREAK_MS}ms.") + return ms + + +# Attribute values are quoted, so anything writing one has to escape a quote +# inside it -- and this has to undo that, or the engine speaks the entity. +# Ampersand is unescaped last so "&quot;" survives as the literal +# """ rather than collapsing into a quote. +_ENTITIES = ((""", '"'), ("'", "'"), ("<", "<"), (">", ">"), ("&", "&")) + + +def _unescape(value: str) -> str: + for entity, char in _ENTITIES: + value = value.replace(entity, char) + return value + + +def _parse_attrs(raw: str) -> dict[str, str]: + return { + m.group("key").lower(): _unescape(m.group("value")) + for m in _ATTR_RE.finditer(raw or "") + } + + +def _rate_from(raw: str | None) -> float | None: + """``"0.9"`` or ``"90%"`` -> a multiplier. + + Rejects zero and negatives rather than clamping: a rate of 0 means audio of + infinite length, and silently substituting 1.0 would hide a typo behind + output that sounds fine. + """ + if raw is None: + return None + text = raw.strip() + try: + value = float(text[:-1]) / 100.0 if text.endswith("%") else float(text) + except ValueError as exc: + raise ProsodyParseError(f"Not a rate: {raw!r}. Use e.g. \"0.9\" or \"90%\".") from exc + if not 0.1 <= value <= 5.0: + raise ProsodyParseError(f"Rate {value} is outside 0.1..5.0.") + return value + + +def _attrs_for(tag: str, attrs: dict[str, str]) -> Attrs: + if tag == "lang": + code = attrs.get("xml:lang") or attrs.get("lang") + if not code: + raise ProsodyParseError(' needs xml:lang, e.g. .') + return Attrs(language=code.strip().lower()) + + if tag == "prosody": + rate = _rate_from(attrs.get("rate")) + if rate is None: + raise ProsodyParseError(' supports rate, e.g. .') + return Attrs(rate=rate) + + if tag == "emphasis": + return Attrs(emphasis=(attrs.get("level") or "moderate").strip().lower()) + + if tag == "sub": + alias = attrs.get("alias") + if not alias or not alias.strip(): + raise ProsodyParseError(' needs alias, e.g. .') + return Attrs(spoken_as=alias.strip()) + + if tag == "phoneme": + ph = attrs.get("ph") + if not ph or not ph.strip(): + raise ProsodyParseError(' needs ph, e.g. .') # noqa: RUF001 + # Carried as a substitution; whether the engine can take phonemes at + # all is a capability question the compiler answers, not the parser. + return Attrs(spoken_as=ph.strip()) + + return Attrs() + + +def parse(markup: str) -> list[Node]: + """Parse *markup* into a node tree. + + Raises: + ProsodyParseError: on a malformed or unbalanced tag. Malformed markup + is an error rather than being passed through as text, because + passing it through means the engine reads it aloud. + """ + if not markup: + return [] + + root = Span(attrs=Attrs(), children=[], tag="") + stack: list[Span] = [root] + cursor = 0 + + def emit_text(chunk: str) -> None: + if chunk: + stack[-1].children.append(Text(chunk)) + + for match in _TAG_RE.finditer(markup): + emit_text(markup[cursor : match.start()]) + cursor = match.end() + + name = match.group("name").lower() + closing = bool(match.group("closing")) + self_closing = bool(match.group("void")) + attrs = _parse_attrs(match.group("attrs")) + + if closing: + if len(stack) == 1 or stack[-1].tag != name: + expected = stack[-1].tag if len(stack) > 1 else "nothing" + raise ProsodyParseError( + f" does not match the open tag ({expected})." + ) + stack.pop() + continue + + if name in _VOID_TAGS or self_closing: + if name == "break": + raw_time = attrs.get("time") + ms = DEFAULT_BREAK_MS if raw_time is None else parse_duration(raw_time) + stack[-1].children.append(Break(ms)) + continue + + span = Span(attrs=_attrs_for(name, attrs), children=[], tag=name) + stack[-1].children.append(span) + stack.append(span) + + emit_text(markup[cursor:]) + + if len(stack) > 1: + raise ProsodyParseError(f"<{stack[-1].tag}> was never closed.") + + return root.children + + +_STRIP_RE = re.compile( + r"<\s*/?\s*(?:" + "|".join(sorted(_ALL_TAGS)) + r")(?:[^<>]*?)/?\s*>", re.IGNORECASE +) + + +def strip_markup(markup: str) -> str: + """Remove every known tag, leaving the words. + + This is what makes LLM annotation safe to accept: strip the model's output + and compare it to the input. If a single word moved, the model rewrote the + script instead of annotating it, and the result is rejected. The model can + fail to help, but it cannot mangle. + + Whitespace is normalised on both sides, since inserting a tag on its own + line is a formatting change rather than a content one. + """ + return re.sub(r"\s+", " ", _STRIP_RE.sub("", markup or "")).strip() + + +def has_markup(text: str) -> bool: + """Whether any known tag is present, so unmarked text can skip the harness.""" + return bool(text) and _TAG_RE.search(text) is not None diff --git a/backend/services/prosody/pipeline.py b/backend/services/prosody/pipeline.py new file mode 100644 index 000000000..d1f55d9f6 --- /dev/null +++ b/backend/services/prosody/pipeline.py @@ -0,0 +1,183 @@ +"""The one entry point that joins the transformer to generation. + +Both generation paths -- the persisted one and the streaming one -- need the +same thing: resolve the dictionary, compile a plan, and either render it or +step aside. Doing that in one place is what keeps the two paths from drifting, +and what makes "unmarked text behaves exactly as before" a property of a single +function rather than a claim repeated twice. + +The step-aside matters. The overwhelmingly common script has no markup and no +dictionary hits, compiles to one plain run, and must take the existing +single-shot path untouched -- same call, same arguments, same audio. The +transformer is only allowed to cost something when it is actually doing +something. +""" + +from __future__ import annotations + +import logging + +import numpy as np + +from .annotate import annotate, rules_from_entries +from .compiler import compile_plan +from .ir import RenderPlan, Speech +from .parser import ProsodyParseError +from .renderer import render + +logger = logging.getLogger(__name__) + + +def engine_capabilities(engine: str) -> tuple[bool, list[str] | None]: + """What *engine* can honour, read from the model registry. + + Derived here rather than assumed, so a plan's warnings describe the engine + that will actually run. Computed from the configs directly because the + `engine_supports_instruct`/`engine_languages` helpers live on another + branch (#1023); once that lands this should call them instead of + re-deriving the same answer. + + An engine whose variants disagree reports no instruct support: a request + names an engine and the model size can change under it, so the + conservative answer is the only one true for every variant. + """ + try: + from ...backends import get_tts_model_configs + + configs = [c for c in get_tts_model_configs() if c.engine == engine] + if not configs: + return False, None + supports_instruct = all(c.supports_instruct for c in configs) + languages: list[str] = [] + for cfg in configs: + languages.extend(lang for lang in cfg.languages if lang not in languages) + return supports_instruct, languages or None + except Exception: + logger.debug("Engine capability lookup failed for %r", engine, exc_info=True) + return False, None + + +def build_plan( + text: str, + *, + engine: str, + language: str, + db=None, + profile_id: str | None = None, + supports_instruct: bool = False, + engine_languages: list[str] | None = None, + instruct: str | None = None, +) -> tuple[RenderPlan, str]: + """Resolve dictionary entries and compile *text* into a plan. + + Returns the plan and the markup it came from, so a caller can log or show + what the dictionary contributed. + + A malformed *hand-written* tag is an error the caller should surface, but + it must not be able to take down a generation for someone who never used + the feature -- so a parse failure falls back to treating the text as + literal, which is what it would have been before any of this existed. + """ + markup = text + if db is not None: + from ..pronunciation import get_entries + + entries = get_entries(db, language=language, profile_id=profile_id) + if entries: + markup, applied = annotate(text, rules_from_entries(entries)) + if applied: + logger.info("Dictionary annotated %d term(s)", len(applied)) + + try: + plan = compile_plan( + markup, + engine=engine, + default_language=language, + supports_instruct=supports_instruct, + engine_languages=engine_languages, + base_instruct=instruct, + ) + except ProsodyParseError as exc: + logger.warning("Prosody markup did not parse (%s); treating it as plain text", exc) + return ( + RenderPlan(nodes=[Speech(text=text, language=language, instruct=instruct)], + engine=engine), + text, + ) + + for warning in plan.warnings: + logger.info("Prosody: %s", warning.detail) + + return plan, markup + + +async def generate_with_prosody( + text: str, + *, + engine: str, + language: str, + generate_chunked_fn, + tts_model, + voice_prompt, + gen_kwargs: dict, + db=None, + profile_id: str | None = None, + supports_instruct: bool = False, + engine_languages: list[str] | None = None, + seed: int | None = None, + enabled: bool = True, +) -> tuple[np.ndarray, int]: + """Generate *text*, taking the transformer only when it has work to do. + + ``generate_chunked_fn`` is passed in rather than imported so this composes + with the existing chunking rather than competing with it: prosody splits by + directive, chunking splits by length, and a single directive run that is + still too long goes through both. + + Args: + gen_kwargs: The arguments the caller would have used for a plain + generation. Per-run values override language, seed and instruct; + everything else -- trim, runaway detection, chunk size -- carries + through unchanged. + enabled: False renders the text literally, for a script that genuinely + contains something shaped like a tag. + """ + if not enabled: + return await generate_chunked_fn(tts_model, text, voice_prompt, **gen_kwargs) + + plan, _markup = build_plan( + text, + engine=engine, + language=language, + db=db, + profile_id=profile_id, + supports_instruct=supports_instruct, + engine_languages=engine_languages, + instruct=gen_kwargs.get("instruct"), + ) + + if plan.is_trivial: + # Nothing to do. Same call the caller would have made, with the one + # difference that a dictionary respelling may have changed the text. + single = plan.nodes[0] + return await generate_chunked_fn(tts_model, single.text, voice_prompt, **gen_kwargs) + + plan = plan.with_seeds(seed) + + async def generate_run(node: Speech): + run_kwargs = dict(gen_kwargs) + run_kwargs["language"] = node.language + run_kwargs["seed"] = node.seed + run_kwargs["instruct"] = node.instruct + return await generate_chunked_fn(tts_model, node.text, voice_prompt, **run_kwargs) + + logger.info( + "Prosody: rendering %d run(s) across %d node(s)", + sum(1 for n in plan.nodes if isinstance(n, Speech)), + len(plan.nodes), + ) + return await render( + plan, + generate_run, + crossfade_ms=gen_kwargs.get("crossfade_ms", 50), + ) diff --git a/backend/services/prosody/renderer.py b/backend/services/prosody/renderer.py new file mode 100644 index 000000000..66e504e2c --- /dev/null +++ b/backend/services/prosody/renderer.py @@ -0,0 +1,304 @@ +"""Turn a RenderPlan into audio. + +Everything here is assembly. The engine is called once per speech run with that +run's own settings; silences, rate and joins are produced by arithmetic on the +resulting arrays, which is why pauses and per-span language work on all eight +engines including the ones that accept no directives at all. + +Two things this must get right, both measured rather than assumed: + +*trim the interior joins only* + Every generation carries its own leading and trailing silence (250-360ms + lead, 100-400ms trail on the voices measured). Where two runs meet, that is + two lots of dead air the author never asked for, and it compounds: a + three-run sentence runs ~0.7s longer than the same words in one shot. + Trimmed at the joins, the difference is ~0.03s. + + But the *outer* edges are the utterance's natural lead-in and release. + Trimming those makes every clip end abruptly after its last sound, which + reads as a forced delivery even on text carrying no markup at all. + +*do not crossfade into a pause* + A crossfade across a Silence eats the pause from both ends. Runs joined to + each other overlap; runs adjacent to silence are butted. +""" + +from __future__ import annotations + +import logging + +import librosa +import numpy as np + +from .ir import RenderPlan, Silence, Speech + +logger = logging.getLogger(__name__) + +DEFAULT_CROSSFADE_MS = 50 + +# Below this level a frame counts as silence for edge trimming. +_SILENCE_DB = -45.0 +_FRAME_MS = 10 +# Left on after trimming so a crossfade has something to work with and the +# speech does not start hard against the join. +_EDGE_CUSHION_MS = 30 + + +def edge_silence_ms(audio: np.ndarray, sr: int) -> tuple[float, float]: + """Leading and trailing near-silence, in milliseconds.""" + if audio.size == 0: + return 0.0, 0.0 + frame = max(1, int(sr * _FRAME_MS / 1000)) + usable = len(audio) - (len(audio) % frame) + if usable < frame: + return 0.0, 0.0 + frames = audio[:usable].reshape(-1, frame) + rms = np.sqrt(np.mean(frames.astype(np.float64) ** 2, axis=1)) + 1e-12 + loud = np.where(20 * np.log10(rms) > _SILENCE_DB)[0] + if len(loud) == 0: + return float(len(audio) / sr * 1000), 0.0 + return float(loud[0] * _FRAME_MS), float((len(frames) - 1 - loud[-1]) * _FRAME_MS) + + +def trim_edges( + audio: np.ndarray, + sr: int, + cushion_ms: int = _EDGE_CUSHION_MS, + *, + lead: bool = True, + trail: bool = True, +) -> np.ndarray: + """Strip the model's own silence from the chosen edges. + + Only *interior* edges should be trimmed. A generation's trailing silence is + the utterance's natural release -- roughly 290ms on the voices measured -- + and cutting it back to the cushion makes every clip end abruptly, which + reads as a forced delivery even on text with no markup at all. The leading + silence at the very start is the same story. + + What genuinely accumulates is the *join*: run N's trail butted against run + N+1's lead is two lots of dead air the author never asked for, and it + compounds with the number of runs. So the renderer trims where runs meet + and leaves the outer boundary of the whole utterance alone. + """ + if audio.size == 0: + return audio + lead_ms, trail_ms = edge_silence_ms(audio, sr) + cushion = int(sr * cushion_ms / 1000) + start = max(0, int(sr * lead_ms / 1000) - cushion) if lead else 0 + end = len(audio) - (max(0, int(sr * trail_ms / 1000) - cushion) if trail else 0) + return audio[start:end] if end > start else audio + + +# WSOLA windowing. 30ms frames are long enough to hold a pitch period at any +# adult speaking F0 and short enough that a splice lands inside one phoneme. +_WSOLA_FRAME_MS = 30 +_WSOLA_SEARCH_MS = 10 + + +def _wsola(audio: np.ndarray, rate: float, sr: int) -> np.ndarray: + """Time-stretch by waveform-similarity overlap-add. + + A phase vocoder reconstructs from magnitudes and re-estimated phase, which + on speech smears transients and leaves the characteristic "phasey" ring. + WSOLA never leaves the time domain: it overlap-adds real waveform segments, + choosing each splice point by cross-correlation so successive pitch periods + line up. Consonants stay crisp because nothing is resynthesised. + + Args: + rate: Playback rate. <1 lengthens (slower), >1 shortens. + """ + frame = max(2, int(sr * _WSOLA_FRAME_MS / 1000)) + search = max(1, int(sr * _WSOLA_SEARCH_MS / 1000)) + synthesis_hop = frame // 2 + analysis_hop = round(synthesis_hop * rate) + if analysis_hop < 1: + return audio + + window = np.hanning(frame).astype(np.float32) + out = np.zeros(int(len(audio) / rate) + frame, dtype=np.float32) + weights = np.zeros_like(out) + + read = 0 + write = 0 + # Kept from the previous frame: the samples the next one should continue + # from, which is what the search is trying to match. + expected = audio[:frame].astype(np.float32) + + while read + frame + search < len(audio) and write + frame < len(out): + # Search near the nominal read position for the segment that best + # continues what was just written. + lo = max(0, read - search) + hi = min(len(audio) - frame, read + search) + if hi <= lo: + offset = read + else: + candidates = np.arange(lo, hi + 1) + scores = [ + float(np.dot(audio[c : c + frame], expected)) for c in candidates + ] + offset = int(candidates[int(np.argmax(scores))]) + + segment = audio[offset : offset + frame].astype(np.float32) + out[write : write + frame] += segment * window + weights[write : write + frame] += window + + expected = audio[offset + synthesis_hop : offset + synthesis_hop + frame] + if len(expected) < frame: + break + # The nominal pointer advances by analysis_hop regardless of where the + # search landed. Folding the offset back in would let a run of + # forward-biased matches accelerate the read and cut the output short. + read += analysis_hop + write += synthesis_hop + + nonzero = weights > 1e-6 + out[nonzero] /= weights[nonzero] + return out[: write + frame] + + +def apply_rate(audio: np.ndarray, rate: float, sr: int = 24000) -> np.ndarray: + """Change tempo without changing pitch. + + WSOLA rather than a phase vocoder: the vocoder resynthesises from magnitude + and estimated phase, which on speech smears consonants and adds a phasey + ring -- audible enough to be rejected in listening. WSOLA overlap-adds real + waveform segments, so nothing is resynthesised. + + Resampling is not an option either; it would transpose the voice, which is + not what a rate directive means. + """ + if rate == 1.0 or audio.size == 0: + return audio + return _wsola(audio.astype(np.float32), rate, sr) + + +def _crossfade(a: np.ndarray, b: np.ndarray, samples: int) -> np.ndarray: + if samples <= 0 or a.size == 0 or b.size == 0: + return np.concatenate([a, b]) + overlap = min(samples, len(a), len(b)) + out = np.array(a, dtype=np.float32, copy=True) + fade_out = np.linspace(1.0, 0.0, overlap, dtype=np.float32) + fade_in = np.linspace(0.0, 1.0, overlap, dtype=np.float32) + out[-overlap:] = out[-overlap:] * fade_out + b[:overlap] * fade_in + return np.concatenate([out, b[overlap:]]) + + +def assemble( + pieces: list[tuple[np.ndarray, bool]], + sr: int, + crossfade_ms: int = DEFAULT_CROSSFADE_MS, +) -> np.ndarray: + """Join rendered pieces. + + ``pieces`` is ``(audio, is_silence)``. A crossfade is applied only between + two speech runs: overlapping a pause with its neighbours would shorten it + from both ends, so a 700ms break would not last 700ms. + """ + if not pieces: + return np.array([], dtype=np.float32) + + samples = int(sr * crossfade_ms / 1000) + out = np.asarray(pieces[0][0], dtype=np.float32) + prev_is_silence = pieces[0][1] + + for audio, is_silence in pieces[1:]: + audio = np.asarray(audio, dtype=np.float32) + if is_silence or prev_is_silence: + out = np.concatenate([out, audio]) + else: + out = _crossfade(out, audio, samples) + prev_is_silence = is_silence + + return out + + +async def render( + plan: RenderPlan, + generate_run, + *, + crossfade_ms: int = DEFAULT_CROSSFADE_MS, + trim_runs: bool = True, +) -> tuple[np.ndarray, int]: + """Render *plan* to audio. + + Args: + plan: A compiled plan. ``plan.is_trivial`` should be handled by the + caller on the existing single-shot path; this still renders it + correctly, just with no benefit. + generate_run: ``async (Speech) -> (audio, sample_rate)``. Injected + rather than imported so the renderer can be tested without a model + and so long runs can still go through ``generate_chunked``. + crossfade_ms: Overlap between adjacent speech runs. 0 for a butt join. + trim_runs: Trim silence at the joins between runs. The first run's + lead-in and the last run's release are always kept -- they are the + utterance's own boundary, not an artefact of cutting. + + Returns: + ``(audio, sample_rate)``. + """ + pieces: list[tuple[np.ndarray, bool]] = [] + sample_rate: int | None = None + pending_silence_ms = 0 + + # Which speech runs sit at the very start and very end of the utterance. + # Those outer edges keep the model's own lead-in and release; everything + # between them is an interior join and gets trimmed. + speech_positions = [i for i, n in enumerate(plan.nodes) if isinstance(n, Speech)] + first_speech = speech_positions[0] if speech_positions else None + last_speech = speech_positions[-1] if speech_positions else None + + for index, node in enumerate(plan.nodes): + if isinstance(node, Silence): + # Held until a sample rate is known -- a plan can legitimately open + # with a pause, before any run has told us the rate. + pending_silence_ms += node.ms + if sample_rate is not None: + pieces.append((_silence(pending_silence_ms, sample_rate), True)) + pending_silence_ms = 0 + continue + + if not isinstance(node, Speech): + continue + + audio, run_sr = await generate_run(node) + audio = np.asarray(audio, dtype=np.float32) + if audio.ndim > 1: + audio = audio.mean(axis=0) + + if sample_rate is None: + sample_rate = int(run_sr) + if pending_silence_ms: + pieces.append((_silence(pending_silence_ms, sample_rate), True)) + pending_silence_ms = 0 + elif int(run_sr) != sample_rate: + # Mixed rates would otherwise concatenate into a pitch shift. + logger.info( + "Resampling a prosody run from %dHz to %dHz", int(run_sr), sample_rate + ) + audio = librosa.resample(audio, orig_sr=int(run_sr), target_sr=sample_rate) + + if trim_runs: + audio = trim_edges( + audio, + sample_rate, + lead=index != first_speech, + trail=index != last_speech, + ) + audio = apply_rate(audio, node.rate, sample_rate) + + if audio.size: + pieces.append((audio, False)) + + if sample_rate is None: + # Silence-only plan; nothing established a rate. + return np.array([], dtype=np.float32), 0 + + if pending_silence_ms: + pieces.append((_silence(pending_silence_ms, sample_rate), True)) + + return assemble(pieces, sample_rate, crossfade_ms=crossfade_ms), sample_rate + + +def _silence(ms: int, sr: int) -> np.ndarray: + return np.zeros(max(0, int(sr * ms / 1000)), dtype=np.float32) diff --git a/backend/tests/test_pronunciation.py b/backend/tests/test_pronunciation.py new file mode 100644 index 000000000..44ce01ae3 --- /dev/null +++ b/backend/tests/test_pronunciation.py @@ -0,0 +1,577 @@ +""" +Tests for the pronunciation dictionary (#827). + +The rules that matter are the ones that are easy to get wrong: replacements +must not cascade into each other, capitalisation has to survive, scope has to +resolve in a defined order, and the rewrite must never reach the stored text. + +Usage: + python -m pytest backend/tests/test_pronunciation.py -v +""" + +import os +import sys +import tempfile +import uuid +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +_DATA_DIR = tempfile.mkdtemp(prefix="voicebox-pronunciation-test-") +os.environ["VOICEBOX_DATA_DIR"] = _DATA_DIR + +from starlette.testclient import TestClient # noqa: E402 + +from backend.app import app # noqa: E402 +from backend.database import PronunciationEntry, get_db # noqa: E402 +from backend.services.pronunciation import apply_pronunciations, build_pattern # noqa: E402 + + +@pytest.fixture(scope="module") +def client(): + with TestClient(app) as c: + yield c + + +@pytest.fixture +def db(client): + # Depends on `client` so the app lifespan has run: SessionLocal is None + # until init_db(), and a test that only asked for `db` would get it unset. + session = next(get_db()) + try: + yield session + finally: + session.query(PronunciationEntry).delete() + session.commit() + session.close() + + +@pytest.fixture +def profile(client): + """A throwaway voice profile. + + Unique name and explicit cleanup because ``VOICEBOX_DATA_DIR`` is not + honoured on this branch (see #981/#1004) — the suite writes to the repo's + ./data, so a fixed name collides with the previous run's leftovers. + """ + name = f"Pronunciation Test Voice {uuid.uuid4().hex[:8]}" + r = client.post("/profiles", json={"name": name, "language": "en"}) + assert r.status_code == 200, r.text + created = r.json() + yield created + client.delete(f"/profiles/{created['id']}") + + +def add(db, term, replacement, language=None, profile_id=None, enabled=True): + entry = PronunciationEntry( + term=term, + replacement=replacement, + language=language, + profile_id=profile_id, + enabled=enabled, + ) + db.add(entry) + db.commit() + db.refresh(entry) + return entry + + +# ── Matching ───────────────────────────────────────────────────────── + + +def test_replaces_a_term(db): + add(db, "bandeja", "ban-DEH-ha") + out, applied = apply_pronunciations("He plays a bandeja here.", "en", db) + assert out == "He plays a ban-DEH-ha here." + assert len(applied) == 1 + + +def test_matching_is_case_insensitive_but_capitalisation_survives(db): + """One lowercase entry has to cover the term at the start of a sentence.""" + add(db, "bandeja", "ban-DEH-ha") + out, _ = apply_pronunciations("Bandeja is the shot.", "en", db) + assert out == "Ban-DEH-ha is the shot." + + +def test_all_caps_stays_all_caps(db): + add(db, "wcag", "W C A G") + out, _ = apply_pronunciations("Follow WCAG rules.", "en", db) + assert out == "Follow W C A G rules." + + +def test_only_whole_words_match(db): + """Substring matching would turn 'brandeja' into nonsense.""" + add(db, "bandeja", "ban-DEH-ha") + out, applied = apply_pronunciations("A brandejapalooza appeared.", "en", db) + assert out == "A brandejapalooza appeared." + assert applied == [] + + +def test_replacements_do_not_cascade(db): + """The single-pass alternation exists for this: a loop of per-term + substitutions would rewrite the output of an earlier rule.""" + add(db, "bandeja", "ban-DEH-ha") + add(db, "ha", "HAH") + out, applied = apply_pronunciations("The bandeja.", "en", db) + assert out == "The ban-DEH-ha." + assert len(applied) == 1 + + +def test_longer_terms_win(db): + add(db, "bandeja", "ban-DEH-ha") + add(db, "bandeja alta", "ban-DEH-ha AL-ta") + out, _ = apply_pronunciations("A bandeja alta lands deep.", "en", db) + assert out == "A ban-DEH-ha AL-ta lands deep." + + +def test_paralinguistic_tags_are_left_alone(db): + """[laugh] is engine syntax, not speech.""" + add(db, "laugh", "LAFF") + out, _ = apply_pronunciations("[laugh] I did laugh.", "en", db) + assert out == "[laugh] I did LAFF." + + +def test_accented_terms_match(db): + add(db, "víbora", "VEE-bo-ra") + out, applied = apply_pronunciations("Then the víbora.", "en", db) + assert out == "Then the VEE-bo-ra." + assert len(applied) == 1 + + +def test_reports_what_it_changed(db): + """A silent rewrite of someone's script is worse than no rewrite.""" + entry = add(db, "bandeja", "ban-DEH-ha") + _, applied = apply_pronunciations("bandeja and bandeja", "en", db) + assert len(applied) == 2 + assert all(a["entry_id"] == entry.id for a in applied) + + +# ── Scope ──────────────────────────────────────────────────────────── + + +def test_language_scoped_entry_only_applies_to_that_language(db): + """The Spanish word needs respelling when the engine reads English, and + must be left alone when it is already reading Spanish.""" + add(db, "bandeja", "ban-DEH-ha", language="en") + assert apply_pronunciations("una bandeja", "es", db)[0] == "una bandeja" + assert apply_pronunciations("a bandeja", "en", db)[0] == "a ban-DEH-ha" + + +def test_wildcard_language_applies_everywhere(db): + add(db, "bandeja", "ban-DEH-ha") + assert "ban-DEH-ha" in apply_pronunciations("una bandeja", "es", db)[0] + + +def test_disabled_entries_are_skipped(db): + add(db, "bandeja", "ban-DEH-ha", enabled=False) + assert apply_pronunciations("a bandeja", "en", db)[0] == "a bandeja" + + +def test_profile_entry_beats_global(db, profile): + add(db, "bandeja", "GLOBAL") + add(db, "bandeja", "PER-VOICE", profile_id=profile["id"]) + + out, _ = apply_pronunciations("a bandeja", "en", db, profile_id=profile["id"]) + assert out == "a PER-VOICE" + + # A different voice still gets the global entry. + assert apply_pronunciations("a bandeja", "en", db)[0] == "a GLOBAL" + + +def test_other_profiles_entries_do_not_leak(db, profile): + add(db, "bandeja", "PER-VOICE", profile_id=profile["id"]) + assert apply_pronunciations("a bandeja", "en", db)[0] == "a bandeja" + + +# ── Degenerate input ───────────────────────────────────────────────── + + +def test_no_entries_is_a_no_op(db): + out, applied = apply_pronunciations("nothing to do", "en", db) + assert out == "nothing to do" + assert applied == [] + + +@pytest.mark.parametrize("text", ["", " "]) +def test_blank_text_is_returned_unchanged(db, text): + add(db, "bandeja", "ban-DEH-ha") + assert apply_pronunciations(text, "en", db) == (text, []) + + +def test_build_pattern_handles_no_usable_terms(): + assert build_pattern([]) is None + assert build_pattern(["", " "]) is None + + +def test_regex_metacharacters_in_a_term_are_literal(db): + """A term is text, not a pattern -- an unescaped '.' would match anything.""" + add(db, "C++", "C plus plus") + out, _ = apply_pronunciations("I write C++ daily.", "en", db) + assert out == "I write C plus plus daily." + + +# ── API ────────────────────────────────────────────────────────────── + + +def test_crud_roundtrip(client, db): + created = client.post( + "/pronunciations", json={"term": "bandeja", "replacement": "ban-DEH-ha", "language": "en"} + ) + assert created.status_code == 200, created.text + entry_id = created.json()["id"] + + listed = client.get("/pronunciations").json() + assert any(e["id"] == entry_id for e in listed) + + updated = client.put(f"/pronunciations/{entry_id}", json={"replacement": "ban-DAY-ha"}) + assert updated.status_code == 200 + assert updated.json()["replacement"] == "ban-DAY-ha" + assert updated.json()["term"] == "bandeja", "omitted fields must be left alone" + + assert client.delete(f"/pronunciations/{entry_id}").status_code == 200 + assert client.get("/pronunciations").json() == [] + + +def test_duplicate_in_the_same_scope_is_rejected(client, db): + body = {"term": "bandeja", "replacement": "ban-DEH-ha", "language": "en"} + assert client.post("/pronunciations", json=body).status_code == 200 + clash = client.post("/pronunciations", json=body) + assert clash.status_code == 409 + assert "already exists" in clash.json()["detail"] + + +def test_same_term_in_a_different_language_is_allowed(client, db): + assert client.post( + "/pronunciations", json={"term": "bandeja", "replacement": "A", "language": "en"} + ).status_code == 200 + assert client.post( + "/pronunciations", json={"term": "bandeja", "replacement": "B", "language": "it"} + ).status_code == 200 + + +def test_unknown_profile_is_rejected(client, db): + r = client.post( + "/pronunciations", + json={"term": "x", "replacement": "y", "profile_id": "no-such-profile"}, + ) + assert r.status_code == 404 + + +def test_preview_shows_the_rewrite(client, db): + client.post("/pronunciations", json={"term": "bandeja", "replacement": "ban-DEH-ha"}) + r = client.post( + "/pronunciations/preview", json={"text": "a bandeja", "language": "en"} + ) + assert r.status_code == 200, r.text + body = r.json() + assert body["original"] == "a bandeja" + assert body["result"] == "a ban-DEH-ha" + assert body["applied"][0]["term"] == "bandeja" + + +def test_missing_entry_404s(client, db): + assert client.put("/pronunciations/nope", json={"replacement": "x"}).status_code == 404 + assert client.delete("/pronunciations/nope").status_code == 404 + + +# ── The property the design rests on ───────────────────────────────── + + +@pytest.mark.asyncio +async def test_engine_gets_the_respelling_but_the_row_keeps_the_original( + client, db, profile, monkeypatch +): + """The property the whole design rests on. + + If the respelling were stored, History would show a reader ``ban-DEH-ha`` + and editing an entry could never change anything already generated. So the + engine must see the respelling and the database must not. + + The model itself is mocked -- this is about which string goes where, and + loading 3.5 GB of weights would not make the assertion any truer. + """ + import numpy as np + + from backend.services import history as history_service + from backend.services.generation import run_generation + + client.post("/pronunciations", json={"term": "bandeja", "replacement": "ban-DEH-ha"}) + + original = "He plays a bandeja." + generation = await history_service.create_generation( + profile_id=profile["id"], + text=original, + language="en", + audio_path="", + duration=0, + seed=None, + db=db, + status="generating", + engine="qwen", + ) + + seen = {} + + class FakeBackend: + def is_loaded(self): + return True + + async def fake_generate_chunked(_model, text, _voice_prompt, **_kwargs): + seen["text"] = text + return np.zeros(2400, dtype=np.float32), 24000 + + async def fake_load(*_a, **_k): + return None + + async def fake_voice_prompt(*_a, **_k): + return {} + + monkeypatch.setattr("backend.backends.get_tts_backend_for_engine", lambda _e: FakeBackend()) + monkeypatch.setattr("backend.backends.load_engine_model", fake_load) + monkeypatch.setattr("backend.utils.chunked_tts.generate_chunked", fake_generate_chunked) + monkeypatch.setattr( + "backend.services.profiles.create_voice_prompt_for_profile", fake_voice_prompt + ) + + await run_generation( + generation_id=generation.id, + profile_id=profile["id"], + text=original, + language="en", + engine="qwen", + model_size="1.7B", + seed=None, + mode="generate", + ) + + assert seen["text"] == "He plays a ban-DEH-ha.", "engine should receive the respelling" + + db.expire_all() + stored = client.get(f"/history/{generation.id}").json() + assert stored["text"] == original, "the stored row must keep the author's text" + assert "ban-DEH-ha" not in stored["text"] + + +def test_preview_is_the_only_way_to_see_the_rewrite(client, db): + """Since the rewrite is never stored, preview is what makes it inspectable + rather than a black box between the text and the audio.""" + client.post("/pronunciations", json={"term": "bandeja", "replacement": "ban-DEH-ha"}) + body = client.post( + "/pronunciations/preview", json={"text": "a bandeja", "language": "en"} + ).json() + assert body["result"] != body["original"] + assert body["applied"] + + +# ── Review findings (CodeRabbit on #1025) ──────────────────────────── + + +def test_sql_wildcards_in_a_term_are_literal(client, db): + """`ilike` read `%` and `_` in a term as wildcards, so `band_ja` collided + with `bandeja` and blocked a legitimate entry as a duplicate.""" + assert client.post( + "/pronunciations", json={"term": "bandeja", "replacement": "ban-DEH-ha"} + ).status_code == 200 + # Distinct term that ilike would have matched against the one above. + assert client.post( + "/pronunciations", json={"term": "band_ja", "replacement": "BAND-ja"} + ).status_code == 200 + assert client.post( + "/pronunciations", json={"term": "band%", "replacement": "BAND"} + ).status_code == 200 + + +def test_untrimmed_term_still_finds_its_duplicate(client, db): + """The stored value is trimmed, so an untrimmed lookup must normalise too or + it misses the duplicate it just created.""" + assert client.post( + "/pronunciations", json={"term": "bandeja", "replacement": "A"} + ).status_code == 200 + clash = client.post("/pronunciations", json={"term": " bandeja ", "replacement": "B"}) + assert clash.status_code == 409 + + +@pytest.mark.parametrize("field", ["term", "replacement"]) +def test_whitespace_only_values_are_rejected(client, db, field): + """`min_length=1` accepted " ", which then stored as empty — a no-op entry, + or a replacement that deletes the matched speech.""" + body = {"term": "bandeja", "replacement": "ban-DEH-ha"} + body[field] = " " + assert client.post("/pronunciations", json=body).status_code == 422 + + +def test_values_are_stored_trimmed(client, db): + created = client.post( + "/pronunciations", json={"term": " bandeja ", "replacement": " ban-DEH-ha "} + ).json() + assert created["term"] == "bandeja" + assert created["replacement"] == "ban-DEH-ha" + + +def test_preview_rejects_an_unknown_profile(client, db): + """Generation 404s on an unknown profile; preview silently fell back to + global scope, so it reported a different result than it would produce.""" + r = client.post( + "/pronunciations/preview", + json={"text": "a bandeja", "profile_id": "no-such-profile"}, + ) + assert r.status_code == 404 + + +def test_the_database_enforces_scope_uniqueness(client, db): + """find_duplicate is check-then-act; two concurrent creates can both pass + it. The constraint is what actually holds — including for the NULL scopes, + which a plain UNIQUE would treat as distinct.""" + from sqlalchemy.exc import IntegrityError + + add(db, "bandeja", "A") + # Differs only by case, and both are global scope — the pair a plain + # UNIQUE(term, language, profile_id) would have let through. + db.add(PronunciationEntry(term="Bandeja", replacement="B")) + with pytest.raises(IntegrityError): + db.commit() + db.rollback() + + +def test_terms_are_not_logged_at_info(client, db, caplog): + """Terms are user-supplied and often names, so the values belong at DEBUG.""" + import logging + + add(db, "Alicia Fernandez", "ah-LEE-see-ah") + with caplog.at_level(logging.INFO, logger="backend.services.pronunciation"): + apply_pronunciations("Ask Alicia Fernandez.", "en", db) + + info = [r for r in caplog.records if r.levelno == logging.INFO] + assert info, "should still report that a rewrite happened" + # Every fragment, not just the first: a message leaking only the surname + # would still be leaking a name. + for fragment in ("Alicia", "Fernandez", "ah-LEE-see-ah"): + assert all(fragment not in r.getMessage() for r in info), fragment + + +def test_dedup_migration_keeps_the_oldest_row(client, db): + """The migration collapses pre-index duplicates, and "oldest" has to mean + created_at -- ids are random UUIDs, so ordering by id keeps an arbitrary + row and would pick a different one on a different machine.""" + from datetime import datetime + + from sqlalchemy import text as sql_text + + # Via the module, not `from ... import engine`: init_db() rebinds the + # global, so a name imported earlier still points at the pre-init None. + from backend.database import session as db_session + + engine = db_session.engine + + with engine.connect() as conn: + conn.execute(sql_text("DROP INDEX IF EXISTS uq_pronunciation_scope")) + conn.commit() + + older = PronunciationEntry( + term="bandeja", replacement="KEEP", created_at=datetime(2020, 1, 1) + ) + newer = PronunciationEntry( + term="Bandeja", replacement="DROP", created_at=datetime(2030, 1, 1) + ) + # Insert newest first so insertion order can't be what makes it pass. + db.add(newer) + db.commit() + db.add(older) + db.commit() + + from backend.database.migrations import run_migrations + + run_migrations(engine) + + db.expire_all() + rows = db.query(PronunciationEntry).all() + assert len(rows) == 1 + assert rows[0].replacement == "KEEP" + + +# ── Strategies (prosody transformer, phase 3) ──────────────────────── + + +def test_a_language_strategy_entry_roundtrips(client, db): + r = client.post( + "/pronunciations", + json={ + "term": "víbora", + "replacement": "víbora", + "strategy": "language", + "spoken_language": "es", + }, + ) + assert r.status_code == 200, r.text + body = r.json() + assert body["strategy"] == "language" + assert body["spoken_language"] == "es" + + +def test_a_phoneme_strategy_entry_roundtrips(client, db): + r = client.post( + "/pronunciations", + json={ + "term": "chiquita", + "replacement": "chi-KEE-ta", + "strategy": "phoneme", + "phonemes": "tʃiˈkita", # noqa: RUF001 + }, + ) + assert r.status_code == 200, r.text + assert r.json()["phonemes"] == "tʃiˈkita" # noqa: RUF001 + + +def test_entries_default_to_respell(client, db): + """Every pre-existing entry is a respelling, so that has to be the default + or the migration would need a backfill.""" + r = client.post("/pronunciations", json={"term": "x", "replacement": "y"}) + assert r.json()["strategy"] == "respell" + + +def test_a_language_strategy_without_a_language_is_rejected(client, db): + """Storing it would show the user the strategy they picked while silently + falling back to the replacement.""" + r = client.post( + "/pronunciations", + json={"term": "x", "replacement": "y", "strategy": "language"}, + ) + assert r.status_code == 400 + assert "spoken_language" in r.json()["detail"] + + +def test_a_phoneme_strategy_without_phonemes_is_rejected(client, db): + r = client.post( + "/pronunciations", + json={"term": "x", "replacement": "y", "strategy": "phoneme"}, + ) + assert r.status_code == 400 + + +def test_an_unknown_strategy_is_rejected(client, db): + r = client.post( + "/pronunciations", json={"term": "x", "replacement": "y", "strategy": "vibes"} + ) + assert r.status_code == 422 + + +def test_switching_strategy_validates_against_the_stored_row(client, db): + """A strategy change can rely on a field set by an earlier request, so the + check has to consider the row as it will be, not just what was sent.""" + created = client.post( + "/pronunciations", + json={"term": "x", "replacement": "y", "spoken_language": "es"}, + ).json() + + # spoken_language is already stored, so switching strategy is legitimate. + r = client.put(f"/pronunciations/{created['id']}", json={"strategy": "language"}) + assert r.status_code == 200, r.text + assert r.json()["strategy"] == "language" + + +def test_switching_to_a_strategy_it_cannot_satisfy_is_rejected(client, db): + created = client.post("/pronunciations", json={"term": "x", "replacement": "y"}).json() + r = client.put(f"/pronunciations/{created['id']}", json={"strategy": "phoneme"}) + assert r.status_code == 400 diff --git a/backend/tests/test_prosody_annotate.py b/backend/tests/test_prosody_annotate.py new file mode 100644 index 000000000..7fdd90e18 --- /dev/null +++ b/backend/tests/test_prosody_annotate.py @@ -0,0 +1,202 @@ +""" +Tests for turning dictionary entries into markup. + +The annotator is the join between the dictionary and the transformer. Its whole +value is that it produces *the same directives an author would have typed*, so +the rules worth pinning are the ones that keep it honest: it must not touch +what the author already marked up, must not reach inside a tag, and must fall +back to something every engine can read when the preferred strategy is not +available. + +No database and no model — rules are plain values. + +Usage: + python -m pytest backend/tests/test_prosody_annotate.py -v +""" + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +from backend.services.prosody import Speech, TermRule, annotate, compile_plan + +BANDEJA = TermRule(term="bandeja", replacement="bandeha") +VIBORA = TermRule( + term="víbora", replacement="víbora", strategy="language", spoken_language="es" +) +CHIQUITA = TermRule( + term="chiquita", + replacement="chi-KEE-ta", + strategy="phoneme", + phonemes="tʃiˈkita", # noqa: RUF001 +) + + +# ── Realisation per strategy ───────────────────────────────────────── + + +def test_respell_becomes_a_sub(): + out, applied = annotate("He plays a bandeja.", [BANDEJA]) + assert out == 'He plays a bandeja.' + assert applied == ["bandeja"] + + +def test_language_becomes_a_lang_span(): + out, _ = annotate("Then the víbora.", [VIBORA]) + assert out == 'Then the víbora.' + + +def test_phoneme_is_used_where_the_engine_accepts_it(): + out, _ = annotate("A chiquita.", [CHIQUITA], supports_phonemes=True) + assert "Bandeja<' in out + + +def test_all_caps_is_carried(): + out, _ = annotate("A BANDEJA lands deep.", [BANDEJA]) + assert 'alias="BANDEHA"' in out + + +def test_only_whole_words_match(): + out, applied = annotate("A brandejapalooza.", [BANDEJA]) + assert out == "A brandejapalooza." + assert applied == [] + + +def test_longer_terms_win(): + alta = TermRule(term="bandeja alta", replacement="bandeha AL-ta") + out, _ = annotate("A bandeja alta.", [BANDEJA, alta]) + assert 'alias="bandeha AL-ta"' in out + + +def test_every_occurrence_is_annotated(): + out, applied = annotate("bandeja and bandeja", [BANDEJA]) + assert out.count("bandeja and another bandeja', [BANDEJA] + ) + assert out.count(" is void, so text after it is still ordinary prose.""" + out, applied = annotate('onea bandeja', [BANDEJA]) + assert applied == ["bandeja"] + assert "Then a smash.', + [BANDEJA], + ) + plan = compile_plan(out, engine="qwen", default_language="en", supports_instruct=True) + assert any(getattr(n, "ms", None) == 700 for n in plan.nodes) + assert any("bandeha" in getattr(n, "text", "") for n in plan.nodes) diff --git a/backend/tests/test_prosody_llm.py b/backend/tests/test_prosody_llm.py new file mode 100644 index 000000000..41f28e9ed --- /dev/null +++ b/backend/tests/test_prosody_llm.py @@ -0,0 +1,314 @@ +""" +Tests for LLM-drafted prosody markup and the plan preview. + +The model is stubbed throughout. What matters here is not what an LLM says but +what happens to what it says: a suggestion that changed the words must be +thrown away, malformed markup must be thrown away, and a missing model must +degrade to "no help offered" rather than to a broken feature. + +Usage: + python -m pytest backend/tests/test_prosody_llm.py -v +""" + +import os +import sys +import tempfile +import uuid +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +_DATA_DIR = tempfile.mkdtemp(prefix="voicebox-prosody-llm-test-") +os.environ["VOICEBOX_DATA_DIR"] = _DATA_DIR + +from starlette.testclient import TestClient # noqa: E402 + +from backend.app import app # noqa: E402 +from backend.database import PronunciationEntry, get_db # noqa: E402 +from backend.services.prosody import llm_annotate # noqa: E402 +from backend.services.prosody.llm_annotate import ( # noqa: E402 + LLMUnavailableError, + annotate_with_llm, + validate_annotation, +) + +ORIGINAL = "He plays a bandeja, not a smash." +GOOD = 'He plays a bandeja, not a smash.' + + +@pytest.fixture(scope="module") +def client(): + with TestClient(app) as c: + yield c + + +@pytest.fixture +def db(client): + session = next(get_db()) + try: + yield session + finally: + session.query(PronunciationEntry).delete() + session.commit() + session.close() + + +@pytest.fixture +def stub_llm(monkeypatch): + """Replace the model with a scripted sequence of replies.""" + + def install(*replies: str): + calls = {"prompts": []} + + class FakeBackend: + def is_loaded(self): + return True + + async def generate(self, prompt, **_kwargs): + calls["prompts"].append(prompt) + index = min(len(calls["prompts"]) - 1, len(replies) - 1) + return replies[index] + + monkeypatch.setattr(llm_annotate, "is_llm_available", lambda *_a, **_k: True) + monkeypatch.setattr( + "backend.services.llm.get_llm_model", lambda: FakeBackend() + ) + return calls + + return install + + +# ── The invariant ──────────────────────────────────────────────────── + + +def test_a_faithful_annotation_is_accepted(): + assert validate_annotation(ORIGINAL, GOOD) is None + + +def test_a_rewritten_script_is_rejected(): + """The whole reason the check exists: an LLM quietly rephrasing a line would + otherwise be discovered only by listening to the audio.""" + tampered = 'He plays a bandeja, obviously not a smash.' + assert "changed the words" in validate_annotation(ORIGINAL, tampered) + + +def test_a_dropped_word_is_rejected(): + assert validate_annotation(ORIGINAL, "He plays a bandeja.") is not None + + +def test_malformed_markup_is_rejected(): + """Passing it through would have the engine read the tag aloud.""" + assert "malformed" in validate_annotation(ORIGINAL, f'{ORIGINAL}') + + +@pytest.mark.parametrize("candidate", ["", " "]) +def test_an_empty_answer_is_rejected(candidate): + assert validate_annotation(ORIGINAL, candidate) is not None + + +def test_reflowed_whitespace_is_accepted(): + """Putting a tag on its own line is formatting, not content.""" + assert validate_annotation("a b", 'a\n\n b') is None + + +# ── Wrappers small models add ──────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_a_fenced_answer_is_unwrapped(stub_llm): + stub_llm(f"```xml\n{GOOD}\n```") + result = await annotate_with_llm(ORIGINAL) + assert result.accepted + assert result.markup == GOOD + + +@pytest.mark.asyncio +async def test_a_quoted_answer_is_unwrapped(stub_llm): + stub_llm(f'"{GOOD}"') + result = await annotate_with_llm(ORIGINAL) + assert result.accepted + assert result.markup == GOOD + + +@pytest.mark.asyncio +async def test_cleaning_never_repairs_a_word_change(stub_llm): + """Only decoration is stripped. Anything that would alter the words has to + reach the invariant rather than being quietly fixed up.""" + stub_llm("```\nHe plays a totally different sentence.\n```") + result = await annotate_with_llm(ORIGINAL) + assert not result.accepted + + +# ── Retry and rejection ────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_a_bad_first_answer_is_retried(stub_llm): + """The usual failure is prose around the answer rather than a + misunderstanding, so a second attempt with the complaint fed back lands.""" + calls = stub_llm("Sure! Here is your annotated script.", GOOD) + result = await annotate_with_llm(ORIGINAL) + + assert result.accepted + assert result.attempts == 2 + assert "rejected" in calls["prompts"][1], "the complaint should be fed back" + + +@pytest.mark.asyncio +async def test_persistent_failure_returns_the_original(stub_llm): + """`markup` is always safe to use, so a caller can apply it unconditionally + and read the reason only to explain why nothing changed.""" + stub_llm("nonsense", "still nonsense") + result = await annotate_with_llm(ORIGINAL) + + assert not result.accepted + assert result.markup == ORIGINAL + assert result.rejected_reason + assert not result.changed + + +@pytest.mark.asyncio +async def test_an_llm_error_is_not_fatal(stub_llm, monkeypatch): + class Exploding: + def is_loaded(self): + return True + + async def generate(self, *_a, **_k): + raise RuntimeError("boom") + + monkeypatch.setattr(llm_annotate, "is_llm_available", lambda *_a, **_k: True) + monkeypatch.setattr("backend.services.llm.get_llm_model", lambda: Exploding()) + + result = await annotate_with_llm(ORIGINAL) + assert not result.accepted + assert result.markup == ORIGINAL + + +@pytest.mark.asyncio +async def test_an_unchanged_script_is_accepted(stub_llm): + """Nothing to annotate is a valid answer, not a failure.""" + stub_llm(ORIGINAL) + result = await annotate_with_llm(ORIGINAL) + assert result.accepted + assert not result.changed + + +# ── Without an LLM ─────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_a_missing_model_raises_rather_than_downloading(monkeypatch): + """Annotation is optional help. A feature that silently pulls gigabytes the + first time it is used is not optional.""" + monkeypatch.setattr(llm_annotate, "is_llm_available", lambda *_a, **_k: False) + with pytest.raises(LLMUnavailableError): + await annotate_with_llm(ORIGINAL) + + +def test_availability_is_reported(client, monkeypatch): + monkeypatch.setattr(llm_annotate, "is_llm_available", lambda *_a, **_k: False) + body = client.get("/prosody/annotate/availability").json() + assert body["available"] is False + + +def test_the_endpoint_409s_without_a_model(client, monkeypatch): + monkeypatch.setattr(llm_annotate, "is_llm_available", lambda *_a, **_k: False) + r = client.post("/prosody/annotate", json={"text": ORIGINAL}) + assert r.status_code == 409 + assert "not downloaded" in r.json()["detail"] + + +@pytest.mark.asyncio +async def test_blank_text_needs_no_model(): + result = await annotate_with_llm(" ") + assert result.accepted + + +# ── Preview ────────────────────────────────────────────────────────── + + +def test_preview_shows_the_compiled_plan(client, db): + r = client.post( + "/prosody/preview", + json={ + "text": 'One.Dos tres cuatro.', + "engine": "qwen", + "language": "en", + }, + ) + assert r.status_code == 200, r.text + body = r.json() + + kinds = [n["kind"] for n in body["nodes"]] + assert kinds == ["speech", "silence", "speech"] + assert body["nodes"][1]["ms"] == 700 + assert body["nodes"][2]["language"] == "es" + assert body["run_count"] == 2 + assert body["is_trivial"] is False + + +def test_preview_resolves_dictionary_entries_into_markup(client, db): + """The dictionary emits the same directives an author would type, so the + preview can show them rather than only their effect.""" + client.post( + "/pronunciations", json={"term": "bandeja", "replacement": "bandeha"} + ) + body = client.post( + "/prosody/preview", json={"text": "He plays a bandeja.", "language": "en"} + ).json() + + assert 'alias="bandeha"' in body["markup"] + assert body["dictionary_terms"] == ["bandeja"] + assert body["nodes"][0]["source_text"] == "He plays a bandeja." + + +def test_preview_reports_what_the_engine_cannot_do(client, db): + """Base Qwen ignores delivery instructions, and saying so is the difference + between a limitation and a model that looks like it is refusing.""" + body = client.post( + "/prosody/preview", + json={"text": "wow there", "engine": "qwen"}, + ).json() + assert any(w["code"] == "emphasis_unsupported" for w in body["warnings"]) + + +def test_preview_marks_plain_text_as_trivial(client, db): + body = client.post( + "/prosody/preview", json={"text": "Just a plain sentence."} + ).json() + assert body["is_trivial"] is True + assert body["run_count"] == 1 + + +def test_preview_rejects_malformed_markup(client, db): + r = client.post( + "/prosody/preview", json={"text": 'unclosed'} + ) + assert r.status_code == 400 + + +def test_preview_scopes_the_dictionary_to_the_profile(client, db): + """A per-voice entry must not leak into a preview for another voice.""" + name = f"Prosody Preview Voice {uuid.uuid4().hex[:8]}" + profile = client.post("/profiles", json={"name": name, "language": "en"}).json() + try: + client.post( + "/pronunciations", + json={ + "term": "bandeja", + "replacement": "PER-VOICE", + "profile_id": profile["id"], + }, + ) + scoped = client.post( + "/prosody/preview", + json={"text": "a bandeja", "profile_id": profile["id"]}, + ).json() + assert "PER-VOICE" in scoped["markup"] + + unscoped = client.post("/prosody/preview", json={"text": "a bandeja"}).json() + assert "PER-VOICE" not in unscoped["markup"] + finally: + client.delete(f"/profiles/{profile['id']}") diff --git a/backend/tests/test_prosody_pipeline.py b/backend/tests/test_prosody_pipeline.py new file mode 100644 index 000000000..5d6e0381e --- /dev/null +++ b/backend/tests/test_prosody_pipeline.py @@ -0,0 +1,278 @@ +""" +Tests for the transformer wired into generation. + +This is the phase that changes what `/generate` does, so the property under +most scrutiny is the one about *not* changing it: a script with no markup and +no dictionary hits must take the same single-shot call it always did, with the +same arguments. + +The model is stubbed. What is being checked is which calls are made with what, +not what an engine does with them. + +Usage: + python -m pytest backend/tests/test_prosody_pipeline.py -v +""" + +import os +import sys +import tempfile +import uuid +from pathlib import Path + +import numpy as np +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +_DATA_DIR = tempfile.mkdtemp(prefix="voicebox-prosody-pipeline-test-") +os.environ["VOICEBOX_DATA_DIR"] = _DATA_DIR + +from starlette.testclient import TestClient # noqa: E402 + +from backend.app import app # noqa: E402 +from backend.database import PronunciationEntry, get_db # noqa: E402 +from backend.services.prosody.pipeline import build_plan, generate_with_prosody # noqa: E402 + +SR = 24000 + + +@pytest.fixture(scope="module") +def client(): + with TestClient(app) as c: + yield c + + +@pytest.fixture +def db(client): + session = next(get_db()) + try: + yield session + finally: + session.query(PronunciationEntry).delete() + session.commit() + session.close() + + +@pytest.fixture +def spy(): + """Stand in for generate_chunked, recording every call.""" + calls: list[dict] = [] + + async def generate_chunked_fn(_model, text, _voice_prompt, **kwargs): + calls.append({"text": text, **kwargs}) + return np.zeros(SR, dtype=np.float32), SR + + generate_chunked_fn.calls = calls + return generate_chunked_fn + + +def add(db, term, replacement, **kwargs): + entry = PronunciationEntry(term=term, replacement=replacement, **kwargs) + db.add(entry) + db.commit() + return entry + + +BASE = dict( + engine="qwen", + language="en", + tts_model=object(), + voice_prompt={}, +) + + +# ── The property that must not change ──────────────────────────────── + + +@pytest.mark.asyncio +async def test_plain_text_takes_the_single_shot_path(spy, db): + """No markup, no dictionary hits: one call, same text, same arguments.""" + kwargs = dict(language="en", seed=7, instruct=None, max_chunk_chars=800) + await generate_with_prosody( + "Just a plain sentence.", + generate_chunked_fn=spy, + gen_kwargs=dict(kwargs), + db=db, + **BASE, + ) + + assert len(spy.calls) == 1 + call = spy.calls[0] + assert call["text"] == "Just a plain sentence." + assert call["seed"] == 7 + assert call["max_chunk_chars"] == 800 + + +@pytest.mark.asyncio +async def test_prose_that_looks_like_markup_is_untouched(spy, db): + """`5 < 6` must not become a parse error for someone who never used this.""" + await generate_with_prosody( + "If 5 < 6 then x > y.", generate_chunked_fn=spy, gen_kwargs={}, db=db, **BASE + ) + assert len(spy.calls) == 1 + assert spy.calls[0]["text"] == "If 5 < 6 then x > y." + + +@pytest.mark.asyncio +async def test_malformed_markup_falls_back_to_literal_text(spy, db): + """A stray tag must not be able to fail a generation. Before this feature + existed the text was literal, so that is what it degrades to.""" + text = 'He said hola and left.' + await generate_with_prosody( + text, generate_chunked_fn=spy, gen_kwargs={}, db=db, **BASE + ) + assert len(spy.calls) == 1 + assert spy.calls[0]["text"] == text + + +@pytest.mark.asyncio +async def test_the_opt_out_speaks_the_text_literally(spy, db): + """For a script that genuinely contains something shaped like a tag.""" + text = 'Write to insert a pause.' + await generate_with_prosody( + text, generate_chunked_fn=spy, gen_kwargs={}, db=db, enabled=False, **BASE + ) + assert len(spy.calls) == 1 + assert spy.calls[0]["text"] == text + + +# ── When it does have work ─────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_a_language_span_becomes_separate_calls(spy, db): + await generate_with_prosody( + 'He plays a bandeja alta here.', + generate_chunked_fn=spy, + gen_kwargs=dict(language="en"), + db=db, + engine_languages=["en", "es"], + **{k: v for k, v in BASE.items() if k != "language"}, + language="en", + ) + + assert len(spy.calls) > 1 + languages = {c["language"] for c in spy.calls} + assert languages == {"en", "es"} + + +@pytest.mark.asyncio +async def test_a_break_costs_no_generation_call(spy, db): + """Silence is assembly. The engine never sees a pause.""" + await generate_with_prosody( + 'One.Two.', + generate_chunked_fn=spy, + gen_kwargs={}, + db=db, + **BASE, + ) + assert len(spy.calls) == 2 + assert all("break" not in c["text"] for c in spy.calls) + + +@pytest.mark.asyncio +async def test_chunking_arguments_survive_into_every_run(spy, db): + """Prosody splits by directive, chunking splits by length. A directive run + that is still long has to go through both.""" + await generate_with_prosody( + 'a uno dos tres b', + generate_chunked_fn=spy, + gen_kwargs=dict(language="en", max_chunk_chars=250, crossfade_ms=30), + db=db, + engine_languages=["en", "es"], + **BASE, + ) + assert all(c["max_chunk_chars"] == 250 for c in spy.calls) + + +@pytest.mark.asyncio +async def test_seeds_are_deterministic_across_runs(spy, db): + """Varied per run so neighbours do not share artefacts, but derived from + the request seed so the same script renders the same way.""" + markup = 'a uno dos tres b' + call = dict( + generate_chunked_fn=spy, gen_kwargs=dict(language="en"), db=db, + engine_languages=["en", "es"], seed=100, **BASE, + ) + await generate_with_prosody(markup, **call) + first = [c["seed"] for c in spy.calls] + + spy.calls.clear() + await generate_with_prosody(markup, **call) + assert [c["seed"] for c in spy.calls] == first + assert len(set(first)) == len(first), "each run should get its own seed" + + +# ── Dictionary integration ─────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_a_dictionary_respelling_reaches_the_engine(spy, db): + add(db, "bandeja", "bandeha") + await generate_with_prosody( + "He plays a bandeja.", generate_chunked_fn=spy, gen_kwargs={}, db=db, **BASE + ) + assert len(spy.calls) == 1, "a respelling must not cut the sentence" + assert spy.calls[0]["text"] == "He plays a bandeha." + + +@pytest.mark.asyncio +async def test_a_dictionary_language_entry_cuts_a_run(spy, db): + add(db, "víbora", "víbora", strategy="language", spoken_language="es") + await generate_with_prosody( + "Then the víbora lands.", + generate_chunked_fn=spy, + gen_kwargs=dict(language="en"), + db=db, + engine_languages=["en", "es"], + **BASE, + ) + assert any(c["language"] == "es" for c in spy.calls) + + +@pytest.mark.asyncio +async def test_no_dictionary_and_no_markup_needs_no_database_work(spy): + """A caller without a session must still work -- the story path builds + plans outside a request.""" + await generate_with_prosody( + "Plain text.", generate_chunked_fn=spy, gen_kwargs={}, db=None, **BASE + ) + assert spy.calls[0]["text"] == "Plain text." + + +# ── Plan building ──────────────────────────────────────────────────── + + +def test_build_plan_reports_the_markup_it_used(db): + add(db, "bandeja", "bandeha") + plan, markup = build_plan( + "He plays a bandeja.", engine="qwen", language="en", db=db + ) + assert 'alias="bandeha"' in markup + assert plan.is_trivial, "a respelling alone should not need the harness" + + +def test_build_plan_survives_malformed_markup(db): + plan, markup = build_plan( + 'unclosed', engine="qwen", language="en", db=db + ) + assert plan.is_trivial + assert markup == 'unclosed' + + +# ── End to end through the API ─────────────────────────────────────── + + +def test_generate_accepts_the_prosody_flag(client): + """Rejecting the field would break the opt-out before it is used.""" + name = f"Prosody Flag Voice {uuid.uuid4().hex[:8]}" + profile = client.post("/profiles", json={"name": name, "language": "en"}).json() + try: + r = client.post( + "/generate", + json={"profile_id": profile["id"], "text": "hello", "prosody": False}, + ) + assert r.status_code == 200, r.text + client.post(f"/generate/{r.json()['id']}/cancel") + finally: + client.delete(f"/profiles/{profile['id']}") diff --git a/backend/tests/test_prosody_renderer.py b/backend/tests/test_prosody_renderer.py new file mode 100644 index 000000000..159cfa0fc --- /dev/null +++ b/backend/tests/test_prosody_renderer.py @@ -0,0 +1,352 @@ +""" +Tests for the prosody renderer — plan to audio. + +Everything here is assembly, so the engine is a stub that returns tones. That +keeps the whole file model-free while still exercising the arithmetic that +actually matters: whether a pause lasts as long as it says, whether cutting +costs duration, and whether joins behave differently around silence. + +The two rules worth pinning are both measured facts rather than preferences: +trimming run edges is what makes segmentation duration-neutral, and crossfading +into a pause would eat it from both ends. + +Usage: + python -m pytest backend/tests/test_prosody_renderer.py -v +""" + +import sys +from pathlib import Path + +import numpy as np +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +from backend.services.prosody import Silence, Speech, compile_plan +from backend.services.prosody.ir import RenderPlan +from backend.services.prosody.renderer import ( + apply_rate, + assemble, + edge_silence_ms, + render, + trim_edges, +) + +SR = 24000 + + +def tone(seconds: float, freq: float = 440.0, amp: float = 0.3) -> np.ndarray: + t = np.linspace(0, seconds, int(SR * seconds), endpoint=False) + return (amp * np.sin(2 * np.pi * freq * t)).astype(np.float32) + + +def padded(speech_s: float, lead_ms: int = 340, trail_ms: int = 100) -> np.ndarray: + """A run shaped like a real generation: speech wrapped in the model's own + silence. The measured figures were 340ms lead and 100ms trail.""" + lead = np.zeros(int(SR * lead_ms / 1000), dtype=np.float32) + trail = np.zeros(int(SR * trail_ms / 1000), dtype=np.float32) + return np.concatenate([lead, tone(speech_s), trail]) + + +def fake_engine(seconds=1.0, sr=SR, lead_ms=340, trail_ms=100): + """A stub engine returning realistically padded audio, fixed length.""" + + async def generate_run(node: Speech): + return padded(seconds, lead_ms, trail_ms), sr + + return generate_run + + +def proportional_engine(secs_per_char=0.1, sr=SR, lead_ms=340, trail_ms=100): + """A stub whose speech length tracks the text. + + Needed for any comparison across different numbers of runs: splitting a + sentence distributes the same words, so total speech is constant and only + the padding multiplies. A fixed-length stub would add a whole extra run of + speech per cut and measure nothing. + """ + + async def generate_run(node: Speech): + return padded(len(node.text) * secs_per_char, lead_ms, trail_ms), sr + + return generate_run + + +def secs(audio: np.ndarray, sr: int = SR) -> float: + return len(audio) / sr + + +# ── Edge trimming: the finding the design rests on ─────────────────── + + +def test_edge_silence_is_measured(): + lead, trail = edge_silence_ms(padded(1.0, 340, 100), SR) + assert lead == pytest.approx(340, abs=20) + assert trail == pytest.approx(100, abs=20) + + +def test_trimming_removes_the_padding_but_keeps_a_cushion(): + trimmed = trim_edges(padded(1.0, 340, 100), SR) + assert secs(trimmed) < secs(padded(1.0, 340, 100)) + lead, _ = edge_silence_ms(trimmed, SR) + assert lead < 100, "most of the leading silence should be gone" + assert secs(trimmed) > 1.0, "the speech itself must survive" + + +def test_silence_only_audio_is_not_destroyed(): + quiet = np.zeros(SR, dtype=np.float32) + assert trim_edges(quiet, SR).size > 0 + + +def test_empty_audio_is_handled(): + assert trim_edges(np.array([], dtype=np.float32), SR).size == 0 + assert edge_silence_ms(np.array([], dtype=np.float32), SR) == (0.0, 0.0) + + +@pytest.mark.asyncio +async def test_cutting_is_duration_neutral_when_trimmed(): + """The measured result: three runs untrimmed ran ~0.7s longer than one; + trimmed, the difference collapses. That is what makes segmentation viable.""" + # Same nine characters either way, so only the number of cuts differs. + one_run = RenderPlan(nodes=[Speech("aaabbbccc", "en")]) + three_runs = RenderPlan( + nodes=[Speech("aaa", "en"), Speech("bbb", "es"), Speech("ccc", "en")] + ) + + engine = proportional_engine() + untrimmed_1, _ = await render(one_run, engine, trim_runs=False) + untrimmed_3, _ = await render(three_runs, engine, trim_runs=False) + trimmed_1, _ = await render(one_run, engine, trim_runs=True) + trimmed_3, _ = await render(three_runs, engine, trim_runs=True) + + added_untrimmed = secs(untrimmed_3) - secs(untrimmed_1) + added_trimmed = secs(trimmed_3) - secs(trimmed_1) + + assert added_untrimmed > 0.6, "two extra cuts should add real dead air" + assert added_trimmed < added_untrimmed / 2, ( + f"trimming should recover most of it: {added_untrimmed:.2f}s -> {added_trimmed:.2f}s" + ) + + +# ── Pauses ─────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_a_pause_lasts_as_long_as_it_says(): + plan = RenderPlan(nodes=[Speech("a", "en"), Silence(700), Speech("b", "en")]) + with_pause, sr = await render(plan, fake_engine(), crossfade_ms=0) + without, _ = await render( + RenderPlan(nodes=[Speech("a", "en"), Speech("b", "en")]), + fake_engine(), + crossfade_ms=0, + ) + assert secs(with_pause, sr) - secs(without, sr) == pytest.approx(0.7, abs=0.02) + + +@pytest.mark.asyncio +async def test_a_crossfade_does_not_eat_the_pause(): + """Overlapping a silence with its neighbours would shorten it from both + ends, so a 700ms break would not last 700ms.""" + plan = RenderPlan(nodes=[Speech("a", "en"), Silence(700), Speech("b", "en")]) + faded, sr = await render(plan, fake_engine(), crossfade_ms=50) + butted, _ = await render(plan, fake_engine(), crossfade_ms=0) + assert secs(faded, sr) == pytest.approx(secs(butted, sr), abs=0.005) + + +@pytest.mark.asyncio +async def test_adjacent_runs_do_overlap(): + """The crossfade must still apply where it is wanted, or joins click.""" + plan = RenderPlan(nodes=[Speech("a", "en"), Speech("b", "en")]) + faded, sr = await render(plan, fake_engine(), crossfade_ms=100) + butted, _ = await render(plan, fake_engine(), crossfade_ms=0) + assert secs(butted, sr) - secs(faded, sr) == pytest.approx(0.1, abs=0.01) + + +@pytest.mark.asyncio +async def test_a_leading_pause_survives(): + """A plan can open with a break, before any run has established the rate.""" + plan = RenderPlan(nodes=[Silence(500), Speech("a", "en")]) + audio, sr = await render(plan, fake_engine(), crossfade_ms=0) + lead, _ = edge_silence_ms(audio, sr) + assert lead >= 450 + + +@pytest.mark.asyncio +async def test_a_trailing_pause_survives(): + plan = RenderPlan(nodes=[Speech("a", "en"), Silence(500)]) + audio, sr = await render(plan, fake_engine(), crossfade_ms=0) + _, trail = edge_silence_ms(audio, sr) + assert trail >= 450 + + +@pytest.mark.asyncio +async def test_a_plan_of_only_silence_renders_nothing(): + """Nothing established a sample rate, so there is no meaningful output.""" + audio, sr = await render(RenderPlan(nodes=[Silence(500)]), fake_engine()) + assert audio.size == 0 + assert sr == 0 + + +# ── Rate ───────────────────────────────────────────────────────────── + + +def test_a_slower_rate_lengthens_without_resampling(): + original = tone(1.0) + slower = apply_rate(original, 0.5) + assert secs(slower) == pytest.approx(2.0, rel=0.05) + + +def test_rate_one_is_a_no_op(): + original = tone(1.0) + assert apply_rate(original, 1.0) is original + + +@pytest.mark.asyncio +async def test_rate_applies_per_run(): + plan = RenderPlan(nodes=[Speech("a", "en"), Speech("b", "en", rate=0.5)]) + audio, sr = await render(plan, fake_engine(seconds=1.0), crossfade_ms=0) + plain, _ = await render( + RenderPlan(nodes=[Speech("a", "en"), Speech("b", "en")]), + fake_engine(seconds=1.0), + crossfade_ms=0, + ) + assert secs(audio, sr) > secs(plain, sr) + + +# ── Mixed engine output ────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_runs_at_different_rates_are_resampled(): + """Concatenating mismatched rates would come out as a pitch shift.""" + calls = {"n": 0} + + async def varying(node: Speech): + calls["n"] += 1 + return (padded(1.0), SR) if calls["n"] == 1 else (padded(1.0), 48000) + + plan = RenderPlan(nodes=[Speech("a", "en"), Speech("b", "en")]) + audio, sr = await render(plan, varying, crossfade_ms=0) + assert sr == SR + # Both runs are ~1s of speech; a mishandled rate would halve or double one. + assert secs(audio, sr) == pytest.approx(2.0, abs=0.4) + + +@pytest.mark.asyncio +async def test_multichannel_output_is_folded_to_mono(): + async def stereo(node: Speech): + mono = padded(1.0) + return np.stack([mono, mono]), SR + + audio, _ = await render(RenderPlan(nodes=[Speech("a", "en")]), stereo) + assert audio.ndim == 1 + + +# ── Assembly primitives ────────────────────────────────────────────── + + +def test_assemble_of_nothing_is_empty(): + assert assemble([], SR).size == 0 + + +def test_assemble_of_one_piece_is_that_piece(): + piece = tone(0.5) + assert np.array_equal(assemble([(piece, False)], SR), piece) + + +# ── End to end from markup ─────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_a_compiled_script_renders(): + """The join the whole feature exists for: English, a Spanish span, a pause.""" + plan = compile_plan( + 'The shot is a bandeja, no un smash, here.' + 'Not a smash.', + engine="qwen", + default_language="en", + engine_languages=["en", "es"], + ) + languages = [n.language for n in plan.nodes if isinstance(n, Speech)] + assert "es" in languages + + audio, sr = await render(plan, fake_engine(seconds=0.5)) + assert sr == SR + assert secs(audio, sr) > 0.7, "should contain every run plus the pause" + + +# ── The outer boundary is not an interior join ─────────────────────── + + +@pytest.mark.asyncio +async def test_the_utterances_own_lead_and_release_survive(): + """Trimming the outer edges made every clip end ~30ms after the last sound, + which reads as a forced delivery even on text with no markup at all. Only + the joins between runs accumulate dead air worth removing.""" + plan = RenderPlan(nodes=[Speech("a", "en"), Speech("b", "en")]) + audio, sr = await render(plan, fake_engine(lead_ms=340, trail_ms=290), crossfade_ms=0) + + lead, trail = edge_silence_ms(audio, sr) + assert lead == pytest.approx(340, abs=40), "the model's lead-in is natural" + assert trail == pytest.approx(290, abs=40), "the release is natural" + + +@pytest.mark.asyncio +async def test_the_join_between_runs_is_still_trimmed(): + """The outer edges survive, but the interior must not double up.""" + two = RenderPlan(nodes=[Speech("aaa", "en"), Speech("bbb", "en")]) + one = RenderPlan(nodes=[Speech("aaabbb", "en")]) + engine = proportional_engine(lead_ms=340, trail_ms=290) + + joined, sr = await render(two, engine, crossfade_ms=0) + single, _ = await render(one, engine, crossfade_ms=0) + # Same words, same outer edges; only the interior join differs. + assert secs(joined, sr) - secs(single, sr) < 0.15 + + +@pytest.mark.asyncio +async def test_a_single_run_is_returned_untouched(): + """A trivial plan takes the single-shot path in production. Rendered here it + must still come back as the engine produced it, or the baseline sounds + processed when nothing was asked for.""" + engine = fake_engine(lead_ms=340, trail_ms=290) + raw, _ = await engine(Speech("a", "en")) + audio, sr = await render(RenderPlan(nodes=[Speech("a", "en")]), engine) + assert secs(audio, sr) == pytest.approx(secs(raw, sr), abs=0.01) + + +# ── Time stretching (WSOLA) ────────────────────────────────────────── + + +@pytest.mark.parametrize("rate", [0.5, 0.8, 0.9, 1.25, 2.0]) +def test_the_stretch_ratio_is_accurate(rate): + """An early version fed the search offset back into the read pointer, so a + run of forward-biased matches accelerated the read and produced 1.47s where + 2.0s was asked for. The nominal pointer must advance independently.""" + out = apply_rate(tone(1.0), rate, SR) + assert secs(out) == pytest.approx(1.0 / rate, rel=0.06) + + +def test_stretching_preserves_pitch(): + """The reason not to just resample: that would transpose the voice, which + is not what a rate directive means.""" + import numpy as np + + original = tone(1.0, freq=220.0) + slower = apply_rate(original, 0.5, SR) + + def dominant_hz(x): + spectrum = np.abs(np.fft.rfft(x)) + return np.fft.rfftfreq(len(x), 1 / SR)[int(np.argmax(spectrum))] + + assert dominant_hz(slower) == pytest.approx(dominant_hz(original), rel=0.05) + + +def test_stretching_does_not_resynthesise_silence_into_noise(): + """WSOLA overlap-adds real waveform, so silence stays silent -- a vocoder + can ring on the transition into one.""" + import numpy as np + + quiet = np.zeros(SR // 2, dtype=np.float32) + out = apply_rate(np.concatenate([tone(0.5), quiet]), 0.8, SR) + assert float(np.max(np.abs(out[-SR // 4 :]))) < 0.02 diff --git a/backend/tests/test_prosody_transformer.py b/backend/tests/test_prosody_transformer.py new file mode 100644 index 000000000..d8c908cd5 --- /dev/null +++ b/backend/tests/test_prosody_transformer.py @@ -0,0 +1,364 @@ +""" +Tests for the prosody transformer's parse and compile stages. + +No model, no audio, no database — the whole point of making the RenderPlan a +plain data object is that the decisions can be asserted on for free. + +The rules worth pinning are the ones that are easy to get subtly wrong: +attributes inherit but substitutions do not, neighbouring runs coalesce so the +model restarts its prosody as rarely as possible, punctuation orphaned by a +span boundary never becomes its own generation, and anything the target engine +cannot honour is said out loud rather than dropped. + +Usage: + python -m pytest backend/tests/test_prosody_transformer.py -v +""" + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +from backend.services.prosody import ( + ProsodyParseError, + RenderPlan, + Silence, + Speech, + compile_plan, + has_markup, + parse, + strip_markup, +) +from backend.services.prosody.parser import parse_duration + + +def plan(markup: str, **kwargs) -> RenderPlan: + kwargs.setdefault("engine", "qwen") + kwargs.setdefault("default_language", "en") + return compile_plan(markup, **kwargs) + + +def texts(p: RenderPlan) -> list[str]: + return [n.text for n in p.nodes if isinstance(n, Speech)] + + +def codes(p: RenderPlan) -> list[str]: + return [w.code for w in p.warnings] + + +# ── Prose is not XML ───────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "text", + ["5 < 6 and 7 > 6", "Tom & Jerry", "a b", "if x 50%"], +) +def test_prose_that_would_break_an_xml_parser_is_literal(text): + """The parser recognises a closed tag set and leaves everything else alone, + so ordinary prose needs no escaping.""" + p = plan(text) + assert texts(p) == [text] + + +def test_unknown_tags_are_spoken_not_stripped(): + """Silently dropping an unknown tag would change the script. It is not ours, + so it is text.""" + text = "Use the tag" + assert texts(plan(text)) == [text] + + +# ── Directives ─────────────────────────────────────────────────────── + + +def test_a_break_becomes_silence_not_a_generation(): + """No engine accepts a pause, so it is pure assembly — which is why pauses + work identically on all eight.""" + p = plan('One.Two.') + assert p.nodes[1] == Silence(700) + assert texts(p) == ["One.", "Two."] + + +@pytest.mark.parametrize( + ("raw", "expected"), [("700ms", 700), ("0.7s", 700), ("700", 700), ("1s", 1000), ("0ms", 0)] +) +def test_duration_forms(raw, expected): + assert parse_duration(raw) == expected + + +@pytest.mark.parametrize("raw", ["", "soon", "-5ms", "999s", "700m"]) +def test_bad_durations_are_rejected(raw): + with pytest.raises(ProsodyParseError): + parse_duration(raw) + + +def test_a_zero_break_emits_nothing(): + p = plan('One.Two.') + assert not any(isinstance(n, Silence) for n in p.nodes) + + +def test_a_language_span_gets_its_own_run(): + p = plan('a bandeja alta b') + langs = [(n.text.strip(), n.language) for n in p.nodes if isinstance(n, Speech)] + assert ("bandeja alta", "es") in langs + assert all(lang == "en" for text, lang in langs if text != "bandeja alta") + + +def test_rate_applies_to_its_span_only(): + p = plan('normal slow normal') + rates = {n.text.strip(): n.rate for n in p.nodes if isinstance(n, Speech)} + assert rates["slow"] == 0.8 + assert rates["normal"] == 1.0 + + +@pytest.mark.parametrize(("raw", "expected"), [("0.9", 0.9), ("90%", 0.9), ("1.5", 1.5)]) +def test_rate_forms(raw, expected): + p = plan(f'x') + assert p.nodes[0].rate == pytest.approx(expected) + + +@pytest.mark.parametrize("raw", ["0", "-1", "fast", "20"]) +def test_bad_rates_are_rejected(raw): + """Rejected rather than clamped: rate 0 is audio of infinite length, and + quietly substituting 1.0 hides a typo behind output that sounds fine.""" + with pytest.raises(ProsodyParseError): + plan(f'x') + + +def test_sub_replaces_what_the_engine_hears_but_records_the_original(): + p = plan('a bandeja b') + run = next(n for n in p.nodes if isinstance(n, Speech) and n.source_text) + assert run.text == "a ban-DEH-ha b", "the engine hears the respelling" + assert run.source_text == "a bandeja b", "the original is kept for preview" + + +def test_a_respelling_does_not_cut_the_sentence(): + """The whole point of : it changes characters, not settings, so the + sentence stays in one piece. Cutting there would buy exactly the seams that + respelling exists to avoid.""" + p = plan('The shot he plays is a bandeja, not a smash.') + assert len([n for n in p.nodes if isinstance(n, Speech)]) == 1 + + +def test_phoneme_is_carried_as_a_substitution(): + p = plan('a bandeja b') # noqa: RUF001 + run = next(n for n in p.nodes if isinstance(n, Speech) and n.source_text) + assert "banˈdexa" in run.text # noqa: RUF001 + assert run.source_text == "a bandeja b" + + +# ── Nesting and inheritance ────────────────────────────────────────── + + +def test_attributes_inherit_through_nesting(): + p = plan('slow lento') + lento = next(n for n in p.nodes if isinstance(n, Speech) and n.language == "es") + assert lento.rate == 0.8, "the inner span should keep the outer rate" + + +def test_the_inner_span_wins_on_conflict(): + p = plan('a b') + langs = {n.text.strip(): n.language for n in p.nodes if isinstance(n, Speech)} + assert langs["b"] == "it" + + +def test_a_substitution_does_not_leak_to_siblings(): + """ applies to the words it wraps. Inheriting it would put unrelated + text through a substitution the author never asked for -- the run merges + with its neighbours, but only the wrapped word is replaced.""" + p = plan('a b') + run = next(n for n in p.nodes if isinstance(n, Speech)) + assert run.text == "X b", "only the wrapped word is substituted" + assert run.source_text == "a b" + + +# ── Cutting as little as possible ──────────────────────────────────── + + +def test_identical_neighbours_coalesce(): + """Every avoided cut is one less place the model restarts its prosody.""" + p = plan("one two three") + assert len(texts(p)) == 1 + + +def test_orphaned_punctuation_never_becomes_its_own_run(): + """A span boundary orphans its trailing punctuation — . leaves a run + holding just ".". Generating that is a wasted call returning noise.""" + p = plan('a bandeja. b') + assert all(any(c.isalnum() for c in t) for t in texts(p)), texts(p) + + +def test_leading_punctuation_is_absorbed_forward(): + p = plan('. hello there') + assert all(any(c.isalnum() for c in t) for t in texts(p)), texts(p) + + +def test_all_punctuation_still_produces_something(): + """Absorbing must not be able to empty the plan.""" + p = plan("...") + assert p.nodes + + +def test_unmarked_text_is_a_single_trivial_run(): + """The common case must cost nothing — the renderer takes the existing + single-shot path for these.""" + p = plan("Just a plain sentence.") + assert p.is_trivial + + +def test_a_language_span_is_not_trivial(): + assert not plan('a b c d').is_trivial + + +# ── Engine capability ──────────────────────────────────────────────── + + +def test_emphasis_reaches_an_engine_that_honours_instruct(): + p = plan("wow", supports_instruct=True) + assert "strong" in (p.nodes[0].instruct or "") + assert not codes(p) + + +def test_emphasis_on_an_engine_that_ignores_instruct_warns(): + """Dropping it silently is what makes a model look like it is refusing to + follow instructions (#579).""" + p = plan("wow", supports_instruct=False) + assert "emphasis_unsupported" in codes(p) + assert p.nodes[0].instruct is None + + +def test_emphasis_composes_with_the_requests_own_instruct(): + p = plan( + "wow", supports_instruct=True, base_instruct="Speak warmly." + ) + assert "Speak warmly." in p.nodes[0].instruct + assert "emphasis" in p.nodes[0].instruct + + +def test_an_unsupported_language_falls_back_and_says_so(): + p = plan('a x y z', engine_languages=["en", "es"]) + assert "language_unsupported" in codes(p) + assert all(n.language == "en" for n in p.nodes if isinstance(n, Speech)) + + +def test_each_unsupported_language_warns_once(): + p = plan( + 'a x y z b p q r', + engine_languages=["en"], + ) + assert codes(p).count("language_unsupported") == 1 + + +def test_a_tight_single_word_span_is_not_flagged(): + """Listening across three voices found tight spans good, so warning against + them would steer people away from what works.""" + assert not codes(plan('a bandeja b')) + + +def test_a_clause_length_span_is_not_flagged(): + assert not codes(plan('a bandeja, no un smash, b')) + + +@pytest.mark.parametrize("alias", ["bandeha", "ban-deh-ha", "W C A G", "Bandeha"]) +def test_a_reasonable_respelling_is_not_flagged(alias): + """Plain substitution, syllable hyphens alone, an acronym expansion, and a + capitalised proper noun are all fine.""" + assert not codes(plan(f'a x b')) + + +def test_hyphens_plus_capitals_are_flagged(): + """`ban-DEH-ha` was judged exaggerated on every voice and measured ~30% + longer than the same sentence unmarked. It is the combination that + misfires, not either alone.""" + assert "over_articulated_respelling" in codes(plan('a x b')) + + +def test_a_bare_break_is_a_good_default_pause(): + """700ms was judged right on every voice; 1500ms too long unless the script + wants a beat to stop and think.""" + p = plan("onetwo") + assert Silence(700) in p.nodes + + +# ── Malformed markup ───────────────────────────────────────────────── + + +def test_an_unclosed_span_is_an_error(): + """Not passed through as text: passing it through means the engine reads + the tag aloud.""" + with pytest.raises(ProsodyParseError): + parse('oops') + + +def test_a_mismatched_close_is_an_error(): + with pytest.raises(ProsodyParseError): + parse('a') + + +@pytest.mark.parametrize( + "markup", ["x", 'x', "x", "x"] +) +def test_a_span_missing_its_required_attribute_is_an_error(markup): + with pytest.raises(ProsodyParseError): + parse(markup) + + +# ── The invariant that makes LLM annotation safe ───────────────────── + + +def test_strip_markup_recovers_the_words(): + """Annotation is accepted only if stripping the model's output reproduces + the input. The model can fail to help; it cannot mangle the script.""" + original = "The shot here is a bandeja. Not a smash." + annotated = ( + 'The shot here is a bandeja.' + ' Not a smash.' + ) + assert strip_markup(annotated) == strip_markup(original) + + +def test_strip_markup_detects_a_rewritten_script(): + original = "The shot here is a bandeja." + tampered = 'The shot here is a bandeja, obviously.' + assert strip_markup(tampered) != strip_markup(original) + + +def test_strip_markup_ignores_whitespace_reflow(): + """Putting a tag on its own line is formatting, not content.""" + assert strip_markup("a\n\n b") == strip_markup("a b") + + +def test_has_markup_detects_directives(): + assert has_markup('a ') + assert not has_markup("a plain sentence") + assert not has_markup("5 < 6") + + +# ── Seeds ──────────────────────────────────────────────────────────── + + +def test_seeds_vary_per_run_but_stay_deterministic(): + p = plan('a bc d e').with_seeds(100) + seeds = [n.seed for n in p.nodes if isinstance(n, Speech)] + assert seeds == [100, 101] + assert p.with_seeds(100) == plan( + 'a bc d e' + ).with_seeds(100) + + +def test_an_unseeded_plan_stays_unseeded(): + """Takes should still vary when no seed was requested.""" + p = plan("a b c").with_seeds(None) + assert all(n.seed is None for n in p.nodes if isinstance(n, Speech)) + + +def test_a_respelled_run_is_still_trivial(): + """The respelling is already in the text by this point, so there is nothing + for the renderer to assemble. Sending it through anyway would contradict + the property that makes respelling preferred: it does not cut.""" + assert plan('a bandeja b').is_trivial + + +def test_a_break_makes_a_plan_non_trivial(): + assert not plan('ab').is_trivial