#!/usr/bin/env python3 """Plan-level validation for chunking v2 (docs/chunking_v2.md). Verifies the transformer now INCLUDES clips that span a chunk boundary and seeks into them (skip_override / src_offset), instead of the v1 behaviour that dropped them. This is a pure plan-level check — the ffmpeg concat-seam (frame alignment via -c copy) still needs a real render on the rig to confirm. Run: ./venv/bin/python tests/test_chunking_v2.py """ import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from gnommo.transformer import ( MarkerTiming, _extract_audio_events, _extract_video_events, AUDIO_OFFSET_SECONDS, ) from gnommo.models import AudioDefinition, VideoSource, CutoutDefinition, SlideDefinition _fails = [] def check(name, cond, detail=""): print(f" {'PASS' if cond else 'FAIL'} {name}" + (f" — {detail}" if detail and not cond else "")) if not cond: _fails.append(name) # ── audio ──────────────────────────────────────────────────────────────────── def test_audio(): print("audio events:") audio = { "music": AudioDefinition(file="music.mp3", loop=True, duration=90.0), "sfx": AudioDefinition(file="sfx.wav", loop=False, duration=500.0), "blip": AudioDefinition(file="blip.wav", loop=False, duration=50.0), } # music triggers at t=0, sfx at t=10, blip at t=10 markers = [ MarkerTiming(marker_id="Amusic", timestamp=0.0, context="", confidence=1.0), MarkerTiming(marker_id="Asfx", timestamp=10.0, context="", confidence=1.0), MarkerTiming(marker_id="Ablip", timestamp=10.0, context="", confidence=1.0), ] # Chunk window [300, 600): all three started earlier. evs = {e.audio_id: e for e in _extract_audio_events(markers, audio, time_range=(300.0, 600.0))} # Looping music: included, resumes at loop phase. astart = max(0, 0-1)=0; into=300; # phase = 300 % 90 = 30. check("looping music spanning boundary is INCLUDED (v1 dropped it)", "music" in evs) if "music" in evs: m = evs["music"] check("music clamped to window start", abs(m.start_time - 300.0) < 1e-6, f"start={m.start_time}") check("music seeks to loop phase 30.0", abs(m.src_offset - 30.0) < 1e-6, f"src_offset={m.src_offset}") # One-shot still playing at the window: included, linear seek. # astart = max(0,10-1)=9; aend=9+500=509 > 300 → spans. into=300-9=291. check("one-shot still playing is INCLUDED", "sfx" in evs) if "sfx" in evs: s = evs["sfx"] check("sfx linear seek 291.0", abs(s.src_offset - 291.0) < 1e-6, f"src_offset={s.src_offset}") # One-shot that ended before the window: excluded (aend=9+50=59 < 300). check("one-shot ended before window is EXCLUDED", "blip" not in evs) # Full render (no range): everything from the start, no seek. full = {e.audio_id: e for e in _extract_audio_events(markers, audio, time_range=None)} check("full render includes music with no seek", "music" in full and full["music"].src_offset == 0.0) def test_crossfade_phase(): print("crossfade loop phase:") # Looping pad with a 15s crossfade overlap: the crossfade stream repeats every # (duration - overlap) = 60 - 15 = 45s, so a chunk starting at into=300 resumes # at crossfade phase 300 % 45 = 30, while the hard-loop phase is 300 % 60 = 60→0. audio = {"pad": AudioDefinition(file="pad.wav", loop=True, duration=60.0, overlap=15.0)} markers = [MarkerTiming(marker_id="Apad", timestamp=0.0, context="", confidence=1.0)] evs = {e.audio_id: e for e in _extract_audio_events(markers, audio, time_range=(300.0, 600.0))} check("looping pad with overlap is included", "pad" in evs) if "pad" in evs: p = evs["pad"] check("crossfade seeks to loop_len phase 30.0", abs(p.crossfade_offset - 30.0) < 1e-6, f"crossfade_offset={p.crossfade_offset}") check("src_offset still uses full-duration phase 0.0", abs(p.src_offset - 0.0) < 1e-6, f"src_offset={p.src_offset}") # Full render: no phase seek on either. full = {e.audio_id: e for e in _extract_audio_events(markers, audio, time_range=None)} check("full render pad has no crossfade seek", "pad" in full and full["pad"].crossfade_offset == 0.0) # ── video ──────────────────────────────────────────────────────────────────── def test_video(): print("video events:") cutouts = {"fullscreen": CutoutDefinition(x=0, y=0, height=1080, width=1920)} videos = { "bg": VideoSource( source_file="bg.mp4", cutout="fullscreen", layer="below", duration=90.0, skip=0.0, end_on="next_video", ) } slides = {f"S{i}": SlideDefinition(image=f"S{i}.png", type="slide") for i in range(1, 11)} markers = [MarkerTiming(marker_id=f"S{i}", timestamp=(i - 1) * 60.0, context="", confidence=1.0) for i in range(1, 11)] # Background overlay starts at slide-7 time (360) and, as the only video with # end_on next_video, runs to total_duration (600). markers.append(MarkerTiming(marker_id="vfm:bg", timestamp=360.0, context="", confidence=1.0)) total = 600.0 # Chunk window that STARTS AFTER the video began: [420, 600) (slides 8-10). evs, _ = _extract_video_events(markers, videos, cutouts, slides, total, time_range=(420.0, 600.0)) bg = next((e for e in evs if e.video_id == "bg"), None) check("spanning background video is INCLUDED in the later chunk (v1 dropped it)", bg is not None) if bg is not None: check("bg clamped to window start", abs(bg.start_time - 420.0) < 1e-6, f"start={bg.start_time}") # into = 420-360 = 60; playable = 90-0 = 90; 60 < 90 → linear seek 60. check("bg linear seek 60.0 (first play-through)", abs((bg.skip_override or 0) - 60.0) < 1e-6, f"skip_override={bg.skip_override}") # Window starting deep enough that the 90s clip has looped once: [480, 600). # into = 480-360 = 120; 120 >= 90 → phase = 120 % 90 = 30. evs2, _ = _extract_video_events(markers, videos, cutouts, slides, total, time_range=(480.0, 600.0)) bg2 = next((e for e in evs2 if e.video_id == "bg"), None) check("looped background resumes at phase 30.0", bg2 is not None and abs((bg2.skip_override or 0) - 30.0) < 1e-6, f"skip_override={getattr(bg2, 'skip_override', None)}") # Full render: no seek. evs3, _ = _extract_video_events(markers, videos, cutouts, slides, total, time_range=None) bg3 = next((e for e in evs3 if e.video_id == "bg"), None) check("full render includes bg with no seek", bg3 is not None and bg3.skip_override is None) def test_pause_cutscene_chunk_ownership(): # A pause_narration cutscene must live WHOLLY in the chunk where it starts: its # end = start + pause_narration overshoots the pre-pause timeline, so it must not be # truncated by range_end nor duplicated into the next chunk (the video6 double-freeze # / one-frame-logo bug). print("pause cutscene chunk ownership:") cutouts = {"fullscreen": CutoutDefinition(x=0, y=0, height=1080, width=1920)} videos = { "logo": VideoSource(source_file="logo.mov", cutout="fullscreen", layer="above", duration=18.0, pause_narration=18.0), } slides = {f"S{i}": SlideDefinition(image=f"S{i}.png", type="slide") for i in range(1, 6)} markers = [MarkerTiming(marker_id=f"S{i}", timestamp=(i - 1) * 30.0, context="", confidence=1.0) for i in range(1, 6)] # Cutscene triggers at t=100 (between S4=90 and S5=120). Its end = 100+18 = 118. markers.append(MarkerTiming(marker_id="vftp:logo", timestamp=100.0, context="", confidence=1.0)) total = 200.0 # Owning chunk [90, 110): starts inside it. End must NOT be clamped to 110 → full 18s. evs, _ = _extract_video_events(markers, videos, cutouts, slides, total, time_range=(90.0, 110.0)) lg = next((e for e in evs if e.video_id == "logo"), None) check("cutscene INCLUDED in the chunk it starts in", lg is not None) if lg is not None: check("cutscene end NOT clamped to range_end (full pause length)", abs((lg.end_time - lg.start_time) - 18.0) < 1e-6, f"duration={lg.end_time - lg.start_time}") # Next chunk [110, 200): the cutscene started earlier → must be EXCLUDED (no dupe freeze). evs2, _ = _extract_video_events(markers, videos, cutouts, slides, total, time_range=(110.0, 200.0)) check("cutscene EXCLUDED from the next chunk (no duplicate freeze)", not any(e.video_id == "logo" for e in evs2)) if __name__ == "__main__": test_audio() test_crossfade_phase() test_video() test_pause_cutscene_chunk_ownership() print() if _fails: print(f"FAILED: {len(_fails)} check(s): {', '.join(_fails)}") sys.exit(1) print("All chunking-v2 plan-level checks passed.")