From dce64c1813f0a019851fa2afe840268f460b99ec Mon Sep 17 00:00:00 2001 From: digital-garbage <133010426+digital-garbage@users.noreply.github.com> Date: Sun, 4 Oct 2026 14:07:03 +0300 Subject: [PATCH] feat: shot camera (cut times, views, camera moves for [Shot N]) with rating-taught memory Co-Authored-By: Claude Sonnet 5.5 --- core/tests/test_shipped_nodes.py | 1 + modules/conditioning/shot_camera/__init__.py | 52 ++ modules/conditioning/shot_camera/engine.py | 724 ++++++++++++++++++ modules/conditioning/shot_camera/memory.py | 286 +++++++ modules/conditioning/shot_camera/nodes.py | 97 +++ .../shot_camera/tests/test_focus_memory.py | 203 +++++ .../shot_camera/tests/test_node.py | 38 + .../shot_camera/tests/test_shot_camera.py | 405 ++++++++++ modules/models/minimax_h3/pipeline.py | 6 +- modules/system/taste/__init__.py | 9 + 10 files changed, 1820 insertions(+), 1 deletion(-) create mode 100644 modules/conditioning/shot_camera/__init__.py create mode 100644 modules/conditioning/shot_camera/engine.py create mode 100644 modules/conditioning/shot_camera/memory.py create mode 100644 modules/conditioning/shot_camera/nodes.py create mode 100644 modules/conditioning/shot_camera/tests/test_focus_memory.py create mode 100644 modules/conditioning/shot_camera/tests/test_node.py create mode 100644 modules/conditioning/shot_camera/tests/test_shot_camera.py diff --git a/core/tests/test_shipped_nodes.py b/core/tests/test_shipped_nodes.py index e6516335..6a65c912 100644 --- a/core/tests/test_shipped_nodes.py +++ b/core/tests/test_shipped_nodes.py @@ -56,6 +56,7 @@ FROZEN = { "FunPackCutOpening": ["images", "frames", "fps", "audio"], "FunPackPromptMarkup": ["text"], "FunPackNegativeErase": ["positive", "negative", "strength", "mode", "keep_size"], + "FunPackShotCamera": ["text", "settings", "length", "frame_rate"], "FunPackEnhancePrompt": [ "clip", "text", "enabled", "instructions", "max_length", "greedy", "temperature", "top_p", "top_k", "min_p", "repetition_penalty", "presence_penalty", "seed", "thinking", "image", "use_image", diff --git a/modules/conditioning/shot_camera/__init__.py b/modules/conditioning/shot_camera/__init__.py new file mode 100644 index 00000000..2dfcc981 --- /dev/null +++ b/modules/conditioning/shot_camera/__init__.py @@ -0,0 +1,52 @@ +"""Shot camera: cut times, views and camera moves for the [Shot N] blocks of an H3 prompt, chosen by rules. + +MiniMax H3 reads shots written into the prompt ("[Shot 2] At 00:03.000, the camera cuts to... The camera pushes in toward the +lamp."). This writes them for you, with no language model: a small spaCy tagger finds what each shot is about, a few words +pick the move. Three independent, off-by-default switches, each with a chance: + +* cut times: shots after the first open with their time; a shot whose point changes between two of your shortcuts is split +* views: your own trusted list (side view, from above, POV...), never on shot 1, never the same twice in a row +* camera moves: one to three per shot, aimed at the most specific thing in it ("'s hand", not "") + +Deterministic: the same prompt gives the same shots; "Variation" re-rolls. Ratings teach it: which views and moves are liked +(a dislike that blames the picture alone teaches nothing here), and the words your prompts keep returning to are favoured. +`[Shot N]` is H3's own shot, never a timeline scene. Unvalidated on a GPU. + + text -> Shot Camera -> text -> (Prompt Markup -> encode) +""" + +from . import memory +from .nodes import FunPackShotCamera + +ID = "conditioning_shot_camera" +TITLE = "Shot camera" +MOUNT = "settings.general" +STAGE = "conditioning" +CATEGORY = "conditioning" +STATUS = "experimental" + +NODES = [FunPackShotCamera] + +SETTINGS = { + "camera_moves": {"type": "bool", "default": False, "label": "Camera moves", + "hint": "Adds a camera move, aimed at the most specific thing in the shot, to every [Shot N] that has none."}, + "camera_moves_chance": {"type": "float", "default": 0.7, "min": 0.05, "max": 1.0, "step": 0.05, "label": "Chance a shot gets a move", + "when": {"camera_moves": True}}, + "shot_cuts": {"type": "bool", "default": False, "label": "Shot cut times", + "hint": "Shots after the first open with their cut time; a shot whose point changes between two shortcuts is split in two."}, + "shot_cuts_chance": {"type": "float", "default": 0.5, "min": 0.0, "max": 1.0, "step": 0.05, "label": "Chance a changing shot is split", + "when": {"shot_cuts": True}}, + "shot_views": {"type": "bool", "default": False, "label": "Shot views", + "hint": "Later shots open on a view (side, from above, POV…), never the same twice in a row."}, + "shot_views_chance": {"type": "float", "default": 0.4, "min": 0.0, "max": 1.0, "step": 0.05, "label": "Chance a shot gets a view", + "when": {"shot_views": True}}, + "variation": {"type": "int", "default": 0, "min": 0, "max": 9999, "label": "Variation", + "hint": "The same prompt always gets the same shots. Change this number to draw different ones."}, +} + + +def on_rating(prompt_id, rating, axis=None): + return memory.on_rating(prompt_id, rating, axis) + + +PROVIDES = {"on_rating": on_rating} diff --git a/modules/conditioning/shot_camera/engine.py b/modules/conditioning/shot_camera/engine.py new file mode 100644 index 00000000..3eb01260 --- /dev/null +++ b/modules/conditioning/shot_camera/engine.py @@ -0,0 +1,724 @@ +"""Add a camera move to each `[Shot N]` block of an H3 prompt that has none, aimed at the +most specific thing in that shot. No language model: a small spaCy tagger (rules-only +fallback when it is not installed) picks the target, a few words pick the move. + +`[Shot N]` here is MiniMax's own in-prompt shot, never a timeline scene. `` tags +are the subjects every shot is about, so they are never a target on their own; a part or +belonging of one ("'s hand") is. A thing named in most shots (or in the text before +the first shot) is a constant, not a target. Sound text (music, score...) is left alone. +""" + +import math +import random +import re + +from ..._core import log + +SHOT = re.compile(r"\[\s*Shot\s+(\d+)\s*\]", re.I) +SUBJECT = re.compile(r"<\s*Subject\s*(\d+)\s*>", re.I) +# A `` tag confuses the tagger ("Subject1 bites" reads as a noun); a real-looking +# rare name parses like a person. Swapped in before tagging, swapped back after. +NAMES = ("Mariel", "Tobias", "Odette", "Rurik", "Sabine", "Teodor", "Ilsa", "Corwin", "Yvette") +STAND_IN = re.compile(r"\b(" + "|".join(NAMES) + r")\b") +AUDIO = re.compile(r"\b(music|soundtrack|score|melody|instrumental|beat|bassline|synth|" + r"song|tempo|audio|ambient sound|sound of)\b", re.I) +# H3's own vocabulary (VIDEO_PROMPT_WRITING_GUIDE_base_en.md): a shot opening "the camera cuts +# to" is a cut, not a camera move, and must not make the shot look like it already has one. +CUT = re.compile(r"\b(?:the )?(?:camera|shot) (?:cuts|transitions|changes|switches) to\b", re.I) +# Whether a shot ALREADY has a camera move. Words like tilt, roll, pull, push, arc, focus are +# ordinary body movement ("tilts 's head up and down", "pulls out a knife"), and +# "the camera" is often only where someone is looking, so those count only when the CAMERA +# does them, or when a sentence opens with the bare camera verb ("Pan right to reveal..."). +_CAMERA_VERBS = (r"(?:moves?|pushes|pulls|pans|tilts|zooms|tracks|follows|drifts|circles|orbits|" + r"rises|lowers|glides|sweeps|racks|arcs|dollies|trucks|rolls|rotates|holds|" + r"shakes|stays|remains|sways|slides|cranes|swings|reframes|widens|tightens|" + r"starts|begins|continues|lingers|settles|zoom|pan|tilt|push|pull)") +CAMERA = re.compile( + r"\b(?:zoom(?:s|ed|ing)?\s+(?:in|out)|zoom-(?:in|out)|dolly|rack(?:s|ing)? focus|close-?up|" + r"handheld|wide shot|wide-angle shot|tracking shot|crane shot|static shot|whip pan|" + r"camera shake)\b" + r"|\bcamera(?:'s)?\s+(?:\w+ly\s+)?" + _CAMERA_VERBS + r"\b" + r"|\bcamera(?:'s)? (?:movement|motion|move)\b" + r"|(?:^|[.!?]\s+)(?:slowly |quickly |smoothly )?(?:pan|tilt|truck|pedestal|arc|orbit|roll|" + r"zoom|track|dolly|crane)\b", + re.I) +GENERIC = {"camera", "video", "scene", "shot", "frame", "background", "foreground", "view", + "image", "screen", "moment", "time", "way", "side", "front", "middle", "one", + "other", "something", "everything", "thing", "music", "sound", "style", "lighting", + "light", "atmosphere", "mood", "setting"} +PARTS = {"lip", "mouth", "tongue", "tooth", "teeth", "eye", "eyes", "face", "cheek", "hand", + "finger", "hair", "neck", "shoulder", "waist", "hip", "leg", "foot", "feet", "arm", + "skin", "expression", "smile", "gaze", "back", "chin", "brow"} +PLACES = {"room", "street", "kitchen", "bedroom", "bed", "table", "floor", "stage", "hall", + "garden", "beach", "forest", "city", "car", "office", "bathroom", "field"} +MOVES_DETAIL = ("The camera pushes in toward {x}.", "The camera racks focus to {x}.", + "The camera zooms in on {x}.") +MOVES_OTHER = ("The camera zooms in on {x}.", "The camera arcs around {x}.", + "The camera pushes in toward {x}.") +MOVES_TRAVEL = ("The camera pans from {x} to {y}.", "The camera racks focus from {x} to {y}.", + "The camera moves from {x} to {y}.") +# Chained moves (varied mode): a follow-up to a first move, and a closing pull-back. +MOVES_THEN = ("Then the camera pans to {y}.", "Then the camera racks focus to {y}.", + "Then the camera pushes in toward {y}.") +MOVES_FINISH = ("Then the camera zooms out.", "Then the camera pulls back.", + "Then the camera pulls back to a wider view.") +COUNT_WEIGHTS = (6, 3, 1) # one move / two / three, in a shot that gets any +TRAVEL_SHARE = 0.25 # of one-move shots with two topics, how many travel X -> Y +HOLD_SHARE = 0.25 # of the other one-move shots, how many stay still on one thing +MOVES_HOLD = ("The camera holds steady on {x}.", "The camera stays still, focused on {x}.", + "The camera stays fixed, focused on {x}.") +DETERMINERS = {"the", "a", "an", "this", "that", "these", "those"} +POSSESSIVE_PRONOUNS = {"his", "her", "their", "its", "my", "your", "our"} + +_nlp = None +_tried = False + + +def _say(what, error, effect): + """One log line, once, when something falls back.""" + log.once(f"shot_camera:{what}", log.ALERT, "FunPack Shot Camera", f"{what}: {type(error).__name__}: {error} -- {effect}") + + +def _spacy(): + """The small English tagger, or None (said once) when it is not installed.""" + global _nlp, _tried + if not _tried: + _tried = True + try: + import spacy + _nlp = spacy.load("en_core_web_sm") + except Exception as e: # noqa: BLE001 + fix = ("spacy not installed, run `pip install spacy`" if isinstance(e, ImportError) + else "model missing, run `python -m spacy download en_core_web_sm`") + _say("spaCy tagger", e, f"simple word rules used; {fix}") + return _nlp + + +def _hide(text): + return SUBJECT.sub(lambda m: NAMES[(int(m.group(1)) - 1) % len(NAMES)], text) + + +def _show(text): + return STAND_IN.sub(lambda m: f"", text) + + +def _is_subject(text): + return bool(STAND_IN.fullmatch(re.sub(r"^(?:the\s+)", "", text.strip(), flags=re.I))) + + +def _split_sound(body): + """-> (picture text, sound text): everything from the first sentence about sound on.""" + for m in re.finditer(r"[^.!?]+[.!?]?\s*", body): + if AUDIO.search(m.group(0)): + return body[:m.start()], body[m.start():] + return body, "" + + +# Nouns that only mean something as part of a bigger thing: "the base" of what? +RELATIONAL = {"base", "tip", "top", "bottom", "side", "edge", "end", "middle", "part", "front", + "back", "surface", "inside", "outside", "center", "centre", "head", "bit", "area", + "spot", "line", "corner", "half"} + + +def _words(tok, with_det=False): + keep = {tok} + for c in tok.children: + if c.i < tok.i and (c.dep_ in ("poss", "compound", "amod") + or (with_det and c.dep_ == "det")): + keep.add(c) + if c.dep_ == "poss": + keep.update(g for g in c.children if g.dep_ == "case") + toks = sorted(keep, key=lambda t: t.i) + return " ".join(t.text for t in toks).strip().replace(" 's", "'s").replace(" ’s", "’s") + + +def _phrase(tok): + """-> (text, owned, has_of): a chunk head with the words that name it ("Subject1's hand", + "red lips", "base of the lamp"). An "of ..." that follows belongs to the noun and is kept: + a target cut off before it ("the base") names nothing.""" + text = _words(tok) + has_of = False + for c in tok.children: + if c.dep_ == "prep" and c.lower_ == "of": + pobj = next((g for g in c.children if g.dep_ == "pobj"), None) + if pobj is not None: + text += " of " + _words(pobj, with_det=True) + has_of = True + return text, any(c.dep_ == "poss" for c in tok.children), has_of + + +def _resolve(lemma, text, owned, has_of, names): + """The candidate's phrase made whole, or None when it cannot be: a relational noun with + nothing to be the base/tip/side OF, or a body part with no owner when the shot has more + than one person ("the skin": whose?). With exactly one person in the shot, they own it.""" + if has_of or owned: + return text, True + if lemma in RELATIONAL: + return None + if lemma in PARTS: + if len(names) == 1: + return f"{next(iter(names))}'s {text}", True + return None + return text, owned + + +def candidates(picture): + """[(lemma, phrase, owned, position, dep)] for the nouns of one shot's picture text.""" + hidden = _hide(picture) + nlp = _spacy() + out = [] + if nlp is not None: + doc = nlp(hidden) + for chunk in doc.noun_chunks: + root = chunk.root + prev = doc[root.i - 1] if root.i else None + if prev is not None and STAND_IN.fullmatch(prev.text) and prev.dep_ != "poss": + continue # " laughs": the tagger misread a verb + if root.pos_ in ("PRON", "NUM") or _is_subject(chunk.text): + continue + lemma = root.lemma_.lower() + if root.dep_ == "pobj" and root.head.lower_ == "of": + continue # "a glass of wine": the glass is the thing + text, owned, has_of = _phrase(root) + if _is_subject(text) or lemma in GENERIC or STAND_IN.fullmatch(root.text): + continue + whole = _resolve(lemma, text, owned, has_of, set(STAND_IN.findall(hidden))) + if whole is None: + continue + dep, head = root.dep_, root + while dep == "conj" and head.head is not head: # "a banana and a cherry": both objects + head = head.head + dep = head.dep_ + out.append((lemma, whole[0], whole[1], chunk.start, dep)) + return out + for m in re.finditer(rf"(?:({'|'.join(NAMES)})'s|\b(his|her|their|the|a|an))\s+((?:\w+\s+)?\w+)", + hidden, re.I): + words = m.group(3).split() + lemma = words[-1].lower().rstrip("s") if words[-1].lower() not in PARTS else words[-1].lower() + if lemma in GENERIC: + continue + owner = m.group(1) or (m.group(2) if (m.group(2) or "").lower() in POSSESSIVE_PRONOUNS else "") + text = (f"{owner}'s " if m.group(1) else f"{owner} " if owner else "") + " ".join(words) + out.append((lemma, text.strip(), bool(owner), m.start(), "obj")) + return out + + +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) + + (prior or {}).get(lemma, 0.0) - pos * 1e-4) + + +def _name(c): + """The phrase of candidate `c` as it reads in the prompt, tags restored, with an article.""" + text = _show(c[1]) + if not re.match(r"(?:the|a|an| 0.5 and c[0] not in seen: + seen.add(c[0]) + out.append(c) + return out + + +def _pick(options, last, rng): + """One of `options`, never the one the previous shot opened with when others exist.""" + pool = [o for o in options if o != last] or list(options) + return rng.choice(pool) if rng else pool[0] + + +HABIT_TEMPERATURE = 0.7 # lower = the habit decides more; higher = closer to a coin toss +MENTION_BONUS = 0.5 # per extra mention inside the shot, capped at 1.0: what a shot keeps returning to + + +def _draw(cands, prior, rng, n): + """`n` distinct candidates. Only those within a role-step of the best plain score compete (a + habit can never lift a background word over a real target); among them, with an `rng`, the + pick is drawn with odds exp(score / T) rather than always taking the top, so a word the user + has used a hundred times does not win every shot it merely appears in.""" + top = max(_score(c) for c in cands) + eligible = [c for c in cands if _score(c) >= top - 0.99] or list(cands) + if rng is None: + return sorted(eligible, key=lambda c: -_score(c, prior))[:n] + chosen, pool = [], list(eligible) + while pool and len(chosen) < n: + weights = [math.exp(_score(c, prior) / HABIT_TEMPERATURE) for c in pool] + pick = rng.choices(range(len(pool)), weights)[0] + chosen.append(pool.pop(pick)) + return chosen + + +def _tilt(base, key, arms): + """`base` odds tilted by how the ratings treated arm `key` (0.5-1.5x; 1 when unrated).""" + g, b = (arms or {}).get(key, (0.0, 0.0)) + return base * (0.5 + (g + 1.0) / (g + b + 2.0)) + + +def _plan(pool, last, rng, prior=None, force=None, mode="auto", arms=None): + """-> (moves text, target description, opening template, style). 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. + `arms`: {"style:hold": (good, bad), ...} what ratings taught; tilts those draws. `style` is + the arm that was drawn ("hold", "travel", "k1".."k3"), None when a person decided.""" + topics = _topics(pool) + 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 force: # the user picked what to aim at, in this order + best, topics = force[0], list(force) + if mode == "hold": + hold = _pick(MOVES_HOLD, last, rng) + target = _name(best) + return hold.format(x=target), target, hold, None + if force and len(force) >= 2: + # "from here to there (and on to...)": the camera visits the picks in the order given. + move = _pick(MOVES_TRAVEL, last, rng) + sentences = [move.format(x=_name(force[0]), y=_name(force[1]))] + sentences += [_pick(MOVES_THEN, None, rng).format(y=_name(c)) for c in force[2:]] + return " ".join(sentences), " -> ".join(_name(c) for c in force), move, None + if mode == "move": + k, travel = 1, False + elif rng is None: + k, travel = 1, len(topics) >= 2 + else: + k = rng.choices((1, 2, 3), [_tilt(w, f"style:k{n}", arms) for n, w in zip((1, 2, 3), COUNT_WEIGHTS)])[0] + travel = k == 1 and len(topics) >= 2 and rng.random() < _tilt(TRAVEL_SHARE, "style:travel", arms) + if travel: + x, y = _name(topics[0]), _name(topics[-1]) + move = _pick(MOVES_TRAVEL, last, rng) + return move.format(x=x, y=y), f"{x} -> {y}", move, "travel" if rng else None + if rng is not None and k == 1 and rng.random() < _tilt(HOLD_SHARE, "style:hold", arms): + hold = _pick(MOVES_HOLD, last, rng) # some shots want one fixed focus + target = _name(best) + return hold.format(x=target), target, hold, "hold" + first = topics[0] if (k > 1 and topics) else best + moves = MOVES_DETAIL if (first[2] or first[0] in PARTS) else MOVES_OTHER + opening = _pick(moves, last, rng) + target = _name(first) + sentences = [opening.format(x=target)] + desc = target + if k > 1: + if len(topics) >= 2: + y = _name(topics[-1]) + sentences.append(_pick(MOVES_THEN, None, rng).format(y=y)) + desc += f" -> {y}" + else: + sentences.append(_pick(MOVES_FINISH, None, rng)) + if k > 2 and len(topics) >= 2: + sentences.append(_pick(MOVES_FINISH, None, rng)) + return " ".join(sentences), desc, opening, (f"k{k}" if rng is not None and mode == "auto" else None) + + +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 [], "", [], [] + 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)] + parts = [_split_sound(b) for b in bodies] + cands = [candidates(own_words_removed(p[0])) for p in parts] + header_lemmas = {c[0] for c in candidates(header)} + # In (nearly) every shot = about the whole prompt, not this shot. 70%: with three shots a + # noun shared by two of them is still a topic of those two, only all three is a constant. + need = max(2, math.ceil(0.7 * len(bodies))) if len(bodies) > 1 else 10 ** 9 + seen = {} + for lst in cands: + 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, arms=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. + `arms`: what ratings taught (see _plan); each report entry's "arms" lists the choices drawn + on their own for the rating that follows ("move:yes"/"move:no", a style, "word:").""" + 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": [], + "key": shot_key(picture), "arms": []} + rng = random.Random(f"{seed}:{i}") if seed is not None else None + 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" + wanted = said.get("lemmas") or ([said["lemma"]] if said.get("lemma") else []) + force = [c for c in (next((c for c in pool if c[0] == l), None) for l in wanted) if c] + 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 = {} + for c in pool: + mentions[c[0]] = mentions.get(c[0], 0) + 1 + boost = {l: min(1.0, MENTION_BONUS * (n - 1)) for l, n in mentions.items() if n > 1} + shot_prior = ({l: (prior or {}).get(l, 0.0) + boost.get(l, 0.0) for l in {*(prior or {}), *boost}} + if boost else prior) + if CAMERA.search(CUT.sub("", picture)): + entry["why"] = "already has a camera move" + elif not pool: + entry["why"] = "nothing specific to aim at" + elif mode == "none": + entry["why"] = "no move, as you chose" + elif rng is not None and mode == "auto" and not force and rng.random() >= chance: + entry["why"] = "left as written by chance" + entry["arms"] = ["move:no"] + else: + moves, desc, opening, style = _plan(pool, last, rng, shot_prior, force=force, mode=mode, arms=arms) + entry.update(move=moves, target=desc) + if rng is not None and mode == "auto" and not force: # only what nobody dictated is learned + entry["arms"] = ["move:yes"] + ([f"style:{style}"] if style else []) + names = desc.split(" -> ") + entry["arms"] += [f"word:{c[0]}" for c in {c[0]: c for c in pool if _name(c) in names}.values()] + last = opening + core = picture.rstrip() + gap = picture[len(core):] or (" " if sound else "") + picture = core + ("" if core.endswith((".", "!", "?")) else ".") + " " \ + + moves + gap + out.append(m.group(0) + picture + sound) + report.append(entry) + return "".join(out), report + + +# ── shot cuts ─────────────────────────────────────────────────────────────────────── +# H3 wants every shot after the first to open with its cut time and a cut phrase +# ("[Shot 2] At 00:03.000, the camera cuts to ..."), times increasing and inside the video. +# Nothing in a prompt knows the video's length, so the caller hands it in (seconds). +STAMP = re.compile(r"\b\d\d:\d\d\.\d{3}\b") +CUT_OPENERS = ("the camera cuts to a new angle", "the shot transitions to the next moment", + "the shot changes to a new view", "the shot switches to the next beat") +LEAD_CUT = re.compile(r"^\s*(?:the\s+)?(?:camera|shot)\s+(?:cuts|transitions|changes|switches)\s+to\b", + re.I) +MIN_SHOT_SECONDS = 2 +SENTENCE = re.compile(r"[^.!?]+[.!?]*\s*") + + +def _stamp(t): + return f"{int(t) // 60:02d}:{int(t) % 60:02d}.000" + + +def _sentence_topics(sentence, constant): + return {c[0] for c in candidates(sentence) if c[0] not in constant and _score(c) > 0.5} + + +def _switch_point(picture, constant, pieces): + """Index of the first sentence that (a) starts exactly where a known shortcut text starts and + (b) shares no topic with everything before it, both sides having some: the shot's main + point changes between two of the user's shortcuts. Never inside one.""" + sentences = SENTENCE.findall(picture) + seen, offset = set(), 0 + for j, sent in enumerate(sentences): + topics = _sentence_topics(sent, constant) + starts_piece = j and any(picture.startswith(r, offset + len(sent) - len(sent.lstrip())) + for r in pieces) + if starts_piece and seen and topics and not (topics & seen): + return j, sentences + seen |= topics + offset += len(sent) + return None, sentences + + +def add_shot_cuts(text, seconds, seed=0, chance=0.5, pieces=()): + """-> (new prompt, info). Shots whose main point changes are split in two; every shot after + the first then opens with its cut time, spread evenly over `seconds` and rounded + to whole seconds. A prompt that already carries cut times is left alone. + `pieces`: the texts of the user's shortcuts; a shot is only ever cut BETWEEN two of them, + so with none given nothing is split (times are still added). + info = {"before", "after", "times": [...], "why": str}.""" + marks = list(SHOT.finditer(text or "")) + info = {"before": len(marks), "after": len(marks), "times": [], "why": "", "arms": []} + if not marks: + return text, info + if not seconds or seconds < MIN_SHOT_SECONDS: + info["why"] = "the video's length is not known here" if not seconds else "the video is too short" + return text, info + bodies = [text[m.end():(marks[i + 1].start() if i + 1 < len(marks) else len(text))] + for i, m in enumerate(marks)] + if any(STAMP.search(b[:60]) for b in bodies[1:]): + info["why"] = "cut times are already written, left as they are" + return text, info + header = text[:marks[0].start()] + parts = [_split_sound(b) for b in bodies] + cands = [candidates(own_words_removed(p[0])) for p in parts] + header_lemmas = {c[0] for c in candidates(header)} + need = max(2, math.ceil(0.7 * len(bodies))) if len(bodies) > 1 else 10 ** 9 + seen = {} + for lst in cands: + 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 + blocks = [] # [picture text, sound text] + for i, (picture, sound) in enumerate(parts): + rng = random.Random(f"{seed}:cut:{i}") + j, sentences = _switch_point(picture, constant, pieces) + split = j is not None and rng.random() < chance + if j is not None: + info["arms"].append("split:yes" if split else "split:no") + if split: + first, second = "".join(sentences[:j]), "".join(sentences[j:]) + blocks.append([first.rstrip() + " ", ""]) + second = re.sub(r"^\s*(?:then|next|after that|afterwards),?\s+", "", second, flags=re.I) + blocks.append([second[:1].upper() + second[1:], sound]) + else: + blocks.append([picture, sound]) + if len(blocks) > len(marks) and (len(blocks) > seconds or seconds / len(blocks) < MIN_SHOT_SECONDS): + blocks = [[p, s] for p, s in parts] # not enough seconds for the splits + info["arms"] = [] + # Equal shares of the scene, rounded to whole seconds, no shot shorter than MIN_SHOT_SECONDS + # (a shortcut's text length says nothing about how long its beat should last: weighting by + # it gave one-second shots next to five-second ones). + n = len(blocks) + gap = MIN_SHOT_SECONDS if n * MIN_SHOT_SECONDS <= seconds else 1 + times = [] + for k in range(1, n): + t = round(k * seconds / n) + t = max(t, (times[-1] if times else 0) + gap) # not too close behind + t = min(t, int(seconds) - (n - k) * gap) # leave room for the shots after + times.append(t) + if len(blocks) > 1 and times[-1] >= seconds: + info["why"] = f"{len(blocks)} shots do not fit in {seconds}s at whole seconds" + info["arms"] = [] + return text, info + out = [header] + for k, (picture, sound) in enumerate(blocks): + label = f"[Shot {k + 1}]" + body = picture + if k: + stamp = f"At {_stamp(times[k - 1])}, " + rng = random.Random(f"{seed}:opener:{k}") + if LEAD_CUT.match(body): + body = " " + stamp + LEAD_CUT.sub(lambda m: m.group(0).strip().lower(), body.lstrip(), 1) + else: + body = " " + stamp + rng.choice(CUT_OPENERS) + ". " + body.lstrip() + else: + body = body if body.startswith(" ") else " " + body + out.append(label + body + sound) + info.update(after=len(blocks), times=[_stamp(t) for t in times]) + return "".join(out), info + + +# ── views ─────────────────────────────────────────────────────────────────────────── +# Chosen, not detected: a prompt seldom says which view it wants. The user's own trusted +# wording, verbatim. Shot 1 is skipped: it may sit on a reference image or a pinned first +# frame, which a new view would contradict; every later shot starts on a new angle anyway. +VIEWS = ("POV view", "POV view from above", "POV view from below", "Side view", + "View from above", "View from below", "View from behind", "Front view") +# A view the shot already asks for. "from behind" / "from above" alone are also body actions +# ("enters from behind", "drips from above"), so they count only when a camera word goes with +# them ("seen from behind", "shot from above", "camera from below"). +VIEW_STATED = re.compile(r"\b(pov|(?:side|front|rear|back|top|overhead|bird'?s[- ]eye)[- ]view|view from|" + r"(?:seen|shot|filmed|viewed|captured|camera|angle|footage)\b[^.!?,;]{0,20}?" + r"\bfrom (?:above|below|behind))\b", re.I) +OPENER = re.compile(r"^\s*At \d\d:\d\d\.\d{3},[^.!?]*[.!?]\s*") + + +# What a shot's own words say about which views can show it. A view that cannot show what the +# shot describes (the back of someone whose face is the point, the front of someone walking +# away) only confuses the model, so it is never offered. Traits also key what the ratings +# teach: a view that works for a face shot may not work for a hands shot. +VIEW_TRAITS = { + "face": re.compile(r"\b(faces?|eyes?|eye contact|mouths?|lips|smil\w*|expressions?|gaze|" + r"cheeks?|facing|looks? (?:at|into))\b", re.I), + "back": re.compile(r"\b(back(?! and| to| into| off| up| down)|rear|spine|shoulder blades?|nape|" + r"walks? away|turned away|from behind)\b", re.I), + "hands": re.compile(r"\b(hands?|fingers?|you|your|reach\w*|hold\w*|touch\w*)\b", re.I), +} +_FRONT_ONLY = ("POV view", "POV view from above", "POV view from below", "Front view") + + +def view_traits(picture): + body = own_words_removed(picture or "") + return sorted(t for t, rx in VIEW_TRAITS.items() if rx.search(body)) + + +def allowed_views(traits): + """The views that can show a shot with these traits: a face excludes the view from behind; + a back (or someone leaving) excludes every view from the front, POV included.""" + out = list(VIEWS) + if "face" in traits: + out = [v for v in out if v != "View from behind"] + if "back" in traits: + out = [v for v in out if v not in _FRONT_ONLY] + return out + + +def _view_weight(view, traits, stats): + """How much a view is liked for a shot like this: the smoothed good-rate of the view, and + of the view on each of this shot's traits, averaged. 0.5 when nothing was rated yet.""" + rates = [] + for key in [view] + [f"{view}@{t}" for t in (traits or ["none"])]: + g, b = (stats or {}).get(key, (0.0, 0.0)) + if g + b > 0: + rates.append((g + 1.0) / (g + b + 2.0)) + return 0.15 + (sum(rates) / len(rates) if rates else 0.5) + + +def shot_texts(text): + """{shot number: its picture text, without what this module wrote into it}: what the + Reactive focus review shows so a choice is about a shot the person can read.""" + out = {} + marks = list(SHOT.finditer(text or "")) + for i, m in enumerate(marks): + end = marks[i + 1].start() if i + 1 < len(marks) else len(text) + picture, _sound = _split_sound(text[m.end():end]) + out[int(m.group(1))] = " ".join(own_words_removed(picture).split()) + return out + + +def view_options(text, stats=None): + """What each `[Shot N]` after the first could open with, for a person to choose from. + -> [{"shot", "key", "traits", "already", "candidates": [{"view", "score"}], "auto"}]""" + marks = list(SHOT.finditer(text or "")) + out = [] + for i, m in enumerate(marks): + if not i and len(marks) > 1: # see add_shot_views: a lone shot may take a view + continue + end = marks[i + 1].start() if i + 1 < len(marks) else len(text) + picture, _sound = _split_sound(text[m.end():end]) + traits = view_traits(picture) + cands = sorted(((v, _view_weight(v, traits, stats)) for v in allowed_views(traits)), + key=lambda c: -c[1]) + stated = VIEW_STATED.search(CUT.sub("", picture)) + out.append({"shot": int(m.group(1)), "key": shot_key(picture), "traits": traits, + "already": bool(stated), "stated": stated.group(0) if stated else "", + "candidates": [{"view": v, "score": round(w, 2)} for v, w in cands], + "auto": cands[0][0] if cands else None}) + return out + + +def add_shot_views(text, seed=0, chance=0.4, stats=None, choices=None, skipped=None): + """-> (new prompt, [{"shot", "view", "traits"}]). Shots 2+ (or the only shot) that state no view get one of the + views that can show them as a sentence of its own (after the cut opener, if there is one), + never the same as the shot before, drawn per shot from the seed and weighted by `stats` + (what ratings taught). `choices` {shot_key: {"mode": "auto"|"none"|"pick", "view": str}}: + a person's pick is always used, "none" leaves the shot alone. `skipped`, if a list, gets one + "shot N: reason" per shot that got no view.""" + marks = list(SHOT.finditer(text or "")) + if not marks: + return text, [] + # Shot 1 of several may sit on a reference image or a pinned first frame, which a view would + # contradict, so views open shots 2+. A prompt with ONE shot has nothing else to open. + lone = len(marks) == 1 + out, added, last = [text[:marks[0].start()]], [], None + for i, m in enumerate(marks): + end = marks[i + 1].start() if i + 1 < len(marks) else len(text) + body = text[m.end():end] + rng = random.Random(f"{seed}:view:{i}") + picture, sound = _split_sound(body) + view = None + stated = VIEW_STATED.search(CUT.sub("", picture)) if (i or lone) else None + if stated and skipped is not None: + skipped.append(f"shot {m.group(1)}: already states “{stated.group(0)}”") + if (i or lone) and not stated: + traits = view_traits(picture) + said = (choices or {}).get(shot_key(picture)) or {} + allowed = allowed_views(traits) + if said.get("mode") == "none": + pass + elif said.get("views") or said.get("view"): + # "fine from these": a pick is always used; with several, one of them is drawn. + wanted = [v for v in (said.get("views") or [said.get("view")]) if v in allowed] + pool = [v for v in wanted if v != last] or wanted + if pool: + view = rng.choices(pool, weights=[_view_weight(v, traits, stats) for v in pool])[0] + elif rng.random() < chance: + pool = [v for v in allowed if v != last] or allowed + if pool: + view = rng.choices(pool, weights=[_view_weight(v, traits, stats) for v in pool])[0] + if not view and skipped is not None: + skipped.append(f"shot {m.group(1)}: " + ("no view, as you chose" if said.get("mode") == "none" + else "left as written by chance")) + if view: + head = OPENER.match(picture) + cut = head.end() if head else len(picture) - len(picture.lstrip()) + picture = picture[:cut].rstrip() + (" " if cut else " ") + view + ". " + picture[cut:].lstrip() + body = picture + sound + added.append({"shot": int(m.group(1)), "view": view, "traits": traits}) + last = view + out.append(m.group(0) + body) + return "".join(out), added + + +_VIEW_SENTENCE = re.compile(r"^\s*(?:" + "|".join(re.escape(v) for v in VIEWS) + r")\.\s*", re.I) + + +def own_words_removed(picture): + """`picture` without what this module wrote into it (its own cut openers and views) and + without the bare cut words of a user-written opener ("At 00:03.500, the camera cuts to"), + whose content after them is kept: those words are not the shot's content and must never + become a target or a topic.""" + picture = _OUR_OPENER.sub("", picture, count=1) + picture = _CUT_WORDS.sub("", picture, count=1) + return _VIEW_SENTENCE.sub("", picture, count=1) + + +_OUR_OPENER = re.compile(r"^\s*At \d\d:\d\d\.\d{3},\s*(?:" + "|".join(re.escape(o) for o in CUT_OPENERS) + + 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/modules/conditioning/shot_camera/memory.py b/modules/conditioning/shot_camera/memory.py new file mode 100644 index 00000000..a226d0c7 --- /dev/null +++ b/modules/conditioning/shot_camera/memory.py @@ -0,0 +1,286 @@ +"""Which words a 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. A prompt +seen again counts nothing, so regenerating does not inflate a word. + +Self-contained: no dependency on the host pack. One store per user, at +`/shot_camera/focus_memory.json` (else `~/.shot_camera/focus_memory.json`); +the SHOT_CAMERA_MEMORY environment variable names another file. Delete the file to forget. +""" + +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 +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 +WORD_RATE_W = 0.5 # max bonus/penalty a word earns from the ratings of runs that aimed at it + + +def _path(): + override = os.environ.get("SHOT_CAMERA_MEMORY") + if override: + return override + try: + import folder_paths + base = folder_paths.get_user_directory() + except Exception: # noqa: BLE001 — not inside ComfyUI, or an older one + base = os.path.join(os.path.expanduser("~"), ".shot_camera") + return os.path.join(base, "focus_memory.json") + return os.path.join(base, "shot_camera", "focus_memory.json") + + +def _read(): + try: + with open(_path(), "r", encoding="utf-8") as f: + data = json.load(f) + return data if isinstance(data, dict) else {} + except (OSError, ValueError): + return {} + + +def prior(): + """{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 {} + 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) + for key, (g, b) in arm_stats().items(): # what ratings taught about aiming at a word + if key.startswith("word:"): + lemma = key[5:] + out[lemma] = out.get(lemma, 0.0) + WORD_RATE_W * 2.0 * (arm_rate(key) - 0.5) + 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") + picked = [picked] if isinstance(picked, str) else list(picked or []) + for word in picked: + picks[word] = picks.get(word, 0) + 1 + if picked and auto and auto not in picked: + rejects[auto] = rejects.get(auto, 0) + 1 + data.update(picks=picks, rejects=rejects, shots=total, kept=kept) + _save(data) + return n + + +def view_stats(): + """{key: (good, bad)} for each view and each "view@trait" the ratings and picks taught.""" + return {k: (float(v[0]), float(v[1])) for k, v in (_read().get("views") or {}).items() + if isinstance(v, (list, tuple)) and len(v) == 2} + + +def _bump_views(data, used, good, bad): + views = data.setdefault("views", {}) + for u in used: + for key in [u["view"]] + [f"{u['view']}@{t}" for t in (u.get("traits") or ["none"])]: + g, b = views.get(key, [0.0, 0.0]) + views[key] = [round(g + good, 4), round(b + bad, 4)] + + +def rate_views(used, sign): + """A rated generation: its views share the credit (sign > 0) or the blame (sign < 0), so a + run with four views does not punish each as hard as a run with one. `used`: [{"view", + "traits"}]. -> views rated.""" + used = [u for u in used or [] if isinstance(u, dict) and u.get("view")] + if not used or not sign: + return 0 + data = _read() + share = 1.0 / len(used) + _bump_views(data, used, share if sign > 0 else 0.0, share if sign < 0 else 0.0) + _save(data) + return len(used) + + +def learn_views(decisions): + """What a person chose: [{"auto": view|None, "picked": view|None, "traits": [...]}]. A pick + counts as a good sign for that view; the view the rewriter proposed instead, as half a bad.""" + data = _read() + n = 0 + for d in decisions or []: + tr = d.get("traits") or [] + picked = [d["picked"]] if isinstance(d.get("picked"), str) else list(d.get("picked") or []) + if picked: + for v in picked: + _bump_views(data, [{"view": v, "traits": tr}], 1.0 / len(picked), 0.0) + n += 1 + if d.get("auto") and d["auto"] not in picked: + _bump_views(data, [{"view": d["auto"], "traits": tr}], 0.0, 0.5) + _save(data) + return n + + +def arm_stats(): + """{arm: (good, bad)} for what ratings taught about the camera's own choices: "move:yes" / + "move:no" (a shot got a move / was left alone), "style:hold|travel|k1|k2|k3", "word:", + "split:yes|no" (a shot whose point changes was cut in two / kept whole).""" + return {k: (float(v[0]), float(v[1])) for k, v in (_read().get("arms") or {}).items() + if isinstance(v, (list, tuple)) and len(v) == 2} + + +def arm_rate(key, stats=None): + """Smoothed good-rate of one arm: 0.5 until rated.""" + g, b = (stats if stats is not None else arm_stats()).get(key, (0.0, 0.0)) + return (g + 1.0) / (g + b + 2.0) + + +def rate_arms(used, sign): + """A rated generation: every choice the camera made in it shares the credit (sign > 0) or + the blame (sign < 0), so a run with many choices does not punish each as hard as a run with + one. `used`: [arm key]. -> arms rated.""" + used = [u for u in used or [] if isinstance(u, str) and u] + if not used or not sign: + return 0 + data = _read() + arms = data.setdefault("arms", {}) + share = 1.0 / len(used) + for u in used: + g, b = arms.get(u, [0.0, 0.0]) + arms[u] = [round(g + (share if sign > 0 else 0.0), 4), round(b + (share if sign < 0 else 0.0), 4)] + _save(data) + return len(used) + + +def summary(limit=40): + """What is remembered, for the person to read and prune: words (picked / replaced / merely + recurring) and views (good / bad, with the per-trait rows folded under their view).""" + data = _read() + picks, rejects, seen = (data.get("picks") or {}), (data.get("rejects") or {}), (data.get("seen") or {}) + words = [{"word": w, "picks": int(picks.get(w, 0)), "rejects": int(rejects.get(w, 0)), "seen": int(seen.get(w, 0))} + for w in {*picks, *rejects, *seen}] + words.sort(key=lambda r: -(r["picks"] * 3 + r["rejects"] * 3 + r["seen"])) + views = [{"view": k, "good": g, "bad": b} for k, (g, b) in view_stats().items() if "@" not in k] + views.sort(key=lambda r: -(r["good"] + r["bad"])) + return {"words": words[:limit], "more": max(0, len(words) - limit), "views": views, + "prompts": int(data.get("prompts", 0)), "shots": int(data.get("shots", 0)), + "kept": int(data.get("kept", 0))} + + +def forget(kind, name=None): + """Drop one remembered word (`kind="word"`), one view and its per-trait rows (`"view"`), or + everything (`"all"`). -> True if anything was removed.""" + data = _read() + if kind == "all": + existed = bool(data) + data = {} + elif kind == "word": + existed = any(name in (data.get(k) or {}) for k in ("picks", "rejects", "seen")) + for k in ("picks", "rejects", "seen"): + (data.get(k) or {}).pop(name, None) + existed = existed or (data.get("arms") or {}).pop(f"word:{name}", None) is not None + elif kind == "view": + views = data.get("views") or {} + gone = [k for k in views if k == name or k.startswith(f"{name}@")] + existed = bool(gone) + for k in gone: + views.pop(k) + else: + raise ValueError(f"unknown kind {kind!r}") + _save(data) + return existed + + +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: + chance = (chance + kept / total) / 2.0 + return max(0.05, min(1.0, chance * _lean("move:yes", "move:no"))) + + +def _lean(yes, no): + """Factor (0.5-2) that tilts an odds toward the arm the ratings liked more, 1 when unrated.""" + stats = arm_stats() + ry, rn = arm_rate(yes, stats), arm_rate(no, stats) + return 2.0 * ry / (ry + rn) + + +def split_chance(chance): + """The chance of cutting a shot in two, tilted by whether split runs were liked.""" + return max(0.0, min(1.0, chance * _lean("split:yes", "split:no"))) + + +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): + """Count each of `lemmas` once for this prompt. -> prompts seen so far (0 = nothing kept).""" + if not lemmas: + return 0 + data = _read() + 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:]) + _save(data) + return data["prompts"] + + +MAX_RUNS = 64 + + +def record_run(prompt_id, chose): + """What one run chose ({"views": [...], "arms": [...]}), kept until it is rated.""" + data = _read() + runs = data.get("runs") or {} + runs[str(prompt_id)] = chose + data["runs"] = dict(list(runs.items())[-MAX_RUNS:]) + _save(data) + + +def on_rating(prompt_id, rating, axis=None): + """A clip was rated: the views and camera choices of the run that made it share the credit or the blame. + A dislike that blames the picture alone teaches nothing here (the camera was not at fault), and only a run's + first rating counts, so changing your mind does not count it twice. -> choices rated.""" + data = _read() + run = (data.get("runs") or {}).get(str(prompt_id)) + if not run or str(prompt_id) in (data.get("rated") or []): + return 0 + sign = 1 if rating == "liked" else -1 if rating == "disliked" and axis != "image" else 0 + if not sign: + return 0 + data["rated"] = ((data.get("rated") or []) + [str(prompt_id)])[-MAX_RUNS:] + _save(data) + return rate_views(run.get("views"), sign) + rate_arms(run.get("arms"), sign) diff --git a/modules/conditioning/shot_camera/nodes.py b/modules/conditioning/shot_camera/nodes.py new file mode 100644 index 00000000..1b6b0c4a --- /dev/null +++ b/modules/conditioning/shot_camera/nodes.py @@ -0,0 +1,97 @@ +"""The Shot Camera node: the prompt text in, the prompt text out with cut times, views and camera moves in its [Shot N] blocks.""" + +import hashlib + +from comfy_api.latest import io + +from ..._core import log, registry as registry_mod, schema as schema_mod, shortcuts +from . import engine, memory + +ID = "conditioning_shot_camera" + + +def _values(settings): + """This module's settings from the pipeline's settings object, checked against its own declaration.""" + spec = registry_mod.current().specs.get(ID) + if spec is None: + return {} + clean, _problems = schema_mod.check_values(spec, (settings or {}).get(ID)) + return clean + + +def _pieces(): + """Every enabled shortcut's replacement text, longest first: how a cut tells where one shortcut ends and the next begins.""" + try: + out = set() + for sc in shortcuts.listing(): + if sc.enabled: + out.update(r.strip() for r in sc.replacements if len(r.strip()) >= 15) + return sorted(out, key=len, reverse=True) + except Exception: # noqa: BLE001 -- no library: nothing is split, times are still added + return [] + + +def _prompt_id(): + try: + from server import PromptServer + return getattr(PromptServer.instance, "last_prompt_id", None) + except Exception: # noqa: BLE001 + return None + + +def rewrite(text, values, seconds=None, pieces=None): + """-> (new text, what happened, what the run chose for the rating that follows).""" + seed = f"{engine.content_fingerprint(text)}:{values.get('variation', 0)}" # the same prompt gives the same shots; 'variation' re-rolls + said, arms, views = [], [], [] + if values.get("shot_cuts") and engine.SHOT.search(text or ""): + chance = memory.split_chance(float(values.get("shot_cuts_chance", 0.5))) + text, info = engine.add_shot_cuts(text, seconds, seed=seed, chance=chance, pieces=_pieces() if pieces is None else pieces) + arms += info.get("arms", []) + said.append(f"cuts at {', '.join(info['times'])}" + (f" ({info['before']}→{info['after']} shots)" if info["after"] != info["before"] else "") if info["times"] else f"cuts: {info['why']}") + if values.get("shot_views") and engine.SHOT.search(text or ""): + skipped = [] + text, added = engine.add_shot_views(text, seed=seed, chance=float(values.get("shot_views_chance", 0.4)), stats=memory.view_stats(), skipped=skipped) + views += [{"view": a["view"], "traits": a["traits"]} for a in added] + said.append("views: " + (", ".join(f"shot {a['shot']} {a['view']}" for a in added) if added else f"none ({'; '.join(skipped) or 'nothing to do'})")) + if values.get("camera_moves") and engine.SHOT.search(text or ""): + chance = memory.effective_chance(float(values.get("camera_moves_chance", 0.7))) + before = text + text, report = engine.add_camera_moves(text, seed=seed, chance=chance, prior=memory.prior(), arms=memory.arm_stats()) + arms += [a for r in report for a in r.get("arms", [])] + if report: + memory.observe(engine.content_fingerprint(before), [w for r in report for w in r.get("lemmas", [])]) + moved = [r for r in report if r["move"]] + said.append(f"moves: {len(moved)} of {len(report)} shots" + ("" if moved else " (" + "; ".join(f"shot {r['shot']} {r['why']}" for r in report) + ")")) + return text, "; ".join(said), {"views": views, "arms": arms} + + +class FunPackShotCamera(io.ComfyNode): + @classmethod + def define_schema(cls) -> io.Schema: + return io.Schema( + node_id="FunPackShotCamera", + display_name="FunPack Shot Camera", + category="FunPack/Conditioning", + description="Cut times, views and camera moves for the [Shot N] blocks of an H3 prompt, chosen by rules (no language model). Each is off until switched on in Settings ▸ Engine.", + inputs=[io.String.Input("text", multiline=True, default=""), + io.Custom("FUNPACK_SETTINGS").Input("settings", optional=True), + io.Int.Input("length", default=124, min=1, max=100000, tooltip="Frames of the video: with the frame rate, how long the shots have between them."), + io.Float.Input("frame_rate", default=25.0, min=1.0, max=240.0)], + outputs=[io.String.Output(display_name="text"), io.String.Output(display_name="status")], + ) + + @classmethod + def execute(cls, text: str, length: int, frame_rate: float, settings=None) -> io.NodeOutput: + values = _values(settings) + if not any(values.get(k) for k in ("camera_moves", "shot_cuts", "shot_views")) or not engine.SHOT.search(text or ""): + return io.NodeOutput(text, "unchanged") + try: + out, said, chose = rewrite(text, values, seconds=length / frame_rate if frame_rate else None) + except Exception as exc: # noqa: BLE001 -- a rewrite is never worth the run + log.failed("FunPack Shot Camera", exc) + return io.NodeOutput(text, f"left as written ({exc})") + pid = _prompt_id() + if pid and (chose["views"] or chose["arms"]): + memory.record_run(pid, chose) + log.info("FunPack Shot Camera", said) + return io.NodeOutput(out, said) diff --git a/modules/conditioning/shot_camera/tests/test_focus_memory.py b/modules/conditioning/shot_camera/tests/test_focus_memory.py new file mode 100644 index 00000000..a04e949e --- /dev/null +++ b/modules/conditioning/shot_camera/tests/test_focus_memory.py @@ -0,0 +1,203 @@ +"""Cross-prompt word frequency as a second score for where the camera aims.""" +from pathlib import Path + +import pytest + +from modules.conditioning.shot_camera import memory as fm +from modules.conditioning.shot_camera import engine as sc + + +@pytest.fixture +def store(tmp_path, monkeypatch): + monkeypatch.setattr(fm, "_path", lambda: str(tmp_path / "focus_memory.json")) + return tmp_path + + +def test_silent_until_enough_prompts_then_more_seen_means_more_bonus(store): + assert fm.prior() == {} + fm.observe("a", ["chin", "lamp"]) + fm.observe("b", ["chin"]) + assert fm.prior() == {} # two prompts: not a habit yet + fm.observe("c", ["chin", "door"]) + p = fm.prior() + 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("same", ["chin"]) + assert fm.observe("same", ["chin"]) == 1 and fm.prior() == {} + + +def test_nothing_to_count_keeps_nothing(store): + assert fm.observe("a", []) == 0 and fm.prior() == {} + + +def test_the_memory_is_its_own_file_named_by_the_environment_or_the_user_dir(monkeypatch, tmp_path): + monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "mine.json")) + assert fm._path() == str(tmp_path / "mine.json") + fm.observe("a", ["chin"]) + assert (tmp_path / "mine.json").is_file() + monkeypatch.delenv("SHOT_CAMERA_MEMORY") + assert fm._path().endswith("focus_memory.json") and "shot_camera" in fm._path() + + +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_a_habit_does_not_win_every_shot_it_merely_appears_in(): + """100 videos with bananas, then a shot with a banana AND a cherry: the cherry still gets + the camera some of the time, and only between equals.""" + pytest.importorskip("spacy") + P = "Intro. [Shot 1] holds a banana and a cherry." + heard = {"banana": 0, "cherry": 0} + for seed in range(200): + _o, rep = sc.add_camera_moves(P, seed=seed, chance=1.0, prior={"banana": fm.FREQ_W}) + t = rep[0]["target"] + heard["banana"] += "banana" in t + heard["cherry"] += "cherry" in t + assert heard["banana"] > heard["cherry"] > 20, heard # favoured, not decisive + + +def test_a_habit_never_beats_a_real_target(): + pytest.importorskip("spacy") + P = "Intro. [Shot 1] touches 's chin and glances at a banana." + for seed in range(60): + _o, rep = sc.add_camera_moves(P, seed=seed, chance=1.0, prior={"banana": fm.FREQ_W}) + assert "chin" in rep[0]["target"], rep[0]["target"] + + +def test_a_word_the_shot_keeps_returning_to_is_what_it_is_about(): + pytest.importorskip("spacy") + 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"] + + +# ── rated, trait-aware views ───────────────────────────────────────────────────────── +# ── several picks per shot, and readable previews ──────────────────────────────────── +def test_several_targets_in_order_become_one_travelling_move(): + P = "[Shot 1] A woman raises the lamp. Her hand holds a cup near the window." + opts = sc.focus_options(P)[0]["candidates"] + assert len(opts) >= 3 + a, b, c = opts[0]["lemma"], opts[1]["lemma"], opts[2]["lemma"] + key = sc.shot_key(P.split("]", 1)[1]) + out, rep = sc.add_camera_moves(P, seed=1, chance=0.0, + choices={key: {"mode": "move", "lemmas": [b, a, c]}}) + assert rep[0]["target"].count("->") == 2 and rep[0]["lemmas"] + names = [t.strip() for t in rep[0]["target"].split("->")] + assert names[0].endswith(next(o["text"] for o in opts if o["lemma"] == b).split()[-1]) + assert "from" in out and out.count("Then the camera") == 1 + one, rep1 = sc.add_camera_moves(P, seed=1, chance=0.0, choices={key: {"mode": "hold", "lemmas": [a, b]}}) + assert "->" not in rep1[0]["target"] # Hold aims at the first pick only + + +def test_shot_texts_name_what_each_shot_says(): + t = sc.shot_texts("[Shot 1] A cat sits. [Shot 2] At 00:03.000, the camera cuts to a new angle. A dog runs.") + assert t == {1: "A cat sits.", 2: "A dog runs."} + + +# ── ratings teach moves, styles, targets and splits ────────────────────────────────── +def test_ratings_teach_the_camera_its_own_choices(store): + assert fm.effective_chance(0.5) == 0.5 and fm.split_chance(0.5) == 0.5 and fm.prior() == {} + for _ in range(4): + fm.rate_arms(["move:yes", "style:hold", "word:lamp", "split:no"], 1) + fm.rate_arms(["move:no", "word:door", "split:yes"], -1) + assert fm.effective_chance(0.5) > 0.5 and fm.split_chance(0.5) < 0.5 + p = fm.prior() + assert p["lamp"] > 0 > p["door"] + assert fm.split_chance(0.0) == 0.0 # a chance the user turned off stays off + assert fm.forget("word", "lamp") and "lamp" not in fm.prior() + + +def test_plan_reports_what_it_drew_and_a_liked_style_is_drawn_more(): + P = "[Shot 1] A woman raises the lamp. Her hand holds a cup near the window." + rep = sc.add_camera_moves(P, seed=3, chance=1.0)[1][0] + assert "move:yes" in rep["arms"] and any(a.startswith("word:") for a in rep["arms"]) + assert sc.add_camera_moves(P, seed=3, chance=0.0)[1][0]["arms"] == ["move:no"] + forced = {sc.shot_key(P.split("]", 1)[1]): {"mode": "move"}} + assert sc.add_camera_moves(P, seed=3, chance=1.0, choices=forced)[1][0]["arms"] == [] # a person's call is not learned + like = {"style:hold": (20.0, 0.0), "style:k1": (0.0, 20.0)} + holds = sum("style:hold" in sc.add_camera_moves(P, seed=s, chance=1.0, arms=like)[1][0]["arms"] for s in range(60)) + plain = sum("style:hold" in sc.add_camera_moves(P, seed=s, chance=1.0)[1][0]["arms"] for s in range(60)) + assert holds > plain + diff --git a/modules/conditioning/shot_camera/tests/test_node.py b/modules/conditioning/shot_camera/tests/test_node.py new file mode 100644 index 00000000..4878ae57 --- /dev/null +++ b/modules/conditioning/shot_camera/tests/test_node.py @@ -0,0 +1,38 @@ +"""The node's rewrite, the rating hook and the module contract.""" +import importlib + +from .. import memory, nodes + + +def _mod(): + return importlib.import_module(nodes.__package__) + + +def test_off_by_default_leaves_text_alone(tmp_path, monkeypatch): + monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json")) + text = "[Shot 1] A woman walks.\n[Shot 2] She sits down." + assert nodes.rewrite(text, {}, seconds=5.0, pieces=[])[0] == text + + +def test_cuts_add_times(tmp_path, monkeypatch): + monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json")) + text = "[Shot 1] A woman walks along the street.\n[Shot 2] She sits down at a table." + out, said, _ = nodes.rewrite(text, {"shot_cuts": True}, seconds=6.0, pieces=[]) + assert out != text and said + + +def test_rating_teaches_once_and_skips_picture_only(tmp_path, monkeypatch): + monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json")) + chose = {"views": [{"view": "side view", "traits": ["none"]}], "arms": ["a1"]} + memory.record_run("p1", chose) + assert memory.on_rating("p1", "disliked", "image") == 0 # the picture was at fault, not the camera + assert memory.on_rating("p1", "liked") == 2 + assert memory.on_rating("p1", "disliked") == 0 # changing your mind does not count twice + assert memory.view_stats()["side view"] == (1.0, 0.0) + assert memory.on_rating("unknown", "liked") == 0 + + +def test_contract(): + m = _mod() + assert m.PROVIDES["on_rating"]("nope", "liked") == 0 + assert m.CATEGORY == "conditioning" and m.NODES diff --git a/modules/conditioning/shot_camera/tests/test_shot_camera.py b/modules/conditioning/shot_camera/tests/test_shot_camera.py new file mode 100644 index 00000000..6d1ad879 --- /dev/null +++ b/modules/conditioning/shot_camera/tests/test_shot_camera.py @@ -0,0 +1,405 @@ +"""Camera moves for [Shot N] blocks. Fixtures are neutral stand-ins with the real shape: + tags everywhere, one shot per block, a music sentence at the end.""" +from pathlib import Path + +import pytest + +from modules.conditioning.shot_camera import engine as sc + +HEAD = ("Two friends in a sunlit loft, is a tall woman with red hair, " + " is a man in a grey coat. Cinematic 35mm look. ") +S1 = "[Shot 1] is waving 's hand, greeting . " +S2 = "[Shot 2] bites 's lower lip while smiles at . " +S3 = "[Shot 3] hands a glass of wine across the table. laughs. " +MUSIC = "Upbeat jazz music with a soft bassline plays throughout." +PROMPT = HEAD + S1 + S2 + S3 + MUSIC + + +def test_no_shot_markers_leaves_the_prompt_alone(): + assert sc.add_camera_moves("A woman walks. Music plays.") == ("A woman walks. Music plays.", []) + + +def test_move_aims_at_a_part_of_the_subject_never_the_subject(): + pytest.importorskip("spacy") + out, rep = sc.add_camera_moves(PROMPT) + assert rep[0]["target"] == "'s hand" + assert rep[1]["target"] == "'s lower lip" + for r in rep: + assert r["move"] and "." not in r["move"].replace("'s", "") + assert rep[0]["move"] in out and "'s hand" in rep[0]["move"] + + +def test_moves_sit_inside_their_shot_and_before_the_music(): + pytest.importorskip("spacy") + out, rep = sc.add_camera_moves(PROMPT) + assert out.startswith(HEAD + "[Shot 1]") # intro untouched + assert out.endswith(" " + MUSIC) # music untouched, last + assert out.index(rep[2]["move"]) < out.index(MUSIC) + assert out.index("[Shot 2]") < out.index(rep[1]["move"]) < out.index("[Shot 3]") + + +def test_a_shot_with_a_move_already_is_left_alone(): + pytest.importorskip("spacy") + p = HEAD + "[Shot 1] The camera zooms in on 's hand as it waves. " + S2 + out, rep = sc.add_camera_moves(p) + assert rep[0]["move"] is None and "already" in rep[0]["why"] + assert "[Shot 1] The camera zooms in on 's hand as it waves. [Shot 2]" in out + + +def test_a_thing_in_most_shots_is_not_a_target(): + pytest.importorskip("spacy") + p = (HEAD + "[Shot 1] holds a rose. [Shot 2] smells the rose. " + "[Shot 3] hugs 's shoulder near the rose.") + _out, rep = sc.add_camera_moves(p) + assert all(r["target"] != "the rose" for r in rep) + assert rep[2]["target"] == "'s shoulder" + + +def test_a_shot_with_only_subjects_gets_nothing(): + pytest.importorskip("spacy") + _out, rep = sc.add_camera_moves(HEAD + "[Shot 1] laughs with . " + "[Shot 2] nods at .") + assert [r["move"] for r in rep] == [None, None] + + +def test_the_same_move_does_not_repeat_between_shots(): + pytest.importorskip("spacy") + p = HEAD + "[Shot 1] raises 's hand. [Shot 2] lifts 's chin." + _out, rep = sc.add_camera_moves(p) + assert rep[0]["move"].split(" ")[1:3] != rep[1]["move"].split(" ")[1:3] + + +def test_without_spacy_the_rules_still_find_owned_parts(monkeypatch): + monkeypatch.setattr(sc, "_nlp", None) + monkeypatch.setattr(sc, "_tried", True) + _out, rep = sc.add_camera_moves(PROMPT) + assert rep[0]["target"] == "'s hand" + + +def test_tags_survive_the_round_trip(): + pytest.importorskip("spacy") + out, _rep = sc.add_camera_moves(PROMPT) + assert out.count("") >= PROMPT.count("") + assert not any(n in out for n in sc.NAMES) + + +def test_two_topics_in_a_shot_become_a_move_from_the_first_to_the_last(): + pytest.importorskip("spacy") + p = HEAD + "[Shot 1] touches 's chin, then reaches for the lamp." + _out, rep = sc.add_camera_moves(p) + assert "'s chin" in rep[0]["move"] and "the lamp" in rep[0]["move"] + assert rep[0]["move"].index("chin") < rep[0]["move"].index("lamp") + assert " from " in rep[0]["move"] + + +def test_three_topics_go_from_the_first_to_the_last(): + pytest.importorskip("spacy") + p = (HEAD + "[Shot 1] touches 's chin, picks up the lamp, " + "then opens the window.") + _out, rep = sc.add_camera_moves(p) + assert "chin" in rep[0]["move"] and "window" in rep[0]["move"] and "lamp" not in rep[0]["move"] + + +TWO = HEAD + "[Shot 1] touches 's chin, then reaches for the lamp." +ONE = HEAD + "[Shot 1] touches 's chin." + + +def test_varied_moves_are_reproducible_from_the_seed(): + pytest.importorskip("spacy") + assert sc.add_camera_moves(PROMPT, seed=7) == sc.add_camera_moves(PROMPT, seed=7) + assert len({sc.add_camera_moves(PROMPT, seed=s)[0] for s in range(20)}) > 5 + + +def test_chance_zero_leaves_every_shot_alone_and_says_why(): + pytest.importorskip("spacy") + out, rep = sc.add_camera_moves(PROMPT, seed=1, chance=0.0) + assert out == PROMPT and all("chance" in r["why"] for r in rep) + + +def test_no_single_pattern_is_every_prompt(): + """Over many seeds a shot with two topics gets one, two and three moves, some travelling + X -> Y and some not: 'zoom in, move to Y, zoom out' is never the rule.""" + pytest.importorskip("spacy") + sentences, travel = set(), set() + for seed in range(200): + _out, rep = sc.add_camera_moves(TWO, seed=seed) + sentences.add(rep[0]["move"].count(". ") + 1) + travel.add(" from " in rep[0]["move"]) + assert sentences == {1, 2, 3} and travel == {True, False} + + +def test_a_single_topic_shot_can_still_chain_with_a_pull_back(): + pytest.importorskip("spacy") + moves = {sc.add_camera_moves(ONE, seed=s)[1][0]["move"] for s in range(60)} + assert any("Then the camera" in m for m in moves) and any("Then" not in m for m in moves) + assert all(m.startswith("The camera") for m in moves) + + +def test_a_shot_that_opens_with_a_cut_is_not_taken_for_having_a_camera_move(): + """H3 writes cuts at a shot's opening: "At 00:03.500, the camera cuts to ...".""" + pytest.importorskip("spacy") + p = (HEAD + "[Shot 1] raises 's hand. " + "[Shot 2] At 00:03.500, the camera cuts to who lifts 's chin.") + _out, rep = sc.add_camera_moves(p) + assert rep[1]["move"] and "already" not in rep[1]["why"] + p2 = HEAD + "[Shot 1] raises 's hand as the camera pushes in slowly." + assert "already" in sc.add_camera_moves(p2)[1][0]["why"] + + +@pytest.mark.parametrize("text", [ + " tilts 's head up and down.", + " rolls onto 's side and pulls out a knife.", + " looks at the camera and smiles, pushes in the last piece.", + "'s back arcs, then focuses on the screen.", + " starts to pull out slowly and pushes in again, then will push in once more.", + "Pull out slowly. Push in deeper.", + " moves up and down, then pans 's gaze across the room.", +]) +def test_body_movement_and_looking_at_the_camera_are_not_camera_moves(text): + assert not sc.CAMERA.search(sc.CUT.sub("", text)) + + +@pytest.mark.parametrize("text", [ + "The camera pushes in slowly toward .", + "Camera pans left across the room.", + "Slow zoom in on 's face.", + " waves. Pan right to reveal the door.", + "A tracking shot follows .", + "Rack focus to the window.", + "The camera slowly drifts upward.", + "The camera pulls back to a wider view.", + "The camera pushes in toward .", +]) +def test_real_camera_moves_are_still_found(text): + assert sc.CAMERA.search(sc.CUT.sub("", text)) + + +def test_a_cut_alone_is_not_a_camera_move(): + assert not sc.CAMERA.search(sc.CUT.sub("", "At 00:03.500, the camera cuts to .")) + + +def test_a_noun_shared_by_two_of_three_shots_is_still_a_target(): + """Only a noun in (nearly) every shot is a constant. Two of three is a shared topic.""" + pytest.importorskip("spacy") + p = (HEAD + "[Shot 1] holds the rose near the door. " + "[Shot 2] smells the rose beside a window. " + "[Shot 3] closes the door behind a curtain.") + _out, rep = sc.add_camera_moves(p) + assert all(r["move"] for r in rep), [r["why"] for r in rep] + + +def test_a_noun_in_every_shot_is_still_a_constant(): + pytest.importorskip("spacy") + p = (HEAD + "[Shot 1] holds the rose. [Shot 2] smells the rose. " + "[Shot 3] drops the rose.") + assert [r["move"] for r in sc.add_camera_moves(p)[1]] == [None, None, None] + + +PA = " touches 's chin and smiles." +PB = "Then opens the window and the curtain sways." +PC = " laughs and pours a glass of wine." +PIECES = [PA, PB, PC] +CUTS = ("Two friends in a sunlit loft, is a tall woman, is a man. " + f"[Shot 1] {PA} {PB} [Shot 2] {PC} Upbeat jazz music plays.") + + +def test_a_shot_whose_point_changes_is_cut_in_two_with_whole_second_times(): + pytest.importorskip("spacy") + out, info = sc.add_shot_cuts(CUTS, 12, seed=1, chance=1.0, pieces=PIECES) + assert (info["before"], info["after"]) == (2, 3) + assert info["times"] == ["00:04.000", "00:08.000"] + assert "[Shot 3] At 00:08.000," in out and "[Shot 2] At 00:04.000," in out + assert out.count("[Shot ") == 3 and out.endswith("Upbeat jazz music plays.") + assert "Then " not in out + + +def test_cut_times_rise_and_stay_inside_the_video(): + pytest.importorskip("spacy") + for secs in (5, 9, 14, 30): + _out, info = sc.add_shot_cuts(CUTS, secs, seed=2, chance=1.0, pieces=PIECES) + ts = [int(t[3:5]) + 60 * int(t[:2]) for t in info["times"]] + assert ts == sorted(set(ts)) and all(0 < t < secs for t in ts) + + +def test_no_length_or_existing_times_or_no_room_leave_the_prompt_alone(): + pytest.importorskip("spacy") + assert sc.add_shot_cuts(CUTS, None)[0] == CUTS and "length" in sc.add_shot_cuts(CUTS, None)[1]["why"] + assert sc.add_shot_cuts(CUTS, 1)[0] == CUTS + stamped = CUTS.replace("[Shot 2] ", "[Shot 2] At 00:05.000, the camera cuts to ") + out, info = sc.add_shot_cuts(stamped, 12, chance=1.0, pieces=PIECES) + assert out == stamped and "already" in info["why"] + assert sc.add_shot_cuts("no shots", 10)[0] == "no shots" + + +def test_chance_zero_still_stamps_the_existing_shots_but_splits_nothing(): + pytest.importorskip("spacy") + out, info = sc.add_shot_cuts(CUTS, 10, seed=1, chance=0.0) + assert info["after"] == 2 and len(info["times"]) == 1 and "[Shot 2] At 00:" in out + + +def test_shot_cuts_are_reproducible_from_the_seed(): + pytest.importorskip("spacy") + assert (sc.add_shot_cuts(CUTS, 12, seed=5, chance=1.0, pieces=PIECES) + == sc.add_shot_cuts(CUTS, 12, seed=5, chance=1.0, pieces=PIECES)) + + +def test_views_use_the_trusted_words_skip_shot_one_and_never_repeat(): + P = HEAD + "[Shot 1] waves. [Shot 2] nods. [Shot 3] smiles. [Shot 4] sits." + for seed in range(30): + out, added = sc.add_shot_views(P, seed=seed, chance=1.0) + assert [a["shot"] for a in added] == [2, 3, 4] + assert all(a["view"] in sc.VIEWS for a in added) + assert all(a["view"] != b["view"] for a, b in zip(added, added[1:])) + assert out.startswith(HEAD + "[Shot 1] waves. [Shot 2]") + assert len({tuple(a["view"] for a in sc.add_shot_views(P, seed=s, chance=1.0)[1]) for s in range(40)}) > 5 + + +def test_views_respect_chance_zero_and_a_view_already_stated(): + P = HEAD + "[Shot 1] waves. [Shot 2] Front view. nods. [Shot 3] smiles." + assert sc.add_shot_views(P, seed=1, chance=0.0) == (P, []) + _out, added = sc.add_shot_views(P, seed=1, chance=1.0) + assert [a["shot"] for a in added] == [3] + assert sc.add_shot_views("[Shot 1] only one.", seed=1, chance=1.0)[1][0]["shot"] == 1 # a lone shot takes a view + + +def test_a_view_goes_after_the_cut_opener_and_before_the_action(): + P = HEAD + "[Shot 1] waves. [Shot 2] At 00:04.000, the camera cuts to a new angle. nods. " + MUSIC + out, added = sc.add_shot_views(P, seed=3, chance=1.0) + v = added[0]["view"] + assert f"a new angle. {v}. nods." in out and out.endswith(MUSIC) + + +def test_a_shot_is_never_cut_inside_one_shortcut(): + """The topic changes between PA and PB. If PA+PB is ONE shortcut, nothing may be split.""" + pytest.importorskip("spacy") + one = f"{PA} {PB}" + out, info = sc.add_shot_cuts(CUTS, 12, seed=1, chance=1.0, pieces=[one, PC]) + assert info["after"] == 2 and out.count("[Shot ") == 2 + out, info = sc.add_shot_cuts(CUTS, 12, seed=1, chance=1.0) # no shortcut texts known + assert info["after"] == 2 + + +@pytest.mark.parametrize("view", sc.VIEWS) +def test_a_view_is_not_a_camera_move(view): + assert not sc.CAMERA.search(sc.CUT.sub("", view + ".")) + + +def test_a_shot_that_was_given_a_view_still_gets_its_move(): + pytest.importorskip("spacy") + p = HEAD + "[Shot 1] waves. [Shot 2] touches 's chin." + for seed in range(12): + viewed, added = sc.add_shot_views(p, seed=seed, chance=1.0) + assert added + _out, rep = sc.add_camera_moves(viewed, seed=seed, chance=1.0) + assert rep[1]["move"], (added, rep[1]["why"]) + + +def test_cut_times_split_the_scene_evenly_with_no_tiny_shots(): + """Text-length weighting gave 00:03 / 00:04 / 00:09 over 15.08s: a one-second shot.""" + pytest.importorskip("spacy") + P = HEAD + "[Shot 1] waves. [Shot 2] nods. [Shot 3] smiles. [Shot 4] sits." + _out, info = sc.add_shot_cuts(P, 15.0833) + secs = [int(t[3:5]) for t in info["times"]] + assert secs == [4, 8, 11] + edges = [0] + secs + [15.0833] + assert all(b - a >= 2 for a, b in zip(edges, edges[1:])) + assert sc.add_shot_cuts(P, 8)[1]["times"] == ["00:02.000", "00:04.000", "00:06.000"] + assert sc.add_shot_cuts(P, 4)[1]["times"] == ["00:01.000", "00:02.000", "00:03.000"] + + +def test_the_phrases_we_write_never_say_the_camera_pulls_out(): + """Only what the rewriter writes; a shortcut may say it as an action.""" + for group in (sc.MOVES_DETAIL, sc.MOVES_OTHER, sc.MOVES_TRAVEL, sc.MOVES_THEN, sc.MOVES_FINISH): + assert not any("pulls out" in m for m in group), group + + +def test_words_the_rewriter_wrote_never_become_targets(): + """'The camera pans from the shot transitions to the tip': the cut opener was read as + the shot's own content.""" + pytest.importorskip("spacy") + P = (HEAD + "[Shot 1] waves. [Shot 2] At 00:04.000, the shot transitions to the next " + "moment. Side view. touches 's chin, then reaches for the lamp.") + _out, rep = sc.add_camera_moves(P, seed=1, chance=1.0) + m = rep[1]["move"] or "" + assert "transition" not in m and "moment" not in m and "view" not in m.lower(), m + assert "chin" in m or "lamp" in m + + +def test_the_full_pipeline_output_has_no_own_words_in_a_move(): + pytest.importorskip("spacy") + P = (HEAD + f"[Shot 1] {PA} [Shot 2] {PB} {PC} " + MUSIC) + for seed in range(20): + a, _ = sc.add_shot_cuts(P, 12, seed=seed, chance=1.0, pieces=PIECES) + b, _ = sc.add_shot_views(a, seed=seed, chance=1.0) + _out, rep = sc.add_camera_moves(b, seed=seed, chance=1.0) + for r in rep: + t = (r["target"] or "").lower() + assert not any(w in t for w in ("transition", "moment", "beat", "new angle", "new view", + "switch", "changes")), t + + +def test_a_relational_noun_is_kept_with_what_it_is_part_of_or_dropped(): + pytest.importorskip("spacy") + c = {x[0]: x[1] for x in sc.candidates(" reaches for the base of the lamp.")} + assert c.get("base") == "base of the lamp" + assert "base" not in {x[0] for x in sc.candidates(" looks at the base, then leaves.")} + + +def test_an_unowned_body_part_needs_a_single_owner(): + pytest.importorskip("spacy") + one = {x[0]: x[1] for x in sc.candidates(" touches the skin.")} + assert one.get("skin") == "'s skin" or "Mariel's skin" in one.values() + two = {x[0] for x in sc.candidates(" touches the skin while watches.")} + assert "skin" not in two + + +def test_a_move_never_ends_on_a_dangling_word(): + pytest.importorskip("spacy") + P = HEAD + "[Shot 1] touches the base, then the skin, while watches." + _out, rep = sc.add_camera_moves(P, seed=1, chance=1.0) + assert rep[0]["move"] is None + + +def test_some_shots_hold_still_on_one_object_and_pans_are_the_minority(): + """Not every shot is a move: a static focus is part of the mix, and X -> Y is rarer than it.""" + pytest.importorskip("spacy") + kinds = {"hold": 0, "pan": 0, "other": 0} + for seed in range(300): + _o, rep = sc.add_camera_moves(TWO, seed=seed) + m = rep[0]["move"] + if "holds steady" in m or "stays still" in m or "stays fixed" in m: + kinds["hold"] += 1 + elif " from " in m and " to " in m and " Then" not in m: + kinds["pan"] += 1 + else: + kinds["other"] += 1 + assert kinds["hold"] > 20 and 0 < kinds["pan"] < kinds["other"], kinds + assert kinds["pan"] < 0.3 * 300, kinds + + +def test_a_body_action_from_behind_is_not_a_stated_view(): + for text in ("She is taken from behind by him.", "Water drips from above onto her.", + "He enters from below the stairs."): + assert not sc.VIEW_STATED.search(text), text + for text in ("POV view, she waves.", "Seen from behind, she walks.", "Shot from above, the room.", + "Side view of the table.", "The camera films from below."): + assert sc.VIEW_STATED.search(text), text + out, added = sc.add_shot_views("[Shot 1] A.\n[Shot 2] She is taken from behind by him.", seed=1, chance=1.0) + assert added and added[0]["shot"] == 2 + + +def test_views_say_why_a_shot_got_none(): + why = [] + sc.add_shot_views("[Shot 1] POV view, a cat.", seed=1, chance=1.0, skipped=why) + assert "already states" in why[0] + why = [] + sc.add_shot_views("[Shot 1] A.\n[Shot 2] POV view, a dog.\n[Shot 3] A bird.", seed=1, chance=0.0, skipped=why) + assert any("already states" in w for w in why) and any("by chance" in w for w in why) + + +def test_a_lone_shot_can_take_a_view_but_shot_one_of_several_still_cannot(): + out, added = sc.add_shot_views("[Shot 1] A cat sits.", seed=1, chance=1.0) + assert added and added[0]["shot"] == 1 and out.startswith("[Shot 1] ") + assert sc.view_options("[Shot 1] A cat sits.")[0]["shot"] == 1 + _o, several = sc.add_shot_views("[Shot 1] A.\n[Shot 2] B.", seed=1, chance=1.0) + assert [a["shot"] for a in several] == [2] diff --git a/modules/models/minimax_h3/pipeline.py b/modules/models/minimax_h3/pipeline.py index 8e1660a2..1ce1a00e 100644 --- a/modules/models/minimax_h3/pipeline.py +++ b/modules/models/minimax_h3/pipeline.py @@ -88,8 +88,12 @@ def h3_reference_to_video(): {"at": "generation.prompt", "input": "enabled", "label": "Enhance prompt"}], "inputs": {"clip": ["clip", 0], "text": "", "enabled": False, "image": ["source_image", 0]}}, + # Off by default (its own switches in Settings): writes cut times, views and camera moves into [Shot N] blocks. + {"id": "shot_camera", "group": "Preparation", "node": "FunPackShotCamera", + "roles": [{"at": "project.video", "input": "length", "label": "Length", "drives": "frames"}], + "inputs": {"text": ["enhance", 0], "settings": ["settings", 0], "length": 124, "frame_rate": 24.0}}, {"id": "markup", "group": "Preparation", "node": "FunPackPromptMarkup", - "inputs": {"text": ["enhance", 0]}}, + "inputs": {"text": ["shot_camera", 0]}}, {"id": "r2v", "group": "Preparation", "node": "MiniMaxH3ReferenceToVideo", "roles": [{"at": "project.video", "input": "length", "label": "Length", "drives": "frames"}], "inputs": { diff --git a/modules/system/taste/__init__.py b/modules/system/taste/__init__.py index 5aba7eca..e7ff868a 100644 --- a/modules/system/taste/__init__.py +++ b/modules/system/taste/__init__.py @@ -12,6 +12,7 @@ import os import tempfile from . import store, value +from ... import _core ID = "taste" TITLE = "Taste key" @@ -165,6 +166,14 @@ def routes(table, base, web): out = store.rate(body.get("prompt_id"), body.get("rating"), body.get("axis")) except (ValueError, AttributeError) as exc: return web.json_response({"why": str(exc)}, status=400) + # any module that learns from ratings (provides "on_rating") hears of it; one failing never loses the rating + for spec in _core.registry.current().specs.values(): + hear = spec.provides.get("on_rating") + if hear: + try: + hear(body.get("prompt_id"), body.get("rating"), body.get("axis")) + except Exception as exc: + _core.log.once(f"on_rating:{spec.id}", _core.log.ALERT, spec.title, f"could not learn from this rating: {exc}") return web.json_response(out) @table.post(base + "/generation")