File size: 9,342 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 | #!/usr/bin/env python3
"""Unit test for H3 Drawn Plan: a drawn plan document -> hold-map ranges.
python tests/test_drawn_plan.py
Synthetic, no GPU, no models, no renders. torch is needed because the
node module draws a preview and the grid law lives in motion.py.
What it checks:
1. THE ROUND TRIP: a plan document loaded by the node emits ranges that
drive H3 Manual Hold Map to the SAME hold map as the ranges a human
would have typed for that shot. Both the file input and the pasted
string reach the same answer.
2. SECONDS SURVIVE: every emitted range parses back to the exact world
frames it came from, at the clip's own fps.
3. THE NODE IS A LOADER: its window, hold map and length are the
compiler's, byte-for-byte, not a second opinion computed here.
4. BAD INPUT IS A SENTENCE: malformed JSON, a missing file, a non-object
document and an uncompilable plan each raise one readable ValueError
naming the problem, with no json/traceback vocabulary in the message.
5. It is registered, so ComfyUI actually shows it.
Exit code 0 = pass.
"""
import inspect
import json
import os
import sys
import tempfile
import types
import torch # noqa: F401
HERE = os.path.dirname(os.path.abspath(__file__))
PACK = os.path.dirname(HERE)
sys.path.insert(0, PACK)
# motion.py's node classes import comfy's nested tensor; stub it exactly as
# tests/test_expand_to_end.py does so the suite runs outside ComfyUI.
if "comfy.nested_tensor" not in sys.modules:
class _StubNested:
def __init__(self, tensors):
self.tensors = list(tensors)
self.is_nested = True
def unbind(self):
return self.tensors
_pkg = sys.modules.setdefault("comfy", types.ModuleType("comfy"))
_mod = types.ModuleType("comfy.nested_tensor")
_mod.NestedTensor = _StubNested
_pkg.nested_tensor = _mod
sys.modules["comfy.nested_tensor"] = _mod
from timeline import nodes, schema # noqa: E402
from timeline.h3 import compile as h3compile # noqa: E402
from timeline.h3.gridlaw import motion as M # noqa: E402
FAILS = []
def check(name, ok, detail=""):
print((" PASS " if ok else " FAIL ") + name + (" " + detail if detail else ""))
if not ok:
FAILS.append(name)
def drawn_plan():
"""What the editor exports for the v3.1 fight-scene shot: real time
through frame 72, then 2x to the end of a 124-frame clip."""
plan = schema.new_plan("/tmp/seedhunt_20260963_00001_.mp4", frames=124,
fps=24, width=1152, height=640,
proposed_by="human:editor",
settings={"regen_strength": 0.45, "steps": 25,
"seed": 20260817,
"output_prefix": "video/drawn"})
plan["lanes"].append(schema.generation_density_lane(
[[0, 1.0], [72, 1.0], [73, 2.0], [123, 2.0]], ceiling=4.0,
proposer="human:editor"))
return plan
def manual_map(ranges, length=124, fps=24, hold=2):
"""The ranges as H3 Manual Hold Map reads them. ramp/bridge off: the
compiler already shaped the map, and the node's report says so."""
hold_map, segments, _report = M.H3ManualHoldMap().build(
length=length, fps=fps, ranges=ranges, hold=hold, ramp=False,
bridge=0)
return json.loads(hold_map), segments
def err_of(fn):
try:
fn()
except Exception as e: # noqa: BLE001
return type(e).__name__, str(e)
return "", ""
def main():
node = nodes.H3DrawnPlan()
plan = drawn_plan()
print("== 1. the round trip: exported plan -> node -> H3 Manual Hold Map")
with tempfile.TemporaryDirectory() as d:
p = os.path.join(d, "drawn.plan.json")
schema.save(json.loads(json.dumps(plan)), p)
# outputs grew in the parity packet; the first six are stable and
# this suite is about them
(ranges, length, fps, w0, wlen, report) = node.load(p)[:6]
pasted = node.load("", json.dumps(plan))
print(" ranges: %s" % ranges)
print(" window: %d..%d (%df), length %d, fps %d" % (w0, w0 + wlen - 1,
wlen, length, fps))
check("the file and the pasted document give the same ranges",
pasted[0] == ranges, ranges)
typed = "73-123:2" # what a human types for this shot
from_node, seg_node = manual_map(ranges)
from_typed, seg_typed = manual_map(typed)
check("node ranges and typed ranges compile to the same hold map",
from_node == from_typed,
"%s vs %s" % (h3compile.G.hold_runs_str(from_node["holds"]),
h3compile.G.hold_runs_str(from_typed["holds"])))
check("...and to the same segments string", seg_node == seg_typed,
seg_node)
print(" hold map: %s (world_len %d), segments %s"
% (h3compile.G.hold_runs_str(from_node["holds"]),
from_node["world_len"], seg_node))
print("\n== 2. the seconds are the world clock's")
parsed = []
for part in ranges.split(","):
span, _, h = part.strip().rpartition(":")
a_s, _, b_s = span.partition("-")
parsed.append((int(round(float(a_s.strip().rstrip("s")) * fps)),
int(round(float(b_s.strip().rstrip("s")) * fps)),
int(h)))
check("every emitted range parses back to whole world frames",
parsed == [(73, 123, 2)], str(parsed))
print("\n== 3. the node is a loader, not a second compiler")
B = h3compile.H3Backend()
art = B.compile(json.loads(json.dumps(plan))).compiled["artifact"]
check("window comes from the compiler",
(w0, wlen) == (art["window"]["start"], art["window"]["len"]),
"%s vs %s" % ((w0, wlen), (art["window"]["start"],
art["window"]["len"])))
check("the ranges are the compiled hold map, formatted",
ranges == h3compile.ranges_from_holds(art["hold_map"]["holds"],
art["window"]["start"], fps))
check("length is the clip's world length, untouched", length == 124,
str(length))
check("the report tells the user to turn ramp off",
"ramp OFF" in report and "bridge 0" in report)
body = inspect.getsource(nodes.H3DrawnPlan.load)
check("the node's code carries no grid law: no 17s, no token arithmetic",
"17" not in body and "token" not in body and "legal" not in body,
"%d lines" % len(body.splitlines()))
print("\n== 4. bad input is one readable sentence")
kind, msg = err_of(lambda: node.load("", "{\"clip\": {\"frames\": 124,}"))
print(" malformed json -> %s: %s" % (kind, msg))
check("malformed JSON: a ValueError that names the line and column",
kind == "ValueError" and "not valid JSON" in msg
and "line 1" in msg and "\n" not in msg, msg)
kind, msg = err_of(lambda: node.load("/nonexistent/nope.plan.json"))
print(" missing file -> %s: %s" % (kind, msg))
check("a missing file says so", kind == "ValueError" and "no plan file" in msg,
msg)
kind, msg = err_of(lambda: node.load("", "[1, 2, 3]"))
print(" json array -> %s: %s" % (kind, msg))
check("valid JSON that is not a plan says what it is",
kind == "ValueError" and "not a plan document" in msg, msg)
kind, msg = err_of(lambda: node.load("", ""))
check("nothing at all asks for something", kind == "ValueError"
and "plan_path" in msg, msg)
flat = drawn_plan()
flat["lanes"] = [] # a plan with nothing to compile
kind, msg = err_of(lambda: node.load("", json.dumps(flat)))
print(" no lane -> %s: %s" % (kind, msg))
check("an uncompilable plan reports the backend's own refusal",
kind == "ValueError" and "cannot be compiled" in msg
and "generation_density" in msg, msg)
print("\n== 5. lanes this backend cannot compile yet")
with_pin = drawn_plan()
with_pin["lanes"].append(schema.pin_lane(68, authority=0.6))
kind, msg = err_of(lambda: node.load("", json.dumps(with_pin)))
print(" pin, default -> %s: %s" % (kind, msg))
check("a pin refuses by default: nothing is dropped behind your back",
kind == "ValueError" and "pin" in msg, msg)
r2 = node.load("", json.dumps(with_pin), ignore_uncompiled_lanes=True)
check("...and compiles to the same ranges when you opt in",
r2[0] == ranges, "%r vs %r" % (r2[0], ranges))
check("...saying out loud which lane it skipped",
"SKIPPED" in r2[5] and "pin" in r2[5],
"; ".join(l for l in r2[5].splitlines() if l.startswith("SKIPPED")))
print("\n== 6. registration")
check("H3DrawnPlan is registered with a display name",
nodes.NODE_CLASS_MAPPINGS.get("H3DrawnPlan") is nodes.H3DrawnPlan
and "H3DrawnPlan" in nodes.NODE_DISPLAY_NAME_MAPPINGS)
print()
if FAILS:
print(f"FAILED: {len(FAILS)} check(s): " + ", ".join(FAILS))
return 1
print("all checks passed")
return 0
if __name__ == "__main__":
sys.exit(main())
|