c9cfbe3d08
Boxes at both sides of every size bound and every column and row bound, past the edges and at the limits of a signed integer; each interpolation mode rising and falling, with floors below zero. The size control now keeps its position, so it fails on the size bound alone. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1344 lines
51 KiB
Python
1344 lines
51 KiB
Python
#!/usr/bin/env python3
|
||
"""Second implementation of extensions/direction.md (`wai.media.direction`), over the corpus
|
||
`live-conformance/direction/`.
|
||
|
||
Written from the extension's text. It shares no code with wai-rs: it reads every record as §2
|
||
has it read, with the strict JSON reader of synthesis's second implementation
|
||
(synthesis_verify.py), verifies every signature with WAI's strict Ed25519 rule
|
||
(claim-conformance/_ed25519.py) and every link hash and frames digest with BLAKE3 written from
|
||
its specification (claim-conformance/_blake3.py), runs `wai.generate.spectral` in its own
|
||
arithmetic after checking it against that capability's conformance vector, renders every clip
|
||
frame and output frame (§9), interpolates every track (§5.2), measures every box (§7), computes
|
||
what every element governs (§10) and writes every statement and C2PA statement (§11.1, §12.4)
|
||
itself. It reads §13's codes from the extension.
|
||
|
||
python3 -I direction_verify.py <live-conformance/direction>
|
||
|
||
It gives every vector of the corpus the outcome, the reason, the place and the statements the
|
||
corpus states, and checks that every reason of §13 refuses a vector. It then mutates copies of
|
||
the corpus and requires every copy to fail, and drops each of its own checks in turn and
|
||
requires the corpus to catch the weakened verifier. Exit status 0 only if all of that holds.
|
||
Python 3.8 or later; standard library only.
|
||
"""
|
||
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, HERE)
|
||
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
||
import synthesis_verify as SV # noqa: E402
|
||
import _ed25519 as ED # noqa: E402
|
||
from _blake3 import blake3 # noqa: E402
|
||
|
||
EXTENSION = os.path.join(HERE, "..", "extensions", "direction.md")
|
||
GENERATOR = os.path.join(HERE, "..", "wai-rs", "tests", "vectors", "generate")
|
||
|
||
SIGN_DOMAIN = b"wai:direction\x01" # §2
|
||
LINK_DOMAIN = b"wai:direction-link\x01" # §2
|
||
FRAMES_DOMAIN = b"wai:direction-frames\x01" # §8.4
|
||
SPECTRAL = "wai.generate.spectral"
|
||
MAX_INT = (1 << 53) - 1
|
||
COORD = 1 << 20
|
||
|
||
LEVELS = ["pixel", "frame", "shot", "sequence"] # §3
|
||
SIZES = [("extreme_wide", 0), ("wide", 250), ("full", 700), ("medium_wide", 1100), ("medium", 1500),
|
||
("medium_close", 2200), ("close", 3200), ("extreme_close", 5000)] # §7
|
||
ROWS = ["top", "middle", "bottom"]
|
||
COLUMNS = ["left", "centre", "right"]
|
||
CELLS = [r + "-" + c for r in ROWS for c in COLUMNS]
|
||
DESCRIPTION = { # §11.1
|
||
"shot": "directed shot: framing, duration and subjects",
|
||
"frame": "directed keyframes: framing over time",
|
||
"pixel": "directed region: samples held to a reference",
|
||
"sequence": "directed sequence: order, cuts and transitions",
|
||
}
|
||
|
||
# The checks this verifier makes, each by name, so the self-test can drop one at a time.
|
||
CHECKS = [
|
||
"members", "levels", "tracks", "timeline_rule", "signature",
|
||
"known", "storyboard_first", "storyboard_link", "signer", "conflict_sequence", "conflict_keys", "consistent",
|
||
"check", "layout_count", "layout_camera", "layout_layer", "layout_output",
|
||
"layer_held", "layer_size", "layer_regenerate", "frames",
|
||
"shot_duration", "shot_subjects", "shot_framing", "key_box", "key_camera", "key_classes", "region",
|
||
"seq_order", "seq_cuts", "seq_transitions", "seq_ranges", "governs", "policy",
|
||
]
|
||
DROPPED = set()
|
||
|
||
|
||
def on(name):
|
||
return name not in DROPPED
|
||
|
||
|
||
class Refuse(Exception):
|
||
def __init__(self, reason, at):
|
||
super().__init__(reason)
|
||
self.reason, self.at = reason, at
|
||
|
||
|
||
Bad = SV.Bad
|
||
need = SV.need
|
||
|
||
|
||
def sha256(b):
|
||
return hashlib.sha256(b).digest()
|
||
|
||
|
||
# ---------------------------------------------------------------- §2 values
|
||
|
||
def integer(v):
|
||
need(isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= MAX_INT)
|
||
return v
|
||
|
||
|
||
def int_in(v, lo, hi):
|
||
integer(v)
|
||
need(lo <= v <= hi)
|
||
return v
|
||
|
||
|
||
def array(v, lo, hi):
|
||
need(isinstance(v, list) and lo <= len(v) <= hi)
|
||
return v
|
||
|
||
|
||
def members(v, names):
|
||
if on("members"):
|
||
need(isinstance(v, dict) and set(v) == set(names))
|
||
else:
|
||
need(isinstance(v, dict) and set(names) <= set(v))
|
||
return v
|
||
|
||
|
||
def hex32(v):
|
||
return SV.hexb(v, 32)
|
||
|
||
|
||
def nullable(v, f):
|
||
return None if v is None else f(v)
|
||
|
||
|
||
def signed(v):
|
||
"""A signed integer (§2)."""
|
||
need(isinstance(v, str) and re.fullmatch(r"0|-?[1-9][0-9]{0,6}", v) is not None)
|
||
n = int(v)
|
||
need(-COORD <= n <= COORD)
|
||
return n
|
||
|
||
|
||
def size_ok(w, h):
|
||
need(1 <= w <= 4096 and 1 <= h <= 4096 and w * h <= 1 << 23)
|
||
|
||
|
||
def pic(v):
|
||
members(v, ["samples", "width", "height"])
|
||
w, h = integer(v["width"]), integer(v["height"])
|
||
size_ok(w, h)
|
||
return (hex32(v["samples"]), w, h)
|
||
|
||
|
||
def box(v):
|
||
members(v, ["x", "y", "w", "h"])
|
||
return (signed(v["x"]), signed(v["y"]), int_in(v["w"], 1, COORD), int_in(v["h"], 1, COORD))
|
||
|
||
|
||
def rect(v):
|
||
members(v, ["x", "y", "w", "h"])
|
||
x, y, w, h = integer(v["x"]), integer(v["y"]), integer(v["w"]), integer(v["h"])
|
||
need(w >= 1 and h >= 1)
|
||
return (x, y, w, h)
|
||
|
||
|
||
def within(r, w, h):
|
||
return r[0] + r[2] <= w and r[1] + r[3] <= h
|
||
|
||
|
||
def classes(v):
|
||
size = nullable(v["size"], lambda s: SV.one_of(s, [n for n, _ in SIZES]))
|
||
position = nullable(v["position"], lambda s: SV.one_of(s, CELLS))
|
||
angle = nullable(v["angle"], SV.term)
|
||
return (size, position, angle)
|
||
|
||
|
||
def link_hash(v):
|
||
return blake3(LINK_DOMAIN + SV.jcs(v).encode("utf-8"))
|
||
|
||
|
||
def signing_bytes(v):
|
||
return SIGN_DOMAIN + SV.jcs({k: x for k, x in v.items() if k != "sig"}).encode("utf-8")
|
||
|
||
|
||
SIG_CACHE = {}
|
||
|
||
|
||
def signature(signer, v, sig):
|
||
if not on("signature"):
|
||
return None
|
||
key = (signer, signing_bytes(v), sig)
|
||
if key not in SIG_CACHE:
|
||
SIG_CACHE[key] = ED.verify(signer, key[1], sig)
|
||
return SIG_CACHE[key]
|
||
|
||
|
||
def read_signed(text, kind, names, fields):
|
||
"""§2: read one record, then its signature. `fields` reads the kind's own members."""
|
||
try:
|
||
v = SV.strict_json(text)
|
||
members(v, names)
|
||
need(v["kind"] == kind)
|
||
signer = hex32(v["signer_pubkey"])
|
||
sig = SV.hexb(v["sig"], 64)
|
||
SV.date_time(v["issued_at"])
|
||
r = fields(v)
|
||
except (Bad, KeyError, TypeError, AttributeError, IndexError, ValueError):
|
||
raise Refuse("record_malformed", None)
|
||
bad = signature(signer, v, sig)
|
||
if bad:
|
||
raise Refuse(bad, None)
|
||
r.update(value=v, signer=signer, link=link_hash(v), kind=kind)
|
||
return r
|
||
|
||
|
||
def check_levels(v, want):
|
||
if on("levels"):
|
||
need(v["levels"] == want)
|
||
|
||
|
||
# ---------------------------------------------------------------- §4–§6 direction records
|
||
|
||
def storyboard_fields(v):
|
||
w, h = integer(v["width"]), integer(v["height"])
|
||
size_ok(w, h)
|
||
members(v["rate"], ["num", "den"])
|
||
rate = (int_in(v["rate"]["num"], 1, COORD), int_in(v["rate"]["den"], 1, COORD))
|
||
subjects = []
|
||
for s in array(v["subjects"], 0, 16):
|
||
members(s, ["views"])
|
||
views = []
|
||
for x in array(s["views"], 1, 8):
|
||
members(x, ["angle", "samples", "width", "height"])
|
||
angle = SV.term(x["angle"])
|
||
need(all(a != angle for a, _ in views))
|
||
views.append((angle, pic({"samples": x["samples"], "width": x["width"], "height": x["height"]})))
|
||
subjects.append(views)
|
||
shots = []
|
||
for s in array(v["shots"], 1, 256):
|
||
members(s, ["duration", "subjects", "framing", "reference"])
|
||
dur = int_in(s["duration"], 1, 65536)
|
||
subs = []
|
||
for x in array(s["subjects"], 0, 16):
|
||
i = integer(x)
|
||
need(i < len(subjects) and i not in subs)
|
||
subs.append(i)
|
||
framing = []
|
||
for k, f in enumerate(array(s["framing"], len(subs), len(subs))):
|
||
members(f, ["size", "position", "angle"])
|
||
c = classes(f)
|
||
need(c[2] is None or any(a == c[2] for a, _ in subjects[subs[k]]))
|
||
framing.append(c)
|
||
ref = nullable(s["reference"], pic)
|
||
shots.append({"duration": dur, "subjects": subs, "framing": framing, "reference": ref})
|
||
order = None
|
||
if v["order"] is not None:
|
||
order = []
|
||
for i in array(v["order"], 1, 1024):
|
||
need(integer(i) < len(shots))
|
||
order.append(i)
|
||
check_levels(v, ["shot", "sequence"] if order is not None else ["shot"])
|
||
return {"width": w, "height": h, "rate": rate, "subjects": subjects, "shots": shots, "order": order}
|
||
|
||
|
||
def key_fields(k):
|
||
members(k, ["t", "interp", "camera", "subjects"])
|
||
subs = []
|
||
for s in array(k["subjects"], 0, 16):
|
||
members(s, ["box", "size", "position", "angle"])
|
||
subs.append({"box": nullable(s["box"], box), "classes": classes(s)})
|
||
return {"t": integer(k["t"]), "interp": SV.one_of(k["interp"], ["hold", "linear", "ease"]),
|
||
"camera": nullable(k["camera"], rect), "subjects": subs}
|
||
|
||
|
||
def entries_fields(v):
|
||
out = []
|
||
for e in array(v, 1, 256):
|
||
members(e, ["shot", "keys"])
|
||
shot = integer(e["shot"])
|
||
need(all(x["shot"] != shot for x in out))
|
||
keys = [key_fields(k) for k in array(e["keys"], 1, 1024)]
|
||
need(all(keys[i]["t"] < keys[i + 1]["t"] for i in range(len(keys) - 1)))
|
||
n = len(keys[0]["subjects"])
|
||
need(all(len(k["subjects"]) == n for k in keys))
|
||
if on("tracks"):
|
||
for s in range(n):
|
||
need(len({k["subjects"][s]["box"] is None for k in keys}) == 1)
|
||
need(len({k["camera"] is None for k in keys}) == 1)
|
||
out.append({"shot": shot, "keys": keys})
|
||
return out
|
||
|
||
|
||
def regions_fields(v):
|
||
out = []
|
||
for r in array(v, 1, 256):
|
||
members(r, ["shot", "from", "to", "rects", "source"])
|
||
f, t = integer(r["from"]), integer(r["to"])
|
||
need(f < t)
|
||
out.append({"shot": integer(r["shot"]), "from": f, "to": t,
|
||
"rects": [rect(x) for x in array(r["rects"], 1, 64)], "source": pic(r["source"])})
|
||
return out
|
||
|
||
|
||
def keyframes_fields(v):
|
||
sb = hex32(v["storyboard"])
|
||
entries = nullable(v["keyframes"], entries_fields)
|
||
regions = nullable(v["regions"], regions_fields)
|
||
need(entries is not None or regions is not None)
|
||
check_levels(v, [x for x, on_ in (("pixel", regions is not None), ("frame", entries is not None)) if on_])
|
||
return {"storyboard": sb, "entries": entries, "regions": regions}
|
||
|
||
|
||
def transition(v):
|
||
need(isinstance(v, dict))
|
||
if v.get("kind") == "cut":
|
||
members(v, ["kind"])
|
||
return 0
|
||
need(v.get("kind") == "dissolve")
|
||
members(v, ["kind", "frames"])
|
||
return int_in(v["frames"], 1, MAX_INT)
|
||
|
||
|
||
def events(v):
|
||
out = []
|
||
for e in array(v, 1, 1024):
|
||
members(e, ["shot", "in", "out", "transition"])
|
||
i, o = integer(e["in"]), integer(e["out"])
|
||
need(i < o <= 65536)
|
||
out.append({"shot": integer(e["shot"]), "in": i, "out": o, "d": transition(e["transition"])})
|
||
if on("timeline_rule"):
|
||
need(out[0]["d"] == 0)
|
||
for i, e in enumerate(out):
|
||
nxt = out[i + 1]["d"] if i + 1 < len(out) else 0
|
||
need(e["d"] + nxt <= e["out"] - e["in"])
|
||
return out
|
||
|
||
|
||
def sequence_fields(v):
|
||
sb = hex32(v["storyboard"])
|
||
ev = events(v["events"])
|
||
check_levels(v, ["sequence"])
|
||
return {"storyboard": sb, "events": ev}
|
||
|
||
|
||
READERS = {
|
||
"wai-storyboard": (["kind", "levels", "width", "height", "rate", "subjects", "shots", "order", "issued_at",
|
||
"signer_pubkey", "sig"], storyboard_fields),
|
||
"wai-keyframes": (["kind", "storyboard", "levels", "keyframes", "regions", "issued_at", "signer_pubkey", "sig"],
|
||
keyframes_fields),
|
||
"wai-sequence": (["kind", "storyboard", "levels", "events", "issued_at", "signer_pubkey", "sig"], sequence_fields),
|
||
}
|
||
|
||
|
||
def read_record(text):
|
||
try:
|
||
kind = SV.strict_json(text).get("kind")
|
||
except (Bad, AttributeError):
|
||
raise Refuse("record_malformed", None)
|
||
if kind not in READERS:
|
||
raise Refuse("record_malformed", None)
|
||
names, fields = READERS[kind]
|
||
return read_signed(text, kind, names, fields)
|
||
|
||
|
||
def levels_of(r):
|
||
return r["value"]["levels"]
|
||
|
||
|
||
def sets_sequence(r):
|
||
return "sequence" in levels_of(r)
|
||
|
||
|
||
# ---------------------------------------------------------------- §8 the directed output
|
||
|
||
def source(v):
|
||
if v is None:
|
||
return None
|
||
need(isinstance(v, dict))
|
||
g = SV.term(v.get("generator"))
|
||
if g == SPECTRAL:
|
||
members(v, ["generator", "model", "prompt", "seed", "steps"])
|
||
p = v["prompt"]
|
||
need(isinstance(p, str) and len(p) <= 256 and all(0x20 <= ord(c) <= 0x7E and c not in '"\\' for c in p))
|
||
return {"generator": g, "model": hex32(v["model"]), "prompt": p, "seed": int_in(v["seed"], 0, 2**32 - 1),
|
||
"steps": int_in(v["steps"], 1, 64)}
|
||
members(v, ["generator", "model"])
|
||
return {"generator": g, "model": hex32(v["model"])}
|
||
|
||
|
||
def clip_fields(c):
|
||
members(c, ["frames", "plate", "camera", "subjects", "holds"])
|
||
n = int_in(c["frames"], 1, 65536)
|
||
cam = nullable(c["camera"], lambda a: [rect(x) for x in array(a, n, n)])
|
||
subs = []
|
||
for s in array(c["subjects"], 0, 16):
|
||
members(s, ["subject", "views", "boxes"])
|
||
subs.append({"subject": integer(s["subject"]), "views": [integer(x) for x in array(s["views"], n, n)],
|
||
"boxes": [box(x) for x in array(s["boxes"], n, n)]})
|
||
holds = []
|
||
for h in array(c["holds"], 0, 64):
|
||
members(h, ["from", "to", "rects", "source"])
|
||
f, t = integer(h["from"]), integer(h["to"])
|
||
need(f < t <= n)
|
||
holds.append({"from": f, "to": t, "rects": [rect(x) for x in array(h["rects"], 1, 64)],
|
||
"source": hex32(h["source"])})
|
||
return {"frames": n, "plate": nullable(c["plate"], hex32), "camera": cam, "subjects": subs, "holds": holds}
|
||
|
||
|
||
def path(v):
|
||
need(isinstance(v, list) and len(v) >= 1)
|
||
if v[0] == "sequence":
|
||
need(len(v) == 1)
|
||
return ("sequence",)
|
||
need(len(v) == 2 and v[0] in ("shot", "keyframes", "region"))
|
||
return (v[0], integer(v[1]))
|
||
|
||
|
||
def frame_set(v):
|
||
out = []
|
||
for r in array(v, 0, MAX_INT):
|
||
need(isinstance(r, list) and len(r) == 2)
|
||
a, b = integer(r[0]), integer(r[1])
|
||
need(a < b and (not out or out[-1][1] < a))
|
||
out.append((a, b))
|
||
return out
|
||
|
||
|
||
def directed_fields(v):
|
||
direction = [hex32(x) for x in array(v["direction"], 1, 64)]
|
||
need(len(set(direction)) == len(direction))
|
||
signer = hex32(v["signer_pubkey"])
|
||
directors = [hex32(x) for x in array(v["directors"], 1, 64)]
|
||
need(len(set(directors)) == len(directors) and signer not in directors)
|
||
check = SV.one_of(v["check"], ["replicate", "attest"])
|
||
att = v["attestation"]
|
||
if check == "replicate":
|
||
need(att is None)
|
||
else:
|
||
members(att, ["generator", "class"])
|
||
att = (SV.term(att["generator"]), SV.term(att["class"]))
|
||
layers = []
|
||
for x in array(v["layers"], 0, 256):
|
||
members(x, ["samples", "width", "height", "source"])
|
||
p = pic({"samples": x["samples"], "width": x["width"], "height": x["height"]})
|
||
need(all(l["pic"][0] != p[0] for l in layers))
|
||
layers.append({"pic": p, "source": source(x["source"])})
|
||
clips = [clip_fields(c) for c in array(v["clips"], 1, 256)]
|
||
timeline = events(v["timeline"])
|
||
o = members(v["output"], ["width", "height", "frames", "digest", "content_hash"])
|
||
w, h = integer(o["width"]), integer(o["height"])
|
||
size_ok(w, h)
|
||
governs = []
|
||
for g in array(v["governs"], 0, 1024):
|
||
members(g, ["record", "element", "frames"])
|
||
governs.append((integer(g["record"]), path(g["element"]), frame_set(g["frames"])))
|
||
return {"direction": direction, "directors": directors, "replicate": check == "replicate", "attestation": att,
|
||
"layers": layers, "clips": clips, "timeline": timeline, "width": w, "height": h,
|
||
"frames": int_in(o["frames"], 1, 1 << 20), "digest": hex32(o["digest"]),
|
||
"content_hash": nullable(o["content_hash"], hex32), "governs": governs}
|
||
|
||
|
||
DIRECTED = ["kind", "direction", "directors", "check", "attestation", "layers", "clips", "timeline", "output",
|
||
"governs", "issued_at", "signer_pubkey", "sig"]
|
||
|
||
|
||
# ---------------------------------------------------------------- §6.2 the timeline
|
||
|
||
def starts(ev):
|
||
s = []
|
||
for i, e in enumerate(ev):
|
||
s.append(0 if i == 0 else s[i - 1] + (ev[i - 1]["out"] - ev[i - 1]["in"]) - e["d"])
|
||
return s, s[-1] + ev[-1]["out"] - ev[-1]["in"]
|
||
|
||
|
||
def shows(ev, s, f):
|
||
return [(i, e["shot"], e["in"] + f - s[i]) for i, e in enumerate(ev) if s[i] <= f < s[i] + e["out"] - e["in"]]
|
||
|
||
|
||
# ---------------------------------------------------------------- §5.2, §7
|
||
|
||
def interpolate(v0, v1, n, d, mode):
|
||
if mode == "hold":
|
||
return v0
|
||
if mode == "linear":
|
||
return v0 + ((v1 - v0) * n) // d
|
||
return v0 + ((v1 - v0) * n * n * (3 * d - 2 * n)) // (d * d * d)
|
||
|
||
|
||
def track_value(keys, t):
|
||
"""keys: [(t, (x, y, w, h), interp)], ascending."""
|
||
k = None
|
||
for i, (tk, _, _) in enumerate(keys):
|
||
if tk <= t:
|
||
k = i
|
||
if k is None:
|
||
return None
|
||
tk, vk, mode = keys[k]
|
||
if k + 1 == len(keys):
|
||
return tuple(vk)
|
||
tn, vn, _ = keys[k + 1]
|
||
return tuple(interpolate(vk[m], vn[m], t - tk, tn - tk, mode) for m in range(4))
|
||
|
||
|
||
def size_class(h, fh):
|
||
q = (1000 * h) // fh
|
||
return [n for n, lo in SIZES if q >= lo][-1]
|
||
|
||
|
||
def position_class(b, fw, fh):
|
||
x, y, w, h = b
|
||
col = min(2, max(0, (3 * (2 * x + w)) // (2 * fw)))
|
||
row = min(2, max(0, (3 * (2 * y + h)) // (2 * fh)))
|
||
return ROWS[row] + "-" + COLUMNS[col]
|
||
|
||
|
||
def matches(c, b, angle, sb):
|
||
size, position, ang = c
|
||
return ((size is None or size == size_class(b[3], sb["height"]))
|
||
and (position is None or position == position_class(b, sb["width"], sb["height"]))
|
||
and (ang is None or ang == angle))
|
||
|
||
|
||
# ---------------------------------------------------------------- the generator, in its own arithmetic
|
||
|
||
def fnv(b):
|
||
h = 14695981039346656037
|
||
for x in b:
|
||
h = ((h ^ x) * 1099511628211) & 0xFFFFFFFFFFFFFFFF
|
||
return h
|
||
|
||
|
||
def splitmix(state):
|
||
state = (state + 0x9E3779B97F4A7C15) & 0xFFFFFFFFFFFFFFFF
|
||
z = state
|
||
z = ((z ^ (z >> 30)) * 0xBF58476D1CE4E5B9) & 0xFFFFFFFFFFFFFFFF
|
||
z = ((z ^ (z >> 27)) * 0x94D049BB133111EB) & 0xFFFFFFFFFFFFFFFF
|
||
return state, z ^ (z >> 31)
|
||
|
||
|
||
def tdiv(a, b):
|
||
"""Division truncating toward zero."""
|
||
q = abs(a) // abs(b)
|
||
return q if (a >= 0) == (b >= 0) else -q
|
||
|
||
|
||
def iatan2(dy, dx, t):
|
||
if dx == 0:
|
||
return 0 if dy == 0 else (t // 4 if dy > 0 else 3 * t // 4)
|
||
if dy == 0:
|
||
return 0 if dx > 0 else t // 2
|
||
steps = [128, 76, 40, 20, 10, 5, 3, 1, 1, 0, 0, 0, 0, 0]
|
||
x, y, ang = abs(dx) << 12, abs(dy) << 12, 0
|
||
for i, a in enumerate(steps):
|
||
if y > 0:
|
||
x, y, ang = x + (y >> i), y - (x >> i), ang + a
|
||
elif y < 0:
|
||
x, y, ang = x - (y >> i), y + (x >> i), ang - a
|
||
if dx >= 0 and dy >= 0:
|
||
q = ang
|
||
elif dx < 0 and dy >= 0:
|
||
q = t // 2 - ang
|
||
elif dx < 0:
|
||
q = t // 2 + ang
|
||
else:
|
||
q = t - ang
|
||
return q % t
|
||
|
||
|
||
def isqrt(n):
|
||
import math
|
||
return math.isqrt(n)
|
||
|
||
|
||
class Model:
|
||
def __init__(self, file_bytes):
|
||
m = json.loads(file_bytes)
|
||
self.lut = m["sine_lut"]
|
||
self.t = m["lut_size"]
|
||
self.lut_bits = m.get("lut_bits", 14)
|
||
self.freq_max = m["freq_max"]
|
||
self.amp_bits = m["amp_bits"]
|
||
self.offsets = m["color_offsets"]
|
||
self.base = m["base_components"]
|
||
self.step = m["step_components"]
|
||
self.lexicon = {k: list(v) + [0] * (10 - len(v)) for k, v in m.get("lexicon", {}).items()}
|
||
|
||
def intent(self, prompt):
|
||
toks = re.findall(r"[a-z0-9]+", "".join(c.lower() if "A" <= c <= "Z" else c for c in prompt))
|
||
sums, n = [0] * 10, 0
|
||
for tok in toks:
|
||
if tok in self.lexicon:
|
||
sums = [a + b for a, b in zip(sums, self.lexicon[tok])]
|
||
n += 1
|
||
if n == 0:
|
||
return dict(hue=[256, 256, 256], freq=256, orient=0, radial=0, mirror=0, gradient=0, angular=0, spiral=0)
|
||
mx = max(max(sums[0:3]), 1)
|
||
clamp = lambda v, lo, hi: max(lo, min(hi, v)) # noqa: E731
|
||
return dict(hue=[tdiv(sums[c] * 256, mx) for c in range(3)], freq=tdiv(sums[3], n),
|
||
orient=clamp(tdiv(sums[4], n), -256, 256), radial=clamp(tdiv(sums[5], n), 0, 256),
|
||
mirror=clamp(tdiv(sums[6], n), 0, 256), gradient=clamp(tdiv(sums[7], n), -256, 256),
|
||
angular=clamp(tdiv(sums[8], n), 0, 256), spiral=clamp(tdiv(sums[9], n), 0, 256))
|
||
|
||
def generate(self, prompt, seed, steps, h, w):
|
||
k = self.base + steps * self.step
|
||
it = self.intent(prompt)
|
||
if it["orient"] < 0:
|
||
sx, sy = max(256 + it["orient"], 0), 256
|
||
else:
|
||
sx, sy = 256, max(256 - it["orient"], 0)
|
||
fmax = max(tdiv(self.freq_max * it["freq"], 256), 1)
|
||
state = fnv(prompt.encode()) ^ seed ^ 0xD1B54A32D192ED03
|
||
amax = 1 << self.amp_bits
|
||
span = 2 * fmax + 1
|
||
thr = [it["radial"], it["mirror"], it["angular"], it["spiral"]]
|
||
plane = max(256 - sum(thr), 48)
|
||
total = plane + sum(thr)
|
||
comps = []
|
||
for _ in range(k):
|
||
state, d = splitmix(state)
|
||
pick = d % total
|
||
typ = 0
|
||
if pick >= plane:
|
||
bound = plane
|
||
for i, wgt in enumerate(thr):
|
||
bound += wgt
|
||
if pick < bound:
|
||
typ = i + 1
|
||
break
|
||
state, d = splitmix(state)
|
||
fx = d % span - fmax
|
||
state, d = splitmix(state)
|
||
fy = d % span - fmax
|
||
state, d = splitmix(state)
|
||
amp = d % amax + 1
|
||
state, d = splitmix(state)
|
||
phase = d % self.t
|
||
if typ == 1:
|
||
comps.append((1, max(abs(fx), 2), 0, amp, phase))
|
||
elif typ == 3:
|
||
comps.append((3, abs(fx) % 11 + 2, 0, amp, phase))
|
||
elif typ == 4:
|
||
comps.append((4, abs(fx) % 9 + 2, max(abs(fy), 2), amp, phase))
|
||
else:
|
||
comps.append((typ, tdiv(fx * sx, 256), tdiv(fy * sy, 256), amp, phase))
|
||
t = self.t
|
||
cx, cy = w // 2, h // 2
|
||
need_r = any(c[0] in (1, 4) for c in comps)
|
||
need_th = any(c[0] in (3, 4) for c in comps)
|
||
gscale = tdiv((1 << self.lut_bits) * amax * k, 2048 * max(cy, 1))
|
||
lut = self.lut
|
||
acc = [[0] * (w * h) for _ in range(3)]
|
||
geo = []
|
||
for y in range(h):
|
||
for x in range(w):
|
||
dx, dy = x - cx, y - cy
|
||
geo.append((x, y, dx, dy, isqrt(dx * dx + dy * dy) if need_r else 0, iatan2(dy, dx, t) if need_th else 0))
|
||
for c, off in enumerate(self.offsets):
|
||
plane_acc = acc[c]
|
||
for p, (x, y, dx, dy, r, th) in enumerate(geo):
|
||
a = it["gradient"] * (cy - y) * gscale
|
||
for typ, fx, fy, amp, phase in comps:
|
||
if typ == 1:
|
||
arg = fx * r + phase + off
|
||
elif typ == 2:
|
||
arg = fx * abs(dx) + fy * abs(dy) + phase + off
|
||
elif typ == 3:
|
||
arg = fx * th + phase + off
|
||
elif typ == 4:
|
||
arg = fx * th + fy * r + phase + off
|
||
else:
|
||
arg = fx * x + fy * y + phase + off
|
||
a += amp * lut[arg % t]
|
||
plane_acc[p] = a
|
||
g0 = min(min(a) for a in acc)
|
||
g1 = max(max(a) for a in acc)
|
||
rng = max(g1 - g0, 1)
|
||
eh = [64 + tdiv(it["hue"][c] * 192, 256) for c in range(3)]
|
||
out = bytearray(w * h * 3)
|
||
for c in range(3):
|
||
for p, v in enumerate(acc[c]):
|
||
q = tdiv((v - g0) * 255 + tdiv(rng, 2), rng)
|
||
out[p * 3 + c] = max(0, min(255, tdiv(q * eh[c], 256)))
|
||
return bytes(out)
|
||
|
||
|
||
MODELS = {}
|
||
GENERATED = {}
|
||
|
||
|
||
def model_of(file_bytes):
|
||
d = sha256(file_bytes)
|
||
if d not in MODELS:
|
||
MODELS[d] = Model(file_bytes)
|
||
return MODELS[d]
|
||
|
||
|
||
def generated(file_bytes, prompt, seed, steps, w, h):
|
||
key = (sha256(file_bytes), prompt, seed, steps, w, h)
|
||
if key not in GENERATED:
|
||
GENERATED[key] = model_of(file_bytes).generate(prompt, seed, steps, h, w)
|
||
return GENERATED[key]
|
||
|
||
|
||
def check_generator():
|
||
"""The generator against its conformance vector before anything else."""
|
||
with open(os.path.join(GENERATOR, "generate_spectral_model.json"), "rb") as f:
|
||
mf = f.read()
|
||
with open(os.path.join(GENERATOR, "generate.json")) as f:
|
||
g = json.load(f)
|
||
with open(os.path.join(GENERATOR, "sha256.json")) as f:
|
||
want = json.load(f)["generate"]["sha256"]
|
||
px = generated(mf, g["prompt"], g["seed"], g["steps"], g["w"], g["h"])
|
||
return sha256(px).hex() == want
|
||
|
||
|
||
# ---------------------------------------------------------------- §9 the render
|
||
|
||
def render_clip_frame(sb, clip, t, pics):
|
||
fw, fh = sb["width"], sb["height"]
|
||
out = bytearray(fw * fh * 3)
|
||
pw, ph, plate = pics[clip["plate"]]
|
||
cam = clip["camera"][t] if clip["camera"] is not None else None
|
||
for y in range(fh):
|
||
for x in range(fw):
|
||
if cam is None:
|
||
sx, sy = x, y
|
||
else:
|
||
sx, sy = cam[0] + x * cam[2] // fw, cam[1] + y * cam[3] // fh
|
||
k, j = (y * fw + x) * 3, (sy * pw + sx) * 3
|
||
out[k:k + 3] = plate[j:j + 3]
|
||
for s in clip["subjects"]:
|
||
angle, (dg, vw, vh) = sb["subjects"][s["subject"]][s["views"][t]]
|
||
_, _, view = pics[dg]
|
||
bx, by, bw, bh = s["boxes"][t]
|
||
for y in range(max(by, 0), min(by + bh, fh)):
|
||
vy = (y - by) * vh // bh
|
||
for x in range(max(bx, 0), min(bx + bw, fw)):
|
||
vx = (x - bx) * vw // bw
|
||
k, j = (y * fw + x) * 3, (vy * vw + vx) * 3
|
||
out[k:k + 3] = view[j:j + 3]
|
||
for h in clip["holds"]:
|
||
if h["from"] <= t < h["to"]:
|
||
_, _, src = pics[h["source"]]
|
||
for (rx, ry, rw, rh) in h["rects"]:
|
||
for y in range(ry, ry + rh):
|
||
k = (y * fw + rx) * 3
|
||
out[k:k + rw * 3] = src[k:k + rw * 3]
|
||
return bytes(out)
|
||
|
||
|
||
def render_output(sb, d, pics):
|
||
clips = [[render_clip_frame(sb, c, t, pics) for t in range(c["frames"])] for c in d["clips"]]
|
||
ev = d["timeline"]
|
||
s, n = starts(ev)
|
||
out = []
|
||
for f in range(n):
|
||
sh = shows(ev, s, f)
|
||
if len(sh) == 1:
|
||
_, shot, t = sh[0]
|
||
out.append(clips[shot][t])
|
||
else:
|
||
(_, sa, ta), (i, sb_, tb) = sh
|
||
dd = ev[i]["d"]
|
||
j = f - s[i] + 1
|
||
a, b = clips[sa][ta], clips[sb_][tb]
|
||
out.append(bytes((x * (dd + 1 - j) + y * j + (dd + 1) // 2) // (dd + 1) for x, y in zip(a, b)))
|
||
return out
|
||
|
||
|
||
RENDERED = {}
|
||
|
||
|
||
def frame_digests(sb, d, pics, key):
|
||
if key not in RENDERED:
|
||
RENDERED[key] = [sha256(f) for f in render_output(sb, d, pics)]
|
||
return RENDERED[key]
|
||
|
||
|
||
# ---------------------------------------------------------------- §10 governance
|
||
|
||
def elements(r):
|
||
if r["kind"] == "wai-storyboard":
|
||
e = [("shot", i) for i in range(len(r["shots"]))]
|
||
return e + ([("sequence",)] if r["order"] is not None else [])
|
||
if r["kind"] == "wai-keyframes":
|
||
return ([("keyframes", i) for i in range(len(r["entries"] or []))]
|
||
+ [("region", j) for j in range(len(r["regions"] or []))])
|
||
return [("sequence",)]
|
||
|
||
|
||
LEVEL = {"shot": "shot", "keyframes": "frame", "region": "pixel", "sequence": "sequence"}
|
||
|
||
|
||
def tracks(entry):
|
||
keys = entry["keys"]
|
||
n = len(keys[0]["subjects"])
|
||
subj = []
|
||
for s in range(n):
|
||
if keys[0]["subjects"][s]["box"] is None:
|
||
subj.append(None)
|
||
else:
|
||
subj.append([(k["t"], k["subjects"][s]["box"], k["interp"]) for k in keys])
|
||
cam = None if keys[0]["camera"] is None else [(k["t"], k["camera"], k["interp"]) for k in keys]
|
||
return subj, cam
|
||
|
||
|
||
def keyed(entry, duration):
|
||
subj, cam = tracks(entry)
|
||
if any(x is not None for x in subj) or cam is not None:
|
||
return set(range(entry["keys"][0]["t"], duration))
|
||
return {k["t"] for k in entry["keys"] if any(any(c is not None for c in s["classes"]) for s in k["subjects"])}
|
||
|
||
|
||
def runs(frames):
|
||
out = []
|
||
for f in sorted(frames):
|
||
if out and out[-1][1] == f:
|
||
out[-1] = (out[-1][0], f + 1)
|
||
else:
|
||
out.append((f, f + 1))
|
||
return out
|
||
|
||
|
||
def governed(sb, d, r, el):
|
||
ev = d["timeline"]
|
||
s, n = starts(ev)
|
||
|
||
def showing(pred):
|
||
return runs(f for f in range(n) if any(pred(shot, t) for _, shot, t in shows(ev, s, f)))
|
||
if el[0] == "shot":
|
||
return showing(lambda shot, t: shot == el[1])
|
||
if el[0] == "sequence":
|
||
return [(0, n)]
|
||
if el[0] == "keyframes":
|
||
e = r["entries"][el[1]]
|
||
ks = keyed(e, sb["shots"][e["shot"]]["duration"])
|
||
return showing(lambda shot, t: shot == e["shot"] and t in ks)
|
||
g = r["regions"][el[1]]
|
||
return showing(lambda shot, t: shot == g["shot"] and g["from"] <= t < g["to"])
|
||
|
||
|
||
# ---------------------------------------------------------------- §12 verification
|
||
|
||
def consistent(sb, r):
|
||
shots = sb["shots"]
|
||
if r["kind"] == "wai-keyframes":
|
||
for e in r["entries"] or []:
|
||
if e["shot"] >= len(shots):
|
||
return False
|
||
shot = shots[e["shot"]]
|
||
for k in e["keys"]:
|
||
if k["t"] >= shot["duration"] or len(k["subjects"]) != len(shot["subjects"]):
|
||
return False
|
||
for i, ks in enumerate(k["subjects"]):
|
||
a = ks["classes"][2]
|
||
if a is not None and not any(x == a for x, _ in sb["subjects"][shot["subjects"][i]]):
|
||
return False
|
||
for g in r["regions"] or []:
|
||
if g["shot"] >= len(shots) or g["to"] > shots[g["shot"]]["duration"]:
|
||
return False
|
||
if not all(within(x, sb["width"], sb["height"]) for x in g["rects"]):
|
||
return False
|
||
if (g["source"][1], g["source"][2]) != (sb["width"], sb["height"]):
|
||
return False
|
||
if r["kind"] == "wai-sequence":
|
||
for e in r["events"]:
|
||
if e["shot"] >= len(shots) or e["out"] > shots[e["shot"]]["duration"]:
|
||
return False
|
||
return True
|
||
|
||
|
||
def layout(sb, d):
|
||
layer = {l["pic"][0]: l for l in d["layers"]}
|
||
if on("layout_count") and len(d["clips"]) != len(sb["shots"]):
|
||
raise Refuse("layout_invalid", ["output"])
|
||
for i, c in enumerate(d["clips"]):
|
||
at = ["clip", i]
|
||
for s in c["subjects"]:
|
||
if s["subject"] >= len(sb["subjects"]) or any(v >= len(sb["subjects"][s["subject"]]) for v in s["views"]):
|
||
raise Refuse("layout_invalid", at)
|
||
if d["replicate"]:
|
||
if c["plate"] is None:
|
||
raise Refuse("layout_invalid", at)
|
||
if c["plate"] not in layer:
|
||
if on("layout_layer"):
|
||
raise Refuse("layer_missing", at)
|
||
else:
|
||
_, pw, ph = layer[c["plate"]]["pic"]
|
||
if c["camera"] is None and (pw, ph) != (sb["width"], sb["height"]):
|
||
raise Refuse("layout_invalid", at)
|
||
if on("layout_camera") and c["camera"] is not None and not all(within(x, pw, ph) for x in c["camera"]):
|
||
raise Refuse("layout_invalid", at)
|
||
for s in c["subjects"]:
|
||
for v in sorted(set(s["views"])):
|
||
_, (dg, vw, vh) = sb["subjects"][s["subject"]][v]
|
||
if dg not in layer:
|
||
raise Refuse("layer_missing", at)
|
||
if layer[dg]["pic"][1:] != (vw, vh):
|
||
raise Refuse("layout_invalid", at)
|
||
elif c["plate"] is not None:
|
||
raise Refuse("layout_invalid", at)
|
||
for h in c["holds"]:
|
||
if not all(within(x, sb["width"], sb["height"]) for x in h["rects"]):
|
||
raise Refuse("layout_invalid", at)
|
||
if d["replicate"]:
|
||
if h["source"] not in layer:
|
||
raise Refuse("layer_missing", at)
|
||
if layer[h["source"]]["pic"][1:] != (sb["width"], sb["height"]):
|
||
raise Refuse("layout_invalid", at)
|
||
for i, e in enumerate(d["timeline"]):
|
||
if e["shot"] >= len(d["clips"]) or e["out"] > d["clips"][e["shot"]]["frames"]:
|
||
raise Refuse("layout_invalid", ["timeline", i])
|
||
_, n = starts(d["timeline"])
|
||
if on("layout_output") and ((d["width"], d["height"]) != (sb["width"], sb["height"]) or d["frames"] != n):
|
||
raise Refuse("layout_invalid", ["output"])
|
||
|
||
|
||
def directed_events(r):
|
||
if r["kind"] == "wai-storyboard":
|
||
return [{"shot": i, "in": 0, "out": r["shots"][i]["duration"], "d": 0} for i in r["order"]]
|
||
return r["events"]
|
||
|
||
|
||
def check_element(sb, d, r, el):
|
||
if el[0] == "shot":
|
||
i = el[1]
|
||
shot, clip = sb["shots"][i], d["clips"][i]
|
||
if on("shot_duration") and clip["frames"] != shot["duration"]:
|
||
raise Refuse("duration_mismatch", ["clip", i])
|
||
if [s["subject"] for s in clip["subjects"]] != shot["subjects"]:
|
||
if on("shot_subjects"):
|
||
raise Refuse("subject_mismatch", ["clip", i])
|
||
return
|
||
for k, c in enumerate(shot["framing"]):
|
||
cs = clip["subjects"][k]
|
||
angle = sb["subjects"][cs["subject"]][cs["views"][0]][0]
|
||
if on("shot_framing") and not matches(c, cs["boxes"][0], angle, sb):
|
||
raise Refuse("framing_mismatch", ["clip", i, 0])
|
||
elif el[0] == "keyframes":
|
||
e = r["entries"][el[1]]
|
||
s = e["shot"]
|
||
clip = d["clips"][s]
|
||
subj, cam = tracks(e)
|
||
for t in sorted(keyed(e, sb["shots"][s]["duration"])):
|
||
at = ["clip", s, t]
|
||
if t >= clip["frames"]:
|
||
raise Refuse("framing_mismatch", at)
|
||
for k, tr in enumerate(subj):
|
||
if tr is not None and on("key_box") and track_value(tr, t) != tuple(clip["subjects"][k]["boxes"][t]):
|
||
raise Refuse("framing_mismatch", at)
|
||
if cam is not None and on("key_camera"):
|
||
got = None if clip["camera"] is None else tuple(clip["camera"][t])
|
||
if got is None or track_value(cam, t) != got:
|
||
raise Refuse("framing_mismatch", at)
|
||
for key in e["keys"]:
|
||
if key["t"] == t:
|
||
for k, ks in enumerate(key["subjects"]):
|
||
cs = clip["subjects"][k]
|
||
angle = sb["subjects"][cs["subject"]][cs["views"][t]][0]
|
||
if on("key_classes") and not matches(ks["classes"], cs["boxes"][t], angle, sb):
|
||
raise Refuse("framing_mismatch", at)
|
||
elif el[0] == "region":
|
||
g = r["regions"][el[1]]
|
||
want = {"from": g["from"], "to": g["to"], "rects": g["rects"], "source": g["source"][0]}
|
||
if on("region") and want not in d["clips"][g["shot"]]["holds"]:
|
||
raise Refuse("region_mismatch", ["clip", g["shot"]])
|
||
else:
|
||
want, got = directed_events(r), d["timeline"]
|
||
if on("seq_order") and [e["shot"] for e in want] != [e["shot"] for e in got]:
|
||
raise Refuse("order_mismatch", ["timeline"])
|
||
sw, _ = starts(want)
|
||
sg, _ = starts(got)
|
||
for i in range(len(want)):
|
||
if on("seq_cuts") and sw[i] != sg[i]:
|
||
raise Refuse("cut_mismatch", ["timeline", i])
|
||
for i in range(len(want)):
|
||
if on("seq_transitions") and want[i]["d"] != got[i]["d"]:
|
||
raise Refuse("transition_mismatch", ["timeline", i])
|
||
for i in range(len(want)):
|
||
if on("seq_ranges") and (want[i]["in"], want[i]["out"]) != (got[i]["in"], got[i]["out"]):
|
||
raise Refuse("duration_mismatch", ["timeline", i])
|
||
|
||
|
||
def verify(case, models, pool):
|
||
# Step 1: reading.
|
||
try:
|
||
d = read_signed(case["output"], "wai-directed", DIRECTED, directed_fields)
|
||
except Refuse as r:
|
||
raise Refuse(r.reason, ["output"])
|
||
held = []
|
||
for i, t in enumerate(case["records"]):
|
||
try:
|
||
held.append(read_record(t))
|
||
except Refuse as r:
|
||
raise Refuse(r.reason, ["record", i])
|
||
|
||
# Step 2: the direction.
|
||
bound, seq_set, keyed_shots = [], False, set()
|
||
for k, link in enumerate(d["direction"]):
|
||
at = ["direction", k]
|
||
rs = [r for r in held if r["link"] == link]
|
||
if not rs:
|
||
if on("known"):
|
||
raise Refuse("record_unknown", at)
|
||
continue
|
||
r = rs[0]
|
||
is_sb = r["kind"] == "wai-storyboard"
|
||
if on("storyboard_first") and (k == 0) != is_sb:
|
||
raise Refuse("storyboard_mismatch", at)
|
||
if on("storyboard_link") and k > 0 and r.get("storyboard") != d["direction"][0]:
|
||
raise Refuse("storyboard_mismatch", at)
|
||
if on("signer") and r["signer"] not in d["directors"]:
|
||
raise Refuse("signer_not_party", at)
|
||
if sets_sequence(r):
|
||
if seq_set and on("conflict_sequence"):
|
||
raise Refuse("direction_conflict", at)
|
||
seq_set = True
|
||
if r["kind"] == "wai-keyframes":
|
||
for e in r["entries"] or []:
|
||
if e["shot"] in keyed_shots and on("conflict_keys"):
|
||
raise Refuse("direction_conflict", at)
|
||
keyed_shots.add(e["shot"])
|
||
if k > 0 and on("consistent") and not consistent(bound[0], r):
|
||
raise Refuse("direction_inconsistent", at)
|
||
bound.append(r)
|
||
sb = bound[0]
|
||
|
||
# Step 3: the check.
|
||
if on("check") and d["replicate"] and any(l["source"] is not None and l["source"]["generator"] != SPECTRAL
|
||
for l in d["layers"]):
|
||
raise Refuse("check_mismatch", ["output"])
|
||
|
||
# Step 4: the layout.
|
||
layout(sb, d)
|
||
|
||
# Steps 5 and 6: the layers and the frames.
|
||
statuses, frame_sha = [], None
|
||
if d["replicate"]:
|
||
pics = {}
|
||
for i, l in enumerate(d["layers"]):
|
||
at = ["layer", i]
|
||
dg, w, h = l["pic"]
|
||
src = l["source"]
|
||
if src is not None and src["generator"] == SPECTRAL:
|
||
mf = models.get(src["model"])
|
||
if mf is None or sha256(mf) != src["model"]:
|
||
raise Refuse("layer_missing", at)
|
||
px = generated(mf, src["prompt"], src["seed"], src["steps"], w, h)
|
||
if on("layer_regenerate") and sha256(px) != dg:
|
||
raise Refuse("regeneration_mismatch", at)
|
||
statuses.append("regenerated")
|
||
elif src is None:
|
||
px = pool.get(dg) if dg in case["pictures"] else None
|
||
if px is None:
|
||
if on("layer_held"):
|
||
raise Refuse("layer_missing", at)
|
||
px = bytes(w * h * 3)
|
||
if on("layer_size") and len(px) != w * h * 3:
|
||
raise Refuse("samples_mismatch", at)
|
||
statuses.append("held")
|
||
else:
|
||
raise Refuse("layer_missing", at) # another generator's picture: never regenerated here
|
||
pics[dg] = (w, h, px)
|
||
key = (SV.jcs(d["value"]["clips"]), SV.jcs(d["value"]["timeline"]), sb["link"],
|
||
tuple(sorted((k, sha256(v[2])) for k, v in pics.items())))
|
||
try:
|
||
digests = frame_digests(sb, d, pics, key)
|
||
except (IndexError, KeyError, ValueError):
|
||
raise Refuse("frame_mismatch", ["output"])
|
||
if on("frames") and blake3(FRAMES_DOMAIN + b"".join(digests)) != d["digest"]:
|
||
raise Refuse("frame_mismatch", ["output"])
|
||
frame_sha = [x.hex() for x in digests]
|
||
else:
|
||
statuses = ["stated"] * len(d["layers"])
|
||
|
||
# Step 7: the direction against the layout.
|
||
for r in bound:
|
||
for el in elements(r):
|
||
check_element(sb, d, r, el)
|
||
|
||
# Step 8: governance.
|
||
computed = []
|
||
for k, r in enumerate(bound):
|
||
for el in elements(r):
|
||
computed.append((k, el, governed(sb, d, r, el)))
|
||
if on("governs"):
|
||
for i in range(max(len(computed), len(d["governs"]))):
|
||
got = d["governs"][i] if i < len(d["governs"]) else None
|
||
want = computed[i] if i < len(computed) else None
|
||
if got != want:
|
||
raise Refuse("governance_mismatch", ["governs", i])
|
||
|
||
# Step 9: policy.
|
||
if on("policy") and case["policy"]["require"] == "exact" and not d["replicate"]:
|
||
raise Refuse("not_exact", ["policy"])
|
||
|
||
return statements(d, bound, computed, frame_sha, statuses)
|
||
|
||
|
||
def statements(d, bound, computed, frame_sha, statuses):
|
||
def director(r):
|
||
return d["directors"].index(r["signer"])
|
||
ev = d["timeline"]
|
||
s, n = starts(ev)
|
||
|
||
def el_json(el):
|
||
return list(el)
|
||
frames = []
|
||
for f in range(n):
|
||
by = {lv: set() for lv in LEVELS}
|
||
for k, el, fr in computed:
|
||
if any(a <= f < b for a, b in fr):
|
||
by[LEVEL[el[0]]].add(director(bound[k]))
|
||
frames.append({"shows": [[shot, t] for _, shot, t in shows(ev, s, f)],
|
||
"directed_by": {lv: sorted(by[lv]) for lv in LEVELS}})
|
||
c2pa = []
|
||
for di in range(len(d["directors"])):
|
||
actions = []
|
||
for k, el, fr in computed:
|
||
r = bound[k]
|
||
if director(r) != di or not fr:
|
||
continue
|
||
changes = []
|
||
for a, b in fr:
|
||
fr_range = {"type": "frame", "frame": {"start": a, "end": b - 1}}
|
||
if el[0] == "region":
|
||
for (x, y, w, h) in r["regions"][el[1]]["rects"]:
|
||
shape = {"type": "spatial", "shape": {"type": "rectangle", "unit": "pixel",
|
||
"origin": {"x": x, "y": y}, "width": w, "height": h}}
|
||
changes.append({"region": [fr_range, shape]})
|
||
else:
|
||
changes.append({"region": [fr_range]})
|
||
actions.append({"action": "c2pa.edited", "changes": changes, "description": DESCRIPTION[LEVEL[el[0]]]})
|
||
if actions:
|
||
c2pa.append({"director": di, "statement": SV.jcs({"actions": actions})})
|
||
att = d["attestation"]
|
||
return {
|
||
"output": d["link"].hex(),
|
||
"status": "exact" if d["replicate"] else "attested",
|
||
"generator": att[0] if att else None,
|
||
"class": att[1] if att else None,
|
||
"records": [{"kind": r["kind"], "link": r["link"].hex(), "director": director(r), "levels": levels_of(r)}
|
||
for r in bound],
|
||
"elements": [{"record": k, "element": el_json(el), "level": LEVEL[el[0]], "director": director(bound[k]),
|
||
"frames": [[a, b] for a, b in fr]} for k, el, fr in computed],
|
||
"cuts": s[1:],
|
||
"frames": frames,
|
||
"frame_sha256": frame_sha,
|
||
"layers": statuses,
|
||
"c2pa": c2pa,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------- the corpus
|
||
|
||
def outcome(v, models, pool):
|
||
try:
|
||
st = verify(v, models, pool)
|
||
return {"outcome": "accept", "statements": st}
|
||
except Refuse as r:
|
||
return {"outcome": "refuse", "reason": r.reason, "at": r.at}
|
||
except Exception as e: # only a verifier with a check dropped gets here
|
||
if not DROPPED:
|
||
raise
|
||
return {"outcome": "error", "error": repr(e)}
|
||
|
||
|
||
def stated(v):
|
||
return {k: v[k] for k in ("outcome", "reason", "at", "statements") if k in v}
|
||
|
||
|
||
def held(doc):
|
||
models = {}
|
||
for m in doc["models"]:
|
||
f = m["file"].encode("utf-8")
|
||
models[bytes.fromhex(m["sha256"])] = f
|
||
pool = {}
|
||
for p in doc["pictures"]:
|
||
px = bytes.fromhex(p["rgb8"])
|
||
pool[sha256(px)] = px # a picture is named by its samples digest, computed here
|
||
return models, pool
|
||
|
||
|
||
def case_of(v, models, pool):
|
||
c = dict(v)
|
||
c["models_held"] = {bytes.fromhex(x): models[bytes.fromhex(x)] for x in v["models"] if bytes.fromhex(x) in models}
|
||
c["pictures"] = {bytes.fromhex(x) for x in v["pictures"]}
|
||
return c
|
||
|
||
|
||
def run_vector(v, models, pool):
|
||
c = case_of(v, models, pool)
|
||
return outcome(c, c["models_held"], pool)
|
||
|
||
|
||
def check(doc):
|
||
models, pool = held(doc)
|
||
fails = []
|
||
for i, m in enumerate(doc.get("measures") or []):
|
||
try:
|
||
fw, fh = m["frame"]
|
||
b = box(m["box"])
|
||
if (size_class(b[3], fh), position_class(b, fw, fh)) != (m["size"], m["position"]):
|
||
fails.append(f"measure {i}")
|
||
except (Bad, KeyError, TypeError, ValueError):
|
||
fails.append(f"measure {i}")
|
||
for i, t in enumerate(doc.get("interpolations") or []):
|
||
try:
|
||
got = interpolate(signed(t["from"]), signed(t["to"]), t["n"], t["d"], t["interp"])
|
||
if str(got) != t["value"]:
|
||
fails.append(f"interpolation {i}")
|
||
except (Bad, KeyError, TypeError, ValueError, ZeroDivisionError):
|
||
fails.append(f"interpolation {i}")
|
||
if not doc.get("measures") or not doc.get("interpolations"):
|
||
fails.append("no measures or no interpolations")
|
||
for v in doc["vectors"]:
|
||
got = run_vector(v, models, pool)
|
||
if got != stated(v):
|
||
fails.append(v["name"])
|
||
return fails
|
||
|
||
|
||
def section(text, head):
|
||
at = text.index("\n" + head)
|
||
end = text.find("\n## ", at + 1)
|
||
return text[at:end if end >= 0 else len(text)]
|
||
|
||
|
||
def reasons():
|
||
with open(EXTENSION, encoding="utf-8") as f:
|
||
text = f.read()
|
||
return re.findall(r"^\| `([a-z_]+)` \|", section(text, "## 13. Reasons"), re.M)
|
||
|
||
|
||
# ---------------------------------------------------------------- mutations
|
||
|
||
def flip_hex_digit(s, at):
|
||
c = s[at]
|
||
return s[:at] + ("1" if c == "0" else "0") + s[at + 1:]
|
||
|
||
|
||
def mutations(doc):
|
||
"""Copies of the corpus, each with one change a conforming verifier must notice."""
|
||
out = []
|
||
names = {v["name"]: i for i, v in enumerate(doc["vectors"])}
|
||
|
||
def mutate(name, about, f):
|
||
m = copy.deepcopy(doc)
|
||
f(m["vectors"][names[name]], m)
|
||
out.append((f"{name}: {about}", m))
|
||
|
||
def edit_output(f):
|
||
def g(v, _):
|
||
o = json.loads(v["output"])
|
||
f(o)
|
||
v["output"] = SV.jcs(o)
|
||
return g
|
||
|
||
mutate("directed", "a box moved in the signed output", edit_output(
|
||
lambda o: o["clips"][1]["subjects"][1]["boxes"][0].update(x="47")))
|
||
mutate("directed", "a record's signature digit changed", lambda v, _: v["records"].__setitem__(
|
||
0, SV.jcs(dict(json.loads(v["records"][0]), sig=flip_hex_digit(json.loads(v["records"][0])["sig"], 5)))))
|
||
mutate("directed", "a cut stated a frame late", lambda v, _: v["statements"]["cuts"].__setitem__(0, 5))
|
||
mutate("directed", "a frame's director dropped", lambda v, _: v["statements"]["frames"][10]["directed_by"]
|
||
.__setitem__("pixel", []))
|
||
mutate("directed", "a frame's SHA-256 changed", lambda v, _: v["statements"]["frame_sha256"].__setitem__(
|
||
0, flip_hex_digit(v["statements"]["frame_sha256"][0], 0)))
|
||
mutate("directed", "a C2PA statement's end frame changed", lambda v, _: v["statements"]["c2pa"][2].__setitem__(
|
||
"statement", v["statements"]["c2pa"][2]["statement"].replace('"end":13', '"end":12')))
|
||
mutate("directed", "the model file not held", lambda v, _: v.__setitem__("models", []))
|
||
mutate("directed", "a byte of the model file changed", lambda v, m: m["models"][0].__setitem__(
|
||
"file", m["models"][0]["file"].replace("1", "2", 1)))
|
||
|
||
def drawing(v, m):
|
||
for p in m["pictures"]:
|
||
if p["sha256"] in v["pictures"]:
|
||
p["rgb8"] = flip_hex_digit(p["rgb8"], 10)
|
||
mutate("directed", "a byte of the drawn reference changed", drawing)
|
||
mutate("attested", "the attestation's class changed in the statement", lambda v, _: v["statements"]
|
||
.__setitem__("class", "detector"))
|
||
mutate("attested", "attested stated as exact", lambda v, _: v["statements"].__setitem__("status", "exact"))
|
||
mutate("undirected_order", "a sequence element stated where no record sets one", lambda v, _: v["statements"]
|
||
["frames"][0]["directed_by"].__setitem__("sequence", [0]))
|
||
mutate("cut_a_frame_early", "its refusal stated at another event", lambda v, _: v.__setitem__("at", ["timeline", 2]))
|
||
mutate("order_swapped", "its refusal stated with another reason", lambda v, _: v.__setitem__("reason", "cut_mismatch"))
|
||
mutate("storyboard_order", "a level dropped from a record's statement", lambda v, _: v["statements"]["records"][0]
|
||
.__setitem__("levels", ["shot"]))
|
||
|
||
def section_mutation(about, f):
|
||
m = copy.deepcopy(doc)
|
||
f(m)
|
||
out.append((about, m))
|
||
section_mutation("a measure's size class moved up a class", lambda m: m["measures"][1].__setitem__(
|
||
"size", SIZES[[n for n, _ in SIZES].index(m["measures"][1]["size"]) + 1][0]))
|
||
section_mutation("an interpolation's value rounded toward zero", lambda m: [
|
||
t.__setitem__("value", str(int(t["value"]) + 1)) for t in m["interpolations"]
|
||
if t["interp"] == "linear" and t["from"] == "-40" and t["n"] == 1][:1])
|
||
section_mutation("the measures left out", lambda m: m.__setitem__("measures", []))
|
||
return out
|
||
|
||
|
||
def self_test(doc):
|
||
"""Drop each check in turn: the corpus must catch every weakened verifier."""
|
||
missed = []
|
||
for name in CHECKS:
|
||
DROPPED.clear()
|
||
DROPPED.add(name)
|
||
if not check(doc):
|
||
missed.append(name)
|
||
DROPPED.clear()
|
||
return missed
|
||
|
||
|
||
def main(corpus_dir):
|
||
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
||
doc = json.load(f)
|
||
ok = True
|
||
if not check_generator():
|
||
print("FAIL the generator does not reproduce its conformance vector")
|
||
return 1
|
||
print("PASS the generator reproduces wai.generate.spectral's conformance vector")
|
||
fails = check(doc)
|
||
for n in fails:
|
||
print("FAIL", n)
|
||
print(("PASS" if not fails else "FAIL") + f" every measure, interpolation and vector gets what the corpus states "
|
||
f"({len(doc['measures'])} measures, {len(doc['interpolations'])} interpolations, {len(doc['vectors'])} vectors)")
|
||
ok &= not fails
|
||
codes = reasons()
|
||
used = {v.get("reason") for v in doc["vectors"]}
|
||
missing = [c for c in codes if c not in used]
|
||
print(("PASS" if not missing else "FAIL") + " every reason of §13 refuses a vector" + (f": {missing}" if missing else ""))
|
||
ok &= not missing
|
||
caught = 0
|
||
muts = mutations(doc)
|
||
for about, m in muts:
|
||
if check(m):
|
||
caught += 1
|
||
else:
|
||
print("FAIL a mutated copy passed:", about)
|
||
ok = False
|
||
print(("PASS" if caught == len(muts) else "FAIL") + f" every mutated copy fails ({caught} of {len(muts)})")
|
||
missed = self_test(doc)
|
||
for n in missed:
|
||
print("FAIL the corpus does not catch a verifier without its check", n)
|
||
print(("PASS" if not missed else "FAIL") + f" the corpus catches the verifier without each of its {len(CHECKS)} checks")
|
||
ok &= not missed
|
||
return 0 if ok else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) != 2:
|
||
print("usage: direction_verify.py <live-conformance/direction>")
|
||
sys.exit(2)
|
||
sys.exit(main(sys.argv[1]))
|