From 5d419c97d9a67255f1f8be7b50efb5bfc1f427e2 Mon Sep 17 00:00:00 2001 From: digital-garbage <133010426+digital-garbage@users.noreply.github.com> Date: Wed, 30 Sep 2026 20:10:19 +0300 Subject: [PATCH] feat: reactive focus backend: per-shot focus options, user choices and learning Co-Authored-By: Claude Sonnet 5.5 --- conditioning.py | 41 +++++++++++++++--- focus_memory.py | 72 +++++++++++++++++++++++++++----- movie_editor/server.py | 42 +++++++++++++++++++ shot_camera.py | 85 ++++++++++++++++++++++++++++++++------ tests/test_focus_memory.py | 65 +++++++++++++++++++++++++++++ 5 files changed, 276 insertions(+), 29 deletions(-) diff --git a/conditioning.py b/conditioning.py index a08feaa2..64753af4 100644 --- a/conditioning.py +++ b/conditioning.py @@ -775,6 +775,30 @@ def split_prompt_by_transitions(prompt, placement="start"): return [(text, None)] +def focus_scene_texts(scene_segments, variables=None, placement="start"): + """Each scene's prompt as the camera-move step will see it: shortcuts expanded, the anchor + and postfix folded in, $variables resolved (before cuts, views and moves, whose random + draws belong to a run). What the Reactive focus review shows, so a choice is about real text.""" + canonical = split_scenes_from_segments( + scene_segments.get("anchor", ""), scene_segments.get("scenes", []), + placement=placement, postfix=scene_segments.get("postfix", "")) + anchor = (canonical.get("anchor_expanded", "") or "").strip() + postfix = (canonical.get("postfix_expanded", "") or "").strip() + texts = [] + for s in canonical.get("scenes", []) or []: + seg = (s.get("expanded") or "").strip() + if anchor and seg: + seg = anchor + (" " if anchor[-1] in ".!?," else ", ") + seg + else: + seg = anchor or seg + if postfix and seg: + seg = seg + (" " if seg[-1] in ".!?," else ", ") + postfix + elif postfix: + seg = postfix + texts.append(_resolve_variables(seg, variables)[0] if variables else seg) + return texts + + def _shortcut_texts(min_len=15): """Every enabled shortcut's replacement text, longest first: how shot cuts tell where one of the user's shortcuts ends and the next begins in an already-expanded prompt.""" @@ -9344,7 +9368,7 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): return out @staticmethod - def _v2_camera_moves(text, where, seed=None, chance=1.0): + def _v2_camera_moves(text, where, seed=None, chance=1.0, choices=None): """`text` with a camera move added to each [Shot N] block that lacks one.""" try: try: @@ -9360,7 +9384,12 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): except Exception as e: # noqa: BLE001 _log.failed("FunPackVideoRefinerV2", "focus memory", e, "targets are scored without it") prior = {} - out, report = _sc.add_camera_moves(text, seed=seed, chance=chance, prior=prior) + try: + chance = _fm.effective_chance(chance) # pulled toward how often moves are kept + except Exception: # noqa: BLE001 + pass + out, report = _sc.add_camera_moves(text, seed=seed, chance=chance, prior=prior, + choices=choices if isinstance(choices, dict) else None) if report: try: _fm.observe(_sc.content_fingerprint(text), @@ -10250,7 +10279,7 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): steer_mode="relative", absolute_strength=0.6, h3_phrase_emphasis=False, h3_phrase_variability=0.0, camera_moves=False, camera_moves_chance=0.7, shot_cuts=False, shot_cut_chance=0.0, - shot_views=False, shot_view_chance=0.4, + shot_views=False, shot_view_chance=0.4, focus_choices=None, prompt_enhance=False, prompt_enhance_system="", prompt_enhance_reference_intro="", prompt_enhance_temperature=0.7, prompt_enhance_top_p=0.92, prompt_enhance_max_length=400, prompt_enhance_thinking=False, @@ -10932,7 +10961,8 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): prompt_to_encode = self._v2_shot_views(prompt_to_encode, "prompt", seed, shot_view_chance) _base_before_camera = prompt_to_encode if camera_moves and not split_by_transitions: - prompt_to_encode = self._v2_camera_moves(prompt_to_encode, "prompt", seed, camera_moves_chance) + prompt_to_encode = self._v2_camera_moves(prompt_to_encode, "prompt", seed, camera_moves_chance, + focus_choices) # ── prompt enhancer ──────────────────────────────────────────────────────────── # Runs BEFORE the prompt becomes conditioning, so what ComfyUI generates from is the @@ -11259,7 +11289,7 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): _scene_before_camera = list(split_scene_texts or []) if camera_moves and split_scene_texts: split_scene_texts = [self._v2_camera_moves(t, f"scene {_i + 1}", f"{seed}:{_i}", - camera_moves_chance) + camera_moves_chance, focus_choices) for _i, t in enumerate(split_scene_texts)] # Enhanced PER SCENE, not as one blob. The editor's scene list is # authoritative on count, so rewriting the scenes as a single paragraph @@ -14278,6 +14308,7 @@ class FunPackStudio: camera_moves_chance=float(rf.get("camera_moves_chance", 0.7)), shot_cuts=bool(rf.get("shot_cuts", False)), shot_cut_chance=float(rf.get("shot_cut_chance", 0.0)), + focus_choices=(rf.get("focus_choices") if isinstance(rf.get("focus_choices"), dict) else None), shot_views=bool(rf.get("shot_views", False)), shot_view_chance=float(rf.get("shot_view_chance", 0.4)), h3_phrase_variability=float(rf.get("h3_phrase_variability", 0.0) or 0.0), diff --git a/focus_memory.py b/focus_memory.py index fa07d9cd..552c0cf4 100644 --- a/focus_memory.py +++ b/focus_memory.py @@ -18,6 +18,9 @@ import os MIN_PROMPTS = 3 FREQ_W = 0.6 # max bonus: reorders near-equals, never outranks an owned part MAX_HASHES = 300 +PICK_W = 0.9 # bonus a word the user has chosen grows toward (tanh of picks) +REJECT_W = 0.6 # penalty a word the rewriter proposed and the user replaced grows toward +MIN_SHOTS_FOR_RATE = 10 # shots reviewed before "how often the user wants a move" is trusted def _path(): @@ -43,13 +46,65 @@ def _read(): def prior(): - """{lemma: bonus}, or {} while too few prompts have been seen.""" + """{lemma: bonus}: words that keep coming back across prompts (once enough prompts were + seen), plus what the user's own choices said: up for words they picked over the rewriter's + proposal, down for the words they replaced.""" data = _read() seen = data.get("seen") or {} - if int(data.get("prompts", 0)) < MIN_PROMPTS or not seen: - return {} - top = math.log1p(max(seen.values())) - return {lemma: FREQ_W * math.log1p(n) / top for lemma, n in seen.items()} + out = {} + if int(data.get("prompts", 0)) >= MIN_PROMPTS and seen: + top = math.log1p(max(seen.values())) + out = {lemma: FREQ_W * math.log1p(n) / top for lemma, n in seen.items()} + for lemma, n in (data.get("picks") or {}).items(): + out[lemma] = out.get(lemma, 0.0) + PICK_W * math.tanh(n / 2.0) + for lemma, n in (data.get("rejects") or {}).items(): + out[lemma] = out.get(lemma, 0.0) - REJECT_W * math.tanh(n / 2.0) + return out + + +def learn(decisions): + """Remember what a person chose. `decisions`: [{"auto": lemma|None, "picked": lemma|None, + "mode": "auto"|"move"|"hold"|"none"}], one per shot shown. A picked word is reinforced; the + word the rewriter proposed instead is marked down when they differ; "none" counts against + moves in general (see move_rate). -> shots remembered.""" + data = _read() + picks, rejects = data.get("picks") or {}, data.get("rejects") or {} + total, kept = int(data.get("shots", 0)), int(data.get("kept", 0)) + n = 0 + for d in decisions or []: + mode = d.get("mode") or "auto" + n += 1 + total += 1 + kept += mode != "none" + if mode == "none": + continue + picked, auto = d.get("picked"), d.get("auto") + if picked: + picks[picked] = picks.get(picked, 0) + 1 + if auto and auto != picked: + rejects[auto] = rejects.get(auto, 0) + 1 + data.update(picks=picks, rejects=rejects, shots=total, kept=kept) + _save(data) + return n + + +def effective_chance(chance): + """The configured chance of a move, pulled toward how often the user actually keeps one + (of the shots they reviewed) once enough were reviewed.""" + data = _read() + total, kept = int(data.get("shots", 0)), int(data.get("kept", 0)) + if total < MIN_SHOTS_FOR_RATE: + return chance + return max(0.05, min(1.0, (chance + kept / total) / 2.0)) + + +def _save(data): + path = _path() + os.makedirs(os.path.dirname(path), exist_ok=True) + tmp = path + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + json.dump(data, f) + os.replace(tmp, path) def observe(prompt_hash, lemmas): @@ -65,10 +120,5 @@ def observe(prompt_hash, lemmas): seen[lemma] = seen.get(lemma, 0) + 1 data.update(seen=seen, prompts=int(data.get("prompts", 0)) + 1, hashes=(hashes + [prompt_hash])[-MAX_HASHES:]) - path = _path() - os.makedirs(os.path.dirname(path), exist_ok=True) - tmp = path + ".tmp" - with open(tmp, "w", encoding="utf-8") as f: - json.dump(data, f) - os.replace(tmp, path) + _save(data) return data["prompts"] diff --git a/movie_editor/server.py b/movie_editor/server.py index af2db475..b43a44ac 100644 --- a/movie_editor/server.py +++ b/movie_editor/server.py @@ -3104,6 +3104,48 @@ if web is not None and PromptServer is not None: raise web.HTTPInternalServerError(reason=f"Could not render the card: {e}") return web.Response(body=png, content_type="image/png") + # ── Reactive focus: the camera targets of every [Shot N], for a person to choose from + # BEFORE Generate. The options come from the same text the camera-move step will see. + @routes.post(UI_PREFIX + "/api/projects/{pid}/focus_options") + async def _focus_options(req): + import asyncio + pid = req.match_info["pid"] + body = await req.json() if req.can_read_body else {} + p = _project_or_404(pid) + scene_ids = body.get("scene_ids") + target = _segment(p, scene_ids) if scene_ids else p + + def _build(): + seg = build_generation_scene_segments(target) + texts = bridge._funpack_attr("conditioning", "focus_scene_texts")( + seg, list(target.variables or []), "start") + sc = bridge._funpack_attr("shot_camera", "focus_options") + prior = bridge._funpack_attr("focus_memory", "prior")() + out = [] + for i, text in enumerate(texts): + shots = sc(text, prior) + if shots: + out.append({"index": i, "preview": " ".join(text.split())[:80], "shots": shots}) + return out + + try: + scenes = await asyncio.to_thread(_build) + except Exception as e: # noqa: BLE001 + return web.json_response({"detail": f"Could not list camera targets: {e}"}, status=500) + return web.json_response({"scenes": scenes}) + + @routes.post(UI_PREFIX + "/api/focus/learn") + async def _focus_learn(req): + import asyncio + body = await req.json() if req.can_read_body else {} + decisions = [d for d in (body.get("decisions") or []) if isinstance(d, dict)] + learn = bridge._funpack_attr("focus_memory", "learn") + try: + n = await asyncio.to_thread(learn, decisions) + except Exception as e: # noqa: BLE001 + return web.json_response({"detail": f"Could not remember the choices: {e}"}, status=500) + return web.json_response({"remembered": n}) + @routes.get(UI_PREFIX + "/api/git/status") async def _git_status(_req): return web.json_response(git_update.status()) diff --git a/shot_camera.py b/shot_camera.py index 8be5d5b5..aa5d087e 100644 --- a/shot_camera.py +++ b/shot_camera.py @@ -263,7 +263,7 @@ def _draw(cands, prior, rng, n): return chosen -def _plan(pool, last, rng, prior=None): +def _plan(pool, last, rng, prior=None, force=None, mode="auto"): """-> (moves text, target description, opening template). Fixed and minimal without `rng` (one move, X -> Y when the shot has two topics); with it, how many moves, whether they travel and which words are all drawn, so no shot pattern repeats by construction.""" @@ -271,7 +271,15 @@ def _plan(pool, last, rng, prior=None): best = _draw(pool, prior, rng, 1)[0] if prior and len(topics) > 2: # two of them, weighted by habit, in text order topics = sorted(_draw(topics, prior, rng, 2), key=lambda c: c[3]) - if rng is None: + if force is not None: # the user picked the thing to aim at + best, topics = force, [force] + if mode == "hold": + hold = _pick(MOVES_HOLD, last, rng) + target = _name(best) + return hold.format(x=target), target, hold + if mode == "move": + k, travel = 1, False + elif rng is None: k, travel = 1, len(topics) >= 2 else: k = rng.choices((1, 2, 3), COUNT_WEIGHTS)[0] @@ -302,15 +310,13 @@ def _plan(pool, last, rng, prior=None): return " ".join(sentences), desc, opening -def add_camera_moves(text, seed=None, chance=1.0, prior=None): - """-> (new prompt, [per-shot report dicts]). Unchanged when there is no `[Shot N]`. - - `seed=None`: one plain move per shot (deterministic). With a `seed`, each shot is left - alone with probability 1 - `chance`, and otherwise gets one to three moves, drawn per - shot from the seed, so the same prompt and seed always give the same text.""" +def _analyse(text): + """-> (marks, header, parts, pools): the `[Shot N]` markers, the text before the first, each + shot's (picture, sound) and its candidate targets with the things that are in (nearly) every + shot, or in the intro, taken out.""" marks = list(SHOT.finditer(text or "")) if not marks: - return text, [] + return [], "", [], [] header = text[:marks[0].start()] bodies = [text[m.end():(marks[i + 1].start() if i + 1 < len(marks) else len(text))] for i, m in enumerate(marks)] @@ -325,12 +331,63 @@ def add_camera_moves(text, seed=None, chance=1.0, prior=None): for lemma in {c[0] for c in lst}: seen[lemma] = seen.get(lemma, 0) + 1 constant = {l for l, n in seen.items() if n >= need} | header_lemmas + return marks, header, parts, [[c for c in lst if c[0] not in constant] for lst in cands] + + +def shot_key(picture): + """A stable name for a shot's content: the same whatever cut opener, view or camera sentence + was added, so a choice made about a shot still finds it after the prompt is regenerated.""" + import hashlib + body = own_words_removed(picture or "") + body = re.sub(r"\s+", " ", body).strip().lower() + return hashlib.md5(body.encode("utf-8", "replace")).hexdigest()[:16] + + +def focus_options(text, prior=None): + """What the camera could aim at in each `[Shot N]`, for a person to choose from. + -> [{"shot", "key", "already", "candidates": [{"lemma", "text", "score"}], "auto": str|None}] + `auto` is what the rewriter would pick with no say from anyone (its plain, top choice).""" + marks, _header, parts, pools = _analyse(text) + out = [] + for i, m in enumerate(marks): + picture = parts[i][0] + pool = pools[i] + opts, seen = [], set() + for c in sorted(pool, key=lambda c: -_score(c, prior)): + if c[0] in seen or _score(c) <= 0.5: + continue + seen.add(c[0]) + opts.append({"lemma": c[0], "text": _name(c), "score": round(_score(c, prior), 2)}) + auto = None + if opts and not CAMERA.search(CUT.sub("", picture)): + auto = _plan(pool, None, None, prior)[1] + out.append({"shot": int(m.group(1)), "key": shot_key(picture), + "auto_lemma": opts[0]["lemma"] if opts else None, + "already": bool(CAMERA.search(CUT.sub("", picture))), + "candidates": opts, "auto": auto}) + return out + + +def add_camera_moves(text, seed=None, chance=1.0, prior=None, choices=None): + """-> (new prompt, [per-shot report dicts]). Unchanged when there is no `[Shot N]`. + + `seed=None`: one plain move per shot (deterministic). With a `seed`, each shot is left + alone with probability 1 - `chance`, and otherwise gets one to three moves, drawn per + shot from the seed, so the same prompt and seed always give the same text.""" + marks, header, parts, pools = _analyse(text) + if not marks: + return text, [] report, out, last = [], [header], None for i, m in enumerate(marks): picture, sound = parts[i] - entry = {"shot": int(m.group(1)), "move": None, "target": None, "why": "", "lemmas": []} + entry = {"shot": int(m.group(1)), "move": None, "target": None, "why": "", "lemmas": [], + "key": shot_key(picture)} rng = random.Random(f"{seed}:{i}") if seed is not None else None - pool = [c for c in cands[i] if c[0] not in constant] + pool = pools[i] + # What the person said about this shot, if anything: "none", or a target and/or a mode. + said = (choices or {}).get(entry["key"]) or {} + mode = said.get("mode") or "auto" + force = next((c for c in pool if c[0] == said.get("lemma")), None) if said.get("lemma") else None entry["lemmas"] = [c[0] for c in pool if _score(c) > 0.5] # the words this shot dwells on # A word the shot keeps coming back to is what it is about: a small lift per extra mention. mentions = {} @@ -343,10 +400,12 @@ def add_camera_moves(text, seed=None, chance=1.0, prior=None): entry["why"] = "already has a camera move" elif not pool: entry["why"] = "nothing specific to aim at" - elif rng is not None and rng.random() >= chance: + elif mode == "none": + entry["why"] = "no move, as you chose" + elif rng is not None and mode == "auto" and force is None and rng.random() >= chance: entry["why"] = "left as written by chance" else: - moves, desc, opening = _plan(pool, last, rng, shot_prior) + moves, desc, opening = _plan(pool, last, rng, shot_prior, force=force, mode=mode) entry.update(move=moves, target=desc) last = opening core = picture.rstrip() diff --git a/tests/test_focus_memory.py b/tests/test_focus_memory.py index 72ba6cf1..951c4361 100644 --- a/tests/test_focus_memory.py +++ b/tests/test_focus_memory.py @@ -127,3 +127,68 @@ def test_a_word_the_shot_keeps_returning_to_is_what_it_is_about(): P = "Intro. [Shot 1] lifts the cherry, then the lamp, then looks at the cherry again." hits = sum("cherry" in sc.add_camera_moves(P, seed=s, chance=1.0)[1][0]["target"] for s in range(80)) assert hits > 50, hits + + +def test_choices_reinforce_what_was_picked_and_mark_down_what_was_replaced(store): + fm.learn([{"auto": "lamp", "picked": "chin", "mode": "auto"}] * 3) + p = fm.prior() + assert p["chin"] > 0.5 > 0 > p["lamp"] + fm.learn([{"auto": "lamp", "picked": "lamp", "mode": "auto"}]) + assert fm.prior()["lamp"] > p["lamp"] - 1 # agreeing does not punish + + +def test_no_move_counts_against_moves_not_against_words(store): + fm.learn([{"auto": "chin", "picked": None, "mode": "none"}] * 12) + assert fm.prior() == {} + assert fm.effective_chance(0.7) < 0.4 + assert fm.effective_chance(1.0) < 0.6 + + +def test_the_chance_is_left_alone_until_enough_shots_were_reviewed(store): + fm.learn([{"auto": "chin", "picked": None, "mode": "none"}] * 3) + assert fm.effective_chance(0.7) == 0.7 + + +def test_a_chosen_target_and_mode_are_obeyed(): + pytest.importorskip("spacy") + P = "Intro. [Shot 1] raises the lamp and lowers the mirror." + opts = sc.focus_options(P)[0] + key = opts["key"] + assert {c["lemma"] for c in opts["candidates"]} >= {"lamp", "mirror"} + for seed in range(10): + _o, rep = sc.add_camera_moves(P, seed=seed, chance=1.0, + choices={key: {"mode": "hold", "lemma": "mirror"}}) + assert "mirror" in rep[0]["move"] and ("holds" in rep[0]["move"] or "stays" in rep[0]["move"]) + _o, rep = sc.add_camera_moves(P, seed=seed, chance=1.0, + choices={key: {"mode": "move", "lemma": "lamp"}}) + assert "lamp" in rep[0]["move"] and "mirror" not in rep[0]["move"] and " Then" not in rep[0]["move"] + _o, rep = sc.add_camera_moves(P, seed=seed, chance=1.0, choices={key: {"mode": "none"}}) + assert rep[0]["move"] is None and "you chose" in rep[0]["why"] + + +def test_a_choice_survives_cut_openers_and_views_added_to_the_shot(): + pytest.importorskip("spacy") + base = "Intro. [Shot 1] waves. [Shot 2] raises the lamp." + viewed = ("Intro. [Shot 1] waves. [Shot 2] At 00:04.000, the camera cuts to a new angle. " + "Side view. raises the lamp.") + assert sc.focus_options(base)[1]["key"] == sc.focus_options(viewed)[1]["key"] + + +def test_focus_scene_texts_and_options_follow_the_generation_path(monkeypatch): + """The review shows the text the camera-move step will see: shortcuts expanded, anchor and + postfix folded in, one text per scene.""" + pytest.importorskip("spacy") + import conditioning + import templates + db = {"shortcuts": {"k": {"name": "k", "enabled": True, "triggers": ["tip"], + "replacements": ["[Shot 1] touches 's chin and raises the lamp."], + "refinement_key": "", "category": "", "sub_category": ""}}} + monkeypatch.setattr(templates, "load_shortcut_db", lambda: db) + monkeypatch.setattr(templates, "load_custom_transition_triggers", lambda: {}) + texts = conditioning.focus_scene_texts( + {"anchor": "Intro .", "scenes": ["tip", "[Shot 1] waves."], "postfix": ""}, + [], "start") + assert len(texts) == 2 and texts[0].startswith("Intro .") + assert "touches 's chin" in texts[0] + opts = sc.focus_options(texts[0]) + assert opts[0]["candidates"] and opts[0]["auto_lemma"]