diff --git a/docs/FunPackLTXAVSceneChainSampler.md b/docs/FunPackLTXAVSceneChainSampler.md index f85d1157..c7820fbe 100644 --- a/docs/FunPackLTXAVSceneChainSampler.md +++ b/docs/FunPackLTXAVSceneChainSampler.md @@ -34,7 +34,7 @@ Important: this sampler is resource heavy. Long chains can produce very large fi ## Outputs -`latent`, `images`, `status`, `scene_count`, `scene_report`, `scene_boundaries`. The `scene_report` and `scene_boundaries` describe how the chain was split and stitched; the boundaries can be used downstream to locate scene cuts in the decoded video. +`latent`, `images`, `status`, `scene_count`, `scene_report`, `scene_boundaries`. `scene_report` is a per-scene text/seed log. `scene_boundaries` is JSON with overlap diagnostics: per-scene pixel ranges, boundary seam positions, `contamination_zones` (where latent overlap and transition effects can mix scene N+1 into scene N's tail), per-scene mechanisms used, and `global_steering` notes (embed guidance / Studio absolute steer affect **every** scene, not just the seam). ## Behavior diff --git a/movie_editor/tests/conftest.py b/movie_editor/tests/conftest.py index e284beae..b4f752b0 100644 --- a/movie_editor/tests/conftest.py +++ b/movie_editor/tests/conftest.py @@ -5,9 +5,21 @@ ComfyUI/torch stack; without this guard pytest climbs into it during collection fails on heavy imports that aren't needed for these light, dependency-free tests. """ import sys +import types from pathlib import Path # Make `movie_editor` importable when pytest is run from the repo root. ROOT = Path(__file__).resolve().parents[2] +MOVIE_EDITOR = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) + +# Pytest package setup imports movie_editor as `.movie_editor` and would +# execute the parent custom-node __init__ (torch/comfy/folder_paths). Stub the +# parent package so only the light movie_editor __init__ runs. +_parent_name = ROOT.name +if _parent_name not in sys.modules: + _parent = types.ModuleType(_parent_name) + _parent.__path__ = [str(ROOT)] + _parent.__file__ = str(ROOT / "__init__.py") + sys.modules[_parent_name] = _parent diff --git a/movie_editor/tests/test_builder.py b/movie_editor/tests/test_builder.py index 1985aded..6fdf4b3e 100644 --- a/movie_editor/tests/test_builder.py +++ b/movie_editor/tests/test_builder.py @@ -81,7 +81,7 @@ def test_core_skeleton_links_and_params(): assert graph["fps"]["inputs"]["value"] == 24 assert graph["sampler"]["inputs"]["seed"] == 42 # with no slots, the open ports are unsatisfied - assert any("studio.model" in u for u in report["unsatisfied"]) + assert any("FunPackStudio" in u and ".model" in u for u in report["unsatisfied"]) def test_explicit_wires_and_autowire(): @@ -171,7 +171,7 @@ def test_ambiguous_type_is_reported_not_guessed(): ]} graph, report = builder.build(OI, models, PARAMS) assert not isinstance(graph["sampler"]["inputs"].get("vae"), list) # left unwired - assert any("sampler.vae" in a for a in report["ambiguous"]) + assert any("FunPackLTXAVSceneChainSampler" in a and ".vae" in a for a in report["ambiguous"]) def test_extract_widgets_handles_control_and_links(): diff --git a/movie_editor/tests/test_nodes.py b/movie_editor/tests/test_nodes.py index 48b159e0..7de08512 100644 --- a/movie_editor/tests/test_nodes.py +++ b/movie_editor/tests/test_nodes.py @@ -71,7 +71,13 @@ def test_ports_from_input_types_derives_connection_points(): } ports = nodes.ports_from_input_types("Studio", "FunPackStudio", it) by_input = {p["input"]: p["type"] for p in ports} - assert by_input == {"model": "MODEL", "clip": "CLIP", "source_image": "IMAGE", "latent": "LATENT"} + assert by_input == { + "model": "MODEL", + "clip": "CLIP", + "source_image": "IMAGE", + "latent": "LATENT", + "positive_prompt": "STRING", + } assert all(p["id"].startswith("FunPackStudio.") for p in ports) # STRING widgets excluded diff --git a/movie_editor/tests/test_timeline.py b/movie_editor/tests/test_timeline.py index 1141b6b6..fef1a646 100644 --- a/movie_editor/tests/test_timeline.py +++ b/movie_editor/tests/test_timeline.py @@ -40,9 +40,9 @@ def test_anchor_plus_two_scenes_emits_separators(): {"text": "landing on a rooftop"}, ], ) - prompt = build_combined_prompt(p) + prompt = build_combined_prompt(p, for_generation=True) # anchor first, intro marker before scene 0, scene marker before scene 1 - assert prompt.splitlines()[0] == "red-haired heroine" + assert prompt.startswith("red-haired heroine") assert "cut to" in prompt assert "blur" in prompt assert prompt.index("cut to") < prompt.index("flying") @@ -54,10 +54,10 @@ def test_missing_markers_fall_back_to_scene_labels(): anchor="a knight", scenes=[{"text": "in a forest"}, {"text": "at a castle"}], ) - prompt = build_combined_prompt(p) + prompt = build_combined_prompt(p, for_generation=True) # No explicit markers -> generic "scene N" boundaries so the split still separates. + assert "scene 1" in prompt assert "scene 2" in prompt - assert "scene 3" in prompt def test_excluded_scenes_dropped_unless_requested(): @@ -76,7 +76,7 @@ def test_excluded_scenes_dropped_unless_requested(): def test_no_anchor_first_scene_has_no_leading_separator(): p = _project(scenes=[{"text": "opening shot", "transition_to_next": "cut"}, {"text": "second"}]) prompt = build_combined_prompt(p) - assert prompt.splitlines()[0] == "opening shot" + assert prompt == "opening shot second" def test_project_roundtrip_preserves_forward_compat_fields(): diff --git a/samplers.py b/samplers.py index 4afbc4c7..bfd5a8d9 100644 --- a/samplers.py +++ b/samplers.py @@ -2557,6 +2557,163 @@ class FunPackLTXAVSceneChainSampler: return text return f"Scene {index + 1}" + def _scene_meta(self, scene_conditioning, index): + """Extract FunPack scene metadata for overlap / contamination diagnostics.""" + meta = {} + if ( + isinstance(scene_conditioning, (list, tuple)) + and len(scene_conditioning) >= 2 + and isinstance(scene_conditioning[1], dict) + ): + meta = dict(scene_conditioning[1]) + text = str(meta.get("funpack_scene_text") or "").strip() or f"Scene {index + 1}" + encode = str(meta.get("funpack_encode_text") or "").strip() or text + return { + "index": index + 1, + "text": text, + "encode_text": encode, + "seed": meta.get("funpack_scene_seed"), + "transition_effect": meta.get("funpack_transition_effect"), + "temporal_mult": meta.get("funpack_temporal_mult"), + } + + def _latent_to_pixel_frame(self, latent_frame, time_scale): + latent_frame = max(0, int(latent_frame)) + time_scale = max(1, int(time_scale)) + if time_scale > 1: + return int((latent_frame - 1) * time_scale + 1) if latent_frame > 0 else 0 + return latent_frame + + def _scene_pixel_ranges(self, scene_count, video_frames, pixel_overlap): + """Contiguous pixel spans in the stitched output (before transition post-FX).""" + if scene_count <= 0: + return [] + pixel_overlap = max(0, int(pixel_overlap)) + stride = max(1, int(video_frames) - pixel_overlap) + ranges = [] + for i in range(scene_count): + start = i * stride + end = start + int(video_frames) - 1 + ranges.append({"scene": i + 1, "start": start, "end": end}) + return ranges + + def _boundary_contamination_zones(self, boundary_pixel, pixel_overlap, transition_duration, effect): + """Pixel ranges where scene N+1 can visibly influence scene N (and vice versa).""" + boundary_pixel = max(0, int(boundary_pixel)) + pixel_overlap = max(0, int(pixel_overlap)) + zones = {} + if pixel_overlap > 0: + zones["latent_blend"] = { + "scene_prev_tail": [max(0, boundary_pixel - pixel_overlap), max(0, boundary_pixel - 1)], + "scene_next_head": [boundary_pixel, boundary_pixel + pixel_overlap - 1], + "note": "Previous scene tail is copied into the next chunk, denoised under next-scene conditioning, then slerp-blended back — motion from scene N+1 can appear in scene N's last overlap frames.", + } + if effect and effect != "none" and transition_duration > 0: + half = max(1, int(transition_duration) // 2) + lo = max(0, boundary_pixel - half) + hi = boundary_pixel + half - 1 + zones["transition_effect"] = { + "effect": effect, + "pixel_range": [lo, hi], + "note": "Post-decode pixel effect mixes frames across the seam — scene N+1 content can show in scene N's tail (crossfade is strongest).", + } + return zones + + def _build_overlap_diagnostics( + self, + *, + scene_count, + video_frames, + pixel_overlap, + latent_overlap, + time_scale, + transition_duration, + boundaries, + scene_runs, + carry_i2v_guides, + mid_scene_guide, + embed_guidance, + embed_guidance_strength, + embed_guidance_source, + ): + """JSON report: where scene conditioning/latent/pixel domains overlap.""" + pixel_ranges = self._scene_pixel_ranges(scene_count, video_frames, pixel_overlap) + boundary_reports = [] + for entry in boundaries or []: + pixel = int(entry.get("pixel_frame") or 0) + effect = entry.get("effect") + between = entry.get("between") or [] + zones = self._boundary_contamination_zones(pixel, pixel_overlap, transition_duration, effect) + boundary_reports.append({ + "between_scenes": between, + "pixel_frame": pixel, + "latent_frame": entry.get("latent_frame"), + "effect": effect, + "contamination_zones": zones, + }) + + global_steering = [] + if embed_guidance: + global_steering.append({ + "mechanism": "embed_guidance", + "strength": float(embed_guidance_strength), + "source": str(embed_guidance_source or "relative"), + "scope": "all_scenes_every_denoise_step", + "note": "Steers each scene's conditioning toward the learned liked direction during sampling. " + "Absolute mode uses a prompt-agnostic global taste — action from ANY liked scene can bias scene 1.", + }) + if carry_i2v_guides: + global_steering.append({ + "mechanism": "carry_i2v_guides", + "scope": "scene_2_onward", + "note": "Scene 1 template frames are prepended as hidden guide tokens in continuation chunks only — " + "does not inject scene 2 into scene 1.", + }) + if mid_scene_guide: + global_steering.append({ + "mechanism": "mid_scene_guide", + "scope": "scene_2_onward", + "note": "Middle frame of scene N guides scene N+1 denoising — does not retroactively change scene N.", + }) + global_steering.append({ + "mechanism": "studio_absolute_steer", + "scope": "all_scenes_pre_sample", + "note": "When Studio steer_mode is absolute/both, the same global taste pull is applied to every " + "scene conditioning entry before the chain sampler runs — can add scene-2-like motion cues to scene 1.", + }) + + scenes_out = [] + for run in scene_runs: + idx = int(run.get("index") or 0) + pr = next((r for r in pixel_ranges if r["scene"] == idx), None) + item = dict(run) + if pr: + item["pixel_range"] = [pr["start"], pr["end"]] + tail_zones = [] + for br in boundary_reports: + between = br.get("between_scenes") or [] + if len(between) == 2 and between[0] == idx: + for zone in (br.get("contamination_zones") or {}).values(): + if isinstance(zone, dict) and "scene_prev_tail" in zone: + tail_zones.append(zone["scene_prev_tail"]) + if tail_zones: + item["tail_contamination_pixels"] = tail_zones + if idx == 1 and embed_guidance: + item["whole_scene_steering"] = True + scenes_out.append(item) + + return { + "scene_count": scene_count, + "frames_per_scene": int(video_frames), + "pixel_overlap": int(pixel_overlap), + "latent_overlap": int(latent_overlap), + "time_scale": int(time_scale), + "scene_pixel_ranges": pixel_ranges, + "boundaries": boundary_reports, + "global_steering": global_steering, + "scenes": scenes_out, + } + def _scene_seed(self, scene_conditioning): if ( isinstance(scene_conditioning, (list, tuple)) @@ -3203,6 +3360,7 @@ class FunPackLTXAVSceneChainSampler: scene_anchors = self._parse_scene_anchors(funpack_scene_anchors) scene_outputs: list = [] scene_media_by_ref = self._parse_scene_anchors(funpack_scene_media_refs) + scene_runs: list = [] for scene_index, scene_cond in enumerate(scene_conditionings): scene_positive = [scene_cond] @@ -3216,51 +3374,62 @@ class FunPackLTXAVSceneChainSampler: carried = 0 soft_carried = 0 guide_tail = 0 + run_mechanisms: list = [] anchor_meta = (scene_anchors or {}).get(str(scene_index)) if output is None: chunk = self._clone_latent(latent_template) elif anchor_meta: + run_mechanisms.append("mixed_i2v_anchor") effect = self._scene_transition_effect(scene_cond) - if effect and transition_duration > 0: - boundary_latent = cumulative_latent_frames - boundary_pixel = int((boundary_latent - 1) * time_scale + 1) if time_scale > 1 else boundary_latent - boundary_entries.append({ - "boundary_latent": boundary_latent, - "pixel_frame": max(0, boundary_pixel), - "effect": effect, - }) + boundary_latent = cumulative_latent_frames + boundary_pixel = self._latent_to_pixel_frame(boundary_latent, time_scale) + boundary_entries.append({ + "between": [scene_index, scene_index + 1], + "boundary_latent": boundary_latent, + "pixel_frame": max(0, boundary_pixel), + "effect": effect if effect and transition_duration > 0 else None, + }) + if video_overlap > 0: + run_mechanisms.append(f"latent_overlap({frame_overlap}px)") chunk = self._build_mixed_anchor_chunk( vae, anchor_meta, latent_template, output, video_overlap, ) else: # Record boundary before blending effect = self._scene_transition_effect(scene_cond) - if effect and transition_duration > 0: - boundary_latent = cumulative_latent_frames - boundary_pixel = int((boundary_latent - 1) * time_scale + 1) if time_scale > 1 else boundary_latent - boundary_entries.append({ - "boundary_latent": boundary_latent, - "pixel_frame": max(0, boundary_pixel), - "effect": effect, - }) + boundary_latent = cumulative_latent_frames + boundary_pixel = self._latent_to_pixel_frame(boundary_latent, time_scale) + boundary_entries.append({ + "between": [scene_index, scene_index + 1], + "boundary_latent": boundary_latent, + "pixel_frame": max(0, boundary_pixel), + "effect": effect if effect and transition_duration > 0 else None, + }) chunk = self._build_continuation_chunk(latent_template, output, video_overlap) if video_overlap == 0: chunk, soft_carried = self._prepend_soft_continuation(chunk, output) + if soft_carried > 0: + run_mechanisms.append(f"soft_continuation({soft_carried})") + elif video_overlap > 0: + run_mechanisms.append(f"latent_overlap({frame_overlap}px)") custom_guides = None if per_scene_guides and scene_index < len(per_scene_guides): custom_guides = per_scene_guides[scene_index] if custom_guides: + run_mechanisms.append("custom_guide_stack") chunk, scene_positive, scene_negative, carried, guide_tail = self._apply_configured_guides( chunk, scene_index, custom_guides, latent_template, scene_outputs, scene_media_by_ref, scene_positive, scene_negative, vae, ) carried_guide_frames = max(carried_guide_frames, carried) elif carry_i2v_guides: + run_mechanisms.append("carry_i2v_guides") chunk, scene_positive, scene_negative, carried = self._append_i2v_guides( chunk, latent_template, scene_positive, scene_negative, ) carried_guide_frames = max(carried_guide_frames, carried) if mid_scene_guide and not custom_guides: + run_mechanisms.append("mid_scene_guide") chunk, scene_positive, scene_negative, guide_tail = self._append_mid_scene_guide( chunk, output, scene_positive, scene_negative, vae, mid_scene_guide_strength, ) @@ -3268,10 +3437,12 @@ class FunPackLTXAVSceneChainSampler: guide_tail = 0 if embed_guidance and _value_fn is not None and _value_fn.is_ready(): + run_mechanisms.append("embed_guidance_vf_ascend") orig_cond, orig_extra = scene_positive[0][0], scene_positive[0][1] ascended = self._protect_audio(_value_fn.ascend(orig_cond), orig_cond) scene_positive = [[ascended, orig_extra]] + list(scene_positive[1:]) if embed_guidance and _liked_dir is not None: + run_mechanisms.append(f"embed_guidance({_eg_source},{embed_guidance_strength})") _eg_old_wrapper = self._build_embed_guidance_wrapper(model, _liked_dir, embed_guidance_strength, value_fn=_value_fn) # Per-scene temporal style (temporal_style="auto"): layer a frame_rate wrapper on # top of whatever is installed (e.g. embed guidance). Restored right after sampling, @@ -3311,7 +3482,16 @@ class FunPackLTXAVSceneChainSampler: scene_outputs.append(self._clone_latent(sampled)) output = sampled if output is None else self._blend_latents(output, sampled, video_overlap) cumulative_latent_frames = self._tensor_frames(self._latent_tensors(output)[0]) - report_lines.append(f"Scene {scene_index + 1}: seed={scene_seed}, text={self._scene_text(scene_cond, scene_index)}") + scene_meta = self._scene_meta(scene_cond, scene_index) + scene_runs.append({ + **scene_meta, + "seed_used": scene_seed, + "mechanisms": run_mechanisms, + }) + report_lines.append( + f"Scene {scene_index + 1}: seed={scene_seed}, text={scene_meta['text']}" + + (f" | encode≠text" if scene_meta["encode_text"] != scene_meta["text"] else "") + ) del scene_cond, scene_positive, scene_negative, scene_conditionings, chunk, sampled @@ -3346,7 +3526,32 @@ class FunPackLTXAVSceneChainSampler: ) if carry_i2v_guides and carried_guide_frames > 0: status += f", i2v guide tokens={carried_guide_frames} latent frame(s)" - boundaries_out = [{"pixel_frame": e["pixel_frame"], "effect": e["effect"]} for e in boundary_entries] + overlap_diag = self._build_overlap_diagnostics( + scene_count=scene_count, + video_frames=video_frames, + pixel_overlap=int(frame_overlap), + latent_overlap=int(video_overlap), + time_scale=time_scale, + transition_duration=int(transition_duration), + boundaries=boundary_entries, + scene_runs=scene_runs, + carry_i2v_guides=bool(carry_i2v_guides), + mid_scene_guide=bool(mid_scene_guide), + embed_guidance=bool(embed_guidance), + embed_guidance_strength=float(embed_guidance_strength), + embed_guidance_source=str(embed_guidance_source or "relative"), + ) + if scene_count > 1 and int(frame_overlap) > 0: + status += ( + f", overlap_blend={int(frame_overlap)}px" + f" (scene N tail may show scene N+1 motion in last {int(frame_overlap)} frames)" + ) + if embed_guidance: + status += ( + f", embed_guidance={_eg_source}@{embed_guidance_strength}" + f" (steers ALL scenes — not boundary-local)" + ) + boundaries_out = overlap_diag if refinement_key_input: try: try: diff --git a/tests/test_scene_chain_sampler.py b/tests/test_scene_chain_sampler.py index 3f57d212..8778acd4 100644 --- a/tests/test_scene_chain_sampler.py +++ b/tests/test_scene_chain_sampler.py @@ -417,3 +417,62 @@ def test_scene_chain_does_not_carry_i2v_guides_by_default(): assert torch.all(second_call["noise_mask"][:, :, :2] == 0.0) assert torch.all(second_call["noise_mask"][:, :, 2:] == 1.0) assert "i2v guide" not in status + + +def test_overlap_diagnostics_report_latent_blend_zone(): + import json + + node = FunPackLTXAVSceneChainSampler() + diag = node._build_overlap_diagnostics( + scene_count=2, + video_frames=97, + pixel_overlap=16, + latent_overlap=2, + time_scale=8, + transition_duration=16, + boundaries=[{ + "between": [1, 2], + "boundary_latent": 12, + "pixel_frame": 89, + "effect": "crossfade", + }], + scene_runs=[ + {"index": 1, "text": "hero walks", "encode_text": "hero walks", "mechanisms": []}, + {"index": 2, "text": "hero runs", "encode_text": "hero runs", "mechanisms": ["latent_overlap(16px)"]}, + ], + carry_i2v_guides=True, + mid_scene_guide=False, + embed_guidance=True, + embed_guidance_strength=0.15, + embed_guidance_source="absolute", + ) + assert diag["pixel_overlap"] == 16 + blend = diag["boundaries"][0]["contamination_zones"]["latent_blend"] + assert blend["scene_prev_tail"] == [73, 88] + assert blend["scene_next_head"] == [89, 104] + assert diag["scenes"][0]["whole_scene_steering"] is True + assert any(g["mechanism"] == "embed_guidance" for g in diag["global_steering"]) + # scene_boundaries output is JSON in the full sample() path + sample_calls.clear() + node = FunPackLTXAVSceneChainSampler() + positive = [scene_cond(0), scene_cond(1)] + latent_template = {"samples": torch.zeros(1, 2, 5, 1, 1)} + _, _, status, _, _, boundaries_json = node.sample( + model=object(), + vae=FakeVAE(), + positive=positive, + negative=[], + sampler=object(), + sigmas=torch.tensor([1.0, 0.0]), + seed=60, + latent_template=latent_template, + num_frames_per_scene=5, + frame_overlap=2, + cfg=1.0, + max_scenes=2, + embed_guidance=False, + ) + parsed = json.loads(boundaries_json) + assert parsed["scene_count"] == 2 + assert "boundaries" in parsed + assert "overlap_blend=2px" in status