diff --git a/CHANGELOG.md b/CHANGELOG.md index bc34bd8f..1d22cdfd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ - Export settings no longer prints a loader input the node no longer has (a stale `sla_enabled: on`) or SLA's settings while `sla` is off. ### Added +- Camera moves remember which words your prompts keep coming back to (per refinement key, counted once per distinct prompt) and favour them when choosing a target; silent until 3 prompts have been seen, and never outranks an owned part. - Shot cut times (H3, off by default): every `[Shot N]` after the first opens with its cut time, spread evenly over the scene's length, rounded to whole seconds, none shorter than 2 s; a shot whose main point changes is split in two. Prompts that already carry times are left alone. - Shot views (H3, off by default): later shots may open with POV / side / front / from above / below / behind, never the same twice in a row; shot 1 is skipped. - Composer ▸ Enhance shows the final prompt of every run (after shortcuts, $variables and camera moves) even with the enhancer off, one card per scene, with the text before camera moves under it. diff --git a/conditioning.py b/conditioning.py index 215666e6..3b63e1c9 100644 --- a/conditioning.py +++ b/conditioning.py @@ -9344,14 +9344,29 @@ 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, key=None): """`text` with a camera move added to each [Shot N] block that lacks one.""" try: try: from . import shot_camera as _sc except ImportError: import shot_camera as _sc - out, report = _sc.add_camera_moves(text, seed=seed, chance=chance) + try: + from . import focus_memory as _fm + except ImportError: + import focus_memory as _fm + try: + prior = _fm.prior(key) + 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) + if key and report: + try: + _fm.observe(key, _sc.content_fingerprint(text), + [w for r in report for w in r.get("lemmas", [])]) + except Exception as e: # noqa: BLE001 + _log.failed("FunPackVideoRefinerV2", "focus memory", e, "this prompt was not counted") except Exception as e: # noqa: BLE001 print(f"[FunPackVideoRefinerV2] camera moves: failed | {where}: {e}; prompt left as written") return text @@ -10917,7 +10932,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, + refinement_key) # ── prompt enhancer ──────────────────────────────────────────────────────────── # Runs BEFORE the prompt becomes conditioning, so what ComfyUI generates from is the @@ -11244,7 +11260,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, refinement_key) 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 diff --git a/focus_memory.py b/focus_memory.py new file mode 100644 index 00000000..45a83ae6 --- /dev/null +++ b/focus_memory.py @@ -0,0 +1,68 @@ +"""Which words the user's prompts keep dwelling on, for choosing where a camera move aims. + +Each candidate target already gets a score from its own prompt (an owned part beats a bare +object, objects of a verb beat doers). This adds a second: how many DIFFERENT prompts the word +turned up in. "It keeps coming back, so it matters" is fused into the first score as a bonus. +Silent until MIN_PROMPTS prompts have been seen: one prompt's words are not a habit. Kept per +refinement key in `.focus_memory.json` (keys are disposable; nothing is migrated). A prompt +seen again counts nothing, so regenerating does not inflate a word. +""" + +import json +import math +import os + +MIN_PROMPTS = 3 +FREQ_W = 0.6 # max bonus: reorders near-equals, never outranks an owned part +MAX_HASHES = 300 + + +def _path(key): + try: + from .conditioning import refinement_state_path + except ImportError: + from conditioning import refinement_state_path + return refinement_state_path(key, "focus_memory", prefix="refine_v2", extension="json") + + +def _read(key): + try: + with open(_path(key), "r", encoding="utf-8") as f: + data = json.load(f) + return data if isinstance(data, dict) else {} + except (FileNotFoundError, ValueError): + return {} + + +def prior(key): + """{lemma: bonus} for `key`, or {} while too few prompts have been seen.""" + if not key: + return {} + data = _read(key) + 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()} + + +def observe(key, prompt_hash, lemmas): + """Count each of `lemmas` once for this prompt. -> prompts seen so far (0 = nothing kept).""" + if not key or not lemmas: + return 0 + data = _read(key) + hashes = data.get("hashes") or [] + if prompt_hash in hashes: + return int(data.get("prompts", 0)) + seen = data.get("seen") or {} + for lemma in set(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(key) + 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) + return data["prompts"] diff --git a/shot_camera.py b/shot_camera.py index 462e17c8..356ae3c7 100644 --- a/shot_camera.py +++ b/shot_camera.py @@ -193,10 +193,12 @@ def candidates(picture): return out -def _score(c): +def _score(c, prior=None): + """How much `c` deserves the camera: what it is in its own prompt (an owned part, an object + of a verb), plus, when given, `prior`'s bonus for words the user's prompts keep returning to.""" lemma, _text, owned, pos, dep = c return ((2 if owned or lemma in PARTS else 0) + (1 if dep in ("dobj", "pobj") else 0) - - pos * 1e-4) + + (prior or {}).get(lemma, 0.0) - pos * 1e-4) def _name(c): @@ -224,12 +226,14 @@ def _pick(options, last, rng): return rng.choice(pool) if rng else pool[0] -def _plan(pool, last, rng): +def _plan(pool, last, rng, prior=None): """-> (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.""" topics = _topics(pool) - best = max(pool, key=_score) + best = max(pool, key=lambda c: _score(c, prior)) + if prior and len(topics) > 2: # the two it cares about most, in text order + topics = sorted(sorted(topics, key=lambda c: -_score(c, prior))[:2], key=lambda c: c[3]) if rng is None: k, travel = 1, len(topics) >= 2 else: @@ -257,7 +261,7 @@ def _plan(pool, last, rng): return " ".join(sentences), desc, opening -def add_camera_moves(text, seed=None, chance=1.0): +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 @@ -283,9 +287,10 @@ def add_camera_moves(text, seed=None, chance=1.0): 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": ""} + entry = {"shot": int(m.group(1)), "move": None, "target": None, "why": "", "lemmas": []} 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] + entry["lemmas"] = [c[0] for c in pool if _score(c) > 0.5] # the words this shot dwells on if CAMERA.search(CUT.sub("", picture)): entry["why"] = "already has a camera move" elif not pool: @@ -293,7 +298,7 @@ def add_camera_moves(text, seed=None, chance=1.0): elif rng is not None and rng.random() >= chance: entry["why"] = "left as written by chance" else: - moves, desc, opening = _plan(pool, last, rng) + moves, desc, opening = _plan(pool, last, rng, prior) entry.update(move=moves, target=desc) last = opening core = picture.rstrip() @@ -470,3 +475,16 @@ _OUR_OPENER = re.compile(r"^\s*At \d\d:\d\d\.\d{3},\s*(?:" + "|".join(re.escape( + r")\.\s*", re.I) _CUT_WORDS = re.compile(r"^\s*(?:At \d\d:\d\d\.\d{3},\s*)?(?:the\s+)?(?:camera|shot)\s+" r"(?:cuts|transitions|changes|switches)\s+to\b\s*", re.I) + + +def content_fingerprint(text): + """A hash of a prompt's own content: the same whatever cut times, openers, views or shot + labels this module added on that run, so regenerating one prompt is one prompt.""" + import hashlib + body = re.sub(r"\[\s*Shot\s+\d+\s*\]", " ", text or "", flags=re.I) + body = re.sub(r"At \d\d:\d\d\.\d{3},\s*(?:" + "|".join(re.escape(o) for o in CUT_OPENERS) + r")\.", + " ", body, flags=re.I) + body = re.sub(r"\s+", " ", body) + body = re.sub(r"(?:^|(?<=[.!?]\s))(?:" + "|".join(re.escape(v) for v in VIEWS) + r")\.", " ", body) + body = re.sub(r"\s+", " ", body).strip().lower() + return hashlib.md5(body.encode("utf-8", "replace")).hexdigest() diff --git a/tests/test_focus_memory.py b/tests/test_focus_memory.py new file mode 100644 index 00000000..81eb232d --- /dev/null +++ b/tests/test_focus_memory.py @@ -0,0 +1,85 @@ +"""Cross-prompt word frequency as a second score for where the camera aims.""" +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import focus_memory as fm # noqa: E402 +import shot_camera as sc # noqa: E402 + + +@pytest.fixture +def store(tmp_path, monkeypatch): + monkeypatch.setattr(fm, "_path", lambda key: str(tmp_path / f"{key}.focus.json")) + return tmp_path + + +def test_silent_until_enough_prompts_then_more_seen_means_more_bonus(store): + assert fm.prior("k") == {} + fm.observe("k", "a", ["chin", "lamp"]) + fm.observe("k", "b", ["chin"]) + assert fm.prior("k") == {} # two prompts: not a habit yet + fm.observe("k", "c", ["chin", "door"]) + p = fm.prior("k") + assert p["chin"] == pytest.approx(fm.FREQ_W) and 0 < p["lamp"] < p["chin"] + assert p["lamp"] == p["door"] + + +def test_the_same_prompt_again_counts_nothing(store): + for _ in range(5): + fm.observe("k", "same", ["chin"]) + assert fm.observe("k", "same", ["chin"]) == 1 and fm.prior("k") == {} + + +def test_no_key_keeps_nothing(store): + assert fm.observe("", "a", ["chin"]) == 0 and fm.prior("") == {} + + +def test_a_recurring_word_wins_between_two_otherwise_equal_targets(): + pytest.importorskip("spacy") + P = "Intro. [Shot 1] raises the lamp and lowers the mirror." + plain = sc.add_camera_moves(P)[1][0]["target"] + lamp = sc.add_camera_moves(P, prior={"lamp": 0.6})[1][0]["target"] + mirror = sc.add_camera_moves(P, prior={"mirror": 0.6})[1][0]["target"] + assert "lamp" in lamp or "->" in lamp + assert "mirror" in mirror or "->" in mirror + assert plain + + +def test_the_prior_never_outranks_an_owned_part(): + pytest.importorskip("spacy") + P = "Intro. [Shot 1] touches 's chin and glances at the lamp." + _o, rep = sc.add_camera_moves(P, prior={"lamp": sc_max()}) + assert "chin" in rep[0]["target"] + + +def sc_max(): + return fm.FREQ_W + + +def test_regenerating_with_other_cuts_and_views_is_still_one_prompt(): + base = "Intro. [Shot 1] waves. [Shot 2] nods." + varied = ("Intro. [Shot 1] waves. [Shot 2] At 00:04.000, the shot changes to a new view. " + "Side view. nods.") + assert sc.content_fingerprint(base) == sc.content_fingerprint(varied) + assert sc.content_fingerprint(base) != sc.content_fingerprint(base + " More.") + + +def test_add_camera_moves_reports_the_words_each_shot_dwells_on(): + pytest.importorskip("spacy") + _o, rep = sc.add_camera_moves("Intro. [Shot 1] touches 's chin.") + assert rep[0]["lemmas"] == ["chin"] + + +def test_refiner_wrapper_counts_prompts_and_uses_the_prior(store, monkeypatch): + pytest.importorskip("spacy") + import conditioning + R = conditioning.FunPackVideoRefinerV2 + monkeypatch.setattr(conditioning, "refinement_state_path", lambda k, *a, **kw: str(store / f"{k}.j")) + monkeypatch.setattr(fm, "_path", lambda k: str(store / f"{k}.j")) + for i in range(3): + R._v2_camera_moves(f"Intro. [Shot 1] touches 's chin{i and ' again' * i}.", + "prompt", 1, 1.0, "k") + assert fm.prior("k").get("chin", 0) > 0