From bc0da79ffcbeb5ef1478a0426d7eb764df4ceae3 Mon Sep 17 00:00:00 2001 From: digital-garbage <133010426+digital-garbage@users.noreply.github.com> Date: Sun, 20 Sep 2026 01:46:26 +0300 Subject: [PATCH] feat: typed (phrase:1.5) weights reach the H3 attention bias The parser existed but nothing called it; the syntax went to Qwen as text. One pass strips both markups so timed and weighted spans share one clean text; both encode paths carry them. Co-Authored-By: Claude Opus 5 --- conditioning.py | 42 +++++++++++++++++++----------- h3_token_weights.py | 43 +++++++++++++++++++++---------- movie_editor/backend/bridge.py | 12 ++++----- movie_editor/frontend/composer.js | 15 ++++++++--- movie_editor/server.py | 3 ++- tests/test_h3_token_weights.py | 27 ++++++++++++++++++- 6 files changed, 101 insertions(+), 41 deletions(-) diff --git a/conditioning.py b/conditioning.py index 49c42fa8..282f9303 100644 --- a/conditioning.py +++ b/conditioning.py @@ -4463,7 +4463,7 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): from . import h3_token_weights as _tw except ImportError: import h3_token_weights as _tw - prompt_text, timed_spans = _tw.parse_timed(prompt_text) + prompt_text, weighted_spans, timed_spans = _tw.parse_markup(prompt_text) try: if h3 and ref_items: # ref2va takes over the presentation: a first-frame anchor and free-floating @@ -4495,14 +4495,16 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): print("[FunPackStudio] Vision encoding returned invalid conditioning") return None, {"pooled_output": None}, "encode returned invalid conditioning" vision_tag = " +vision" if use_vision else "" - if timed_spans: + if timed_spans or weighted_spans: meta = dict(meta) meta["funpack_h3_timed"] = timed_spans + meta["funpack_h3_weighted"] = weighted_spans meta["funpack_h3_timed_text"] = prompt_text if not h3: _log.note_on_change("studio:timed_phrases", "FunPackStudio", - f"{len(timed_spans)} timed phrase(s) stripped from the " - f"prompt: timing is MiniMax H3 only, the words stay.") + f"{len(timed_spans)} timed / {len(weighted_spans)} weighted " + f"phrase(s) stripped from the prompt: the markup is MiniMax " + f"H3 only, the words stay.") if resolved_refs: # The resolved order travels with the conditioning: the sampler must encode these # exact references, in this exact order, or "" points at the wrong one. @@ -11985,13 +11987,15 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): (`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): + link_weighted = (link_texts or {}).get("weighted") if isinstance(link_texts, dict) else None + if (link_timed or link_weighted) and isinstance(conditioning_list, list): conditioning_list = [ - ([e[0], dict(e[1], funpack_h3_timed=[tuple(t) for t in link_timed], + ([e[0], dict(e[1], funpack_h3_timed=[tuple(t) for t in (link_timed or [])], + funpack_h3_weighted=[tuple(t) for t in (link_weighted or [])], 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) + and not (e[1].get("funpack_h3_timed") or e[1].get("funpack_h3_weighted")) else e) for e in conditioning_list] try: try: @@ -12004,10 +12008,11 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): return conditioning_list tokenizer = None out = [] - placed = 0 + placed = typed_n = 0 for entry in conditioning_list or []: if not (isinstance(entry, (list, tuple)) and len(entry) >= 2 - and isinstance(entry[1], dict) and entry[1].get("funpack_h3_timed")): + and isinstance(entry[1], dict) + and (entry[1].get("funpack_h3_timed") or entry[1].get("funpack_h3_weighted"))): out.append(entry) continue cond, meta = entry[0], dict(entry[1]) @@ -12036,14 +12041,18 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): enc = tokenizer(text, add_special_tokens=False, return_offsets_mapping=True) offsets = list(enc["offset_mapping"]) spans = [] - for a, b, w, t0, t1 in meta["funpack_h3_timed"]: + for a, b, w, t0, t1 in meta.get("funpack_h3_timed") or []: toks = _tw.token_spans_from_offsets(offsets, [(a, b, w)]) if toks: spans.append((toks[0][0], toks[0][1], float(w), float(t0), float(t1))) + # Typed `(phrase:w)` weights join the rating-learned spans: same bias, same + # cap. A typed weight is the user's call, so it is not scaled by evidence. + typed = _tw.token_spans_from_offsets( + offsets, [(a, b, w) for a, b, w in meta.get("funpack_h3_weighted") or []]) prompt_tokens = len(offsets) - if spans and prompt_tokens and cond_len: + if (spans or typed) and prompt_tokens and cond_len: tw = dict(meta.get("funpack_h3_token_weights") or {}) - tw.setdefault("spans", []) + tw["spans"] = list(tw.get("spans") or []) + [tuple(t) for t in typed] tw["prompt_tokens"] = prompt_tokens if "base" not in tw: _t, _n, verified = _tw.choose_encoded_text( @@ -12052,12 +12061,15 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): meta.get("minimax_token_tags"), cond_len, prompt_tokens) if base is not None: tw["base"] = int(base) - tw["timed"] = spans + if spans: + tw["timed"] = spans meta["funpack_h3_token_weights"] = tw placed += len(spans) + typed_n += len(typed) out.append([cond, meta] if isinstance(entry, list) else (cond, meta)) - if placed: - print(f"[FunPackStudio] H3 timed phrases: {placed} window(s) tagged for the sampler.") + if placed or typed_n: + print(f"[FunPackStudio] H3 prompt markup: {placed} timed window(s), " + f"{typed_n} weighted phrase(s) tagged for the sampler.") return out except Exception as _e: # noqa: BLE001 _log.failed("FunPackStudio", "H3 timed phrases", _e, diff --git a/h3_token_weights.py b/h3_token_weights.py index 697bf5db..1e9728fa 100644 --- a/h3_token_weights.py +++ b/h3_token_weights.py @@ -76,27 +76,42 @@ FPS = 24 FRAME_PER_TOKEN = (1, 4, 4, 4, 4) -def parse_timed(text: str): - """Strip timed-phrase syntax. -> (clean_text, [(start_char, end_char, weight, t0, t1)]). +_MARKUP = re.compile(_TIMED.pattern + "|" + _WEIGHTED.pattern) - The model must never see `[…@2.0-3.5]`: Qwen would read it as punctuation. Spans index - into `clean_text` (same contract as `parse`). A window with t1 <= t0 is dropped. + +def parse_markup(text: str): + """Strip BOTH prompt markups in one left-to-right pass. + + -> (clean_text, weighted [(start, end, weight)], timed [(start, end, weight, t0, t1)]). + One pass, so both span lists index the same clean text; stripping them one after the + other would leave the first list's offsets pointing into a string that no longer exists. """ - if not text or "@" not in text: - return text, [] - out, spans, pos = [], [], 0 - for m in _TIMED.finditer(text): + if not text or ("(" not in text and "[" not in text): + return text, [], [] + out, weighted, timed, pos = [], [], [], 0 + for m in _MARKUP.finditer(text): out.append(text[pos:m.start()]) - phrase = m.group(1) start = sum(len(p) for p in out) + if m.group(1) is not None: # [phrase(:w)?@t0-t1] + phrase = m.group(1) + t0, t1 = float(m.group(3)), float(m.group(4)) + if t1 > t0 and phrase.strip(): + timed.append((start, start + len(phrase), + float(m.group(2)) if m.group(2) else 1.0, t0, t1)) + else: # (phrase:w) + phrase = m.group(5) + weighted.append((start, start + len(phrase), float(m.group(6)))) out.append(phrase) - t0, t1 = float(m.group(3)), float(m.group(4)) - if t1 > t0 and phrase.strip(): - spans.append((start, start + len(phrase), - float(m.group(2)) if m.group(2) else 1.0, t0, t1)) pos = m.end() out.append(text[pos:]) - return "".join(out), spans + return "".join(out), weighted, timed + + +def parse_timed(text: str): + """`parse_markup` for callers that only want the windows; the clean text has BOTH + markups removed (the model must never see either).""" + clean, _weighted, timed = parse_markup(text) + return clean, timed def video_row_window(latent_t: int, frame_rows: int, t0: float, t1: float): diff --git a/movie_editor/backend/bridge.py b/movie_editor/backend/bridge.py index 09a77bb4..c3c05178 100644 --- a/movie_editor/backend/bridge.py +++ b/movie_editor/backend/bridge.py @@ -109,20 +109,20 @@ def expand_prompt_with_windows(text: str, variables=None): `[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), ...]) + text. -> (clean_expanded_text, timed [(start, end, weight, t0, t1)], weighted [(start, end, weight)]) """ expanded = expand_prompt_fragment(text) if expanded and variables: resolve_variables = _funpack_attr("templates", "resolve_variables") expanded = resolve_variables(expanded, variables)[0] if not expanded: - return 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] + parse_markup = _funpack_attr("h3_token_weights", "parse_markup") + clean, weighted, timed = parse_markup(expanded) + return clean, [list(t) for t in timed], [list(t) for t in weighted] except Exception: # noqa: BLE001 - FunPack unreachable: the text goes out as typed - return expanded, [] + return expanded, [], [] def parse_timeline_raw(prompt: str) -> dict: diff --git a/movie_editor/frontend/composer.js b/movie_editor/frontend/composer.js index 3fb7d039..cdc37f75 100644 --- a/movie_editor/frontend/composer.js +++ b/movie_editor/frontend/composer.js @@ -451,16 +451,23 @@ if (!phrase) continue; found.push({ phrase, t0, t1, w: m[2] ? parseFloat(m[2]) : 1, bad: !(t1 > t0) }); } + // (phrase:1.5) — a weight with no window. Mirrors h3_token_weights._WEIGHTED. + const weighted = [...txt.matchAll(/(? ({ phrase: m[1].trim(), w: parseFloat(m[2]) })).filter((f) => f.phrase); const PC = window.PipelineCaps; const h3 = !!(PC && PC.isH3 && PC.isH3(S.get())); let msg = ""; let warn = false; - if (found.length) { + if (found.length || weighted.length) { const bad = found.filter((f) => f.bad); - if (!h3) { msg = "⏱ Timed phrases are MiniMax H3 only — the words stay, the times are dropped."; warn = true; } + if (!h3) { msg = "⏱ Timed and weighted phrases are MiniMax H3 only — the words stay, the markup is dropped."; warn = true; } else if (bad.length) { msg = `⚠ "${bad[0].phrase}" ends before it starts (${bad[0].t0}-${bad[0].t1}s) — window ignored.`; warn = true; } - else msg = "⏱ " + found.map((f) => `${f.phrase} ${f.t0}-${f.t1}s${f.w !== 1 ? ` ×${f.w}` : ""}`).join(" · ") - + " — only shapes the picture inside its window; keep the subject outside the brackets."; + else { + const parts = found.map((f) => `${f.phrase} ${f.t0}-${f.t1}s${f.w !== 1 ? ` ×${f.w}` : ""}`) + .concat(weighted.map((f) => `${f.phrase} ×${f.w}`)); + msg = "⏱ " + parts.join(" · ") + + (found.length ? " — a window only shapes the picture inside it; keep the subject outside the brackets." : ""); + } } timedHintEl.textContent = msg; timedHintEl.classList.toggle("compose-var-warn", warn); diff --git a/movie_editor/server.py b/movie_editor/server.py index 0e9e23e3..5716c60c 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, timed = bridge.expand_prompt_with_windows(prompt, variables) + body, timed, weighted = bridge.expand_prompt_with_windows(prompt, variables) postfix = ex(effective_postfix(target)) return { "prompt": body, @@ -755,6 +755,7 @@ def _expanded_link_texts(target: Project, prompt: str) -> dict[str, str]: # Timed phrases stripped from `prompt`, as char spans on it. `full_prompt` starts with # `prompt`, so the same spans hold there. "timed": timed, + "weighted": weighted, } diff --git a/tests/test_h3_token_weights.py b/tests/test_h3_token_weights.py index 7e953906..af55d5d1 100644 --- a/tests/test_h3_token_weights.py +++ b/tests/test_h3_token_weights.py @@ -802,10 +802,17 @@ def test_variability_out_of_range_is_clamped_not_inverted(refiner, monkeypatch): def test_timed_syntax_is_stripped_and_windows_kept(): clean, spans = tw.parse_timed("a man [walks left@1.0-2.5] then [waves:1.5 @ 3-4] (x:1.2)") - assert clean == "a man walks left then waves (x:1.2)" + assert clean == "a man walks left then waves x" assert spans == [(6, 16, 1.0, 1.0, 2.5), (22, 27, 1.5, 3.0, 4.0)] +def test_both_markups_are_parsed_in_one_pass_on_one_clean_text(): + clean, weighted, timed = tw.parse_markup("a (cat:1.5) [runs@1-2] (dog:0.5)") + assert clean == "a cat runs dog" + assert weighted == [(2, 5, 1.5), (11, 14, 0.5)] + assert timed == [(6, 10, 1.0, 1.0, 2.0)] + + def test_an_empty_window_is_dropped(): assert tw.parse_timed("[walks@2-2]")[1] == [] @@ -917,3 +924,21 @@ def test_a_wired_conditioning_without_editor_windows_is_left_alone(refiner, monk 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] + + +def test_typed_weights_reach_the_sampler_next_to_the_windows(refiner, monkeypatch): + _quiet(monkeypatch) + clean, weighted, timed = tw.parse_markup("(cat:2) [runs@1-2]") + meta = {"funpack_h3_timed": timed, "funpack_h3_weighted": weighted, + "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["spans"] == [(0, 3, 2.0)] and tag["timed"] == [(4, 8, 1.0, 1.0, 2.0)] + + +def test_a_typed_weight_alone_is_enough(refiner, monkeypatch): + _quiet(monkeypatch) + meta = {"funpack_h3_weighted": [(0, 3, 1.5)], "funpack_h3_timed_text": "cat runs", + "minimax_token_tags": [1] * 8} + out = refiner._v2_apply_h3_timed_phrases([[torch.zeros(1, 8, 4), meta]], _H3Clip()) + assert out[0][1]["funpack_h3_token_weights"]["spans"] == [(0, 3, 1.5)]