diff --git a/samplers.py b/samplers.py index 9bfc80a9..ee6ee426 100644 --- a/samplers.py +++ b/samplers.py @@ -2424,6 +2424,12 @@ class FunPackLTXAVSceneChainSampler: "default": False, "tooltip": "Print per-scene identity-transfer shape/status logs to the console.", }), + # NOTE: append-only — keep new sampler widgets at the END of this block so the + # builder's positional reference-workflow mapping (extract_widgets) stays aligned. + "carry_overlap_through_anchor": ("BOOLEAN", { + "default": False, + "tooltip": "When a scene switches to its own i2v anchor (funpack_scene_anchors — a different reference image/character than the previous scene), still carry frame_overlap latent frames from the previous scene's tail into the frames right after the anchor image, instead of the default hard cut with no carried context. Preserves background/environment continuity through an anchor change (e.g. a Best-FaceID identity_transfer scene swapping the reference face mid-chain). The anchor image's own leading frame is never touched by the carried tail. No effect on scenes without a per-scene anchor.", + }), }, "hidden": { "unique_id": "UNIQUE_ID", @@ -3794,18 +3800,44 @@ class FunPackLTXAVSceneChainSampler: out = self._invoke_img2video_inplace(vae, image, chunk, strength) return out if out is not None else chunk - def _build_mixed_anchor_chunk(self, vae, anchor_meta, latent_template, previous, video_overlap): - """Mixed source: Img2Video anchor latent only — prior scene overlap is not used.""" + def _build_mixed_anchor_chunk(self, vae, anchor_meta, latent_template, previous, video_overlap, + carry_overlap=False): + """Mixed source: Img2Video anchor latent. By default prior scene overlap is not used + (hard cut). When carry_overlap is set and a previous scene exists, the chunk is first + seeded as a normal continuation (previous scene's tail copied into the leading + video_overlap frames, protected) and the anchor image is then written on top — since + LTXVImgToVideoInplace only overwrites the single encoded image frame's worth of samples + and preserves the rest of the incoming noise_mask, the anchor's own leading frame stays a + hard cut while the remaining carried frames keep the old scene's background/environment.""" filename = (anchor_meta or {}).get("filename") strength = float((anchor_meta or {}).get("strength", 1.0)) image = self._load_image_tensor(filename) if filename else None if image is None: if previous is None: return self._clone_latent(latent_template) - return self._build_continuation_chunk(latent_template, previous, video_overlap) - base = self._clone_latent(latent_template) + return self._build_continuation_chunk(latent_template, previous, 0) + if carry_overlap and previous is not None and video_overlap > 0: + base = self._build_continuation_chunk(latent_template, previous, video_overlap) + else: + base = self._clone_latent(latent_template) return self._apply_img2video_to_video_latent(vae, image, base, strength) + def _identity_pin_filename(self, guide_list, scene_media_by_ref, identity_transfer_enabled): + """First identity_pin-tagged image guide's filename, or None. Mirrors the matching + inside _apply_configured_guides, but callable standalone for branches (like the mixed + i2v anchor path) that don't run the rest of the guide stack against their chunk.""" + if not identity_transfer_enabled: + return None + for g in guide_list or []: + if not g or not g.get("enabled", True): + continue + if str(g.get("source", "template")) == "image" and g.get("identity_pin"): + ref = g.get("media_ref") + fn = (scene_media_by_ref or {}).get(ref) if ref else None + if fn: + return fn + return None + def _encode_image_guide_frame(self, filename, vae, ref_tensor): import os try: @@ -4454,6 +4486,7 @@ class FunPackLTXAVSceneChainSampler: alg_guide_blur_strength=2.0, alg_guide_blur_sigma_threshold=0.975, identity_transfer_enabled=False, identity_projector="None", source_id=2.0, phase_scale=1.0, id_strength=1.0, arcface_mode="auto_adjust", debug_log=False, + carry_overlap_through_anchor=False, unique_id=None, prompt=None): if not isinstance(positive, list) or not positive: raise ValueError("positive conditioning must contain at least one scene entry.") @@ -4674,8 +4707,20 @@ class FunPackLTXAVSceneChainSampler: "effect": effect if effect and transition_duration > 0 else None, }) chunk = self._build_mixed_anchor_chunk( - vae, anchor_meta, latent_template, output, 0, + vae, anchor_meta, latent_template, output, video_overlap, + carry_overlap=carry_overlap_through_anchor, ) + if carry_overlap_through_anchor and video_overlap > 0: + run_mechanisms.append(f"latent_overlap_through_anchor({frame_overlap}px)") + # The anchor branch skips _apply_configured_guides entirely, so an identity_pin + # guide configured for this scene would otherwise never resolve — Best-FaceID + # identity_transfer needs this to fire on the exact scenes that swap anchors. + if per_scene_guides and scene_index < len(per_scene_guides): + identity_ref_filename = self._identity_pin_filename( + per_scene_guides[scene_index], scene_media_by_ref, identity_transfer_enabled, + ) + if identity_ref_filename: + run_mechanisms.append("identity_pin_on_anchor_scene") else: # Record boundary before blending effect = self._scene_transition_effect(scene_cond) @@ -4836,6 +4881,11 @@ class FunPackLTXAVSceneChainSampler: a_frame = self._harvest_joyai_audio(sampled, joyai_frame_select) if joyai_audio_memory else None joyai_bank.add(v_frame, a_frame) scene_outputs.append(self._clone_latent(sampled)) + # Stays 0 for anchor scenes even with carry_overlap_through_anchor on: the post-sample + # slerp blend would reach into the anchor image's own leading frame (position 0 of + # `sampled`) and fade it against the previous scene's tail, undermining the hard cut. + # The carried frames beyond it are already seeded pre-sample (see + # _build_mixed_anchor_chunk), so no post-hoc smoothing is needed there. blend_overlap = 0 if anchor_meta else video_overlap output = sampled if output is None else self._blend_latents(output, sampled, blend_overlap) cumulative_latent_frames = self._tensor_frames(self._latent_tensors(output)[0]) @@ -4961,6 +5011,7 @@ class FunPackLTXAVSceneChainSampler: "transitions_enabled": scene_count > 1, "joyai_memory": bool(joyai_memory), "joyai_audio_memory": bool(joyai_audio_memory), + "carry_overlap_through_anchor": bool(carry_overlap_through_anchor), }) except Exception as e: print(f"[FunPackLTXAVSceneChainSampler] Failed to write sampler context: {e}") diff --git a/tests/test_identity_transfer.py b/tests/test_identity_transfer.py index 3af38d48..bbdd1d84 100644 --- a/tests/test_identity_transfer.py +++ b/tests/test_identity_transfer.py @@ -136,6 +136,27 @@ def test_identity_pin_appends_normally_when_disabled(monkeypatch): assert identity_ref_filename is None +# ── _identity_pin_filename (standalone lookup for the mixed i2v anchor branch, +# which skips _apply_configured_guides entirely) ───────────────────────────── + +def test_identity_pin_filename_finds_first_pin_entry(): + s = _sampler() + guide_list = [ + {"enabled": True, "source": "image", "media_ref": "prior", "strength": 0.35}, + {"enabled": True, "source": "image", "media_ref": "pin", "identity_pin": True, "strength": 0.35}, + ] + scene_media_by_ref = {"pin": "pin.png", "prior": "prior.png"} + assert s._identity_pin_filename(guide_list, scene_media_by_ref, True) == "pin.png" + + +def test_identity_pin_filename_none_when_disabled_or_missing(): + s = _sampler() + guide_list = [{"enabled": True, "source": "image", "media_ref": "pin", "identity_pin": True}] + assert s._identity_pin_filename(guide_list, {"pin": "pin.png"}, False) is None + assert s._identity_pin_filename(guide_list, {}, True) is None + assert s._identity_pin_filename(None, {"pin": "pin.png"}, True) is None + + # ── _install_identity_overlap / _strip_identity_overlap ──────────────────────── class _FakePatchifier: diff --git a/tests/test_scene_chain_sampler.py b/tests/test_scene_chain_sampler.py index ad63a4f4..8bb1392b 100644 --- a/tests/test_scene_chain_sampler.py +++ b/tests/test_scene_chain_sampler.py @@ -118,6 +118,16 @@ class FakeVAE: downscale_index_formula = (1, 1, 1) +class FakeModel: + """Minimal stand-in with the one attribute the per-scene wrapper snapshot/restore + (samplers.py, around _scene_base_wrapper) needs. Plain object() lacks model_options + entirely, which pre-dates this feature (every sample()-calling test in this file + currently fails on that AttributeError — a known stale-mock gap, not something this + change introduces or fixes file-wide).""" + def __init__(self): + self.model_options = {} + + def scene_cond(index): return ( torch.ones(1, 2, 3) * float(index + 1), @@ -557,3 +567,110 @@ def test_mixed_anchor_skips_frame_overlap(): mechs = parsed["scenes"][1]["mechanisms"] assert "mixed_i2v_anchor" in mechs assert not any("latent_overlap" in m for m in mechs) + + +def test_mixed_anchor_carries_overlap_when_enabled(): + """carry_overlap_through_anchor=True: the chunk fed to scene 1's sampler call is seeded + with scene 0's tail (frame_overlap latent frames) instead of a bare template, even though + scene 1 has its own i2v anchor. The mocked _apply_img2video_to_video_latent is an identity + passthrough here, so whatever _build_mixed_anchor_chunk hands it is exactly what gets + sampled — letting us assert on the carried values and protected mask directly.""" + import json + + sample_calls.clear() + node = FunPackLTXAVSceneChainSampler() + node._load_image_tensor = lambda _fn: torch.ones(1, 8, 8, 3) + node._apply_img2video_to_video_latent = lambda _vae, _img, chunk, _strength: node._clone_latent(chunk) + + positive = [scene_cond(0), scene_cond(1)] + latent_template = {"samples": torch.zeros(1, 2, 5, 3, 3)} + anchors = json.dumps({"1": {"filename": "anchor.png", "strength": 1.0}}) + + latent, _images, status, scene_count, _report, boundaries_json = node.sample( + model=FakeModel(), + vae=FakeVAE(), + positive=positive, + negative=[], + sampler=object(), + sigmas=torch.tensor([1.0, 0.0]), + seed=80, + latent_template=latent_template, + num_frames_per_scene=5, + frame_overlap=2, + cfg=1.0, + max_scenes=2, + funpack_scene_anchors=anchors, + carry_overlap_through_anchor=True, + prompt={"n1": {"inputs": {}}}, + unique_id="test-node", + ) + + assert scene_count == 2 + assert latent["samples"].shape[2] == 10 + parsed = json.loads(boundaries_json) + mechs = parsed["scenes"][1]["mechanisms"] + assert "mixed_i2v_anchor" in mechs + assert "latent_overlap_through_anchor(2px)" in mechs + + # Scene 0 (no anchor) samples with mask=None, so _sample_like adds its seed (80) uniformly: + # scene 0's full output is all 80s. Scene 1's chunk should carry the last 2 frames of that + # (80, 80) into its own leading frames, protected (mask=0), template's remaining 3 frames + # untouched (0) and free to denoise (mask=1). + scene1_chunk = sample_calls[1]["latent_image"] + scene1_mask = sample_calls[1]["noise_mask"] + samples = scene1_chunk["samples"] if isinstance(scene1_chunk, dict) else scene1_chunk + mask = scene1_mask["samples"] if isinstance(scene1_mask, dict) else scene1_mask + assert torch.allclose(samples[:, :, :2], torch.full_like(samples[:, :, :2], 80.0)) + assert torch.allclose(samples[:, :, 2:], torch.zeros_like(samples[:, :, 2:])) + assert torch.allclose(mask[:, :, :2], torch.zeros_like(mask[:, :, :2])) + assert torch.allclose(mask[:, :, 2:], torch.ones_like(mask[:, :, 2:])) + + +def test_mixed_anchor_resolves_identity_pin_when_configured(): + """The mixed_i2v_anchor branch skips _apply_configured_guides entirely, so without the + explicit lookup an identity_pin guide configured for the anchor scene would never resolve + and Best-FaceID identity_transfer could never fire on an anchor-swap scene.""" + import json + + sample_calls.clear() + node = FunPackLTXAVSceneChainSampler() + node._load_image_tensor = lambda _fn: torch.ones(1, 8, 8, 3) + node._apply_img2video_to_video_latent = lambda _vae, _img, chunk, _strength: node._clone_latent(chunk) + + positive = [scene_cond(0), scene_cond(1)] + latent_template = {"samples": torch.zeros(1, 2, 5, 3, 3)} + anchors = json.dumps({"1": {"filename": "anchor.png", "strength": 1.0}}) + guides = json.dumps({ + "stack_enabled": True, + "scenes": [ + [], + [{"enabled": True, "source": "image", "media_ref": "pin", "identity_pin": True, "strength": 0.35}], + ], + }) + media_refs = json.dumps({"pin": "pin.png"}) + + _latent, _images, _status, scene_count, _report, boundaries_json = node.sample( + model=FakeModel(), + vae=FakeVAE(), + positive=positive, + negative=[], + sampler=object(), + sigmas=torch.tensor([1.0, 0.0]), + seed=80, + latent_template=latent_template, + num_frames_per_scene=5, + frame_overlap=2, + cfg=1.0, + max_scenes=2, + funpack_scene_anchors=anchors, + funpack_scene_guides=guides, + funpack_scene_media_refs=media_refs, + identity_transfer_enabled=True, + prompt={"n1": {"inputs": {}}}, + unique_id="test-node", + ) + + assert scene_count == 2 + parsed = json.loads(boundaries_json) + mechs = parsed["scenes"][1]["mechanisms"] + assert "identity_pin_on_anchor_scene" in mechs