fix: timed phrases reach the sampler when a wired conditioning owns the prompt
The editor strips the brackets before the encoding node sees them and hands the windows to Studio as char spans on the text it gave that node. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
25e5cc8de0
commit
9a3ea8c35c
+35
-3
@@ -11961,14 +11961,29 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner):
|
||||
out.append([cond, meta] if isinstance(entry, list) else (cond, meta))
|
||||
return out
|
||||
|
||||
def _v2_apply_h3_timed_phrases(self, conditioning_list, clip):
|
||||
def _v2_apply_h3_timed_phrases(self, conditioning_list, clip, link_texts=None):
|
||||
"""Locate each scene's timed phrases as token spans for the sampler's attention bias.
|
||||
|
||||
`_v2_encode_prompt` stripped the `@t0-t1` syntax and recorded the char spans against
|
||||
the clean text it encoded; this maps them onto that text's tokens (offset mapping,
|
||||
no re-encode) and stores them next to the rating-learned spans in
|
||||
`funpack_h3_token_weights`. Always on: the user typed the window.
|
||||
|
||||
A WIRED conditioning was encoded by another node from the editor's expanded text
|
||||
(`link_texts`): the editor stripped the syntax before that node saw it and hands the
|
||||
windows over as `link_texts["timed"]`, char spans on `link_texts["prompt"]`. The
|
||||
text that was actually encoded is verified against the tensor's own tag run
|
||||
(`choose_encoded_text`), same as phrase emphasis, so the spans land on real tokens.
|
||||
"""
|
||||
link_timed = (link_texts or {}).get("timed") if isinstance(link_texts, dict) else None
|
||||
if link_timed and isinstance(conditioning_list, list):
|
||||
conditioning_list = [
|
||||
([e[0], dict(e[1], funpack_h3_timed=[tuple(t) for t in link_timed],
|
||||
funpack_h3_timed_link=True)]
|
||||
if isinstance(e, (list, tuple)) and len(e) >= 2 and isinstance(e[1], dict)
|
||||
and e[1].get("funpack_conditioning_owner") == "wired"
|
||||
and not e[1].get("funpack_h3_timed") else e)
|
||||
for e in conditioning_list]
|
||||
try:
|
||||
try:
|
||||
from . import minimax_h3 as _h3
|
||||
@@ -11990,8 +12005,25 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner):
|
||||
tokenizer = tokenizer or _tw.h3_tokenizer(clip)
|
||||
if tokenizer is None:
|
||||
return conditioning_list
|
||||
text = str(meta.get("funpack_h3_timed_text") or "")
|
||||
cond_len = int(cond.shape[1]) if hasattr(cond, "shape") and cond.dim() >= 2 else 0
|
||||
if meta.pop("funpack_h3_timed_link", False):
|
||||
text, _n, verified = _tw.choose_encoded_text(
|
||||
tokenizer, [link_texts.get("full_prompt"), link_texts.get("prompt")],
|
||||
meta.get("minimax_token_tags"), cond_len)
|
||||
if text is None:
|
||||
_log.feature("FunPackStudio", "H3 timed phrases", False,
|
||||
"the positive CONDITIONING is wired and neither editor "
|
||||
"text tokenizes to the tensor's prompt run, so the "
|
||||
"windows cannot be placed.")
|
||||
out.append(entry)
|
||||
continue
|
||||
if verified is not None:
|
||||
meta.setdefault("funpack_h3_token_weights", {})
|
||||
tw0 = dict(meta["funpack_h3_token_weights"])
|
||||
tw0.setdefault("base", int(verified))
|
||||
meta["funpack_h3_token_weights"] = tw0
|
||||
else:
|
||||
text = str(meta.get("funpack_h3_timed_text") or "")
|
||||
enc = tokenizer(text, add_special_tokens=False, return_offsets_mapping=True)
|
||||
offsets = list(enc["offset_mapping"])
|
||||
spans = []
|
||||
@@ -12269,7 +12301,7 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner):
|
||||
auto_strength=auto_strength,
|
||||
variables=variables,
|
||||
link_texts=link_texts)
|
||||
out = self._v2_apply_h3_timed_phrases(out, clip)
|
||||
out = self._v2_apply_h3_timed_phrases(out, clip, link_texts=link_texts)
|
||||
if mode in ("relative", "both"):
|
||||
out = _step("relative steering", out, lambda c: self._v2_apply_scene_refinement_keys(
|
||||
c, scene_refinement_keys, refinement_key,
|
||||
|
||||
@@ -100,11 +100,29 @@ def expand_prompt_for_node(text: str, variables=None) -> str:
|
||||
Peeks at the shortcut revolver rather than committing it: the run's own expansion inside
|
||||
Studio is what advances the no-repeat cycle, and doing it twice would skip a slot.
|
||||
"""
|
||||
return expand_prompt_with_windows(text, variables)[0]
|
||||
|
||||
|
||||
def expand_prompt_with_windows(text: str, variables=None):
|
||||
"""`expand_prompt_for_node`, also returning the timed-phrase windows it stripped.
|
||||
|
||||
`[walks left@2.0-3.5]` is an instruction for the sampler, not prompt text: a node that
|
||||
encodes the string itself must never see the brackets, and Studio - which never sees the
|
||||
encoded string - needs the windows handed to it separately, as char spans on the clean
|
||||
text. -> (clean_expanded_text, [(start, end, weight, t0, t1), ...])
|
||||
"""
|
||||
expanded = expand_prompt_fragment(text)
|
||||
if not expanded or not variables:
|
||||
return expanded
|
||||
resolve_variables = _funpack_attr("templates", "resolve_variables")
|
||||
return resolve_variables(expanded, variables)[0]
|
||||
if expanded and variables:
|
||||
resolve_variables = _funpack_attr("templates", "resolve_variables")
|
||||
expanded = resolve_variables(expanded, variables)[0]
|
||||
if not expanded:
|
||||
return expanded, []
|
||||
try:
|
||||
parse_timed = _funpack_attr("h3_token_weights", "parse_timed")
|
||||
clean, timed = parse_timed(expanded)
|
||||
return clean, [list(t) for t in timed]
|
||||
except Exception: # noqa: BLE001 - FunPack unreachable: the text goes out as typed
|
||||
return expanded, []
|
||||
|
||||
|
||||
def parse_timeline_raw(prompt: str) -> dict:
|
||||
|
||||
@@ -521,7 +521,8 @@ def build(object_info: dict, models_config: dict, params: dict, media: dict | No
|
||||
# anything that has to line up with those tokens is guessing.
|
||||
_exp = params.get("expanded")
|
||||
if isinstance(_exp, dict) and _exp:
|
||||
_rf["link_texts"] = {k: str(v) for k, v in _exp.items() if isinstance(v, str)}
|
||||
_rf["link_texts"] = {k: (str(v) if isinstance(v, str) else v)
|
||||
for k, v in _exp.items() if isinstance(v, (str, list))}
|
||||
# Project `$name` variables — Studio resolves them dead-last (after split), per scene.
|
||||
_vars = params.get("variables")
|
||||
if isinstance(_vars, (list, dict)) and _vars:
|
||||
|
||||
@@ -744,7 +744,7 @@ def _expanded_link_texts(target: Project, prompt: str) -> dict[str, str]:
|
||||
def ex(text):
|
||||
return bridge.expand_prompt_for_node(text, variables)
|
||||
|
||||
body = ex(prompt)
|
||||
body, timed = bridge.expand_prompt_with_windows(prompt, variables)
|
||||
postfix = ex(effective_postfix(target))
|
||||
return {
|
||||
"prompt": body,
|
||||
@@ -752,6 +752,9 @@ def _expanded_link_texts(target: Project, prompt: str) -> dict[str, str]:
|
||||
"postfix": postfix,
|
||||
"negative_prompt": ex(effective_negative_prompt(target)),
|
||||
"full_prompt": " ".join(t for t in (body, postfix) if t),
|
||||
# Timed phrases stripped from `prompt`, as char spans on it. `full_prompt` starts with
|
||||
# `prompt`, so the same spans hold there.
|
||||
"timed": timed,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -873,3 +873,47 @@ def test_a_phrase_may_contain_parentheses_colons_and_at_signs():
|
||||
clean, spans = tw.parse_timed("[camera: (slow) push-in, mail@x.com @1-2] [fast:0.5@2-3]")
|
||||
assert clean == "camera: (slow) push-in, mail@x.com fast"
|
||||
assert spans == [(0, 34, 1.0, 1.0, 2.0), (35, 39, 0.5, 2.0, 3.0)]
|
||||
|
||||
|
||||
# --- timed phrases reach the sampler on both encode paths ----------------------
|
||||
|
||||
def _quiet(monkeypatch):
|
||||
import conditioning
|
||||
import minimax_h3
|
||||
monkeypatch.setattr(minimax_h3, "is_h3_clip", lambda c: True)
|
||||
monkeypatch.setattr(conditioning, "_log",
|
||||
types.SimpleNamespace(failed=lambda *a, **k: None,
|
||||
note_on_change=lambda *a, **k: None,
|
||||
feature=lambda *a, **k: None), raising=False)
|
||||
|
||||
|
||||
def test_studio_encoded_timed_phrases_become_token_windows(refiner, monkeypatch):
|
||||
_quiet(monkeypatch)
|
||||
clean, timed = tw.parse_timed("cat [runs@1-2]")
|
||||
meta = {"funpack_h3_timed": timed, "funpack_h3_timed_text": clean,
|
||||
"minimax_token_tags": [1] * 8}
|
||||
out = refiner._v2_apply_h3_timed_phrases([[torch.zeros(1, 8, 4), meta]], _H3Clip())
|
||||
tag = out[0][1]["funpack_h3_token_weights"]
|
||||
assert tag["prompt_tokens"] == 8 and tag["base"] == 0
|
||||
assert tag["timed"] == [(4, 8, 1.0, 1.0, 2.0)]
|
||||
|
||||
|
||||
def test_wired_conditioning_takes_its_windows_from_the_editor(refiner, monkeypatch):
|
||||
"""The editor stripped the syntax before the encoding node saw it and hands the windows
|
||||
over as char spans on the text it gave that node."""
|
||||
_quiet(monkeypatch)
|
||||
link = {"prompt": "cat runs", "full_prompt": "cat runs", "timed": [[4, 8, 1.0, 1.0, 2.0]]}
|
||||
# reference block (tag 0) in front, then the 8-token prompt
|
||||
meta = {"funpack_conditioning_owner": "wired", "minimax_token_tags": [0, 0, 0] + [1] * 8}
|
||||
out = refiner._v2_apply_h3_timed_phrases([[torch.zeros(1, 11, 4), meta]], _H3Clip(),
|
||||
link_texts=link)
|
||||
tag = out[0][1]["funpack_h3_token_weights"]
|
||||
assert tag["base"] == 3 and tag["timed"] == [(4, 8, 1.0, 1.0, 2.0)]
|
||||
|
||||
|
||||
def test_a_wired_conditioning_without_editor_windows_is_left_alone(refiner, monkeypatch):
|
||||
_quiet(monkeypatch)
|
||||
meta = {"funpack_conditioning_owner": "wired", "minimax_token_tags": [1] * 8}
|
||||
out = refiner._v2_apply_h3_timed_phrases([[torch.zeros(1, 8, 4), meta]], _H3Clip(),
|
||||
link_texts={"prompt": "cat runs"})
|
||||
assert "funpack_h3_token_weights" not in out[0][1]
|
||||
|
||||
Reference in New Issue
Block a user