| """ |
| Scene window construction from a video duration and its VTT subtitles. |
| |
| A *scene window* is a fixed-length slice of the timeline (default 30s with a |
| 15s stride) plus the subtitle text that overlaps it. Windows are the unit of |
| work for the VLM: each one yields exactly one scene description and one |
| embedding. |
| |
| We use fixed sliding windows rather than shot boundaries (TransNetV2) because |
| narrative arcs — *disagreement then forgiveness* in the same conversation — |
| usually span multiple shots. Shot-aligned windows would slice the arc into |
| pieces and lose the cross-shot context that makes the description useful. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field |
| from typing import Sequence |
|
|
| DEFAULT_WINDOW_SECONDS = 30.0 |
| DEFAULT_STRIDE_SECONDS = 15.0 |
|
|
|
|
| @dataclass |
| class SceneWindow: |
| """A single window of the timeline with its subtitle text.""" |
|
|
| index: int |
| start_seconds: float |
| end_seconds: float |
| subtitle_text: str |
| subtitle_entries: list[dict] = field(default_factory=list) |
|
|
| @property |
| def duration_seconds(self) -> float: |
| return self.end_seconds - self.start_seconds |
|
|
| @property |
| def label(self) -> str: |
| """Stable identifier safe to use as a filename component.""" |
| return f"window-{self.index:04d}-start-{self.start_seconds:07.1f}s" |
|
|
| def to_dict(self) -> dict: |
| return { |
| "index": self.index, |
| "start_seconds": round(self.start_seconds, 3), |
| "end_seconds": round(self.end_seconds, 3), |
| "duration_seconds": round(self.duration_seconds, 3), |
| "subtitle_text": self.subtitle_text, |
| "subtitle_entry_count": len(self.subtitle_entries), |
| } |
|
|
|
|
| def build_scene_windows( |
| duration_seconds: float, |
| subtitle_entries: Sequence[dict], |
| *, |
| window_seconds: float = DEFAULT_WINDOW_SECONDS, |
| stride_seconds: float = DEFAULT_STRIDE_SECONDS, |
| ) -> list[SceneWindow]: |
| """Slice a video into overlapping fixed-length scene windows. |
| |
| Args: |
| duration_seconds: Total duration of the video in seconds. |
| subtitle_entries: Parsed VTT entries (see ``subtitles_process_vtt.parse_vtt``). |
| Each entry must have ``start``, ``end``, and ``text`` keys in seconds. |
| window_seconds: Length of each window. 30s is a good default for |
| narrative arcs; shorter values fragment them, longer values |
| blur emotional shifts together. |
| stride_seconds: Distance between window starts. With the default |
| (15s stride, 30s window) every moment of the video is covered |
| by two windows, giving the VLM a chance to catch arcs that |
| straddle a window boundary. |
| |
| Returns: |
| Ordered list of ``SceneWindow``. The final window is clipped to |
| ``duration_seconds``. A video shorter than one window still |
| produces exactly one window covering the whole video. |
| """ |
| if window_seconds <= 0: |
| raise ValueError(f"window_seconds must be > 0, got {window_seconds}") |
| if stride_seconds <= 0: |
| raise ValueError(f"stride_seconds must be > 0, got {stride_seconds}") |
| if duration_seconds <= 0: |
| raise ValueError(f"duration_seconds must be > 0, got {duration_seconds}") |
|
|
| windows: list[SceneWindow] = [] |
| start = 0.0 |
| index = 0 |
| while start < duration_seconds: |
| end = min(start + window_seconds, duration_seconds) |
| overlapping = _entries_overlapping(subtitle_entries, start, end) |
| text = _join_entries(overlapping) |
| windows.append( |
| SceneWindow( |
| index=index, |
| start_seconds=start, |
| end_seconds=end, |
| subtitle_text=text, |
| subtitle_entries=list(overlapping), |
| ) |
| ) |
| index += 1 |
| next_start = start + stride_seconds |
| if next_start >= duration_seconds: |
| break |
| start = next_start |
|
|
| return windows |
|
|
|
|
| def _entries_overlapping( |
| entries: Sequence[dict], start: float, end: float |
| ) -> list[dict]: |
| """Return entries whose [start, end) interval intersects [start, end).""" |
| matched: list[dict] = [] |
| for entry in entries: |
| entry_start = float(entry.get("start", 0.0)) |
| entry_end = float(entry.get("end", entry_start)) |
| if entry_end <= start: |
| continue |
| if entry_start >= end: |
| continue |
| matched.append(entry) |
| return matched |
|
|
|
|
| def _join_entries(entries: Sequence[dict]) -> str: |
| """Join the entries' text into a single readable paragraph.""" |
| pieces = [] |
| for entry in entries: |
| text = (entry.get("text") or "").strip() |
| if text: |
| pieces.append(text) |
| return " ".join(pieces) |
|
|
