Add Pulse temporal style with per-scene ease-down segments

Introduces a new temporal_style option that cycles motion→hold ramps
across each scene's denoise trajectory via a segmented frame_rate wrapper.
Wired through Studio, engine settings, chain sampler conditioning tags,
and build_enhancements for single-scene workflows.
This commit is contained in:
digital-garbage
2026-06-10 16:50:19 +03:00
parent e514c8f178
commit b72878f866
6 changed files with 176 additions and 29 deletions
+20 -3
View File
@@ -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"):
+85 -20
View File
@@ -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:
+1 -1
View File
@@ -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" },
];
+26 -4
View File
@@ -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
+42
View File
@@ -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]
+2 -1
View File
@@ -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);