diff --git a/conditioning.py b/conditioning.py index 04f60a59..101e58ea 100644 --- a/conditioning.py +++ b/conditioning.py @@ -6194,10 +6194,10 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): "default": "", "tooltip": "Optional user feedback describing what was specifically wrong with the previous output (e.g. 'he was supposed to hold her hand, not her head'). Has highest priority in the advisor system prompt.", }), - "temporal_style": (["natural", "auto", "accelerate", "decelerate", "loop", "freeze"], { + "temporal_style": (["natural", "auto", "accelerate", "decelerate", "loop", "freeze", "pulse"], { "default": "natural", "label": "Temporal Style", - "tooltip": "Controls how the model perceives motion timing via frame_rate RoPE manipulation. natural=no change, auto=per-scene director picks motion energy from each scene's prompt (needs the Scene Chain Sampler), accelerate=faster motion, decelerate=heavier motion, loop=circular temporal coords, freeze=highly compressed time.", + "tooltip": "Controls how the model perceives motion timing via frame_rate RoPE manipulation. natural=no change, auto=per-scene director picks motion energy from each scene's prompt (needs the Scene Chain Sampler), accelerate=faster motion, decelerate=heavier motion, loop=circular temporal coords, freeze=highly compressed time, pulse=repeated ease-down motion holds per scene (needs the Scene Chain Sampler).", }), "split_by_transitions": ("BOOLEAN", { "default": False, @@ -12945,6 +12945,22 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): ) return out + def _v2_apply_pulse_temporal(self, conditioning_list, temporal_style): + """temporal_style="pulse": tag each scene entry so the Scene Chain Sampler installs + a segmented ease-down temporal wrapper per scene.""" + if str(temporal_style or "natural").strip().lower() != "pulse": + return conditioning_list + out = [] + for entry in conditioning_list or []: + if not (isinstance(entry, (list, tuple)) and len(entry) >= 2 and isinstance(entry[1], dict)): + out.append(entry) + continue + cond, meta = entry[0], dict(entry[1]) + meta["funpack_temporal_mode"] = "pulse" + meta["funpack_temporal_label"] = "pulse" + out.append([cond, meta] if isinstance(entry, list) else (cond, meta)) + return out + def _v2_apply_auto_temporal(self, conditioning_list, temporal_style, fallback_text=""): """temporal_style="auto": bake a per-scene frame_rate multiplier into each conditioning entry's metadata so the Scene Chain Sampler can apply per-scene motion @@ -12985,7 +13001,8 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): Guessing has learned a safe-spread ceiling, clamp the output conditioning's video-channel spread to it (the backstop half of the dual auto-cap).""" mode = str(steer_mode or "relative").lower() - out = self._v2_apply_auto_temporal(conditioning_list, temporal_style, fallback_text=temporal_fallback_text) + out = self._v2_apply_pulse_temporal(conditioning_list, temporal_style) + out = self._v2_apply_auto_temporal(out, temporal_style, fallback_text=temporal_fallback_text) if mode in ("relative", "both"): out = self._v2_ascend_conditioning(out, refinement_key, apply=value_guidance) if mode in ("absolute", "both"): diff --git a/ltx_enhancements.py b/ltx_enhancements.py index 102c0e4f..319ffcdb 100644 --- a/ltx_enhancements.py +++ b/ltx_enhancements.py @@ -6,7 +6,7 @@ Technique 2: Per-block attention temperature Technique 3: Temporal RoPE style Manipulate frame_rate in the model's positional embedding to change motion character. - Styles: natural / accelerate / decelerate / loop / freeze + Styles: natural / accelerate / decelerate / loop / freeze / pulse Technique 4: Denoise creativity mask Spatial noise weighting derived from latent variance: high-variance regions get more freedom. @@ -25,7 +25,7 @@ from hashlib import md5 import torch -TEMPORAL_STYLES = ["natural", "auto", "accelerate", "decelerate", "loop", "freeze"] +TEMPORAL_STYLES = ["natural", "auto", "accelerate", "decelerate", "loop", "freeze", "pulse"] # frame_rate multiplier fed to the LTX RoPE per style. >1 => model assumes more # frames cover the same wall-clock time => smaller inter-frame deltas => smoother / @@ -39,6 +39,12 @@ TEMPORAL_STYLE_MULT = { "freeze": 2.0, } +# Pulse: repeated ease-down segments per scene. Each segment starts punchy (peak mult) +# and eases toward a temporal hold (floor mult) before the next segment resets. +PULSE_SEGMENT_COUNT = 3 +PULSE_PEAK_MULT = 0.88 +PULSE_FLOOR_MULT = 1.65 + # Keyword -> motion-energy intent for the "auto" director. The classifier scans a # scene's prompt and emits a continuous frame_rate multiplier (plus a loop flag). @@ -95,6 +101,43 @@ def classify_temporal_intent(text): return {"mult": round(float(mult), 3), "loop": bool(loop), "label": label} +def _ease_down(t): + """Smooth ease from 0→1 for pulse segment ramps.""" + t = max(0.0, min(1.0, float(t))) + return t * t * (3.0 - 2.0 * t) + + +def pulse_mult_for_progress(progress, segment_count=None, peak_mult=None, floor_mult=None): + """Map denoise progress [0,1] to a frame_rate multiplier with repeated ease-down segments.""" + segs = PULSE_SEGMENT_COUNT if segment_count is None else max(1, int(segment_count)) + peak = PULSE_PEAK_MULT if peak_mult is None else float(peak_mult) + floor = PULSE_FLOOR_MULT if floor_mult is None else float(floor_mult) + progress = max(0.0, min(1.0, float(progress))) + seg_idx = min(int(progress * segs), segs - 1) + local_t = progress * segs - seg_idx + return peak + (floor - peak) * _ease_down(local_t) + + +def _scale_frame_rate_in_args(args, mult): + """Return a shallow-copied args dict with frame_rate scaled by mult (no-op on failure).""" + c = args.get("c") + if not (isinstance(c, dict) and "frame_rate" in c): + return args + try: + fr_cond = c["frame_rate"] + if hasattr(fr_cond, "cond"): + original_fr = float(fr_cond.cond) + new_cond = type(fr_cond)(original_fr * float(mult)) + new_c = dict(c) + new_c["frame_rate"] = new_cond + new_args = dict(args) + new_args["c"] = new_c + return new_args + except Exception: + pass + return args + + def make_temporal_wrapper(old_wrapper, mult): """Build a model_function_wrapper that scales the LTX `frame_rate` conditioning by `mult`, chaining any existing wrapper. Returns None when mult is effectively 1.0 @@ -108,25 +151,43 @@ def make_temporal_wrapper(old_wrapper, mult): return None def _temporal_wrapper(apply_fn, args, _mult=mult, _old=old_wrapper): - c = args.get("c") - if isinstance(c, dict) and "frame_rate" in c: - try: - fr_cond = c["frame_rate"] - if hasattr(fr_cond, "cond"): - original_fr = float(fr_cond.cond) - new_cond = type(fr_cond)(original_fr * _mult) - new_c = dict(c) - new_c["frame_rate"] = new_cond - args = dict(args) - args["c"] = new_c - except Exception: - pass + args = _scale_frame_rate_in_args(args, _mult) if _old is not None: return _old(apply_fn, args) return apply_fn(args["input"], args["timestep"], **args.get("c", {})) return _temporal_wrapper + +def make_pulse_temporal_wrapper(old_wrapper, segment_count=None, peak_mult=None, floor_mult=None): + """Build a model_function_wrapper that cycles ease-down temporal segments per scene. + + Denoise progress (derived from sigma) is divided into `segment_count` segments. Within + each segment the frame_rate multiplier eases from `peak_mult` (punchier motion) toward + `floor_mult` (temporal hold). A fresh wrapper should be installed per scene so sigma + bounds reset between scenes.""" + segs = PULSE_SEGMENT_COUNT if segment_count is None else max(1, int(segment_count)) + peak = PULSE_PEAK_MULT if peak_mult is None else float(peak_mult) + floor = PULSE_FLOOR_MULT if floor_mult is None else float(floor_mult) + sigma_start = [None] + + def _pulse_wrapper(apply_fn, args, _old=old_wrapper, _segs=segs, _peak=peak, _floor=floor): + ts = args.get("timestep") + try: + sigma = float(ts.max().item()) if ts is not None else 1.0 + except Exception: + sigma = 1.0 + if sigma_start[0] is None: + sigma_start[0] = max(sigma, 1e-6) + progress = max(0.0, min(1.0, 1.0 - sigma / sigma_start[0])) + mult = pulse_mult_for_progress(progress, _segs, _peak, _floor) + args = _scale_frame_rate_in_args(args, mult) + if _old is not None: + return _old(apply_fn, args) + return apply_fn(args["input"], args["timestep"], **args.get("c", {})) + + return _pulse_wrapper + # Confirmed semantic focal points (PAG default=14, STG defaults=14,19) ANCHOR_BLOCKS = [14, 19] @@ -963,11 +1024,15 @@ def build_enhancements(model, rating_profile, temporal_style, refinement_key, re print(f"[FunPackEnhancements] Hook setup failed: {e}") # --- Technique 3: temporal RoPE via model_function_wrapper --- - # "auto" is per-scene and owned by the chain sampler (it bakes a per-scene - # multiplier into each scene's conditioning and applies it there), so the global - # path here only fires for the concrete manual styles. "loop" stays a no-op until - # the RoPE coordinate trick is wired. - if temporal_style not in ("natural", "auto"): + # "auto" and "pulse" are per-scene and owned by the chain sampler (auto bakes a + # per-scene multiplier; pulse installs a segmented ease-down wrapper per scene). + # The global path here fires for concrete manual styles and pulse on single-scene + # workflows. "loop" stays a no-op until the RoPE coordinate trick is wired. + if temporal_style == "pulse": + wrapper = make_pulse_temporal_wrapper(model.model_options.get("model_function_wrapper")) + if wrapper is not None: + model.model_options["model_function_wrapper"] = wrapper + elif temporal_style not in ("natural", "auto"): mult = TEMPORAL_STYLE_MULT.get(temporal_style, 1.0) wrapper = make_temporal_wrapper(model.model_options.get("model_function_wrapper"), mult) if wrapper is not None: diff --git a/movie_editor/frontend/engine_settings.js b/movie_editor/frontend/engine_settings.js index 04ca2424..602e0f64 100644 --- a/movie_editor/frontend/engine_settings.js +++ b/movie_editor/frontend/engine_settings.js @@ -74,7 +74,7 @@ { name: "absolute_strength", label: "Absolute strength", kind: "float", default: 0.6, min: 0, max: 1, step: 0.05, dependsOn: "steer_mode", dependsVals: ["absolute", "both"] }, { name: "temporal_style", label: "Temporal style", kind: "combo", - choices: ["natural", "auto", "accelerate", "decelerate", "loop", "freeze"], default: "natural" }, + choices: ["natural", "auto", "accelerate", "decelerate", "loop", "freeze", "pulse"], default: "natural" }, { name: "split_transition_placement", label: "Transition placement", kind: "combo", choices: ["start", "end", "silent"], default: "start" }, ]; diff --git a/samplers.py b/samplers.py index 3bf8857b..9792a97b 100644 --- a/samplers.py +++ b/samplers.py @@ -2750,6 +2750,17 @@ class FunPackLTXAVSceneChainSampler: return None return None + def _scene_temporal_mode(self, scene_conditioning): + """Per-scene temporal mode tag (e.g. pulse). None when unset.""" + if ( + isinstance(scene_conditioning, (list, tuple)) + and len(scene_conditioning) >= 2 + and isinstance(scene_conditioning[1], dict) + ): + mode = str(scene_conditioning[1].get("funpack_temporal_mode") or "").strip().lower() + return mode or None + return None + def _decode_last_frame(self, latent, vae): try: tensors = self._latent_tensors(latent) @@ -3449,18 +3460,29 @@ class FunPackLTXAVSceneChainSampler: 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, + # Per-scene temporal style (auto / pulse): layer a frame_rate wrapper on top of + # whatever is installed (e.g. embed guidance). Restored right after sampling, # before the embed-guidance restore, so the wrappers unwind in install order. _temporal_applied = False _temporal_prev_wrapper = None + _scene_mode = self._scene_temporal_mode(scene_cond) _scene_mult = self._scene_temporal_mult(scene_cond) - if _scene_mult is not None and abs(_scene_mult - 1.0) >= 1e-3: + _cur_wrapper = model.model_options.get("model_function_wrapper") + if _scene_mode == "pulse": + try: + from .ltx_enhancements import make_pulse_temporal_wrapper + except ImportError: + from ltx_enhancements import make_pulse_temporal_wrapper + _tw = make_pulse_temporal_wrapper(_cur_wrapper) + if _tw is not None: + _temporal_prev_wrapper = _cur_wrapper + model.model_options["model_function_wrapper"] = _tw + _temporal_applied = True + elif _scene_mult is not None and abs(_scene_mult - 1.0) >= 1e-3: try: from .ltx_enhancements import make_temporal_wrapper except ImportError: from ltx_enhancements import make_temporal_wrapper - _cur_wrapper = model.model_options.get("model_function_wrapper") _tw = make_temporal_wrapper(_cur_wrapper, _scene_mult) if _tw is not None: _temporal_prev_wrapper = _cur_wrapper diff --git a/tests/test_pulse_temporal.py b/tests/test_pulse_temporal.py new file mode 100644 index 00000000..1a056c74 --- /dev/null +++ b/tests/test_pulse_temporal.py @@ -0,0 +1,42 @@ +"""Unit tests for Pulse temporal style schedule math.""" + +import importlib.util +from pathlib import Path + +_root = Path(__file__).resolve().parents[1] +_spec = importlib.util.spec_from_file_location("ltx_enhancements", _root / "ltx_enhancements.py") +_mod = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_mod) + +pulse_mult_for_progress = _mod.pulse_mult_for_progress +PULSE_PEAK_MULT = _mod.PULSE_PEAK_MULT +PULSE_FLOOR_MULT = _mod.PULSE_FLOOR_MULT + + +def test_pulse_starts_at_peak(): + mult = pulse_mult_for_progress(0.0, segment_count=3) + assert abs(mult - PULSE_PEAK_MULT) < 1e-6 + + +def test_pulse_segment_boundaries_reset_toward_peak(): + # End of segment 0 (progress ~1/3) should be near floor; start of segment 1 resets toward peak. + end_seg0 = pulse_mult_for_progress(1.0 / 3.0 - 1e-6, segment_count=3) + start_seg1 = pulse_mult_for_progress(1.0 / 3.0, segment_count=3) + assert end_seg0 > PULSE_PEAK_MULT + assert abs(start_seg1 - PULSE_PEAK_MULT) < 1e-6 + + +def test_pulse_eases_down_within_segment(): + start = pulse_mult_for_progress(0.0, segment_count=1) + end = pulse_mult_for_progress(1.0, segment_count=1) + mid = pulse_mult_for_progress(0.5, segment_count=1) + assert start < mid < end + assert abs(end - PULSE_FLOOR_MULT) < 1e-6 + + +def test_pulse_three_segments(): + samples = [pulse_mult_for_progress(p, segment_count=3) for p in (0.0, 0.16, 0.33, 0.5, 0.66, 0.99)] + # Each segment should show an upward jump at boundaries (reset to peak). + assert samples[0] < samples[1] < samples[2] + assert samples[2] > samples[3] + assert samples[3] < samples[4] \ No newline at end of file diff --git a/web/funpack_studio.js b/web/funpack_studio.js index 2f8a58b0..b41fda19 100644 --- a/web/funpack_studio.js +++ b/web/funpack_studio.js @@ -8,7 +8,7 @@ const LORA_TYPES = ["general", "action", "style", "quality", "character"]; const ADVISOR_DTYPES = ["bfloat16", "float16", "float32"]; const REFINER_MODES = ["Refine", "Prompt only", "Learning"]; const ADVISOR_MODES = ["Off", "Only diagnostics", "Only prompt", "Full"]; -const TEMPORAL_STYLES = ["natural", "auto", "accelerate", "decelerate", "loop", "freeze"]; +const TEMPORAL_STYLES = ["natural", "auto", "accelerate", "decelerate", "loop", "freeze", "pulse"]; const SB_MODES = ["Pass-through", "Manual", "Auto", "Learning"]; const CATEGORY_ORDER = ["action", "camera", "subject", "appearance", "environment", "style", "quality", "details"]; const TABS = ["Session", "Scene", "Shortcuts", "Transitions", "Refiner", "Advisor", "LoRA", "Sampler", "Adjustments", "Timeline"]; @@ -1209,6 +1209,7 @@ function openPanel(node) { "natural = no change. auto = per-scene director picks motion energy from each scene's prompt (needs the Scene Chain Sampler). " + "accelerate = faster, snappier motion. decelerate = heavier, slower motion. " + "loop = circular temporal coords (experimental). freeze = highly compressed time (experimental). " + + "pulse = repeated ease-down motion holds per scene (needs the Scene Chain Sampler). " + "Attention anchors from Perfect-rated outputs are automatically captured and injected into future runs.")); if (!settings.refiner.temporal_style) settings.refiner.temporal_style = "natural"; const temporalSelect = selectEl(TEMPORAL_STYLES, settings.refiner.temporal_style);