diff --git a/conditioning.py b/conditioning.py index b192ff49..9fd3a43d 100644 --- a/conditioning.py +++ b/conditioning.py @@ -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, diff --git a/movie_editor/backend/bridge.py b/movie_editor/backend/bridge.py index 207fbb12..09a77bb4 100644 --- a/movie_editor/backend/bridge.py +++ b/movie_editor/backend/bridge.py @@ -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: diff --git a/movie_editor/backend/builder.py b/movie_editor/backend/builder.py index f13563f0..25c5097f 100644 --- a/movie_editor/backend/builder.py +++ b/movie_editor/backend/builder.py @@ -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: diff --git a/movie_editor/server.py b/movie_editor/server.py index 5d311a53..0e9e23e3 100644 --- a/movie_editor/server.py +++ b/movie_editor/server.py @@ -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, } diff --git a/tests/test_h3_token_weights.py b/tests/test_h3_token_weights.py index 162a3f51..7e953906 100644 --- a/tests/test_h3_token_weights.py +++ b/tests/test_h3_token_weights.py @@ -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]