From 5ce9d0e51da7d71419d4598099ececaf687923a7 Mon Sep 17 00:00:00 2001 From: Palash Debnath Date: Fri, 10 Jul 2026 18:40:16 +0530 Subject: [PATCH] =?UTF-8?q?feat(dub):=20predict=20segment=20fit=20before?= =?UTF-8?q?=20synthesis=20=E2=80=94=20tight/impossible=20badges=20+=20opt-?= =?UTF-8?q?in=20shorter=20rewrites=20(#1051)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New pure planning layer (services/duration_planner.py) runs after translation, before TTS: estimates each translated line's natural speech duration (self- calibrating from the job's already-synthesized segments, static per-language rates as cold-start fallback) and classifies it fits/tight/impossible against slot + capped gap borrow, with thresholds derived from fit_planner's own caps so "impossible" means "would be trimmed". Verdicts ride the /dub/translate response and badge the segment table; an opt-in (default OFF) LLM pass attaches one-click shorter-rewrite suggestions for impossible lines. Never blocks generation — informs before GPU time is burned. Co-authored-by: mergetest Co-authored-by: Claude Fable 5 --- backend/api/routers/dub_generate.py | 23 ++ backend/api/routers/dub_translate.py | 137 ++++++- backend/schemas/requests.py | 12 + backend/services/duration_planner.py | 329 +++++++++++++++++ docs/dubbing/translation-engines.md | 18 + frontend/src/api/types.ts | 11 + frontend/src/components/DubSegmentRow.jsx | 36 ++ frontend/src/components/dub/DubLeftColumn.jsx | 19 + frontend/src/hooks/useDubWorkflow.js | 14 + frontend/src/i18n/locales/en.json | 10 +- frontend/src/store/index.ts | 1 + frontend/src/store/prefsSlice.ts | 10 + .../src/test/DubSegmentRowPlanBadge.test.jsx | 108 ++++++ tests/test_dub_duration_plan_api.py | 335 ++++++++++++++++++ tests/test_duration_planner.py | 314 ++++++++++++++++ 15 files changed, 1375 insertions(+), 2 deletions(-) create mode 100644 backend/services/duration_planner.py create mode 100644 frontend/src/test/DubSegmentRowPlanBadge.test.jsx create mode 100644 tests/test_dub_duration_plan_api.py create mode 100644 tests/test_duration_planner.py diff --git a/backend/api/routers/dub_generate.py b/backend/api/routers/dub_generate.py index c87efdce..be17d10d 100644 --- a/backend/api/routers/dub_generate.py +++ b/backend/api/routers/dub_generate.py @@ -338,6 +338,13 @@ async def dub_generate(job_id: str, req: DubRequest): # retain every generated tensor in RAM until final assembly. _pending_seg_writes: list[tuple] = [] + # Calibration records for the pre-synthesis duration planner + # (services/duration_planner.py): text length + the NATURAL-rate TTS + # duration of every freshly synthesized segment. Only meaningful for + # the natural-rate strategies — strict_slot forces the audio to the + # slot length, which would poison the observed chars-per-second. + _natural_dur_records: dict[str, dict] = {} + # Phase 4.1 bench instrumentation: measure where incremental time goes. # Only prints when regen_only is active (real-user incremental path). _t_start = time.perf_counter() @@ -679,6 +686,15 @@ async def dub_generate(job_id: str, req: DubRequest): sync_scores.append(sync_ratio) + # Duration-planner calibration sample: this text length spoke + # for this long at natural rate. Keyed by stable seg id and + # merged into the per-language job map after the loop. + if strategy != "strict_slot" and seg.text.strip() and generated_dur > 0: + _natural_dur_records[str(seg_id)] = { + "chars": len(seg.text.strip()), + "dur": round(generated_dur, 4), + } + # Build the fingerprint now (cheap) but defer the disk write # and job flush to the batch-write phase after the GPU loop. _seg_fp = None @@ -779,6 +795,13 @@ async def dub_generate(job_id: str, req: DubRequest): hashes[_sid] = _fp quality_map[_sid] = _nstep job["seg_hashes"] = dict(hashes) + # Duration-planner calibration: per-language (chars, natural dur) + # records. update() (not replace) so partial regens keep accumulating + # samples from earlier runs of this track. + if _natural_dur_records: + job.setdefault("seg_natural_durs_by_lang", {}).setdefault( + lang_code, {}, + ).update(_natural_dur_records) # Single job flush instead of one per 8 segments. _save_job(job_id, job) _t_diskw = time.perf_counter() - _t_diskw_0 diff --git a/backend/api/routers/dub_translate.py b/backend/api/routers/dub_translate.py index c98207f3..3b75fe0d 100644 --- a/backend/api/routers/dub_translate.py +++ b/backend/api/routers/dub_translate.py @@ -718,6 +718,132 @@ async def dub_translate(req: TranslateRequest): return JSONResponse(status_code=500, content={"error": str(e)}) +def _stamp_duration_plan(rows, req) -> None: + """Attach a pre-synthesis duration-plan verdict to every row (in place). + + Pure planning (services/duration_planner.py): estimate the natural + speech duration of each row's FINAL text — self-calibrated from this + job's already-synthesized segments when possible — and classify it + against slot + borrowable gap using fit_planner's own caps. The verdict + rides on the row as ``plan`` so the segment table can badge tight/ + impossible segments BEFORE any GPU time is spent. Informational only — + generation is never blocked. Never raises. + """ + try: + from services.duration_planner import calibration_from_job, classify_segments + + timed = [ + s for s in req.segments + if getattr(s, "start", None) is not None and getattr(s, "end", None) is not None + ] + if not timed: + return # old client — no timeline info, no plan + text_by_id = {str(r["id"]): (r.get("text") or "") for r in rows} + segs = sorted( + ( + { + "id": str(s.id), + "start": float(s.start), + "end": float(s.end), + "text": text_by_id.get(str(s.id), ""), + } + for s in timed + ), + key=lambda d: d["start"], + ) + calib = None + total_dur = 0.0 + if getattr(req, "job_id", None): + job = _get_job(req.job_id) + if job: + calib = calibration_from_job(job, req.target_lang) + total_dur = float(job.get("duration") or 0.0) + verdicts = { + v["id"]: v + for v in classify_segments( + segs, req.target_lang, calibration=calib, total_dur_s=total_dur, + ) + } + for row in rows: + v = verdicts.get(str(row["id"])) + if v is None or row.get("error") or not (row.get("text") or "").strip(): + continue + row["plan"] = { + "status": v["status"], + "est_dur_s": v["est_dur_s"], + "available_s": v["available_s"], + "est_overrun_s": v["est_overrun_s"], + "calibrated": v["calibrated"], + } + except Exception as e: # noqa: BLE001 — planning must never sink a translate + logger.debug("duration-plan stamping skipped: %s", e) + + +async def _apply_condense_pass(rows, req, loop) -> None: + """Opt-in LLM condensation for ``impossible`` rows (in place). + + Fans ``condense_for_slot`` out on the CPU pool under the same wall-clock + budget the cinematic phase uses, so a slow LLM can't hang the translate. + Suggestions land as ``plan.suggested_text`` — the user applies them per + segment; the row's ``text`` is never touched here. Every failure mode + (no LLM, LLM error, divergent reply, budget) degrades to no suggestion. + """ + targets = [ + row for row in rows + if (row.get("plan") or {}).get("status") == "impossible" + and (row.get("text") or "").strip() and not row.get("error") + ] + if not targets: + return + try: + from services.duration_planner import calibration_from_job, condense_for_slot + + calib = None + if getattr(req, "job_id", None): + job = _get_job(req.job_id) + if job: + calib = calibration_from_job(job, req.target_lang) + source_by_id = {str(s.id): s.text for s in req.segments} + sem = asyncio.Semaphore(int(os.environ.get("OMNIVOICE_LLM_CONCURRENCY", "6"))) + + async def _one(row): + async with sem: + res = await loop.run_in_executor( + _cpu_pool, + lambda: condense_for_slot( + row["text"], + available_s=float(row["plan"]["available_s"]), + target_lang=req.target_lang, + source_text=source_by_id.get(str(row["id"])), + calibration=calib, + ), + ) + if res.get("applied") and res.get("text"): + row["plan"]["suggested_text"] = res["text"] + row["plan"]["suggested_est_dur_s"] = res.get("est_dur_s") + + tasks = [asyncio.ensure_future(_one(row)) for row in targets] + budget = _cinematic_budget() + done, pending = await asyncio.wait( + tasks, timeout=budget if budget and budget > 0 else None, + ) + for task in pending: + task.cancel() # abandon the executor thread (#730 pattern) + for task in done: + exc = task.exception() + if exc is not None: + logger.warning("condense pass segment failed: %s", exc) + except Exception as e: # noqa: BLE001 — a suggestion pass must never sink a translate + logger.warning("condense pass skipped: %s", e) + + +async def _finalize_duration_plan(rows, req, loop) -> None: + """Stamp plan verdicts on the FINAL row texts, then (opt-in) condense.""" + _stamp_duration_plan(rows, req) + if getattr(req, "condense", False): + await _apply_condense_pass(rows, req, loop) + + def _stamp_predicted_rate_ratio(translated, req) -> None: """Stamp a predicted ``rate_ratio`` on every row that has a known slot. @@ -811,8 +937,10 @@ async def _maybe_cinematic(translated, req, src_lang, loop, *, already_llm=False "quality_used": "fast", **_dialect_flags(req, applied=(already_llm and bool(dialect_hint)))} - # Fast (and anything unrecognised) returns the plain translation unchanged. + # Fast (and anything unrecognised) returns the plain translation unchanged + # (plus the pre-synthesis duration-plan badges — no LLM needed for those). if quality not in ("cinematic", "autofit"): + await _finalize_duration_plan(translated, req, loop) return base source_by_id: dict[str, str] = {str(s.id): s.text for s in req.segments} @@ -843,6 +971,8 @@ async def _maybe_cinematic(translated, req, src_lang, loop, *, already_llm=False out["rate_ratio"] = row["rate_ratio"] merged.append(out) await _apply_fit_pass(merged, req, slots_by_id, source_by_id, quality, loop, deadline) + # Plan AFTER the fit pass — verdicts must describe the final text. + await _finalize_duration_plan(merged, req, loop) return {"translated": merged, "target_lang": req.target_lang, "source_lang": src_lang, "quality_used": quality, **_dialect_flags(req, applied=bool(dialect_hint))} @@ -852,6 +982,7 @@ async def _maybe_cinematic(translated, req, src_lang, loop, *, already_llm=False if not cinematic_available(): logger.warning("%s requested but no LLM configured — returning Fast result.", quality) base["cinematic_skipped"] = "no-llm-configured" + await _finalize_duration_plan(translated, req, loop) return base directions: dict[str, str] = { @@ -870,6 +1001,7 @@ async def _maybe_cinematic(translated, req, src_lang, loop, *, already_llm=False pairs.append((seg_id, source_by_id.get(seg_id, ""), literal)) if not pairs: + await _finalize_duration_plan(translated, req, loop) return base refined = await cinematic_refine_many( @@ -906,6 +1038,9 @@ async def _maybe_cinematic(translated, req, src_lang, loop, *, already_llm=False # Phase 4.4 speech-rate fit pass — now concurrent + bounded (see helper). await _apply_fit_pass(merged, req, slots_by_id, source_by_id, quality, loop, deadline) + # Plan AFTER the fit pass — verdicts must describe the final text. + await _finalize_duration_plan(merged, req, loop) + return { "translated": merged, "target_lang": req.target_lang, diff --git a/backend/schemas/requests.py b/backend/schemas/requests.py index 715a7b5a..de1b992b 100644 --- a/backend/schemas/requests.py +++ b/backend/schemas/requests.py @@ -122,6 +122,12 @@ class TranslateSegment(BaseModel): # Available time slot (end - start, seconds) for rate-ratio prediction # and the cinematic slot-fit pass. Same silent-drop fix as `direction`. slot_seconds: Optional[float] = None + # Timeline position (seconds) — lets the duration planner borrow silence + # from the gap to the NEXT segment when classifying fits/tight/impossible + # (services/duration_planner.py). Optional: old clients that only send + # slot_seconds still get rate_ratio badges, just no plan verdicts. + start: Optional[float] = None + end: Optional[float] = None class TranslateRequest(BaseModel): segments: List[TranslateSegment] @@ -147,6 +153,12 @@ class TranslateRequest(BaseModel): # the direct translation). auto_glossary: Optional[bool] = None reflect: Optional[bool] = None + # Opt-in LLM condensation (default OFF): for segments the duration + # planner classifies "impossible", ask the configured LLM for a shorter + # meaning-preserving rewrite and attach it as plan.suggested_text — a + # per-segment suggestion the user applies manually, never auto-applied. + # No LLM configured / LLM failure → silently no suggestion. + condense: Optional[bool] = False class DubIngestUrlRequest(BaseModel): url: str diff --git a/backend/services/duration_planner.py b/backend/services/duration_planner.py new file mode 100644 index 00000000..bee89847 --- /dev/null +++ b/backend/services/duration_planner.py @@ -0,0 +1,329 @@ +"""Pre-synthesis duration planning for dub segments. + +The Smart Fit planner (services/fit_planner.py) reconciles dubbed audio +with the timeline AFTER synthesis — by then a doomed segment has already +burned GPU time and can only be sped up or trimmed. This module predicts +BEFORE TTS whether a translated segment can possibly fit its slot, so the +UI can badge it (and optionally offer a shorter rewrite) while the text is +still cheap to change. It never blocks generation — it informs. + +Three pieces, all pure and unit-testable: + +1. **Estimator** — predict the natural speech duration of target-language + text. Self-calibrating: segments already synthesized in this job carry + ``(chars, natural duration)`` records (written by dub_generate for every + natural-rate strategy), and the median chars-per-second of those is a + far better predictor for *this* voice/engine/language than any table. + With no (or too little) calibration data it falls back to the + conservative static per-language rate table in ``services.speech_rate`` + (the same one the rate-ratio badge uses). + +2. **Classifier** — per segment, compare the estimate against the + *available* time: the slot plus silence borrowable from the gap to the + next segment (mirroring fit_planner's slack absorption, but with a + deliberate cap — see ``GAP_BORROW_MAX_S``). The verdict thresholds are + derived from the SAME ``FitParams`` caps fit_planner enforces, so: + + fits need ≤ max_audio_only_rate — absorbed imperceptibly + tight need ≤ what the caps absorb — audible speed-up and/or + video slow-down + impossible beyond the caps — fit_planner will trim + +3. **Condensation** (optional, caller-gated) — for ``impossible`` segments, + ask the configured LLM for a meaning-preserving shorter rewrite + targeting the available duration. Strictly best-effort: no LLM, an LLM + error, or a divergent reply all degrade to a no-op. + +No I/O, no torch; the only side-effectful function is ``condense_for_slot`` +(network LLM call), which callers opt into explicitly. +""" +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import Iterable, Optional + +from services.fit_planner import MAX_AUDIO_RATE_HARD, FitParams +from services.llm_backend import OffBackend, get_active_llm_backend +from services.speech_rate import expected_duration +# Shared LLM-output divergence guard (target-script + length window + +# critique-echo) — same seam speech_rate's Autofit pass uses. +from services.translator import refine_output_ok + +logger = logging.getLogger("omnivoice.duration_planner") + +# LLM Skills registry id — condensation is the same "make the line fit its +# slot" skill family as the Autofit pass, so it routes (and can be disabled) +# through the same Settings → LLM Skills entry. +_SKILL_ID = "slot_fitting" + +# ── Calibration ───────────────────────────────────────────────────────── + +# A calibration only counts once this many usable samples exist — below +# that, one odd segment (a sound effect, a mumbled clone ref) would swing +# the estimate more than the static table's error. +MIN_CALIBRATION_SAMPLES = 3 +# Per-sample sanity floor: shorter/tinier segments carry more silence +# padding and TTS ramp-up than speech, so their chars/sec is noise. +MIN_SAMPLE_DUR_S = 0.4 +MIN_SAMPLE_CHARS = 4 + +# How far a segment may borrow into the silent gap before the next segment +# (or the video tail). fit_planner itself absorbs the WHOLE gap, so this cap +# makes the pre-synthesis verdict deliberately conservative: a huge gap +# (scene change, music bed) is real slack at mix time, but planning speech +# to sprawl seconds past its slot is rarely what the user wants — and the +# estimate is fuzzy enough that promising it would over-sell. +GAP_BORROW_MAX_S = 3.0 + + +@dataclass(frozen=True) +class Calibration: + """Observed speech rate for one (job, language) pair.""" + cps: float # chars per second at natural TTS rate + samples: int # how many segments backed it + + +def calibrate_cps(samples: Iterable[tuple[float, float]]) -> Optional[Calibration]: + """Derive a chars-per-second calibration from ``(chars, natural_dur_s)`` + pairs of already-synthesized segments. + + Median of the per-segment rates — robust against the occasional outlier + (a segment that's mostly a breath, an engine hiccup) that would drag a + mean. Returns None when fewer than ``MIN_CALIBRATION_SAMPLES`` usable + samples exist; callers then fall back to the static table. + """ + rates: list[float] = [] + for chars, dur in samples: + try: + chars = float(chars) + dur = float(dur) + except (TypeError, ValueError): + continue + if dur >= MIN_SAMPLE_DUR_S and chars >= MIN_SAMPLE_CHARS: + rates.append(chars / dur) + if len(rates) < MIN_CALIBRATION_SAMPLES: + return None + rates.sort() + n = len(rates) + mid = n // 2 + median = rates[mid] if n % 2 else (rates[mid - 1] + rates[mid]) / 2.0 + if median <= 0: + return None + return Calibration(cps=median, samples=n) + + +def calibration_from_job(job: dict, lang: str) -> Optional[Calibration]: + """Build a Calibration from the ``seg_natural_durs_by_lang`` records + dub_generate persists on the job. Tolerates any legacy/partial shape.""" + try: + recs = (job.get("seg_natural_durs_by_lang") or {}).get(lang) or {} + return calibrate_cps( + (r.get("chars", 0), r.get("dur", 0)) + for r in recs.values() + if isinstance(r, dict) + ) + except Exception as e: # noqa: BLE001 — calibration is best-effort by design + logger.debug("calibration_from_job skipped: %s", e) + return None + + +# ── Estimator ─────────────────────────────────────────────────────────── + + +def estimate_natural_duration( + text: str, lang: str, calibration: Optional[Calibration] = None, +) -> float: + """Predicted natural-rate speech duration (seconds) of ``text``. + + Calibrated rate when available, else the static per-language table + (``speech_rate.expected_duration``, 13 cps default for unknown codes). + """ + text = (text or "").strip() + if not text: + return 0.0 + if calibration is not None and calibration.cps > 0: + return len(text) / calibration.cps + return expected_duration(text, lang) + + +# ── Classifier ────────────────────────────────────────────────────────── + + +def absorb_caps(params: FitParams) -> tuple[float, float]: + """(fits_cap, absorb_cap) need-ratios aligned with fit_planner. + + ``fits_cap``: up to here the audio-only speed-up is imperceptible. + ``absorb_cap``: up to here fit_planner's knobs absorb the overrun + (audio cap × video cap in hybrid mode; the legacy hard audio ceiling + when video retiming is off). Beyond it, fit_planner trims. + """ + if params.allow_video_retime: + return params.max_audio_only_rate, params.audio_rate_cap * params.video_slow_cap + return params.max_audio_only_rate, MAX_AUDIO_RATE_HARD + + +def classify_segments( + segments: list[dict], + target_lang: str, + *, + calibration: Optional[Calibration] = None, + fit_params: Optional[FitParams] = None, + total_dur_s: float = 0.0, + gap_borrow_max_s: float = GAP_BORROW_MAX_S, +) -> list[dict]: + """Classify each segment's translated text against its timeline slot. + + ``segments``: chronological dicts with ``id``, ``start``, ``end`` + (seconds) and ``text`` (the translated text about to be synthesized). + ``total_dur_s``: original video duration (0/unknown → the last segment + gets no tail borrow), mirroring ``fit_planner.plan_fit``. + + Returns one dict per segment:: + + {id, status, est_dur_s, available_s, est_overrun_s, calibrated} + + ``status`` ∈ {"fits", "tight", "impossible"}; ``est_overrun_s`` is the + predicted seconds of speech past the available time (0 when it fits). + Pure function: no I/O, deterministic. + """ + params = fit_params or FitParams() + fits_cap, cap = absorb_caps(params) + n = len(segments) + out: list[dict] = [] + for i, seg in enumerate(segments): + start = float(seg["start"]) + end = float(seg["end"]) + slot = max(0.0, end - start) + + # Borrowable silence — fit_planner's slack absorption, capped. + if i + 1 < n: + gap = max(0.0, float(segments[i + 1]["start"]) - end) + borrow = min(max(0.0, gap - params.gap_guard_s), gap_borrow_max_s) + elif total_dur_s > 0: + borrow = min(max(0.0, float(total_dur_s) - end), gap_borrow_max_s) + else: + borrow = 0.0 + available = slot + borrow + + est = estimate_natural_duration(seg.get("text") or "", target_lang, calibration) + if est <= 0.0: + status = "fits" + overrun = 0.0 + elif available <= 0.0: + status = "impossible" + overrun = est + else: + need = est / available + # Same boundary tolerance as fit_planner's _EPS: a need that + # lands exactly on a cap is absorbed, not escalated. + if need <= fits_cap + 1e-9: + status = "fits" + elif need <= cap + 1e-9: + status = "tight" + else: + status = "impossible" + overrun = max(0.0, est - available) + + out.append({ + "id": str(seg.get("id", f"seg_{i}")), + "status": status, + "est_dur_s": round(est, 3), + "available_s": round(available, 3), + "est_overrun_s": round(overrun, 3), + "calibrated": calibration is not None, + }) + return out + + +# ── Optional LLM condensation ─────────────────────────────────────────── + +_CONDENSE_PROMPT = """\ +You are a dubbing writer. The user will give you a translated line that is +TOO LONG for its time slot. Rewrite it shorter so it can be read aloud +within the target duration: cut filler words, tighten phrasing, and drop +the least essential clauses — but preserve the meaning. Never change +character names, proper nouns, numbers, or technical terms. Stay in the +same language as the line. +Reply with ONLY the rewritten line. No quotes, no commentary.""" + +# Bound the LLM loop — condensation is a per-segment *suggestion*, not a +# fit guarantee, so two shots are plenty before degrading to a no-op. +_CONDENSE_ATTEMPTS = 2 + + +def condense_for_slot( + text: str, + *, + available_s: float, + target_lang: str, + source_text: Optional[str] = None, + calibration: Optional[Calibration] = None, +) -> dict: + """Meaning-preserving shorter rewrite of ``text`` targeting ``available_s``. + + Returns ``{"text", "applied", "est_dur_s"}`` (+ ``"error"`` on the no-op + paths). ``applied=False`` keeps the input text untouched — no LLM + configured, LLM failure, and divergent/too-aggressive replies all + degrade there. The best (shortest-estimate) candidate that passes the + divergence guard AND is actually shorter than the input wins; a reply + that fits ``available_s`` returns immediately. + """ + text = (text or "").strip() + base_est = estimate_natural_duration(text, target_lang, calibration) + if not text or available_s <= 0: + return {"text": text, "applied": False, "est_dur_s": round(base_est, 3), + "error": "nothing-to-condense"} + if base_est <= available_s: + return {"text": text, "applied": False, "est_dur_s": round(base_est, 3), + "error": "already-fits"} + + from services import llm_skills + # `active=` forwards this module's (monkeypatch-able) name so the + # no-override path matches the plain get_active_llm_backend behavior. + llm = llm_skills.skill_backend(_SKILL_ID, active=lambda: get_active_llm_backend()) + if isinstance(llm, OffBackend): + return {"text": text, "applied": False, "est_dur_s": round(base_est, 3), + "error": "no-llm"} + + best: Optional[tuple[str, float]] = None # (candidate, est) + for attempt in range(1, _CONDENSE_ATTEMPTS + 1): + user_lines = [ + f"Target language: {target_lang}", + f"Target duration: {available_s:.2f}s", + f"Current line: {text}", + f"Current reading duration: ~{base_est:.2f}s", + ] + if source_text: + user_lines.append(f"Source line (for meaning): {source_text}") + if attempt > 1 and best is not None: + user_lines.append( + f"Your previous rewrite was still ~{best[1]:.2f}s. Cut further." + ) + try: + reply = llm.chat( + system=_CONDENSE_PROMPT, user="\n".join(user_lines), + temperature=0.2, # pinned like Autofit — default 1.0 drifts/invents + ) + except Exception as e: # noqa: BLE001 — LLM failure must no-op, never raise + logger.warning("condense attempt %d failed: %s", attempt, e) + break + candidate = (reply or "").strip() + if not candidate: + continue + ok, reason = refine_output_ok(text, candidate, target_lang) + if not ok: + logger.warning("condense attempt %d rejected (%s)", attempt, reason) + continue + est = estimate_natural_duration(candidate, target_lang, calibration) + if est >= base_est: + continue # not actually shorter — useless as a suggestion + if best is None or est < best[1]: + best = (candidate, est) + if est <= available_s: + break # fits — done + + if best is None: + return {"text": text, "applied": False, "est_dur_s": round(base_est, 3), + "error": "condense-failed"} + return {"text": best[0], "applied": True, "est_dur_s": round(best[1], 3)} diff --git a/docs/dubbing/translation-engines.md b/docs/dubbing/translation-engines.md index 0576bee0..e588d95a 100644 --- a/docs/dubbing/translation-engines.md +++ b/docs/dubbing/translation-engines.md @@ -107,6 +107,24 @@ MT engines can't run either stage): turn it off for long videos on slow or metered providers. If any refinement step fails or times out, the direct translation is kept silently; refinement can never fail a segment. +### Fit prediction (all quality levels) + +Every translation additionally gets a **pre-synthesis fit check** — no LLM +needed. For each segment, OmniVoice predicts how long the translated line will +take to speak (self-calibrating to your voice/engine from segments already +generated in the job, with a per-language rate table as the cold-start +fallback) and compares it against the slot plus the silence it can borrow +before the next line. Segments the Smart Fit caps can only absorb with an +audible speed-up get a **Tight fit** badge; segments no fitting can save get a +**Won't fit +Ns** badge — so you can shorten the text *before* burning GPU +time on a line that would end up trimmed. Badges are informational only: +generation is never blocked. + +**Suggest shorter lines** (checkbox under Quality, off by default) goes one +step further: for every "Won't fit" segment it asks the configured LLM for a +meaning-preserving shorter rewrite and offers it on the row as a one-click +**Use shorter rewrite** suggestion. It never rewrites anything automatically, +and with no LLM configured (or on any LLM error) it simply does nothing. ## LLM Providers (for Cinematic / Autofit) diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index de447aef..cb71c9d5 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -244,6 +244,17 @@ export interface DubTranslateResponse { text_original?: string; rate_ratio?: number; rate_error?: string; + /** Pre-synthesis duration plan (backend services/duration_planner.py). */ + plan?: { + status: 'fits' | 'tight' | 'impossible'; + est_dur_s: number; + available_s: number; + est_overrun_s: number; + calibrated: boolean; + /** Opt-in LLM condensation suggestion (request condense=true only). */ + suggested_text?: string; + suggested_est_dur_s?: number; + }; }[]; } diff --git a/frontend/src/components/DubSegmentRow.jsx b/frontend/src/components/DubSegmentRow.jsx index c6f372d2..06888798 100644 --- a/frontend/src/components/DubSegmentRow.jsx +++ b/frontend/src/components/DubSegmentRow.jsx @@ -275,6 +275,42 @@ function DubSegmentRow({ 📖 {seg.rate_ratio.toFixed(2)}× )} + {/* Pre-synthesis duration plan (backend duration_planner): warn about + tight/impossible segments BEFORE GPU time is spent. Informational + only — generation is never blocked. */} + {seg.plan && (seg.plan.status === 'tight' || seg.plan.status === 'impossible') && ( + + {' '} + {seg.plan.status === 'impossible' + ? t('segment.plan_impossible', { + seconds: (seg.plan.est_overrun_s || 0).toFixed(1), + }) + : t('segment.plan_tight')} + + )} + {seg.plan && seg.plan.suggested_text && seg.plan.suggested_text !== seg.text && ( + + )} s.setAutoGlossary); const reflectPass = useAppStore((s) => s.reflectPass); const setReflectPass = useAppStore((s) => s.setReflectPass); + // Opt-in LLM condensation suggestions for segments the duration planner + // classifies as impossible to fit (default OFF — needs an LLM). + const condenseSuggest = useAppStore((s) => s.condenseSuggest); + const setCondenseSuggest = useAppStore((s) => s.setCondenseSuggest); // Frozen-build (packaged/signed, read-only site-packages) escape-hatch // popover: pip install is impossible, so we surface the copyable command + // a one-click switch to the always-bundled Argos engine + a docs deeplink. @@ -652,6 +656,21 @@ export default function DubLeftColumn({ { value: 'cinematic', label: t('dub.cinematic_quality') }, ]} /> + {/* Opt-in (default OFF): when the duration planner marks a + translated line "impossible" for its slot, ask the LLM for a + shorter rewrite the user can apply per segment. */} + {/* LLM engine only: auto-glossary + reflect pass. Both default ON; the reflect tooltip is explicit that it multiplies LLM calls. */} diff --git a/frontend/src/hooks/useDubWorkflow.js b/frontend/src/hooks/useDubWorkflow.js index d64e002b..2f6d05d6 100644 --- a/frontend/src/hooks/useDubWorkflow.js +++ b/frontend/src/hooks/useDubWorkflow.js @@ -83,6 +83,7 @@ export default function useDubWorkflow({ const cfg = useAppStore((s) => s.cfg); const speed = useAppStore((s) => s.speed); const translateQuality = useAppStore((s) => s.translateQuality); + const condenseSuggest = useAppStore((s) => s.condenseSuggest); const timingStrategy = useAppStore((s) => s.timingStrategy); const fitOptions = useAppStore((s) => s.fitOptions); const glossaryTerms = useAppStore((s) => s.glossaryTerms); @@ -718,6 +719,11 @@ export default function useDubWorkflow({ target_lang: s.target_lang, direction: s.direction || undefined, slot_seconds: s.end != null && s.start != null ? s.end - s.start : undefined, + // Timeline position — lets the backend's duration planner borrow + // silence from the gap to the next segment when classifying + // fits/tight/impossible before any GPU time is spent. + start: s.start != null ? s.start : undefined, + end: s.end != null ? s.end : undefined, })), target_lang: targetLang, provider: translateProvider, @@ -731,6 +737,9 @@ export default function useDubWorkflow({ // loop reuses this callback long after the click-time closure. auto_glossary: useAppStore.getState().autoGlossary, reflect: useAppStore.getState().reflectPass, + // Opt-in (default OFF): ask the LLM for shorter rewrites of + // segments the planner marks impossible — suggestions only. + condense: condenseSuggest || undefined, // #280: regional dialect — only sent when it matches the target // language so a stale "es-AR" never rides on a French translate. dialect: dialectMatchesLang(dubDialect, targetLang) ? dubDialect : undefined, @@ -765,6 +774,10 @@ export default function useDubWorkflow({ // the user clicks Generate Dub. rate_ratio: hit.rate_ratio != null ? hit.rate_ratio : s.rate_ratio, rate_error: hit.rate_error || s.rate_error, + // Pre-synthesis duration plan (fits/tight/impossible + optional + // condensed-rewrite suggestion) — drives the row badge so a + // doomed segment is visible before Generate Dub is clicked. + plan: hit.plan != null ? hit.plan : s.plan, }; }), ); @@ -822,6 +835,7 @@ export default function useDubWorkflow({ dubJobId, translateProvider, translateQuality, + condenseSuggest, glossaryTerms, setIsTranslating, setDubSegments, diff --git a/frontend/src/i18n/locales/en.json b/frontend/src/i18n/locales/en.json index 127df09e..ecab9ccc 100644 --- a/frontend/src/i18n/locales/en.json +++ b/frontend/src/i18n/locales/en.json @@ -859,6 +859,8 @@ "auto_glossary_title": "One extra LLM pass reads the whole transcript and extracts the video's theme plus key terms, so names and terminology stay consistent across segments. Your manual glossary entries always win.", "reflect_label": "Reflect pass", "reflect_title": "After each segment's direct translation, the LLM critiques and rewrites it into natural spoken dialogue. Uses 3 LLM calls per segment instead of 1; if refinement fails, the direct translation is kept.", + "condense_label": "Suggest shorter lines", + "condense_title": "When a translated line is predicted not to fit its time slot, ask the LLM for a shorter meaning-preserving rewrite you can apply per segment. Off by default; needs a configured LLM.", "restore": "↺ Restore", "restore_title": "Restore all segments to the original transcribed text", "clean_up": "Clean Up", @@ -1024,7 +1026,13 @@ "fit_stretched_title": "Stretch Video mode: this segment's video was slowed to {{ratio}}× to fit the natural dub audio.", "fit_compressed_title": "TTS audio is {{pct}}% of the slot — heavily compressed.", "fit_audio_title": "Audio fit inside the slot.", - "fit_ratio_title": "TTS audio is {{pct}}% of the slot." + "fit_ratio_title": "TTS audio is {{pct}}% of the slot.", + "plan_tight": "Tight fit", + "plan_tight_title": "Predicted speech ≈{{est}}s vs {{avail}}s available — Smart Fit can absorb this, but only with an audible speed-up or video slow-down. Shorten the text for a cleaner result.", + "plan_impossible": "Won't fit +{{seconds}}s", + "plan_impossible_title": "Predicted speech ≈{{est}}s vs {{avail}}s available — about {{seconds}}s more than any fitting can absorb, so the audio would be trimmed. Shorten the text before generating.", + "plan_apply": "Use shorter rewrite", + "plan_apply_title": "Replace this line with the suggested shorter rewrite: \"{{text}}\"" }, "timeline": { "track_label": "Segment timeline", diff --git a/frontend/src/store/index.ts b/frontend/src/store/index.ts index 63ab71e2..2df2bce4 100644 --- a/frontend/src/store/index.ts +++ b/frontend/src/store/index.ts @@ -85,6 +85,7 @@ export const useAppStore = create()( translateQuality: s.translateQuality, autoGlossary: s.autoGlossary, reflectPass: s.reflectPass, + condenseSuggest: s.condenseSuggest, dualSubs: s.dualSubs, burnSubs: s.burnSubs, glossaryVisible: s.glossaryVisible, diff --git a/frontend/src/store/prefsSlice.ts b/frontend/src/store/prefsSlice.ts index 1d3eb673..af157431 100644 --- a/frontend/src/store/prefsSlice.ts +++ b/frontend/src/store/prefsSlice.ts @@ -74,6 +74,13 @@ interface FitOptions { export interface PrefsSlice { translateQuality: TranslateQuality; + /** + * Opt-in LLM condensation for doomed dub segments (default OFF). When on, + * Translate asks the configured LLM for a shorter meaning-preserving + * rewrite of every segment the duration planner marks "impossible" and + * attaches it as a per-segment suggestion — never applied automatically. + */ + condenseSuggest: boolean; dualSubs: boolean; burnSubs: boolean; glossaryVisible: boolean; @@ -138,6 +145,7 @@ export interface PrefsSlice { setReflectPass: (on: boolean) => void; setTranslateQuality: (q: TranslateQuality) => void; + setCondenseSuggest: (on: boolean) => void; setDualSubs: (on: boolean) => void; setBurnSubs: (on: boolean) => void; setGlossaryVisible: (on: boolean) => void; @@ -222,6 +230,7 @@ export const createPrefsSlice: StateCreator = (s translateQuality: 'fast', autoGlossary: true, reflectPass: true, + condenseSuggest: false, dualSubs: false, burnSubs: false, glossaryVisible: true, @@ -243,6 +252,7 @@ export const createPrefsSlice: StateCreator = (s setTranslateQuality: (q) => set({ translateQuality: q }), setAutoGlossary: (on) => set({ autoGlossary: on }), setReflectPass: (on) => set({ reflectPass: on }), + setCondenseSuggest: (on) => set({ condenseSuggest: on }), setDualSubs: (on) => set({ dualSubs: on }), setBurnSubs: (on) => set({ burnSubs: on }), setGlossaryVisible: (on) => set({ glossaryVisible: on }), diff --git a/frontend/src/test/DubSegmentRowPlanBadge.test.jsx b/frontend/src/test/DubSegmentRowPlanBadge.test.jsx new file mode 100644 index 00000000..678a0222 --- /dev/null +++ b/frontend/src/test/DubSegmentRowPlanBadge.test.jsx @@ -0,0 +1,108 @@ +import React from 'react'; +import { describe, it, expect, vi } from 'vitest'; +import { render, screen, fireEvent } from '@testing-library/react'; +import '../i18n'; + +// Pre-synthesis duration-plan badge — the /dub/translate response stamps a +// per-segment `plan` (fits/tight/impossible + optional condensed rewrite); +// the row must warn on tight/impossible BEFORE any GPU time is spent, stay +// silent on fits, and let the user apply a suggested rewrite in one click. + +import DubSegmentRow from '../components/DubSegmentRow'; + +function makeProps(plan, over = {}) { + return { + seg: { + id: 's1', + start: 0, + end: 2, + text: 'hola mundo', + plan, + }, + idx: 0, + disabled: false, + isActive: false, + isDone: false, + isPlaying: false, + previewLoading: false, + selected: false, + profiles: [], + speakerClones: {}, + onEditField: vi.fn(), + onDelete: vi.fn(), + onRestore: vi.fn(), + onPreview: vi.fn(), + onSelect: vi.fn(), + onSplit: vi.fn(), + onMerge: vi.fn(), + canMerge: false, + onDirect: vi.fn(), + onSeek: vi.fn(), + timelineSelected: false, + ...over, + }; +} + +describe('DubSegmentRow duration-plan badge', () => { + it('warns on an impossible segment with the estimated overrun', () => { + render( + , + ); + expect(screen.getByText(/Won't fit \+4\.9s/)).toBeTruthy(); + }); + + it('flags a tight segment', () => { + render( + , + ); + expect(screen.getByText(/Tight fit/)).toBeTruthy(); + }); + + it('stays silent when the plan says fits', () => { + render( + , + ); + expect(screen.queryByText(/Tight fit/)).toBeNull(); + expect(screen.queryByText(/Won't fit/)).toBeNull(); + }); + + it('applies the condensed rewrite via onEditField — never automatically', () => { + const props = makeProps({ + status: 'impossible', + est_dur_s: 6.9, + available_s: 2.0, + est_overrun_s: 4.9, + calibrated: false, + suggested_text: 'hola', + }); + render(); + // The row still shows the original text — suggestions are opt-in per click. + expect(screen.getByDisplayValue('hola mundo')).toBeTruthy(); + fireEvent.click(screen.getByText(/Use shorter rewrite/)); + expect(props.onEditField).toHaveBeenCalledWith('s1', 'text', 'hola'); + }); +}); diff --git a/tests/test_dub_duration_plan_api.py b/tests/test_dub_duration_plan_api.py new file mode 100644 index 00000000..a4efae0b --- /dev/null +++ b/tests/test_dub_duration_plan_api.py @@ -0,0 +1,335 @@ +"""Duration planning on the dub API surface. + +Two halves of the pre-synthesis planning loop (services/duration_planner.py): + + - /dub/translate stamps a per-segment ``plan`` verdict (fits/tight/ + impossible + estimated overrun) on the rows the segment table consumes, + self-calibrated from the job's synthesized-segment records when present, + and — opt-in via ``condense`` (default OFF) — attaches an LLM + shorter-rewrite suggestion to impossible rows; + - dub_generate records the (chars, natural duration) calibration samples + those verdicts feed on, for every natural-rate strategy but never for + strict_slot (whose slot-forced WAVs would poison the observed rate). +""" +from __future__ import annotations + +import os +os.environ.setdefault("OMNIVOICE_DISABLE_FILE_LOG", "1") + +import asyncio +import json + +import pytest +import torch + +from schemas.requests import DubRequest, TranslateRequest, TranslateSegment + + +def _install_fake_argos(monkeypatch): + """Fake `argostranslate` translating en→es to `[es]` offline.""" + import sys + import types + + class _Pkg: + from_code = "en" + to_code = "es" + + pkg = types.ModuleType("argostranslate.package") + pkg.get_installed_packages = lambda: [_Pkg()] + pkg.update_package_index = lambda: None + pkg.get_available_packages = lambda: [] + pkg.install_from_path = lambda p: None + tr = types.ModuleType("argostranslate.translate") + tr.translate = lambda text, frm, to: f"[{to}]{text}" + root = types.ModuleType("argostranslate") + root.package = pkg + root.translate = tr + monkeypatch.setitem(sys.modules, "argostranslate", root) + monkeypatch.setitem(sys.modules, "argostranslate.package", pkg) + monkeypatch.setitem(sys.modules, "argostranslate.translate", tr) + + +# Three slots sized so the fake-argos output ("[es]" + text, es ≈ 15.5 cps) +# lands squarely in each verdict bucket: +# s1: 24 chars / 2.0s → need ~0.8 → fits +# s2: 44 chars / 1.0s → need ~2.8 → tight (hybrid caps absorb ≤ 3.0) +# s3: 84 chars / 0.5s → need ~10.8 → impossible +def _timed_segments(): + return [ + TranslateSegment(id="s1", text="A" * 20, slot_seconds=2.0, start=0.0, end=2.0), + TranslateSegment(id="s2", text="B" * 40, slot_seconds=1.0, start=2.05, end=3.05), + TranslateSegment(id="s3", text="C" * 80, slot_seconds=0.5, start=3.1, end=3.6), + ] + + +def _rows_by_id(resp): + return {r["id"]: r for r in resp["translated"]} + + +@pytest.mark.asyncio +async def test_translate_stamps_plan_per_segment(monkeypatch): + """The response the segment table consumes carries one plan per row, + with statuses aligned to fit_planner's caps and a truthful overrun.""" + from api.routers import dub_translate + _install_fake_argos(monkeypatch) + + req = TranslateRequest( + segments=_timed_segments(), + target_lang="es", provider="argos", source_lang="en", quality="fast", + ) + rows = _rows_by_id(await dub_translate.dub_translate(req)) + + assert rows["s1"]["plan"]["status"] == "fits" + assert rows["s2"]["plan"]["status"] == "tight" + assert rows["s3"]["plan"]["status"] == "impossible" + p3 = rows["s3"]["plan"] + # Shape contract for the UI badge + tooltip. + assert set(p3) >= {"status", "est_dur_s", "available_s", "est_overrun_s", "calibrated"} + assert p3["calibrated"] is False # no job → static table + assert p3["est_overrun_s"] > 0 + assert "suggested_text" not in p3 # condense is opt-in, default OFF + + +@pytest.mark.asyncio +async def test_translate_plan_uses_job_calibration(monkeypatch): + """With synthesized-segment records on the job, the estimate calibrates + to THIS voice's observed rate — a fast voice flips s2 tight → fits.""" + from api.routers import dub_translate + _install_fake_argos(monkeypatch) + + job = { + "duration": 4.0, + "seg_natural_durs_by_lang": {"es": { + "a": {"chars": 40, "dur": 1.0}, # 40 cps — a very fast voice + "b": {"chars": 80, "dur": 2.0}, + "c": {"chars": 120, "dur": 3.0}, + }}, + } + monkeypatch.setattr(dub_translate, "_get_job", lambda job_id: job) + + req = TranslateRequest( + segments=_timed_segments(), job_id="j1", + target_lang="es", provider="argos", source_lang="en", quality="fast", + ) + rows = _rows_by_id(await dub_translate.dub_translate(req)) + + assert rows["s2"]["plan"]["status"] == "fits" # 44 chars / 40 cps ≈ 1.1s + assert rows["s2"]["plan"]["calibrated"] is True + # s3: 84/40 = 2.1s over 0.9s available (0.5s slot + 0.4s video tail from + # job duration) → need ~2.3: the caps absorb it now — impossible → tight. + assert rows["s3"]["plan"]["status"] == "tight" + assert rows["s3"]["plan"]["available_s"] == pytest.approx(0.9) + + +@pytest.mark.asyncio +async def test_translate_without_timing_stamps_no_plan(monkeypatch): + """Old clients that only send slot_seconds keep the exact pre-feature + response shape — rate_ratio badge yes, plan no.""" + from api.routers import dub_translate + _install_fake_argos(monkeypatch) + + req = TranslateRequest( + segments=[TranslateSegment(id="s1", text="Hello", slot_seconds=2.0)], + target_lang="es", provider="argos", source_lang="en", quality="fast", + ) + row = _rows_by_id(await dub_translate.dub_translate(req))["s1"] + assert "rate_ratio" in row + assert "plan" not in row + + +@pytest.mark.asyncio +async def test_condense_toggle_attaches_suggestion_to_impossible_only(monkeypatch): + """condense=true + a configured LLM → impossible rows get a shorter + suggested_text; the row's text itself is never rewritten.""" + from api.routers import dub_translate + from services import duration_planner as dp + _install_fake_argos(monkeypatch) + + class _Shortener: + def chat(self, *, system, user, timeout=None, temperature=None): + return "D" * 36 # shorter than s3's 84 chars, passes the guard + + monkeypatch.setattr(dp, "get_active_llm_backend", lambda: _Shortener()) + + req = TranslateRequest( + segments=_timed_segments(), condense=True, + target_lang="es", provider="argos", source_lang="en", quality="fast", + ) + rows = _rows_by_id(await dub_translate.dub_translate(req)) + + assert rows["s3"]["plan"]["suggested_text"] == "D" * 36 + assert rows["s3"]["text"] == "[es]" + "C" * 80 # untouched — user applies it + assert "suggested_text" not in rows["s1"]["plan"] # fits: no suggestion + assert "suggested_text" not in rows["s2"]["plan"] # tight: no suggestion + + +@pytest.mark.asyncio +async def test_condense_llm_failure_degrades_to_no_suggestion(monkeypatch): + from api.routers import dub_translate + from services import duration_planner as dp + _install_fake_argos(monkeypatch) + + class _Broken: + def chat(self, **kw): + raise RuntimeError("provider down") + + monkeypatch.setattr(dp, "get_active_llm_backend", lambda: _Broken()) + + req = TranslateRequest( + segments=_timed_segments(), condense=True, + target_lang="es", provider="argos", source_lang="en", quality="fast", + ) + rows = _rows_by_id(await dub_translate.dub_translate(req)) + assert rows["s3"]["plan"]["status"] == "impossible" + assert "suggested_text" not in rows["s3"]["plan"] # no-op, translate intact + + +@pytest.mark.asyncio +async def test_cinematic_path_plans_the_refined_text(monkeypatch): + """The cinematic/autofit merge path stamps plans too — on the FINAL + (refined) text, after the fit pass.""" + from api.routers import dub_translate + _install_fake_argos(monkeypatch) + + async def fake_refine_many(pairs, **kw): + return [{"id": sid, "text": f"CINE:{lit}", "literal": lit, "critique": ""} + for sid, _src, lit in pairs] + + monkeypatch.setattr(dub_translate, "cinematic_available", lambda: True) + monkeypatch.setattr(dub_translate, "cinematic_refine_many", fake_refine_many) + + req = TranslateRequest( + segments=_timed_segments(), + target_lang="es", provider="argos", source_lang="en", quality="cinematic", + ) + rows = _rows_by_id(await dub_translate.dub_translate(req)) + for sid in ("s1", "s2", "s3"): + assert rows[sid]["plan"]["status"] in ("fits", "tight", "impossible") + # 5 extra "CINE:" chars push s3 even further past its 0.5s slot. + assert rows["s3"]["plan"]["status"] == "impossible" + + +# ── dub_generate calibration recording ────────────────────────────────── + +SR = 24000 + + +class _FakeBackend: + """TTS stand-in: text encodes its natural duration as ':'.""" + + applies_own_mastering = False + sample_rate = SR + + def generate(self, text=None, **kwargs): + dur = float(text.split(":", 1)[0]) + return torch.full((1, int(dur * SR)), 0.25) + + +async def _fake_stretch(wav, target_samples, sr): + if target_samples <= 0 or wav.shape[-1] == target_samples: + return wav + return torch.nn.functional.interpolate( + wav.unsqueeze(0), size=target_samples, mode="linear", align_corners=False, + ).squeeze(0) + + +@pytest.fixture +def patched_generate(monkeypatch, tmp_path): + import api.routers.dub_generate as dg + + async def _fake_resolve(**kwargs): + return _FakeBackend() + + job = {"duration": 6.0, "dubbed_tracks": {}, "speaker_clones": {}} + job_dir = tmp_path / "jobX" + job_dir.mkdir() + + monkeypatch.setattr(dg, "resolve_generation_backend", _fake_resolve) + monkeypatch.setattr(dg, "_get_job", lambda job_id: job) + monkeypatch.setattr(dg, "_save_job", lambda job_id, j: None) + monkeypatch.setattr(dg, "DUB_DIR", str(tmp_path)) + monkeypatch.setattr( + dg, "dub_seg_path", lambda job_id, seg_id: str(job_dir / f"seg_{seg_id}.wav"), + ) + monkeypatch.setattr(dg, "rvc_is_enabled", lambda: False) + monkeypatch.setattr(dg, "embed_watermark", lambda wav, sr: wav) + monkeypatch.setattr(dg, "apply_mastering", lambda a, sample_rate=None: a) + monkeypatch.setattr(dg, "get_effect_chain", lambda preset: None) + monkeypatch.setattr(dg, "apply_effects_chain", lambda a, **k: a) + monkeypatch.setattr(dg, "normalize_audio", lambda a, target_dBFS=None: a) + monkeypatch.setattr(dg, "_pitch_preserving_stretch", _fake_stretch) + + events: list[str] = [] + + class _StubTaskManager: + def is_cancelled(self, task_id): + return False + + async def add_task(self, task_id, task_type, func, *args, **kwargs): + async for evt in func(*args): + events.append(evt) + + monkeypatch.setattr(dg, "task_manager", _StubTaskManager()) + + def run(body: dict): + events.clear() + asyncio.run(dg.dub_generate("jobX", DubRequest(**body))) + parsed = [ + json.loads(e.strip()[len("data: "):]) + for e in events if e.strip().startswith("data: ") + ] + assert any(p.get("type") == "done" for p in parsed), parsed + return parsed + + return run, job + + +def _body(segments, **extra): + return { + "segments": segments, + "segment_ids": [str(i) for i in range(len(segments))], + "language": "Auto", + "language_code": "es", + "num_step": 4, + **extra, + } + + +def test_generate_records_natural_durations_for_calibration(patched_generate): + """Natural-rate strategies persist (chars, natural dur) per segment — + the raw material calibration_from_job turns into a chars/sec rate.""" + run, job = patched_generate + run(_body( + [ + {"start": 0.0, "end": 2.0, "text": "1.5:hola"}, + {"start": 3.0, "end": 4.0, "text": "2.0:adios"}, + ], + timing_strategy="concise", + )) + recs = job["seg_natural_durs_by_lang"]["es"] + assert recs["0"] == {"chars": len("1.5:hola"), "dur": 1.5} + assert recs["1"] == {"chars": len("2.0:adios"), "dur": 2.0} + + # The records feed a usable calibration once enough samples exist. + from services.duration_planner import calibration_from_job + run(_body( + [ + {"start": 0.0, "end": 2.0, "text": "1.5:hola"}, + {"start": 3.0, "end": 4.0, "text": "2.0:adios"}, + {"start": 4.5, "end": 5.0, "text": "1.0:si"}, + ], + timing_strategy="smart_fit", + )) + assert len(job["seg_natural_durs_by_lang"]["es"]) == 3 # merged, not replaced + assert calibration_from_job(job, "es") is not None + + +def test_strict_slot_never_records_calibration(patched_generate): + """strict_slot pads/trims the audio to the slot — recording those + durations would poison the observed chars-per-second.""" + run, job = patched_generate + run(_body( + [{"start": 0.0, "end": 2.0, "text": "1.5:hola"}], + timing_strategy="strict_slot", + )) + assert "seg_natural_durs_by_lang" not in job diff --git a/tests/test_duration_planner.py b/tests/test_duration_planner.py new file mode 100644 index 00000000..5bd1dea1 --- /dev/null +++ b/tests/test_duration_planner.py @@ -0,0 +1,314 @@ +"""Pre-synthesis duration planning (services/duration_planner.py). + +Covers the pure planning layer that runs after translation, before TTS: + + - estimator: static per-language fallback vs. self-calibrated rate; + - calibration: median chars-per-second from synthesized samples, garbage + filtering, the minimum-sample gate, and the job-record reader; + - classifier: fits/tight/impossible thresholds derived from fit_planner's + caps, gap borrowing (and its cap), the tail borrow, audio-only mode; + - alignment: an "impossible" verdict must mean fit_planner would trim, + and "tight"/"fits" must mean it would not; + - condensation: opt-in LLM shorter-rewrite suggestions — no-LLM no-op, + already-fits no-op (no LLM call), divergence-guard rejection, LLM + failure no-op, and the retry-then-accept path. +""" +from __future__ import annotations + +import os +os.environ.setdefault("OMNIVOICE_DISABLE_FILE_LOG", "1") + +import pytest + +from services import duration_planner as dp +from services.fit_planner import MAX_AUDIO_RATE_HARD, FitParams, plan_fit + + +# ── Estimator ─────────────────────────────────────────────────────────── + + +def test_static_fallback_matches_rate_table(): + # 30 chars @ en 15 cps → 2.0s (same table as speech_rate's badge). + assert dp.estimate_natural_duration("A" * 30, "en") == pytest.approx(2.0) + + +def test_unknown_language_uses_conservative_default(): + # Unknown code falls back to 13 cps — never a crash, never zero. + assert dp.estimate_natural_duration("A" * 26, "xx") == pytest.approx(2.0) + + +def test_calibrated_rate_overrides_table(): + calib = dp.Calibration(cps=10.0, samples=5) + assert dp.estimate_natural_duration("A" * 30, "en", calib) == pytest.approx(3.0) + + +def test_empty_text_estimates_zero(): + assert dp.estimate_natural_duration("", "en") == 0.0 + assert dp.estimate_natural_duration(" ", "en", dp.Calibration(10.0, 3)) == 0.0 + + +# ── Calibration ───────────────────────────────────────────────────────── + + +def test_calibrate_cps_median_of_sample_rates(): + # Rates 10, 15, 20 cps → median 15; the outlier-resistant middle wins. + calib = dp.calibrate_cps([(20, 2.0), (30, 2.0), (40, 2.0)]) + assert calib is not None + assert calib.cps == pytest.approx(15.0) + assert calib.samples == 3 + + +def test_calibrate_cps_even_count_averages_middle_pair(): + calib = dp.calibrate_cps([(10, 1.0), (20, 1.0), (30, 1.0), (40, 1.0)]) + assert calib.cps == pytest.approx(25.0) + + +def test_calibrate_cps_requires_min_samples(): + assert dp.calibrate_cps([(30, 2.0), (30, 2.0)]) is None + assert dp.calibrate_cps([]) is None + + +def test_calibrate_cps_filters_garbage_samples(): + # Sub-0.4s durations and tiny texts are TTS ramp/silence noise, not + # speech rate — after filtering only 2 usable samples remain → None. + assert dp.calibrate_cps([ + (30, 2.0), (30, 2.0), # usable + (100, 0.1), (2, 5.0), # garbage: too short / too few chars + ("x", 1.0), (None, None), # garbage: non-numeric + ]) is None + # With a third usable sample the garbage no longer blocks calibration. + calib = dp.calibrate_cps([(30, 2.0), (30, 2.0), (30, 2.0), (100, 0.1)]) + assert calib is not None and calib.cps == pytest.approx(15.0) + + +def test_calibration_from_job_reads_generate_records(): + job = {"seg_natural_durs_by_lang": {"es": { + "a": {"chars": 20, "dur": 2.0}, + "b": {"chars": 30, "dur": 2.0}, + "c": {"chars": 40, "dur": 2.0}, + "legacy-junk": "not-a-dict", # tolerated, skipped + }}} + calib = dp.calibration_from_job(job, "es") + assert calib is not None and calib.cps == pytest.approx(15.0) + # Other language / missing map / legacy job → None (static fallback). + assert dp.calibration_from_job(job, "fr") is None + assert dp.calibration_from_job({}, "es") is None + + +# ── Classifier thresholds (aligned with FitParams caps) ───────────────── + + +def _seg(i, start, end, text): + return {"id": f"s{i}", "start": start, "end": end, "text": text} + + +def _classify_one(text, slot, **kw): + return dp.classify_segments([_seg(0, 0.0, slot, text)], "en", **kw)[0] + + +def test_fits_when_need_within_audio_only_cap(): + # est 2.4s over 2.0s slot → need 1.2 == max_audio_only_rate → still fits + # (fit_planner absorbs it with an imperceptible audio-only speed-up). + v = _classify_one("A" * 36, 2.0) + assert v["status"] == "fits" + assert v["est_dur_s"] == pytest.approx(2.4) + # Overrun is still reported truthfully even for a "fits" verdict. + assert v["est_overrun_s"] == pytest.approx(0.4) + + +def test_tight_between_audio_only_and_absorb_cap(): + # need = est/slot: just above 1.2 → tight; at the hybrid absorb cap + # (audio 1.5 × video 2.0 = 3.0) → still tight (planner absorbs, no trim). + assert _classify_one("A" * 39, 2.0)["status"] == "tight" # need 1.3 + assert _classify_one("A" * 90, 2.0)["status"] == "tight" # need 3.0 + + +def test_impossible_beyond_absorb_cap_with_overrun(): + # need 4.0 > 3.0 → even hybrid caps can't absorb it: fit_planner trims. + v = _classify_one("A" * 120, 2.0) # est 8.0s vs 2.0s available + assert v["status"] == "impossible" + assert v["est_overrun_s"] == pytest.approx(6.0) + + +def test_audio_only_mode_caps_at_legacy_hard_ceiling(): + params = FitParams(allow_video_retime=False) + # need 2.0: hybrid would absorb it, audio-only (hard cap 1.8) cannot. + text = "A" * 60 # est 4.0s over 2.0s + assert _classify_one(text, 2.0)["status"] == "tight" + assert _classify_one(text, 2.0, fit_params=params)["status"] == "impossible" + assert MAX_AUDIO_RATE_HARD == pytest.approx(1.8) + + +def test_calibration_changes_the_verdict(): + # 45 chars over 2.0s: static en (15 cps) → est 3.0, need 1.5 → tight. + # A fast calibrated voice (30 cps) → est 1.5 → fits. + text = "A" * 45 + assert _classify_one(text, 2.0)["status"] == "tight" + v = _classify_one(text, 2.0, calibration=dp.Calibration(cps=30.0, samples=4)) + assert v["status"] == "fits" + assert v["calibrated"] is True + + +def test_empty_text_and_zero_slot_edge_cases(): + assert _classify_one("", 2.0)["status"] == "fits" + # Speech but literally no available time → impossible, overrun = est. + v = dp.classify_segments([_seg(0, 1.0, 1.0, "A" * 30)], "en")[0] + assert v["status"] == "impossible" + assert v["est_overrun_s"] == pytest.approx(2.0) + + +# ── Gap borrowing ─────────────────────────────────────────────────────── + + +def test_gap_borrow_extends_available_time(): + # est 3.0s over a 2.0s slot (need 1.5 → tight)… but a 1.05s gap to the + # next segment lends 1.0s (gap − guard) → available 3.0 → need 1.0 → fits. + segs = [_seg(0, 0.0, 2.0, "A" * 45), _seg(1, 3.05, 4.0, "hi")] + v = dp.classify_segments(segs, "en") + assert v[0]["status"] == "fits" + assert v[0]["available_s"] == pytest.approx(3.0) + # Without the gap the same segment is tight. + assert _classify_one("A" * 45, 2.0)["status"] == "tight" + + +def test_gap_borrow_is_capped(): + # A 60s gap must not promise 60s of slack: borrow caps at GAP_BORROW_MAX_S. + segs = [_seg(0, 0.0, 2.0, "A" * 45), _seg(1, 62.0, 63.0, "hi")] + v = dp.classify_segments(segs, "en")[0] + assert v["available_s"] == pytest.approx(2.0 + dp.GAP_BORROW_MAX_S) + + +def test_last_segment_borrows_capped_tail(): + v = dp.classify_segments([_seg(0, 0.0, 2.0, "A" * 45)], "en", total_dur_s=2.5)[0] + assert v["available_s"] == pytest.approx(2.5) # tail 0.5s, under the cap + v = dp.classify_segments([_seg(0, 0.0, 2.0, "A" * 45)], "en", total_dur_s=60.0)[0] + assert v["available_s"] == pytest.approx(2.0 + dp.GAP_BORROW_MAX_S) + # Unknown video duration → no tail borrow (mirrors plan_fit). + v = dp.classify_segments([_seg(0, 0.0, 2.0, "A" * 45)], "en")[0] + assert v["available_s"] == pytest.approx(2.0) + + +# ── Alignment with fit_planner ────────────────────────────────────────── + + +@pytest.mark.parametrize("chars,expected_status", [ + (30, "fits"), # est 2.0s / 2.95s avail → need <1.2 + (90, "tight"), # est 6.0s → need ~2.03: hybrid absorbs + (140, "impossible"), # est ~9.3s → need >3.0: beyond the caps +]) +def test_verdict_matches_what_fit_planner_would_do(chars, expected_status): + """"impossible" must mean "fit_planner will trim" — feed the classifier's + own estimate to plan_fit as the natural duration and cross-check.""" + segs = [_seg(0, 0.0, 2.0, "A" * chars), _seg(1, 3.0, 4.0, "hi")] + verdict = dp.classify_segments(segs, "en")[0] + assert verdict["status"] == expected_status + + est = dp.estimate_natural_duration("A" * chars, "en") + plan = plan_fit( + [{"id": s["id"], "start": s["start"], "end": s["end"]} for s in segs], + [est, 0.5], + total_dur_s=4.0, + ) + if expected_status == "impossible": + assert plan.segments[0].status == "overflow_trimmed" + assert plan.segments[0].overflow_s > 0 + else: + assert plan.segments[0].status != "overflow_trimmed" + assert plan.segments[0].overflow_s == 0 + + +# ── Condensation ──────────────────────────────────────────────────────── + + +class _FakeLLM: + """Non-Off LLM stand-in returning scripted replies; counts calls.""" + + def __init__(self, replies): + self.replies = list(replies) + self.calls = 0 + self.last_temperature = None + + def chat(self, *, system, user, timeout=None, temperature=None): + self.calls += 1 + self.last_temperature = temperature + if not self.replies: + raise RuntimeError("no more scripted replies") + reply = self.replies.pop(0) + if isinstance(reply, Exception): + raise reply + return reply + + +@pytest.fixture +def fake_llm(monkeypatch): + def _install(replies): + llm = _FakeLLM(replies) + monkeypatch.setattr(dp, "get_active_llm_backend", lambda: llm) + return llm + return _install + + +def test_condense_no_llm_is_noop(monkeypatch): + monkeypatch.setenv("OMNIVOICE_LLM_BACKEND", "off") + text = "A" * 90 + res = dp.condense_for_slot(text, available_s=2.0, target_lang="en") + assert res["applied"] is False + assert res["text"] == text + assert res["error"] == "no-llm" + + +def test_condense_already_fitting_never_calls_llm(fake_llm): + llm = fake_llm(["SHOULD-NOT-BE-CALLED"]) + text = "A" * 15 # est 1.0s, well inside 2.0s + res = dp.condense_for_slot(text, available_s=2.0, target_lang="en") + assert res["applied"] is False + assert res["error"] == "already-fits" + assert llm.calls == 0 + + +def test_condense_accepts_shorter_rewrite(fake_llm): + llm = fake_llm(["B" * 28]) # est ~1.87s ≤ 2.0s available + text = "A" * 60 # est 4.0s + res = dp.condense_for_slot(text, available_s=2.0, target_lang="en") + assert res["applied"] is True + assert res["text"] == "B" * 28 + assert res["est_dur_s"] == pytest.approx(28 / 15.0, abs=0.01) + assert llm.calls == 1 + assert llm.last_temperature == 0.2 # pinned, like the Autofit pass + + +def test_condense_retries_then_keeps_best(fake_llm): + # First reply is shorter but still overruns; second fits — second wins. + fake_llm(["B" * 45, "C" * 25]) + res = dp.condense_for_slot("A" * 60, available_s=2.0, target_lang="en") + assert res["applied"] is True + assert res["text"] == "C" * 25 + + +def test_condense_llm_failure_is_noop(fake_llm): + fake_llm([RuntimeError("provider down")]) + text = "A" * 60 + res = dp.condense_for_slot(text, available_s=2.0, target_lang="en") + assert res["applied"] is False + assert res["text"] == text + assert res["error"] == "condense-failed" + + +def test_condense_divergent_reply_rejected(fake_llm): + # 6 chars from a 100-char line → length-ratio 0.06 < the divergence + # guard's floor: the "rewrite" nuked the meaning, so it's discarded. + llm = fake_llm(["short!", "gone!!"]) + text = "A" * 100 + res = dp.condense_for_slot(text, available_s=1.0, target_lang="en") + assert res["applied"] is False + assert res["text"] == text + assert res["error"] == "condense-failed" + assert llm.calls == 2 # both attempts burned, none accepted + + +def test_condense_longer_reply_is_useless(fake_llm): + # A reply that isn't actually shorter can't be suggested. + fake_llm(["A" * 80, "A" * 70]) + res = dp.condense_for_slot("A" * 60, available_s=2.0, target_lang="en") + assert res["applied"] is False + assert res["error"] == "condense-failed"