feat: shot camera (cut times, views, camera moves for [Shot N]) with rating-taught memory
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
digital-garbage
co-authored by
Claude Sonnet 5.5
Claude Sonnet 5.5
parent
fd5e072826
commit
dce64c1813
@@ -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",
|
||||
|
||||
@@ -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 ("<Subject 1>'s hand", not "<Subject 1>")
|
||||
|
||||
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}
|
||||
@@ -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. `<Subject N>` tags
|
||||
are the subjects every shot is about, so they are never a target on their own; a part or
|
||||
belonging of one ("<Subject 1>'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 `<Subject N>` 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 <Subject 1>'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"<Subject {NAMES.index(m.group(1)) + 1}>", 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 # "<Subject 1> 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|<Subject|his|her|their)\b", text, re.I) and "'s " not in text:
|
||||
text = "the " + text
|
||||
return text
|
||||
|
||||
|
||||
def _topics(pool):
|
||||
"""The distinct things a shot dwells on, in the order the text reaches them. Only things
|
||||
with a role beyond being the doer (owned parts, objects of a verb or preposition)."""
|
||||
seen, out = set(), []
|
||||
for c in sorted(pool, key=lambda c: c[3]):
|
||||
if _score(c) > 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:<lemma>")."""
|
||||
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()
|
||||
@@ -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
|
||||
`<ComfyUI user dir>/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:<lemma>",
|
||||
"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)
|
||||
@@ -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)
|
||||
@@ -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] <Subject 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] <Subject 1> touches <Subject 1>'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] <Subject 1> waves. [Shot 2] <Subject 2> nods."
|
||||
varied = ("Intro. [Shot 1] <Subject 1> waves. [Shot 2] At 00:04.000, the shot changes to a new view. "
|
||||
"Side view. <Subject 2> 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] <Subject 1> touches <Subject 1>'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] <Subject 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] <Subject 1> touches <Subject 1>'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] <Subject 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] <Subject 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] <Subject 1> waves. [Shot 2] <Subject 2> raises the lamp."
|
||||
viewed = ("Intro. [Shot 1] <Subject 1> waves. [Shot 2] At 00:04.000, the camera cuts to a new angle. "
|
||||
"Side view. <Subject 2> 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
|
||||
|
||||
@@ -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
|
||||
@@ -0,0 +1,405 @@
|
||||
"""Camera moves for [Shot N] blocks. Fixtures are neutral stand-ins with the real shape:
|
||||
<Subject N> 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, <Subject 1> is a tall woman with red hair, "
|
||||
"<Subject 2> is a man in a grey coat. Cinematic 35mm look. ")
|
||||
S1 = "[Shot 1] <Subject 1> is waving <Subject 1>'s hand, greeting <Subject 2>. "
|
||||
S2 = "[Shot 2] <Subject 1> bites <Subject 1>'s lower lip while <Subject 2> smiles at <Subject 1>. "
|
||||
S3 = "[Shot 3] <Subject 2> hands <Subject 1> a glass of wine across the table. <Subject 1> 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"] == "<Subject 1>'s hand"
|
||||
assert rep[1]["target"] == "<Subject 1>'s lower lip"
|
||||
for r in rep:
|
||||
assert r["move"] and "<Subject 1>." not in r["move"].replace("'s", "")
|
||||
assert rep[0]["move"] in out and "<Subject 1>'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 <Subject 1>'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 <Subject 1>'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] <Subject 1> holds a rose. [Shot 2] <Subject 1> smells the rose. "
|
||||
"[Shot 3] <Subject 2> hugs <Subject 1>'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"] == "<Subject 1>'s shoulder"
|
||||
|
||||
|
||||
def test_a_shot_with_only_subjects_gets_nothing():
|
||||
pytest.importorskip("spacy")
|
||||
_out, rep = sc.add_camera_moves(HEAD + "[Shot 1] <Subject 1> laughs with <Subject 2>. "
|
||||
"[Shot 2] <Subject 2> nods at <Subject 1>.")
|
||||
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] <Subject 1> raises <Subject 1>'s hand. [Shot 2] <Subject 2> lifts <Subject 2>'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"] == "<Subject 1>'s hand"
|
||||
|
||||
|
||||
def test_tags_survive_the_round_trip():
|
||||
pytest.importorskip("spacy")
|
||||
out, _rep = sc.add_camera_moves(PROMPT)
|
||||
assert out.count("<Subject 1>") >= PROMPT.count("<Subject 1>")
|
||||
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] <Subject 1> touches <Subject 1>'s chin, then reaches for the lamp."
|
||||
_out, rep = sc.add_camera_moves(p)
|
||||
assert "<Subject 1>'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] <Subject 1> touches <Subject 1>'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] <Subject 1> touches <Subject 1>'s chin, then reaches for the lamp."
|
||||
ONE = HEAD + "[Shot 1] <Subject 1> touches <Subject 1>'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] <Subject 1> raises <Subject 1>'s hand. "
|
||||
"[Shot 2] At 00:03.500, the camera cuts to <Subject 2> who lifts <Subject 2>'s chin.")
|
||||
_out, rep = sc.add_camera_moves(p)
|
||||
assert rep[1]["move"] and "already" not in rep[1]["why"]
|
||||
p2 = HEAD + "[Shot 1] <Subject 1> raises <Subject 1>'s hand as the camera pushes in slowly."
|
||||
assert "already" in sc.add_camera_moves(p2)[1][0]["why"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("text", [
|
||||
"<Subject 1> tilts <Subject 1>'s head up and down.",
|
||||
"<Subject 1> rolls onto <Subject 1>'s side and pulls out a knife.",
|
||||
"<Subject 1> looks at the camera and smiles, pushes in the last piece.",
|
||||
"<Subject 1>'s back arcs, then <Subject 2> focuses on the screen.",
|
||||
"<Subject 1> starts to pull out slowly and pushes in again, then will push in once more.",
|
||||
"Pull out slowly. Push in deeper.",
|
||||
"<Subject 1> moves up and down, then pans <Subject 1>'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 <Subject 1>.",
|
||||
"Camera pans left across the room.",
|
||||
"Slow zoom in on <Subject 1>'s face.",
|
||||
"<Subject 1> waves. Pan right to reveal the door.",
|
||||
"A tracking shot follows <Subject 1>.",
|
||||
"Rack focus to the window.",
|
||||
"The camera slowly drifts upward.",
|
||||
"The camera pulls back to a wider view.",
|
||||
"The camera pushes in toward <Subject 1>.",
|
||||
])
|
||||
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 <Subject 2>."))
|
||||
|
||||
|
||||
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] <Subject 1> holds the rose near the door. "
|
||||
"[Shot 2] <Subject 2> smells the rose beside a window. "
|
||||
"[Shot 3] <Subject 1> 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] <Subject 1> holds the rose. [Shot 2] <Subject 2> smells the rose. "
|
||||
"[Shot 3] <Subject 1> drops the rose.")
|
||||
assert [r["move"] for r in sc.add_camera_moves(p)[1]] == [None, None, None]
|
||||
|
||||
|
||||
PA = "<Subject 1> touches <Subject 1>'s chin and smiles."
|
||||
PB = "Then <Subject 2> opens the window and the curtain sways."
|
||||
PC = "<Subject 1> laughs and pours a glass of wine."
|
||||
PIECES = [PA, PB, PC]
|
||||
CUTS = ("Two friends in a sunlit loft, <Subject 1> is a tall woman, <Subject 2> 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 <Subject 2>" 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] <Subject 1> waves. [Shot 2] <Subject 2> nods. [Shot 3] <Subject 1> smiles. [Shot 4] <Subject 2> 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] <Subject 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] <Subject 1> waves. [Shot 2] Front view. <Subject 2> nods. [Shot 3] <Subject 1> 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] <Subject 1> waves. [Shot 2] At 00:04.000, the camera cuts to a new angle. <Subject 2> nods. " + MUSIC
|
||||
out, added = sc.add_shot_views(P, seed=3, chance=1.0)
|
||||
v = added[0]["view"]
|
||||
assert f"a new angle. {v}. <Subject 2> 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] <Subject 1> waves. [Shot 2] <Subject 2> touches <Subject 2>'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] <Subject 1> waves. [Shot 2] <Subject 2> nods. [Shot 3] <Subject 1> smiles. [Shot 4] <Subject 2> 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] <Subject 1> waves. [Shot 2] At 00:04.000, the shot transitions to the next "
|
||||
"moment. Side view. <Subject 2> touches <Subject 2>'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(" <Subject 1> 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(" <Subject 1> 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(" <Subject 1> touches the skin.")}
|
||||
assert one.get("skin") == "<Subject 1>'s skin" or "Mariel's skin" in one.values()
|
||||
two = {x[0] for x in sc.candidates(" <Subject 1> touches the skin while <Subject 2> watches.")}
|
||||
assert "skin" not in two
|
||||
|
||||
|
||||
def test_a_move_never_ends_on_a_dangling_word():
|
||||
pytest.importorskip("spacy")
|
||||
P = HEAD + "[Shot 1] <Subject 1> touches the base, then the skin, while <Subject 2> 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]
|
||||
@@ -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": {
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user