File size: 10,016 Bytes
e488ce7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 | #!/usr/bin/env python3
"""MAI Video Compare node side: the viewer extras, and the promise that adding
them changed nothing for a workflow that predates them.
python tests/test_video_compare.py
Synthetic, no GPU, no models, no ComfyUI running. folder_paths and
comfy_api are stubbed; comfy.nested_tensor is stubbed for motion.py,
which window_expand imports for the burst arithmetic.
What it checks:
1. HOLD MAP -> SPANS on a real map: the pier graph's H3ManualHoldMap
(length 243, ranges "132-150", hold 4, ramp on, bridge 8) run through
the node's converter, which must agree frame for frame with
window_expand._bursts (the arithmetic is imported, so a drift here
means someone copied it).
2. THE MANIFEST carries `spans` and `curves` when they are wired, and
carries neither key when they are not.
3. BYTE COMPATIBILITY: a two-video call with no hold_map and no curves
emits exactly the manifest string the pre-change node emitted, and
`winner_video` / `winner_index` are untouched.
4. WIDGET ORDER: INPUT_TYPES before vs after for both classes is
append-only, including MAISeedHunter, whose seeds must stay ahead of
the new inputs.
Exit code 0 = pass.
"""
import importlib
import json
import os
import sys
import tempfile
import types
HERE = os.path.dirname(os.path.abspath(__file__))
PACK = os.path.dirname(HERE)
sys.path.insert(0, os.path.dirname(PACK))
sys.path.insert(0, os.environ.get("COMFYUI_DIR", "/mnt/work/ai/apps/ComfyUI"))
# The pack directory name is not an importable identifier, so bind it to one:
# the modules use relative imports and must be loaded AS a package.
_pkg = types.ModuleType("mainodes_pack")
_pkg.__path__ = [PACK]
sys.modules["mainodes_pack"] = _pkg
if "comfy.nested_tensor" not in sys.modules: # motion.py's only comfy import
class _StubNested:
def __init__(self, tensors, *a, **k):
self.tensors = tensors
def unbind(self):
return self.tensors
_c = sys.modules.setdefault("comfy", types.ModuleType("comfy"))
_m = types.ModuleType("comfy.nested_tensor")
_m.NestedTensor = _StubNested
_c.nested_tensor = _m
sys.modules["comfy.nested_tensor"] = _m
vc = importlib.import_module("mainodes_pack.video_compare")
we = importlib.import_module("mainodes_pack.window_expand")
motion = importlib.import_module("mainodes_pack.motion")
FAILS = []
def check(name, ok, detail=""):
print((" ok " if ok else " FAIL ") + name + ((" " + detail) if detail else ""))
if not ok:
FAILS.append(name)
# ---------------------------------------------------------------------------
# 1. hold map -> regenerated-window spans, on the pier graph's real map
# ---------------------------------------------------------------------------
print("1. hold_map -> spans (pier: length 243, ranges '132-150', hold 4, ramp, bridge 8)")
PIER = motion.H3ManualHoldMap().build(length=243, fps=24, ranges="132-150",
hold=4, ramp=True, bridge=8)[0]
holds = json.loads(PIER)["holds"]
spans = vc._spans_from_hold_map(PIER)
runs = []
for i, h in enumerate(holds):
if not runs or runs[-1][0] != h:
runs.append([h, i, i])
else:
runs[-1][2] = i
print(" holds: " + ", ".join(f"{r[0]}x{r[2] - r[1] + 1}f (f{r[1]}-f{r[2]})" for r in runs))
print(" spans: " + json.dumps(spans))
check("world length preserved", len(holds) == 243, f"{len(holds)}")
check("one regenerated window", len(spans) == 1, json.dumps(spans))
check("span agrees with window_expand._bursts",
spans == [[a, b] for a, b in we._bursts(holds)],
json.dumps([list(x) for x in we._bursts(holds)]))
check("span covers every dilated frame and no real-time frame",
all(holds[f] > 1 for f in range(spans[0][0], spans[0][1] + 1))
and (spans[0][0] == 0 or holds[spans[0][0] - 1] == 1)
and (spans[0][1] == 242 or holds[spans[0][1] + 1] == 1))
check("the typed range 132-150 lies inside the span",
spans[0][0] <= 132 and spans[0][1] >= 150,
f"typed 132-150, snapped {spans[0][0]}-{spans[0][1]}")
TWO = json.dumps({"holds": [1] * 4 + [3] * 6 + [1] * 5 + [2] * 3 + [1] * 2, "world_len": 20})
check("two bursts -> two spans", vc._spans_from_hold_map(TWO) == [[4, 9], [15, 17]],
json.dumps(vc._spans_from_hold_map(TWO)))
check("a flat map has no span",
vc._spans_from_hold_map(json.dumps({"holds": [1] * 39, "world_len": 39})) == [])
# ---------------------------------------------------------------------------
# 2 + 3. the manifest, and the old call byte for byte
# ---------------------------------------------------------------------------
print("2. manifest extras / 3. byte compatibility")
class _Img:
def __init__(self, n):
self.shape = (n, 480, 864, 3)
class _Comp:
def __init__(self, n, fps, audio):
self.images = _Img(n)
self.frame_rate = fps
self.audio = audio
class FakeVideo:
"""Enough VIDEO to drive the node: dimensions, a file write, components."""
def __init__(self, n=243, fps=24.0, w=864, h=480, audio=None):
self.n, self.fps, self.w, self.h, self.audio = n, fps, w, h, audio
def get_dimensions(self):
return self.w, self.h
def save_to(self, path, **kw):
with open(path, "wb") as f:
f.write(b"\x00")
def get_components(self):
return _Comp(self.n, self.fps, self.audio)
TMP = tempfile.mkdtemp(prefix="mai_compare_test_")
_fp = types.ModuleType("folder_paths")
_fp.get_temp_directory = lambda: TMP
sys.modules["folder_paths"] = _fp
_api = types.ModuleType("comfy_api")
_latest = types.ModuleType("comfy_api.latest")
_latest.Types = types.SimpleNamespace(VideoContainer=lambda s: s, VideoCodec=lambda s: s)
_api.latest = _latest
sys.modules["comfy_api"] = _api
sys.modules["comfy_api.latest"] = _latest
vc.time.strftime = lambda f: "121212" # freeze the preview filename stamp
A, B = FakeVideo(audio=object()), FakeVideo(audio=None)
old = vc.MAIVideoCompare().compare(winner=2, preview_crf=23, unique_id="7",
video_1=A, video_2=B, label_1="ctrl", label_2="warm")
EXPECT = json.dumps({"items": [
{"index": 1, "label": "ctrl", "filename": "cmp_7_121212_1.mp4", "subfolder": "mai_compare",
"type": "temp", "frames": 243, "fps": 24.0, "width": 864, "height": 480, "audio": True},
{"index": 2, "label": "warm", "filename": "cmp_7_121212_2.mp4", "subfolder": "mai_compare",
"type": "temp", "frames": 243, "fps": 24.0, "width": 864, "height": 480, "audio": False}],
"winner": 2})
check("old two-video manifest is byte-identical", old["result"][2] == EXPECT,
"got " + old["result"][2][:70] + "...")
check("old manifest has no new keys", list(json.loads(old["result"][2])) == ["items", "winner"])
check("winner passthrough unchanged", old["result"][0] is B and old["result"][1] == 2)
empty = vc.MAIVideoCompare().compare(winner=2, preview_crf=23, unique_id="7", hold_map="",
curves="", video_1=A, video_2=B, label_1="ctrl", label_2="warm")
check("wiring the new inputs empty changes nothing", empty["result"][2] == EXPECT)
CURVES = json.dumps({"A": [0.0, 0.5, 1.0, 0.25], "B": [1.0, 0.5, 0.0, 0.75]})
wired = vc.MAIVideoCompare().compare(winner=1, preview_crf=23, unique_id="7", hold_map=PIER,
curves=CURVES, video_1=A, video_2=B)
m = json.loads(wired["result"][2])
check("manifest carries spans", m.get("spans") == spans, json.dumps(m.get("spans")))
check("manifest carries curves", m.get("curves") == {"A": [0.0, 0.5, 1.0, 0.25], "B": [1.0, 0.5, 0.0, 0.75]})
check("ui payload is the same dict", wired["ui"]["mai_compare"][0]["spans"] == spans)
junk = vc.MAIVideoCompare().compare(winner=1, preview_crf=23, unique_id="7",
hold_map="{not json", curves="[1,2]", video_1=A, video_2=B)
check("garbage in the new inputs never fails the render",
list(json.loads(junk["result"][2])) == ["items", "winner"])
sh = vc.MAISeedHunter().compare(winner=2, preview_crf=23, unique_id="9", video_1=A, video_2=B,
seed_1=11, seed_2=22, hold_map=PIER, curves=CURVES)
msh = json.loads(sh["result"][3])
check("seed hunter keeps the extras and its seeds",
msh.get("spans") == spans and [it["seed"] for it in msh["items"]] == [11, 22]
and sh["result"][1] == 22)
# ---------------------------------------------------------------------------
# 4. widget order audit: append-only, both classes
# ---------------------------------------------------------------------------
print("4. widget order")
OLD_REQ = ["winner", "preview_crf"]
OLD_OPT = [f"video_{i}" for i in range(1, 7)] + [f"label_{i}" for i in range(1, 7)]
OLD_OPT_SH = OLD_OPT + [f"seed_{i}" for i in range(1, 7)]
for cls, old_req, old_opt in ((vc.MAIVideoCompare, OLD_REQ, OLD_OPT),
(vc.MAISeedHunter, OLD_REQ, OLD_OPT_SH)):
t = cls.INPUT_TYPES()
req, opt = list(t["required"]), list(t["optional"])
print(f" {cls.__name__}")
print(f" required before {old_req}")
print(f" required after {req}")
print(f" optional before {old_opt}")
print(f" optional after {opt}")
check(f"{cls.__name__}: required untouched", req == old_req)
check(f"{cls.__name__}: optional is the old order plus a tail",
opt[:len(old_opt)] == old_opt, f"prefix {opt[:len(old_opt)] == old_opt}")
check(f"{cls.__name__}: the tail is exactly the two new inputs",
opt[len(old_opt):] == ["hold_map", "curves"], json.dumps(opt[len(old_opt):]))
for k in ("hold_map", "curves"):
spec = t["optional"][k]
check(f"{cls.__name__}: {k} is an optional forced input with an empty default",
spec[0] == "STRING" and spec[1].get("forceInput") is True and spec[1].get("default") == "")
print()
print("FAILURES: " + (", ".join(FAILS) if FAILS else "none"))
sys.exit(1 if FAILS else 0)
|