digital-garbageandClaude Sonnet 5.5
259900e84a
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
200 lines
10 KiB
Python
200 lines
10 KiB
Python
"""The node's rewrite, the rating hook and the module contract."""
|
|
import importlib
|
|
|
|
import pytest
|
|
|
|
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_cut_within_the_same_shot_is_its_own_switch(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json"))
|
|
m = _mod()
|
|
assert m.SETTINGS["cut_same_shot"]["type"] == "bool" and "when" not in m.SETTINGS["cut_same_shot"]
|
|
assert m.SETTINGS["shot_cuts_chance"]["when"] == {"cut_same_shot": True}
|
|
dance = "She sways her hips from side to side, smiling at the room."
|
|
text = f"[Shot 1] {dance} {dance} {dance}"
|
|
kept, _, _ = nodes.rewrite(text, {"shot_cuts": True, "cut_same_shot": False, "shot_cuts_chance": 1},
|
|
seconds=12, pieces=[dance])
|
|
assert kept.count("[Shot ") == 1
|
|
cut, said, _ = nodes.rewrite(text, {"cut_same_shot": True, "shot_cuts_chance": 1}, seconds=12, pieces=[dance])
|
|
assert cut.count("[Shot ") == 3 and "00:04.000" in cut and "00:08.000" in cut and "cuts at" in said
|
|
|
|
|
|
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", "liked") == 0 # the same rating again counts once
|
|
assert memory.view_stats()["side view"] == (1.0, 0.0)
|
|
assert memory.on_rating("unknown", "liked") == 0
|
|
|
|
|
|
def test_a_like_teaches_a_detail_and_a_bad_composition_can_retire_it(tmp_path, monkeypatch):
|
|
pytest.importorskip("spacy")
|
|
monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json"))
|
|
m = _mod()
|
|
assert m.SETTINGS["detail_notes"]["type"] == "bool" and "when" not in m.SETTINGS["detail_notes"]
|
|
assert m.SETTINGS["detail_notes_chance"]["when"] == {"detail_notes": True}
|
|
text = "[Shot 1] She kisses his lips."
|
|
memory.record_run("liked", {"views": [], "arms": [], "text": "[Shot 1] She has detailed lips.", "details": []})
|
|
assert memory.on_rating("liked", "disliked", "image") == 0
|
|
assert memory.detail_bank() == []
|
|
assert memory.on_rating("liked", "liked") == 1
|
|
assert memory.detail_bank()[0]["phrase"] == "detailed lips"
|
|
out, said, chose = nodes.rewrite(text, {"detail_notes": True, "detail_notes_chance": 1})
|
|
assert "Detailed lips." in out and "shot 1 detailed lips" in said
|
|
assert out.startswith("[Shot 1] She kisses his lips.")
|
|
memory.record_run("bad1", {"views": [], "arms": [], "text": out, "details": chose["details"]})
|
|
memory.record_run("bad2", {"views": [], "arms": [], "text": out, "details": chose["details"]})
|
|
memory.on_rating("bad1", "disliked", "composition")
|
|
memory.on_rating("bad2", "disliked", "composition")
|
|
assert memory.detail_bank() == []
|
|
quiet, _, _ = nodes.rewrite(text, {"detail_notes": True, "detail_notes_chance": 1})
|
|
assert quiet == text
|
|
|
|
|
|
def test_contract():
|
|
m = _mod()
|
|
assert m.PROVIDES["on_rating"]("nope", "liked") == 0
|
|
assert m.CATEGORY == "conditioning" and m.NODES
|
|
|
|
|
|
def test_last_rating_wins_and_forgetting_takes_it_back(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json"))
|
|
memory.record_run("a", {"views": [], "arms": ["move:yes"]})
|
|
assert memory.on_rating("a", "disliked") == 1
|
|
assert memory.arm_stats()["move:yes"] == (0.0, 1.0)
|
|
memory.on_rating("a", "disliked", "image") # refined: the picture, not the camera
|
|
assert memory.arm_stats()["move:yes"] == (0.0, 0.0)
|
|
memory.on_rating("a", "liked")
|
|
memory.on_rating("a", None) # forgotten
|
|
assert memory.arm_stats()["move:yes"] == (0.0, 0.0)
|
|
|
|
|
|
def test_wrong_shaped_memory_and_threads_do_not_break(tmp_path, monkeypatch):
|
|
import threading
|
|
f = tmp_path / "m.json"
|
|
monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(f))
|
|
f.write_text('{"runs":[1],"seen":{"x":"a"},"prompts":5,"hashes":3}')
|
|
memory.record_run("p", {"views": [], "arms": ["a"]})
|
|
assert memory.prior() is not None or True
|
|
def work(i):
|
|
for j in range(40):
|
|
memory.record_run(f"{i}-{j}", {"views": [], "arms": ["a"]})
|
|
memory.on_rating(f"{i}-{j}", "liked")
|
|
ts = [threading.Thread(target=work, args=(i,)) for i in range(4)]
|
|
[t.start() for t in ts]
|
|
[t.join() for t in ts]
|
|
assert memory.arm_stats()["a"][0] > 0
|
|
|
|
|
|
def test_more_shots_than_seconds_refuses_instead_of_writing_bad_times(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json"))
|
|
text = "\n".join(f"[Shot {i}] A woman walks along street number {i}." for i in range(1, 8))
|
|
out, said, _ = nodes.rewrite(text, {"shot_cuts": True}, seconds=5.17, pieces=[])
|
|
assert out == text and "do not fit" in said
|
|
|
|
|
|
def test_memory_routes_read_and_forget(tmp_path, monkeypatch):
|
|
import asyncio, json
|
|
monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json"))
|
|
memory.record_run("a", {"views": [{"view": "side view", "traits": ["none"]}], "arms": []})
|
|
memory.on_rating("a", "liked")
|
|
handlers = {}
|
|
|
|
class Table:
|
|
def get(self, path): return lambda fn: handlers.setdefault(("GET", path), fn)
|
|
def post(self, path): return lambda fn: handlers.setdefault(("POST", path), fn)
|
|
|
|
class web:
|
|
@staticmethod
|
|
def json_response(data, status=200):
|
|
return type("R", (), {"status": status, "data": data})()
|
|
|
|
class Req:
|
|
def __init__(self, body=None): self._b = body
|
|
async def json(self): return self._b
|
|
|
|
_mod().routes(Table(), "/b", web)
|
|
got = asyncio.run(handlers[("GET", "/b/memory")](Req()))
|
|
assert [v["view"] for v in got.data["views"]] == ["side view"]
|
|
assert asyncio.run(handlers[("POST", "/b/forget")](Req({"kind": "view", "name": "side view"}))).data == {"forgotten": True}
|
|
assert asyncio.run(handlers[("GET", "/b/memory")](Req())).data["views"] == []
|
|
assert asyncio.run(handlers[("POST", "/b/forget")](Req({"kind": "nonsense"}))).status == 400
|
|
|
|
|
|
def test_forget_is_complete_and_stays_forgotten(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json"))
|
|
memory.observe("h1", ["lamp"])
|
|
memory.record_run("a", {"views": [{"view": "side view", "traits": ["none"]}], "arms": ["word:lamp", "split:yes"]})
|
|
memory.on_rating("a", "liked")
|
|
assert memory.forget("word", "lamp") and "word:lamp" not in memory.arm_stats()
|
|
assert memory.forget("view", "side view")
|
|
memory.on_rating("a", "disliked") # the run that taught it is gone too
|
|
assert memory.view_stats() == {}
|
|
assert [a["arm"] for a in memory.summary()["arms"]] == ["split:yes"]
|
|
assert memory.forget("arm", "split:yes") and memory.summary()["arms"] == []
|
|
|
|
|
|
def test_a_bad_composition_dislike_blames_the_camera_choices(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json"))
|
|
memory.record_run("c", {"views": [{"view": "low angle", "traits": ["none"]}], "arms": ["move:yes", "style:travel"]})
|
|
assert memory.on_rating("c", "disliked", "composition") == 3
|
|
assert memory.arm_stats()["style:travel"] == (0.0, 0.5) # two camera choices share one blame
|
|
assert memory.view_stats()["low angle"] == (0.0, 1.0)
|
|
assert memory.split_chance(0.5) == 0.5 and memory.effective_chance(0.7) < 0.7 # the next run moves the camera less often
|
|
|
|
|
|
def test_a_node_that_will_not_rewrite_does_not_invalidate_what_comes_after_it(monkeypatch):
|
|
""""Always changed" also re-runs the prompt/reference encode downstream (~7s on H3) on every run."""
|
|
import math
|
|
shots = "[Shot 1] A woman walks.\n[Shot 2] She sits."
|
|
monkeypatch.setattr(nodes, "_values", lambda settings: dict(settings or {}))
|
|
assert nodes.FunUIShotCamera.fingerprint_inputs(text=shots, settings={}) == ""
|
|
assert nodes.FunUIShotCamera.fingerprint_inputs(text="no shots here", settings={"camera_moves": True}) == ""
|
|
assert math.isnan(nodes.FunUIShotCamera.fingerprint_inputs(text=shots, settings={"camera_moves": True}))
|
|
|
|
|
|
def test_camera_switch_turns_off_the_camera_rewrites_and_only_those(tmp_path, monkeypatch):
|
|
"""_off.camera: moves, cuts, views, reframes, liked details (and their chances) sit out; the plain pipeline stays."""
|
|
import types
|
|
monkeypatch.setenv("SHOT_CAMERA_MEMORY", str(tmp_path / "m.json"))
|
|
m = _mod()
|
|
spec = types.SimpleNamespace(id=nodes.ID, settings=m.SETTINGS, defaults=lambda: {k: v.get("default") for k, v in m.SETTINGS.items()})
|
|
monkeypatch.setattr(nodes.registry_mod, "current", lambda *a, **k: types.SimpleNamespace(specs={nodes.ID: spec}))
|
|
every = {k: True for k in nodes.CAMERA_SWITCHES}
|
|
every |= {k: 1.0 for k in nodes.CAMERA_CHANCES}
|
|
on = {nodes.ID: every}
|
|
text = "[Shot 1] A woman walks.\n[Shot 2] She sits down."
|
|
assert nodes._rewrites(nodes._values(on), text)
|
|
off = nodes._values({**on, "_off": {"camera": True}})
|
|
assert all(off[k] is False for k in nodes.CAMERA_SWITCHES) and all(off[k] == 0.0 for k in nodes.CAMERA_CHANCES)
|
|
assert off["detail_notes"] is False and off["detail_notes_chance"] == 0.0
|
|
assert nodes.FunUIShotCamera.execute(text, 124, 24.0, settings={nodes.ID: {k: True for k in nodes.CAMERA_SWITCHES}, "_off": {"camera": True}}).args[0] == text
|
|
assert nodes._values({**on, "_off": {"all": True}}) == {} # all enhancements off: liked details too
|
|
# Flag off, or only "camera": false: the camera works as before.
|
|
assert nodes._values({**on, "_off": {"camera": False}}) == nodes._values(on)
|
|
# Every camera key is a real setting, and nothing outside the list is touched.
|
|
assert set(nodes.CAMERA_SWITCHES + nodes.CAMERA_CHANCES) <= set(m.SETTINGS)
|
|
untouched = {k for k in m.SETTINGS} - set(nodes.CAMERA_SWITCHES + nodes.CAMERA_CHANCES)
|
|
assert all(off[k] == nodes._values(on)[k] for k in untouched)
|