feat(dub): predict segment fit before synthesis — tight/impossible badges + opt-in shorter rewrites (#1051)
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 <test@local> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Palash Debnath
co-authored by
mergetest
Claude Fable 5
mergetest
Claude Fable 5
parent
7dbb95fa15
commit
5ce9d0e51d
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)}
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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;
|
||||
};
|
||||
}[];
|
||||
}
|
||||
|
||||
|
||||
@@ -275,6 +275,42 @@ function DubSegmentRow({
|
||||
📖 {seg.rate_ratio.toFixed(2)}×
|
||||
</span>
|
||||
)}
|
||||
{/* 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') && (
|
||||
<span
|
||||
className="text-[0.48rem] mt-[1px] inline-flex items-center gap-[1px]"
|
||||
style={{ color: seg.plan.status === 'impossible' ? '#fb4934' : '#fabd2f' }}
|
||||
title={t(
|
||||
seg.plan.status === 'impossible'
|
||||
? 'segment.plan_impossible_title'
|
||||
: 'segment.plan_tight_title',
|
||||
{
|
||||
est: (seg.plan.est_dur_s || 0).toFixed(1),
|
||||
avail: (seg.plan.available_s || 0).toFixed(1),
|
||||
seconds: (seg.plan.est_overrun_s || 0).toFixed(1),
|
||||
},
|
||||
)}
|
||||
>
|
||||
<AlertCircle size={8} />{' '}
|
||||
{seg.plan.status === 'impossible'
|
||||
? t('segment.plan_impossible', {
|
||||
seconds: (seg.plan.est_overrun_s || 0).toFixed(1),
|
||||
})
|
||||
: t('segment.plan_tight')}
|
||||
</span>
|
||||
)}
|
||||
{seg.plan && seg.plan.suggested_text && seg.plan.suggested_text !== seg.text && (
|
||||
<button
|
||||
onClick={() => onEditField(seg.id, 'text', seg.plan.suggested_text)}
|
||||
disabled={disabled}
|
||||
title={t('segment.plan_apply_title', { text: seg.plan.suggested_text })}
|
||||
className="bg-transparent border-none text-[#83a598] cursor-pointer p-0 mt-[1px] text-[0.48rem] text-left"
|
||||
>
|
||||
✂ {t('segment.plan_apply')}
|
||||
</button>
|
||||
)}
|
||||
</span>
|
||||
|
||||
<input
|
||||
|
||||
@@ -115,6 +115,10 @@ export default function DubLeftColumn({
|
||||
const setAutoGlossary = useAppStore((s) => 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. */}
|
||||
<label
|
||||
className="flex items-center gap-[4px] mt-[3px] text-[0.55rem] text-fg-muted cursor-pointer select-none"
|
||||
title={t('dub.condense_title')}
|
||||
>
|
||||
<input
|
||||
type="checkbox"
|
||||
checked={condenseSuggest}
|
||||
onChange={(e) => setCondenseSuggest(e.target.checked)}
|
||||
className="cursor-pointer"
|
||||
/>
|
||||
{t('dub.condense_label')}
|
||||
</label>
|
||||
</div>
|
||||
{/* LLM engine only: auto-glossary + reflect pass. Both default ON;
|
||||
the reflect tooltip is explicit that it multiplies LLM calls. */}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -85,6 +85,7 @@ export const useAppStore = create<AppStore>()(
|
||||
translateQuality: s.translateQuality,
|
||||
autoGlossary: s.autoGlossary,
|
||||
reflectPass: s.reflectPass,
|
||||
condenseSuggest: s.condenseSuggest,
|
||||
dualSubs: s.dualSubs,
|
||||
burnSubs: s.burnSubs,
|
||||
glossaryVisible: s.glossaryVisible,
|
||||
|
||||
@@ -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<PrefsSlice, [], [], PrefsSlice> = (s
|
||||
translateQuality: 'fast',
|
||||
autoGlossary: true,
|
||||
reflectPass: true,
|
||||
condenseSuggest: false,
|
||||
dualSubs: false,
|
||||
burnSubs: false,
|
||||
glossaryVisible: true,
|
||||
@@ -243,6 +252,7 @@ export const createPrefsSlice: StateCreator<PrefsSlice, [], [], PrefsSlice> = (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 }),
|
||||
|
||||
@@ -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(
|
||||
<DubSegmentRow
|
||||
{...makeProps({
|
||||
status: 'impossible',
|
||||
est_dur_s: 6.9,
|
||||
available_s: 2.0,
|
||||
est_overrun_s: 4.9,
|
||||
calibrated: false,
|
||||
})}
|
||||
/>,
|
||||
);
|
||||
expect(screen.getByText(/Won't fit \+4\.9s/)).toBeTruthy();
|
||||
});
|
||||
|
||||
it('flags a tight segment', () => {
|
||||
render(
|
||||
<DubSegmentRow
|
||||
{...makeProps({
|
||||
status: 'tight',
|
||||
est_dur_s: 3.0,
|
||||
available_s: 2.0,
|
||||
est_overrun_s: 1.0,
|
||||
calibrated: true,
|
||||
})}
|
||||
/>,
|
||||
);
|
||||
expect(screen.getByText(/Tight fit/)).toBeTruthy();
|
||||
});
|
||||
|
||||
it('stays silent when the plan says fits', () => {
|
||||
render(
|
||||
<DubSegmentRow
|
||||
{...makeProps({
|
||||
status: 'fits',
|
||||
est_dur_s: 1.0,
|
||||
available_s: 2.0,
|
||||
est_overrun_s: 0,
|
||||
calibrated: false,
|
||||
})}
|
||||
/>,
|
||||
);
|
||||
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(<DubSegmentRow {...props} />);
|
||||
// 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');
|
||||
});
|
||||
});
|
||||
@@ -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]<text>` 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 '<seconds>:'."""
|
||||
|
||||
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
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user