From 768838c3d7bbddacd7adade18f7276dfe27fda7f Mon Sep 17 00:00:00 2001 From: digital-garbage <133010426+digital-garbage@users.noreply.github.com> Date: Thu, 1 Oct 2026 23:07:52 +0300 Subject: [PATCH] feat: camera moves, styles, targets and shot splits learn from ratings on their own Co-Authored-By: Claude Sonnet 5.5 --- CHANGELOG.md | 1 + conditioning.py | 38 +++++++++++++++++++------- focus_memory.py | 55 +++++++++++++++++++++++++++++++++++--- shot_camera.py | 54 +++++++++++++++++++++++++------------ tests/test_focus_memory.py | 38 ++++++++++++++++++++++++++ 5 files changed, 157 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e2af2ff5..612c861f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## [Unreleased] ### Added +- Camera moves and shot cuts now learn from your ratings on their own, no Reactive focus needed: how often a shot gets a move, whether it holds / travels / chains moves, which words it aims at, and whether a shot whose point changes gets split. Liked runs raise those choices, disliked ones lower them (picture-only dislikes teach nothing). - The MiniMax H3 X2 Detail VAE works as the Chain Sampler's video VAE: its packed output is detected and decoded to 2x frames (final decode, previews, continuation anchors, detailing detection). ### Changed diff --git a/conditioning.py b/conditioning.py index bc2f221a..ed731749 100644 --- a/conditioning.py +++ b/conditioning.py @@ -776,6 +776,7 @@ def split_prompt_by_transitions(prompt, placement="start"): _RUN_VIEWS = [] # the views this run added, for the rating that follows it +_RUN_ARMS = [] # camera-move / split choices this run drew on its own, same purpose def focus_scene_raw(scene_segments, variables=None): @@ -8819,25 +8820,28 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): @staticmethod def _v2_rate_shot_views(previous_run, profile): - """The rating of the last generation, passed on to the views it carried: liked = good, - disliked = bad. A picture-only dislike and the neutral 'Wrong ...' ratings say nothing - about a view, so they teach nothing.""" + """The rating of the last generation, passed on to what the camera chose in it (views, + whether a shot got a move, its style and target, splits): liked = good, disliked = bad. + A picture-only dislike and the neutral 'Wrong ...' ratings say nothing about a camera + choice, so they teach nothing.""" used = (previous_run or {}).get("shot_views") or [] + arms = (previous_run or {}).get("shot_arms") or [] reward = float(profile.get("reward", 0.0) or 0.0) - if (not used or profile.get("skip_learning") or profile.get("skip_value_function") + if ((not used and not arms) or profile.get("skip_learning") or profile.get("skip_value_function") or profile.get("axis") == "image" or abs(reward) < 0.05): return + sign = 1 if reward > 0 else -1 try: try: from . import focus_memory as _fm except ImportError: import focus_memory as _fm - n = _fm.rate_views(used, 1 if reward > 0 else -1) + n = _fm.rate_views(used, sign) + _fm.rate_arms(arms, sign) if n: - print(f"[FunPackVideoRefinerV2] shot views: rated {n} view(s) " - f"{'good' if reward > 0 else 'bad'} ({profile.get('label') or profile.get('key')})") + print(f"[FunPackVideoRefinerV2] shot camera: rated {n} choice(s) " + f"{'good' if sign > 0 else 'bad'} ({profile.get('label') or profile.get('key')})") except Exception as e: # noqa: BLE001 - print(f"[FunPackVideoRefinerV2] shot views: could not rate | {e}") + print(f"[FunPackVideoRefinerV2] shot camera: could not rate | {e}") def _v2_rating_is_discard(self, learning_profile): """True when the run must leave NO trace in refiner state — the user's @@ -9409,8 +9413,17 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): from . import shot_camera as _sc except ImportError: import shot_camera as _sc + try: + from . import focus_memory as _fm + except ImportError: + import focus_memory as _fm + try: + chance = _fm.split_chance(chance) # tilted by how split runs were rated + except Exception: # noqa: BLE001 + pass out, info = _sc.add_shot_cuts(text, seconds, seed=seed, chance=chance, pieces=_shortcut_texts() if chance > 0 else ()) + _RUN_ARMS.extend(info.get("arms", [])) except Exception as e: # noqa: BLE001 print(f"[FunPackVideoRefinerV2] shot cuts: failed | {where}: {e}; prompt left as written") return text @@ -9446,8 +9459,13 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): chance = _fm.effective_chance(chance) # pulled toward how often moves are kept except Exception: # noqa: BLE001 pass - out, report = _sc.add_camera_moves(text, seed=seed, chance=chance, prior=prior, + try: + arms = _fm.arm_stats() + except Exception: # noqa: BLE001 + arms = {} + out, report = _sc.add_camera_moves(text, seed=seed, chance=chance, prior=prior, arms=arms, choices=choices if isinstance(choices, dict) else None) + _RUN_ARMS.extend(a for r in report for a in r.get("arms", [])) if report: try: _fm.observe(_sc.content_fingerprint(text), @@ -10468,6 +10486,7 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): global_state.setdefault("bad_dir", {}) global_state.setdefault("successful_seed_memory", {}) _RUN_VIEWS.clear() + _RUN_ARMS.clear() previous_run = state.get("last_run") previous_run_refusal = self._v2_run_looks_like_refusal(previous_run) has_previous_run = isinstance(previous_run, dict) and not previous_run_refusal @@ -11435,6 +11454,7 @@ class FunPackVideoRefinerV2(FunPackVideoRefiner): "scene_refinement_keys": [sorted(s) for s in scene_refinement_keys], "gen_context": current_gen_context, "shot_views": list(_RUN_VIEWS), + "shot_arms": list(_RUN_ARMS), } # "-Just forget it-" ("Just ignore it") and auto-discarded refusals must leave NO # trace: skip the feedback-learning sinks too, not only the rating-learning ones above. diff --git a/focus_memory.py b/focus_memory.py index 9d0a87c5..55f5857f 100644 --- a/focus_memory.py +++ b/focus_memory.py @@ -21,6 +21,7 @@ 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(): @@ -59,6 +60,10 @@ def prior(): 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 @@ -135,6 +140,37 @@ def learn_views(decisions): return n +def arm_stats(): + """{arm: (good, bad)} for what ratings taught about the camera's own choices: "move:yes" / + "move:no" (a shot got a move / was left alone), "style:hold|travel|k1|k2|k3", "word:", + "split:yes|no" (a shot whose point changes was cut in two / kept whole).""" + return {k: (float(v[0]), float(v[1])) for k, v in (_read().get("arms") or {}).items() + if isinstance(v, (list, tuple)) and len(v) == 2} + + +def arm_rate(key, stats=None): + """Smoothed good-rate of one arm: 0.5 until rated.""" + g, b = (stats if stats is not None else arm_stats()).get(key, (0.0, 0.0)) + return (g + 1.0) / (g + b + 2.0) + + +def rate_arms(used, sign): + """A rated generation: every choice the camera made in it shares the credit (sign > 0) or + the blame (sign < 0), so a run with many choices does not punish each as hard as a run with + one. `used`: [arm key]. -> arms rated.""" + used = [u for u in used or [] if isinstance(u, str) and u] + if not used or not sign: + return 0 + data = _read() + arms = data.setdefault("arms", {}) + share = 1.0 / len(used) + for u in used: + g, b = arms.get(u, [0.0, 0.0]) + arms[u] = [round(g + (share if sign > 0 else 0.0), 4), round(b + (share if sign < 0 else 0.0), 4)] + _save(data) + return len(used) + + def summary(limit=40): """What is remembered, for the person to read and prune: words (picked / replaced / merely recurring) and views (good / bad, with the per-trait rows folded under their view).""" @@ -161,6 +197,7 @@ def forget(kind, name=None): 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}@")] @@ -178,9 +215,21 @@ def effective_chance(chance): (of the shots they reviewed) once enough were reviewed.""" data = _read() total, kept = int(data.get("shots", 0)), int(data.get("kept", 0)) - if total < MIN_SHOTS_FOR_RATE: - return chance - return max(0.05, min(1.0, (chance + kept / total) / 2.0)) + 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): diff --git a/shot_camera.py b/shot_camera.py index c6e88bf4..d9fab84d 100644 --- a/shot_camera.py +++ b/shot_camera.py @@ -263,10 +263,18 @@ def _draw(cands, prior, rng, n): return chosen -def _plan(pool, last, rng, prior=None, force=None, mode="auto"): - """-> (moves text, target description, opening template). Fixed and minimal without +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.""" + 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 @@ -276,28 +284,28 @@ def _plan(pool, last, rng, prior=None, force=None, mode="auto"): if mode == "hold": hold = _pick(MOVES_HOLD, last, rng) target = _name(best) - return hold.format(x=target), target, hold + 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 + 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), COUNT_WEIGHTS)[0] - travel = k == 1 and len(topics) >= 2 and rng.random() < TRAVEL_SHARE + 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 - if rng is not None and k == 1 and rng.random() < HOLD_SHARE: + 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 + 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) @@ -313,7 +321,7 @@ def _plan(pool, last, rng, prior=None, force=None, mode="auto"): 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 + return " ".join(sentences), desc, opening, (f"k{k}" if rng is not None and mode == "auto" else None) def _analyse(text): @@ -374,12 +382,14 @@ def focus_options(text, prior=None): return out -def add_camera_moves(text, seed=None, chance=1.0, prior=None, choices=None): +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.""" + shot from the seed, so the same prompt and seed always give the same text. + `arms`: what ratings taught (see _plan); each report entry's "arms" lists the choices drawn + on their own for the rating that follows ("move:yes"/"move:no", a style, "word:").""" marks, header, parts, pools = _analyse(text) if not marks: return text, [] @@ -387,7 +397,7 @@ def add_camera_moves(text, seed=None, chance=1.0, prior=None, choices=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)} + "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. @@ -411,9 +421,14 @@ def add_camera_moves(text, seed=None, chance=1.0, prior=None, choices=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 = _plan(pool, last, rng, shot_prior, force=force, mode=mode) + 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 "") @@ -470,7 +485,7 @@ def add_shot_cuts(text, seconds, seed=0, chance=0.5, pieces=()): 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": ""} + info = {"before": len(marks), "after": len(marks), "times": [], "why": "", "arms": []} if not marks: return text, info if not seconds or seconds < MIN_SHOT_SECONDS: @@ -495,7 +510,10 @@ def add_shot_cuts(text, seconds, seed=0, chance=0.5, pieces=()): for i, (picture, sound) in enumerate(parts): rng = random.Random(f"{seed}:cut:{i}") j, sentences = _switch_point(picture, constant, pieces) - if j is not None and rng.random() < chance: + 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) @@ -504,6 +522,7 @@ def add_shot_cuts(text, seconds, seed=0, chance=0.5, pieces=()): 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). @@ -517,6 +536,7 @@ def add_shot_cuts(text, seconds, seed=0, chance=0.5, pieces=()): 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): diff --git a/tests/test_focus_memory.py b/tests/test_focus_memory.py index d3f73322..35d11bf1 100644 --- a/tests/test_focus_memory.py +++ b/tests/test_focus_memory.py @@ -305,3 +305,41 @@ def test_what_is_remembered_can_be_read_and_taken_back(tmp_path, monkeypatch): assert not fm.forget("word", "never-heard-of-it") fm.forget("all") assert fm.summary()["words"] == [] and fm.summary()["shots"] == 0 + + +# ── 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 + + +def test_a_rating_reaches_the_camera_choices_the_run_made(tmp_path, monkeypatch): + import conditioning + monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json")) + run = {"shot_arms": ["move:yes", "word:lamp"]} + rate = conditioning.FunPackVideoRefinerV2._v2_rate_shot_views + rate(run, {"reward": -0.9, "axis": "image"}) + assert fm.arm_stats() == {} + rate(run, {"reward": -0.9, "axis": "composition"}) + rate(run, {"reward": 1.0}) + assert fm.arm_stats()["word:lamp"] == (0.5, 0.5)