160 lines
6.3 KiB
Python
160 lines
6.3 KiB
Python
"""Deterministic narration scheduling for render-time segment concatenation.
|
|
|
|
Rather than pre-concatenating segments into one file, the render stage
|
|
concatenates the processed segments directly. From narration.json + the cached
|
|
per-segment transcripts this module computes two things:
|
|
|
|
1. an ordered segment schedule (processed file, skip, take, and offset in the
|
|
combined timeline) — this drives the ffmpeg concat at render time; and
|
|
2. the merged word-level transcript, with every word re-timed into the
|
|
combined timeline — this drives slide alignment, exactly what
|
|
re-transcribing a pre-concatenated narration file used to produce, but derived
|
|
deterministically (no re-transcription, no separate combined file).
|
|
|
|
The processed files share framerate and format and are uncompressed, so the
|
|
computed offsets match the concatenated audio timeline sample-for-sample.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
|
|
from .models import VideoSource
|
|
from .preprocessor import get_preprocessed_path
|
|
from .transcriber import TranscribedWord, load_transcript
|
|
|
|
|
|
@dataclass
|
|
class NarrationSegment:
|
|
"""One narration segment placed on the combined render timeline."""
|
|
|
|
seg_id: str
|
|
source_path: Path # processed file to concatenate
|
|
skip: float # seconds trimmed from the segment's start
|
|
take: Optional[float] # seconds kept from `skip` (None → to end)
|
|
duration: float # effective seconds contributed to the timeline
|
|
offset: float # start time of this segment in the combined timeline
|
|
|
|
|
|
def segment_order(narration: dict) -> list[str]:
|
|
"""Natural sort of segment ids (S2 before S10, s1-15 before s16-end)."""
|
|
return sorted(
|
|
narration.keys(),
|
|
key=lambda s: [int(t) if t.isdigit() else t.lower() for t in re.split(r"(\d+)", s)],
|
|
)
|
|
|
|
|
|
def build_narration_schedule(
|
|
narration: dict[str, VideoSource],
|
|
narration_dir: Path,
|
|
get_duration: Callable[[Path], float],
|
|
transcripts_dir: Optional[Path] = None,
|
|
verbose: bool = False,
|
|
) -> tuple[list[NarrationSegment], list[TranscribedWord]]:
|
|
"""Return (ordered segments with offsets, merged transcript in timeline).
|
|
|
|
Args:
|
|
narration: seg_id -> VideoSource (from parse_narration).
|
|
narration_dir: base dir the processed files resolve against
|
|
(media/narration; output_file is like processed/…mov).
|
|
get_duration: callable(Path) -> float (ffprobe duration), used only
|
|
when a segment has no explicit take.
|
|
transcripts_dir: where per-segment {seg_id}.json transcripts live
|
|
(defaults to narration_dir/transcripts).
|
|
"""
|
|
if transcripts_dir is None:
|
|
transcripts_dir = narration_dir / "transcripts"
|
|
|
|
segments: list[NarrationSegment] = []
|
|
merged: list[TranscribedWord] = []
|
|
offset = 0.0
|
|
|
|
for seg_id in segment_order(narration):
|
|
vs = narration[seg_id]
|
|
source_path = get_preprocessed_path(narration_dir, vs)
|
|
|
|
skip = vs.skip or 0.0
|
|
take = vs.take
|
|
if take is not None:
|
|
eff = max(0.0, take)
|
|
else:
|
|
full_dur = get_duration(source_path) if source_path.exists() else 0.0
|
|
eff = max(0.0, full_dur - skip)
|
|
seg_end = skip + eff # kept window end in the segment's own timeline
|
|
|
|
segments.append(
|
|
NarrationSegment(
|
|
seg_id=seg_id,
|
|
source_path=source_path,
|
|
skip=skip,
|
|
take=take,
|
|
duration=eff,
|
|
offset=offset,
|
|
)
|
|
)
|
|
|
|
# Re-time this segment's transcript into the combined timeline: keep only
|
|
# words inside [skip, seg_end], subtract skip, and add the running offset.
|
|
tpath = transcripts_dir / f"{seg_id}.json"
|
|
if tpath.exists():
|
|
kept = 0
|
|
for w in load_transcript(tpath):
|
|
if w.end <= skip or w.start >= seg_end:
|
|
continue # entirely outside the kept window
|
|
new_start = max(w.start, skip) - skip + offset
|
|
new_end = min(w.end, seg_end) - skip + offset
|
|
merged.append(
|
|
TranscribedWord(word=w.word, start=round(new_start, 3), end=round(new_end, 3))
|
|
)
|
|
kept += 1
|
|
if verbose:
|
|
print(f" {seg_id}: +{kept} words (skip={skip:.2f}s take={eff:.2f}s → offset {offset:.2f}s)")
|
|
elif verbose:
|
|
print(f" ⚠ {seg_id}: no transcript at {tpath} — words missing from merged transcript")
|
|
|
|
offset += eff
|
|
|
|
return segments, merged
|
|
|
|
|
|
def slice_schedule(
|
|
schedule: list[NarrationSegment],
|
|
window_start: float,
|
|
window_end: float,
|
|
) -> list[NarrationSegment]:
|
|
"""Return the sub-schedule covering ``[window_start, window_end]`` of the
|
|
combined narration timeline — for a partial (chunked) render.
|
|
|
|
Each kept segment's ``skip``/``take``/``offset`` is adjusted so it seeks within
|
|
its OWN file: segments entirely outside the window are dropped, the first/last
|
|
kept segments are trimmed to the window edges, and offsets are re-based so the
|
|
sliced narration starts at 0. ``input_seek_time`` therefore stays 0 in concat
|
|
mode instead of a combined-timeline offset being (wrongly) applied to every
|
|
segment file.
|
|
|
|
This is what makes chunking bulletproof: because each chunk seeks its first
|
|
segment to the true source sample, ``render(A:B) ++ render(B:C)`` lands on the
|
|
exact same narration as ``render(A:C)``. Slicing to the full ``[0, total]`` is a
|
|
no-op, so full renders are unaffected.
|
|
"""
|
|
import copy
|
|
|
|
out: list[NarrationSegment] = []
|
|
for seg in schedule:
|
|
seg_start = seg.offset
|
|
seg_end = seg.offset + seg.duration
|
|
keep_start = max(window_start, seg_start)
|
|
keep_end = min(window_end, seg_end)
|
|
if keep_end - keep_start <= 1e-6:
|
|
continue # segment lies entirely outside the window
|
|
new = copy.copy(seg)
|
|
new.skip = round(seg.skip + (keep_start - seg_start), 6)
|
|
new.take = round(keep_end - keep_start, 6)
|
|
new.duration = new.take
|
|
new.offset = round(keep_start - window_start, 6)
|
|
out.append(new)
|
|
return out
|