bcef9a5671
The extension, the wai-rs module behind the feature exploration, the corpus live-conformance/exploration/ written by wai_exploration_vectors gen, its second implementation exploration_verify.py, the CI step, and the registry, SPEC and CHANGELOG entries. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1230 lines
48 KiB
Python
1230 lines
48 KiB
Python
#!/usr/bin/env python3
|
||
"""Second implementation of extensions/exploration.md (`wai.media.exploration`), over the corpus
|
||
`live-conformance/exploration/`.
|
||
|
||
Written from the extension's text. It shares no code with wai-rs: it makes every picture with
|
||
the generator as Appendix A restates it, computes the descriptors of §3, runs the schedule of
|
||
§5.2, and reads every record as §2 has it read (strict JSON, the encodings of jwp-receipts
|
||
§2.4.3). It verifies every signature with WAI's strict Ed25519 rule
|
||
(claim-conformance/_ed25519.py), every link hash with BLAKE3 written from its specification
|
||
(claim-conformance/_blake3.py), and reads a taste prior's usage statement with the AI usage
|
||
corpus's second implementation (ai_usage_verify.py). It reads §13's codes from the extension.
|
||
|
||
python3 -I exploration_verify.py <live-conformance/exploration>
|
||
|
||
It checks every generation and descriptor vector of the corpus, gives every case the outcome,
|
||
the reason, the record and the statements the corpus states, and checks that every reason of
|
||
§13 refuses a case. Then it 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 math
|
||
import os
|
||
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 ai_usage_verify as AU # noqa: E402
|
||
import _ed25519 as ED # noqa: E402
|
||
from _blake3 import blake3 # noqa: E402
|
||
|
||
WAI = os.path.join(HERE, "..")
|
||
EXTENSION = os.path.join(WAI, "extensions", "exploration.md")
|
||
|
||
SIGN_DOMAIN = b"wai:exploration\x01" # §2
|
||
LINK_DOMAIN = b"wai:exploration-link\x01" # §2
|
||
DESCRIPTOR_SET = "wai-exploration:image6" # §2
|
||
SCHEDULE = "wai-exploration:map-elites/1" # §5.1
|
||
GENERATOR = "wai.generate.spectral" # §4.1
|
||
NAMES = ["luma", "chroma", "detail", "palette", "centroid_x", "centroid_y"] # §3.2, in order
|
||
SHAPES = ["rise", "fall", "near", "far"] # §8.2
|
||
DEFAULT_SEEDS, DEFAULT_STEPS = 16, 4 # §6.1
|
||
MAX_INT = (1 << 53) - 1
|
||
M64 = (1 << 64) - 1
|
||
VOCABULARIES = ["aipref", "cawg", "tdmrep"] # ai-usage §6.1
|
||
AIPREF_LABELS = ["train-ai", "ai-use", "search"]
|
||
CAWG_KEYS = ["cawg.data_mining", "cawg.ai_inference", "cawg.ai_generative_training", "cawg.ai_training"]
|
||
WHITE_SPACE = set("\t\n\x0b\x0c\r \x85\xa0 "
|
||
"
")
|
||
|
||
# A check named here is left out: the verifier weakened on purpose, to show the corpus catches it.
|
||
SKIP = set()
|
||
|
||
|
||
class Refuse(Exception):
|
||
def __init__(self, reason):
|
||
super().__init__(reason)
|
||
self.reason = reason
|
||
|
||
|
||
class Bad(Exception):
|
||
"""A record that does not read: `record_malformed`."""
|
||
|
||
|
||
def need(c):
|
||
if not c:
|
||
raise Bad()
|
||
|
||
|
||
def sha(b):
|
||
return hashlib.sha256(b).digest()
|
||
|
||
|
||
# ---------------------------------------------------------------- §2 records
|
||
|
||
def strict_json(text):
|
||
"""RFC 8259, no member named twice, no number but an unsigned integer, no boolean, no unpaired
|
||
surrogate, at most 32 levels."""
|
||
def pairs(ps):
|
||
d = {}
|
||
for k, x in ps:
|
||
if k in d:
|
||
raise Bad()
|
||
d[k] = x
|
||
return d
|
||
|
||
def no(s):
|
||
raise Bad()
|
||
|
||
def integer(s):
|
||
if s.startswith("-"):
|
||
raise Bad()
|
||
return int(s)
|
||
|
||
try:
|
||
v = json.loads(text, object_pairs_hook=pairs, parse_float=no, parse_int=integer, parse_constant=no)
|
||
except (ValueError, RecursionError):
|
||
raise Bad()
|
||
|
||
def walk(x, depth):
|
||
if isinstance(x, (dict, list)):
|
||
need(depth <= 32)
|
||
for y in (x.values() if isinstance(x, dict) else x):
|
||
walk(y, depth + 1)
|
||
for k in (x if isinstance(x, dict) else []):
|
||
walk(k, depth)
|
||
elif isinstance(x, str):
|
||
need(not any(0xD800 <= ord(c) <= 0xDFFF for c in x))
|
||
elif isinstance(x, bool):
|
||
raise Bad()
|
||
elif isinstance(x, int):
|
||
need(x <= MAX_INT)
|
||
walk(v, 1)
|
||
return v
|
||
|
||
|
||
def jstr(s):
|
||
out = ['"']
|
||
for c in s:
|
||
o = ord(c)
|
||
if c == '"':
|
||
out.append('\\"')
|
||
elif c == "\\":
|
||
out.append("\\\\")
|
||
elif c == "\b":
|
||
out.append("\\b")
|
||
elif c == "\f":
|
||
out.append("\\f")
|
||
elif c == "\n":
|
||
out.append("\\n")
|
||
elif c == "\r":
|
||
out.append("\\r")
|
||
elif c == "\t":
|
||
out.append("\\t")
|
||
elif o < 0x20:
|
||
out.append("\\u%04x" % o)
|
||
else:
|
||
out.append(c)
|
||
out.append('"')
|
||
return "".join(out)
|
||
|
||
|
||
def jcs(v):
|
||
"""RFC 8785 over strings, unsigned integers, null, arrays and objects."""
|
||
if v is None:
|
||
return "null"
|
||
if isinstance(v, bool):
|
||
raise Bad()
|
||
if isinstance(v, int):
|
||
return str(v)
|
||
if isinstance(v, str):
|
||
return jstr(v)
|
||
if isinstance(v, list):
|
||
return "[" + ",".join(jcs(x) for x in v) + "]"
|
||
keys = sorted(v, key=lambda k: k.encode("utf-16-be"))
|
||
return "{" + ",".join(jstr(k) + ":" + jcs(v[k]) for k in keys) + "}"
|
||
|
||
|
||
def link_hash(v):
|
||
return blake3(LINK_DOMAIN + jcs(v).encode("utf-8"))
|
||
|
||
|
||
def signing_bytes(v):
|
||
return SIGN_DOMAIN + jcs({k: x for k, x in v.items() if k != "sig"}).encode("utf-8")
|
||
|
||
|
||
def is_uint(x):
|
||
return isinstance(x, int) and not isinstance(x, bool) and 0 <= x <= MAX_INT
|
||
|
||
|
||
def uint(x, lo=0, hi=MAX_INT):
|
||
need(is_uint(x) and lo <= x <= hi)
|
||
return x
|
||
|
||
|
||
def hexb(x, n):
|
||
need(isinstance(x, str) and len(x) == 2 * n and all(c in "0123456789abcdef" for c in x))
|
||
return bytes.fromhex(x)
|
||
|
||
|
||
def is_term(s):
|
||
return (isinstance(s, str) and 1 <= len(s.encode("utf-8")) <= 2048
|
||
and not any(c in WHITE_SPACE or ord(c) < 0x20 or 0x7F <= ord(c) <= 0x9F for c in s))
|
||
|
||
|
||
def label(s):
|
||
need(isinstance(s, str) and 1 <= len(s.encode("utf-8")) <= 256)
|
||
return s
|
||
|
||
|
||
def date_time(s):
|
||
"""RFC 3339 date-time, the day within its month."""
|
||
need(isinstance(s, str) and len(s) >= 20 and s.isascii())
|
||
d = lambda i: s[i].isdigit()
|
||
need(all(d(i) for i in (0, 1, 2, 3, 5, 6, 8, 9, 11, 12, 14, 15, 17, 18)))
|
||
need(s[4] == "-" and s[7] == "-" and s[10] in "Tt" and s[13] == ":" and s[16] == ":")
|
||
yr, mo, da, hh, mi, ss = int(s[0:4]), int(s[5:7]), int(s[8:10]), int(s[11:13]), int(s[14:16]), int(s[17:19])
|
||
leap = yr % 4 == 0 and (yr % 100 != 0 or yr % 400 == 0)
|
||
days = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
|
||
need(1 <= mo <= 12 and 1 <= da <= days[mo - 1] and hh <= 23 and mi <= 59 and ss <= 60)
|
||
i = 19
|
||
if i < len(s) and s[i] == ".":
|
||
j = i + 1
|
||
while j < len(s) and s[j].isdigit():
|
||
j += 1
|
||
need(j > i + 1)
|
||
i = j
|
||
rest = s[i:]
|
||
if rest in ("Z", "z"):
|
||
return
|
||
need(len(rest) == 6 and rest[0] in "+-" and rest[1:3].isdigit() and rest[3] == ":" and rest[4:6].isdigit())
|
||
need(int(rest[1:3]) <= 23 and int(rest[4:6]) <= 59)
|
||
|
||
|
||
def exact(v, names):
|
||
need(isinstance(v, dict) and set(v) == set(names))
|
||
return v
|
||
|
||
|
||
def read_record(text, kind, names, optional=None):
|
||
"""§2: the value, its signer and its link hash; raises Bad."""
|
||
v = strict_json(text)
|
||
need(isinstance(v, dict))
|
||
need(set(v) == set(names) or (optional is not None and set(v) == set(names) | {optional}))
|
||
need(v["kind"] == kind)
|
||
signer = hexb(v["signer_pubkey"], 32)
|
||
hexb(v["sig"], 64)
|
||
date_time(v["issued_at"])
|
||
return v, signer, link_hash(v)
|
||
|
||
|
||
SIG_MEMO = {}
|
||
|
||
|
||
def check_signature(v, signer):
|
||
if "signature" in SKIP:
|
||
return
|
||
key = (signer, signing_bytes(v), bytes.fromhex(v["sig"]))
|
||
if key not in SIG_MEMO:
|
||
SIG_MEMO[key] = ED.verify(*key)
|
||
if SIG_MEMO[key]:
|
||
raise Refuse(SIG_MEMO[key])
|
||
|
||
|
||
# ---------------------------------------------------------------- Appendix A: the generator
|
||
|
||
def tdiv(a, b):
|
||
q = abs(a) // abs(b)
|
||
return q if (a >= 0) == (b >= 0) else -q
|
||
|
||
|
||
def splitmix(state):
|
||
"""§5.2: the next state and the draw."""
|
||
state = (state + 0x9E3779B97F4A7C15) & M64
|
||
z = state
|
||
z = ((z ^ (z >> 30)) * 0xBF58476D1CE4E5B9) & M64
|
||
z = ((z ^ (z >> 27)) * 0x94D049BB133111EB) & M64
|
||
return state, z ^ (z >> 31)
|
||
|
||
|
||
class Rng:
|
||
def __init__(self, s):
|
||
self.s = s & M64
|
||
|
||
def draw(self):
|
||
self.s, z = splitmix(self.s)
|
||
return z
|
||
|
||
|
||
ATAN = [128, 76, 40, 20, 10, 5, 3, 1, 1, 0, 0, 0, 0, 0]
|
||
|
||
|
||
def iatan2(dy, dx, t):
|
||
"""A.5."""
|
||
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
|
||
x, y, ang = abs(dx) << 12, abs(dy) << 12, 0
|
||
for i, a in enumerate(ATAN):
|
||
if y > 0:
|
||
x, y = x + (y >> i), y - (x >> i)
|
||
ang += a
|
||
elif y < 0:
|
||
x, y = 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
|
||
|
||
|
||
class Model:
|
||
"""A.1."""
|
||
def __init__(self, v):
|
||
self.lut = v["sine_lut"]
|
||
self.t = v["lut_size"]
|
||
self.lut_bits = v.get("lut_bits", 14)
|
||
self.freq_max = v["freq_max"]
|
||
self.amp_bits = v["amp_bits"]
|
||
self.offsets = v["color_offsets"]
|
||
self.base = v["base_components"]
|
||
self.step = v["step_components"]
|
||
self.lexicon = {w: (list(e) + [0] * 10)[:10] for w, e in v.get("lexicon", {}).items()}
|
||
|
||
|
||
def tokens(prompt):
|
||
out, cur = [], []
|
||
for c in prompt.lower():
|
||
if c.isascii() and c.isalnum():
|
||
cur.append(c)
|
||
elif cur:
|
||
out.append("".join(cur))
|
||
cur = []
|
||
if cur:
|
||
out.append("".join(cur))
|
||
return out
|
||
|
||
|
||
def intent(m, prompt):
|
||
"""A.2."""
|
||
s, n = [0] * 10, 0
|
||
for tok in tokens(prompt):
|
||
e = m.lexicon.get(tok)
|
||
if e is not None:
|
||
s = [a + b for a, b in zip(s, e)]
|
||
n += 1
|
||
if n == 0:
|
||
return [256, 256, 256], 256, 0, 0, 0, 0, 0, 0
|
||
mx = max(max(s[0:3]), 1)
|
||
cl = lambda x, lo, hi: max(lo, min(hi, x))
|
||
hue = [tdiv(s[c] * 256, mx) for c in range(3)]
|
||
return (hue, tdiv(s[3], n), cl(tdiv(s[4], n), -256, 256), cl(tdiv(s[5], n), 0, 256), cl(tdiv(s[6], n), 0, 256),
|
||
cl(tdiv(s[7], n), -256, 256), cl(tdiv(s[8], n), 0, 256), cl(tdiv(s[9], n), 0, 256))
|
||
|
||
|
||
def fnv(b):
|
||
h = 14695981039346656037
|
||
for x in b:
|
||
h = ((h ^ x) * 1099511628211) & M64
|
||
return h
|
||
|
||
|
||
GEN_MEMO = {}
|
||
|
||
|
||
def gen(m, prompt, seed, steps, w, h):
|
||
"""§4.1: `gen(model, prompt, seed, steps, w, h)`, as Appendix A restates it."""
|
||
key = (id(m), prompt, seed, steps, w, h)
|
||
if key in GEN_MEMO:
|
||
return GEN_MEMO[key]
|
||
hue, freq, orient, radial, mirror, gradient, angular, spiral = intent(m, prompt)
|
||
k = m.base + steps * m.step
|
||
sx, sy = (max(256 + orient, 0), 256) if orient < 0 else (256, max(256 - orient, 0))
|
||
f = max(tdiv(m.freq_max * freq, 256), 1)
|
||
span, amp_max = 2 * f + 1, 1 << m.amp_bits
|
||
rng = Rng(fnv(prompt.encode("ascii")) ^ seed ^ 0xD1B54A32D192ED03)
|
||
thr = [radial, mirror, angular, spiral]
|
||
plane = max(256 - sum(thr), 48)
|
||
total = plane + sum(thr)
|
||
comps = []
|
||
for _ in range(k):
|
||
pick = rng.draw() % total
|
||
fx = rng.draw() % span - f
|
||
fy = rng.draw() % span - f
|
||
amp = rng.draw() % amp_max + 1
|
||
phase = rng.draw() % m.t
|
||
typ, bound = 0, plane
|
||
if pick >= plane:
|
||
for i, wt in enumerate(thr):
|
||
bound += wt
|
||
if pick < bound:
|
||
typ = i + 1
|
||
break
|
||
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))
|
||
n, t, lut = w * h, m.t, m.lut
|
||
cx, cy = w // 2, h // 2
|
||
gscale = ((1 << m.lut_bits) * amp_max * k) // (2048 * max(cy, 1))
|
||
geo = []
|
||
for y in range(h):
|
||
for x in range(w):
|
||
dx, dy = x - cx, y - cy
|
||
geo.append((x, y, dx, dy, math.isqrt(dx * dx + dy * dy), iatan2(dy, dx, t)))
|
||
acc = [0] * (3 * n)
|
||
for c, o in enumerate(m.offsets):
|
||
for p, (x, y, dx, dy, r, th) in enumerate(geo):
|
||
a = gradient * (cy - y) * gscale
|
||
for typ, fx, fy, amp, ph in comps:
|
||
if typ == 0:
|
||
arg = fx * x + fy * y + ph + o
|
||
elif typ == 1:
|
||
arg = fx * r + ph + o
|
||
elif typ == 2:
|
||
arg = fx * abs(dx) + fy * abs(dy) + ph + o
|
||
elif typ == 3:
|
||
arg = fx * th + ph + o
|
||
else:
|
||
arg = fx * th + fy * r + ph + o
|
||
a += amp * lut[arg % t]
|
||
acc[c * n + p] = a
|
||
g0, g1 = min(acc), max(acc)
|
||
rg = max(g1 - g0, 1)
|
||
e = [64 + tdiv(hue[c] * 192, 256) for c in range(3)]
|
||
out = bytearray(3 * n)
|
||
for c in range(3):
|
||
for p in range(n):
|
||
q = ((acc[c * n + p] - g0) * 255 + rg // 2) // rg
|
||
out[3 * p + c] = min(max(tdiv(q * e[c], 256), 0), 255)
|
||
out = bytes(out)
|
||
GEN_MEMO[key] = out
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------- §3 descriptors
|
||
|
||
DESC_MEMO = {}
|
||
|
||
|
||
def descriptors(b, w, h):
|
||
"""§3.2: the six, in order."""
|
||
key = (sha(b), w, h)
|
||
if key in DESC_MEMO:
|
||
return DESC_MEMO[key]
|
||
n = w * h
|
||
assert len(b) == 3 * n
|
||
Y = [(77 * b[i] + 150 * b[i + 1] + 29 * b[i + 2] + 128) >> 8 for i in range(0, 3 * n, 3)]
|
||
luma = 1024 * sum(Y) // (255 * n)
|
||
chroma = 1024 * sum(max(b[i:i + 3]) - min(b[i:i + 3]) for i in range(0, 3 * n, 3)) // (255 * n)
|
||
P = (w - 1) * h + w * (h - 1)
|
||
D = 0
|
||
for y in range(h):
|
||
for x in range(w):
|
||
p = y * w + x
|
||
if x + 1 < w:
|
||
D += abs(Y[p + 1] - Y[p])
|
||
if y + 1 < h:
|
||
D += abs(Y[p + w] - Y[p])
|
||
detail = 0 if P == 0 else min(1024, 1024 * D // (16 * P))
|
||
k = [0] * 64
|
||
for i in range(0, 3 * n, 3):
|
||
k[(b[i] >> 6) * 16 + (b[i + 1] >> 6) * 4 + (b[i + 2] >> 6)] += 1
|
||
palette = 1024 * (n * n - sum(c * c for c in k)) // (n * n)
|
||
V = [y + 1 for y in Y]
|
||
SV = sum(V)
|
||
cx = 1024 * sum((2 * (p % w) + 1) * V[p] for p in range(n)) // (2 * w * SV)
|
||
cy = 1024 * sum((2 * (p // w) + 1) * V[p] for p in range(n)) // (2 * h * SV)
|
||
out = [luma, chroma, detail, palette, cx, cy]
|
||
DESC_MEMO[key] = out
|
||
return out
|
||
|
||
|
||
def distance(a, b):
|
||
"""§3.4."""
|
||
return sum(abs(x - y) for x, y in zip(a, b))
|
||
|
||
|
||
def read_vector(x):
|
||
need(isinstance(x, list) and len(x) == 6)
|
||
return [uint(d, 0, 1024) for d in x]
|
||
|
||
|
||
def read_grid(x):
|
||
"""§3.5."""
|
||
need(isinstance(x, list) and 1 <= len(x) <= 4)
|
||
grid, seen = [], set()
|
||
for a in x:
|
||
exact(a, ["descriptor", "lo", "hi", "bins"])
|
||
need(a["descriptor"] in NAMES and a["descriptor"] not in seen)
|
||
seen.add(a["descriptor"])
|
||
lo, hi, bins = uint(a["lo"], 0, 1024), uint(a["hi"], 0, 1024), uint(a["bins"], 1, 64)
|
||
need(lo < hi)
|
||
grid.append((NAMES.index(a["descriptor"]), lo, hi, bins))
|
||
need(cells(grid) <= 4096)
|
||
return grid
|
||
|
||
|
||
def cells(grid):
|
||
c = 1
|
||
for a in grid:
|
||
c *= a[3]
|
||
return c
|
||
|
||
|
||
def niche(grid, d):
|
||
k = 0
|
||
for i, lo, hi, bins in grid:
|
||
x = d[i]
|
||
k = k * bins + (0 if x < lo else bins - 1 if x >= hi else (x - lo) * bins // (hi - lo))
|
||
return k
|
||
|
||
|
||
def size_ok(w, h):
|
||
return 1 <= w <= 4096 and 1 <= h <= 4096 and w * h <= 1 << 23
|
||
|
||
|
||
def read_picture(x, with_d):
|
||
exact(x, ["samples", "width", "height", "descriptors"] if with_d else ["samples", "width", "height"])
|
||
w, h = uint(x["width"]), uint(x["height"])
|
||
need(size_ok(w, h))
|
||
return {"samples": hexb(x["samples"], 32), "w": w, "h": h, "d": read_vector(x["descriptors"]) if with_d else None}
|
||
|
||
|
||
def is_prompt(s):
|
||
return isinstance(s, str) and len(s.encode("utf-8")) <= 256 and all(0x20 <= ord(c) <= 0x7E and c not in '"\\' for c in s)
|
||
|
||
|
||
def model_digest(s):
|
||
need(isinstance(s, str) and s.startswith("sha256:"))
|
||
return hexb(s[7:], 32)
|
||
|
||
|
||
# ---------------------------------------------------------------- what the verifier holds
|
||
|
||
class Held:
|
||
def __init__(self, models, priors, samples):
|
||
self.models, self.priors, self.samples = models, priors, samples
|
||
|
||
def model(self, digest):
|
||
if digest not in self.models:
|
||
raise Refuse("model_unheld")
|
||
return self.models[digest]
|
||
|
||
def samples_of(self, p):
|
||
"""A picture's held samples, of its size, or None."""
|
||
b = self.samples.get(p["samples"])
|
||
if b is None:
|
||
return None
|
||
if len(b) != p["w"] * p["h"] * 3 and "sample-size" not in SKIP:
|
||
raise Refuse("samples_mismatch")
|
||
return b
|
||
|
||
|
||
# ---------------------------------------------------------------- §8 taste priors and §10.1 objectives
|
||
|
||
def read_pin(x):
|
||
exact(x, ["id", "sha256", "determinism"])
|
||
label(x["id"])
|
||
label(x["determinism"])
|
||
return hexb(x["sha256"], 32)
|
||
|
||
|
||
def read_objective(x):
|
||
"""§10.1: ("none",), ("taste", digest) or ("model", digest)."""
|
||
if x == "none":
|
||
return ("none", None)
|
||
need(isinstance(x, dict) and "type" in x)
|
||
if x["type"] == "taste":
|
||
exact(x, ["type", "prior"])
|
||
return ("taste", read_pin(x["prior"]))
|
||
need(x["type"] == "model")
|
||
exact(x, ["type", "sha256", "id"])
|
||
label(x["id"])
|
||
return ("model", hexb(x["sha256"], 32))
|
||
|
||
|
||
def objective_report(o):
|
||
return "none" if o[0] == "none" else {"type": o[0], "sha256": o[1].hex()}
|
||
|
||
|
||
TASTE = ["kind", "owner", "scope", "descriptors", "terms", "usage", "issued_at", "signer_pubkey", "sig"]
|
||
|
||
|
||
def read_taste(b):
|
||
"""§8.1, §8.4: the prior, read and verified."""
|
||
try:
|
||
text = b.decode("utf-8")
|
||
v, signer, _ = read_record(text, "wai-taste", TASTE)
|
||
need(jcs(v).encode("utf-8") == b or "canonical-prior" in SKIP)
|
||
owner = hexb(v["owner"], 32)
|
||
need(v["scope"] in ("personal", "community"))
|
||
need(v["descriptors"] == DESCRIPTOR_SET)
|
||
need(isinstance(v["terms"], list) and 1 <= len(v["terms"]) <= 32)
|
||
terms = []
|
||
for t in v["terms"]:
|
||
exact(t, ["descriptor", "shape", "at", "weight"])
|
||
need(t["descriptor"] in NAMES and t["shape"] in SHAPES)
|
||
at, weight = uint(t["at"], 0, 1024), uint(t["weight"], 1, 1024)
|
||
need(t["shape"] in ("near", "far") or at == 0)
|
||
terms.append((NAMES.index(t["descriptor"]), t["shape"], at, weight))
|
||
usage = None
|
||
if v["usage"] is not None:
|
||
try:
|
||
usage = AU.read_statement(v["usage"])
|
||
except AU.Refuse:
|
||
raise Bad()
|
||
except (Bad, UnicodeDecodeError, KeyError, TypeError):
|
||
raise Refuse("record_malformed")
|
||
check_signature(v, signer)
|
||
if signer != owner and "owner" not in SKIP:
|
||
raise Refuse("taste_not_owner")
|
||
return {"digest": sha(b), "owner": owner, "scope": v["scope"], "terms": terms, "usage": usage}
|
||
|
||
|
||
def score(prior, d):
|
||
"""§8.2."""
|
||
s = 0
|
||
for i, shape, at, weight in prior["terms"]:
|
||
x = d[i]
|
||
s += weight * (x if shape == "rise" else 1024 - x if shape == "fall" else 1024 - abs(x - at) if shape == "near" else abs(x - at))
|
||
return s
|
||
|
||
|
||
def resolve(held, digest):
|
||
"""§8.3: the prior a pin names, verified."""
|
||
if digest not in held.priors:
|
||
raise Refuse("objective_unresolved")
|
||
return read_taste(held.priors[digest])
|
||
|
||
|
||
# ---------------------------------------------------------------- §4 and §5 archives
|
||
|
||
def derived(prompt, modifiers, mods):
|
||
"""§4.2."""
|
||
return prompt + "".join(" " + modifiers[i] for i in range(len(modifiers)) if mods >> i & 1)
|
||
|
||
|
||
def run_schedule(m, req, prior):
|
||
"""§5.2: the elites, ascending by niche."""
|
||
rng = Rng(req["seed"])
|
||
A, V = {}, {}
|
||
M = len(req["modifiers"])
|
||
for t in range(req["trials"]):
|
||
if t < req["init"]:
|
||
g = (rng.draw() % (1 << 32), req["steps"], 0)
|
||
else:
|
||
O = sorted(A)
|
||
a = O[rng.draw() % len(O)]
|
||
b = O[rng.draw() % len(O)]
|
||
parent = A[b]["genome"] if V[b] < V[a] else A[a]["genome"]
|
||
mods = parent[2]
|
||
for i in range(M):
|
||
if rng.draw() % M == 0:
|
||
mods ^= 1 << i
|
||
d = rng.draw() % 8
|
||
steps = max(1, min(64, parent[1] + (d - 4 if d < 4 else d - 3)))
|
||
g = (rng.draw() % (1 << 32), steps, mods)
|
||
pic = gen(m, derived(req["prompt"], req["modifiers"], g[2]), g[0], g[1], req["w"], req["h"])
|
||
dv = descriptors(pic, req["w"], req["h"])
|
||
k = niche(req["grid"], dv)
|
||
V[k] = V.get(k, 0) + 1
|
||
s = score(prior, dv) if prior is not None else None
|
||
e = {"niche": k, "trial": t, "genome": g, "samples": sha(pic), "d": dv, "score": s}
|
||
if k not in A or (s is not None and s > A[k]["score"]):
|
||
A[k] = e
|
||
return [A[k] for k in sorted(A)]
|
||
|
||
|
||
ARCHIVE = ["kind", "generator", "model", "prompt", "modifiers", "width", "height", "steps", "descriptors", "grid", "search",
|
||
"objective", "elites", "coverage", "issued_at", "signer_pubkey", "sig"]
|
||
|
||
|
||
def read_archive(v):
|
||
need(v["generator"] == GENERATOR and v["descriptors"] == DESCRIPTOR_SET)
|
||
req = {"model": model_digest(v["model"])}
|
||
need(is_prompt(v["prompt"]))
|
||
req["prompt"] = v["prompt"]
|
||
mods = v["modifiers"]
|
||
need(isinstance(mods, list) and len(mods) <= 8)
|
||
for x in mods:
|
||
need(isinstance(x, str) and 1 <= len(x) <= 32 and all(c in "abcdefghijklmnopqrstuvwxyz0123456789" for c in x))
|
||
req["modifiers"] = mods
|
||
need(len(derived(req["prompt"], mods, (1 << len(mods)) - 1)) <= 256)
|
||
req["w"], req["h"] = uint(v["width"]), uint(v["height"])
|
||
need(size_ok(req["w"], req["h"]))
|
||
req["steps"] = uint(v["steps"], 1, 64)
|
||
req["grid"] = read_grid(v["grid"])
|
||
s = exact(v["search"], ["schedule", "seed", "trials", "init"])
|
||
need(s["schedule"] == SCHEDULE)
|
||
req["seed"], req["trials"] = uint(s["seed"]), uint(s["trials"], 1, 4096)
|
||
req["init"] = uint(s["init"], 1, req["trials"])
|
||
req["objective"] = read_objective(v["objective"])
|
||
need(req["objective"][0] != "model")
|
||
elites, trials = [], set()
|
||
need(isinstance(v["elites"], list) and 1 <= len(v["elites"]) <= req["trials"])
|
||
for e in v["elites"]:
|
||
exact(e, ["niche", "trial", "seed", "steps", "mods", "samples", "descriptors", "score"])
|
||
k = uint(e["niche"], 0, cells(req["grid"]) - 1)
|
||
t = uint(e["trial"], 0, req["trials"] - 1)
|
||
g = (uint(e["seed"], 0, (1 << 32) - 1), uint(e["steps"], 1, 64), uint(e["mods"], 0, (1 << len(mods)) - 1))
|
||
sc = None if e["score"] is None else uint(e["score"])
|
||
need((sc is not None) == (req["objective"][0] == "taste"))
|
||
need(t not in trials and (not elites or elites[-1]["niche"] < k))
|
||
trials.add(t)
|
||
elites.append({"niche": k, "trial": t, "genome": g, "samples": hexb(e["samples"], 32), "d": read_vector(e["descriptors"]), "score": sc})
|
||
c = exact(v["coverage"], ["occupied", "cells"])
|
||
return req, elites, uint(c["occupied"]), uint(c["cells"])
|
||
|
||
|
||
def verify_archive(text, case, archives):
|
||
try:
|
||
v, signer, link = read_record(text, "wai-archive", ARCHIVE)
|
||
req, elites, occupied, n_cells = read_archive(v)
|
||
except (Bad, KeyError, TypeError):
|
||
raise Refuse("record_malformed")
|
||
check_signature(v, signer)
|
||
held, policy = case["held"], case["policy"]
|
||
if (occupied != len(elites) or n_cells != cells(req["grid"])) and "coverage" not in SKIP: # §5.4 step 2
|
||
raise Refuse("coverage_mismatch")
|
||
prior = resolve(held, req["objective"][1]) if req["objective"][0] == "taste" else None # step 3
|
||
m = held.model(req["model"]) # step 4
|
||
for e in elites: # step 5
|
||
pic = gen(m, derived(req["prompt"], req["modifiers"], e["genome"][2]), e["genome"][0], e["genome"][1], req["w"], req["h"])
|
||
if sha(pic) != e["samples"] and "regenerate" not in SKIP:
|
||
raise Refuse("generation_mismatch")
|
||
d = descriptors(pic, req["w"], req["h"])
|
||
if d != e["d"] and "descriptors" not in SKIP:
|
||
raise Refuse("descriptor_mismatch")
|
||
if niche(req["grid"], d) != e["niche"] and "niche" not in SKIP:
|
||
raise Refuse("niche_mismatch")
|
||
if e["score"] != (score(prior, d) if prior else None) and "score" not in SKIP:
|
||
raise Refuse("score_mismatch")
|
||
if policy["replay"] and "replay" not in SKIP: # step 6
|
||
if run_schedule(m, req, prior) != elites:
|
||
raise Refuse("archive_replay_mismatch")
|
||
archives[link] = (req, elites)
|
||
return {"kind": "wai-archive", "link": link.hex(), "level": "replayed" if policy["replay"] else "elites",
|
||
"occupied": occupied, "cells": n_cells, "objective": objective_report(req["objective"])}
|
||
|
||
|
||
# ---------------------------------------------------------------- §6 default-distance
|
||
|
||
DISTANCE = ["kind", "work", "generator", "model", "prompt", "descriptors", "defaults", "class", "descriptor_distance",
|
||
"nearest_descriptor", "sample_distance", "nearest_sample", "issued_at", "signer_pubkey", "sig"]
|
||
FIGURES = ["descriptor_distance", "nearest_descriptor", "sample_distance", "nearest_sample"]
|
||
|
||
|
||
def nearest(xs):
|
||
best = None
|
||
for i, x in enumerate(xs):
|
||
if best is None or x < best[1]:
|
||
best = (i, x)
|
||
return best
|
||
|
||
|
||
def sample_distance(a, b):
|
||
"""§6.2: ⌈1024 · SAD / (765 · n)⌉ = ⌈1024 · SAD / (255 · bytes)⌉."""
|
||
sad = sum(abs(x - y) for x, y in zip(a, b))
|
||
return -(-1024 * sad // (255 * len(a)))
|
||
|
||
|
||
def verify_distance(text, case):
|
||
held = case["held"]
|
||
try:
|
||
v, signer, link = read_record(text, "wai-distance", DISTANCE)
|
||
work = read_picture(v["work"], True)
|
||
need(is_term(v["generator"]) and is_term(v["model"]) and is_prompt(v["prompt"]))
|
||
need(v["descriptors"] == DESCRIPTOR_SET)
|
||
cls = v["class"]
|
||
need(cls in ("measured", "estimate", "unknown"))
|
||
need(isinstance(v["defaults"], list) and len(v["defaults"]) <= DEFAULT_SEEDS)
|
||
defaults = []
|
||
for x in v["defaults"]:
|
||
exact(x, ["samples", "descriptors"])
|
||
defaults.append((None if x["samples"] is None else hexb(x["samples"], 32), read_vector(x["descriptors"])))
|
||
stated = [None if v[f] is None else uint(v[f]) for f in FIGURES]
|
||
if cls == "measured": # §6.3
|
||
need(len(defaults) == DEFAULT_SEEDS and all(s is not None for s, _ in defaults) and all(x is not None for x in stated))
|
||
elif cls == "estimate":
|
||
need(defaults and stated[0] is not None and stated[1] is not None and stated[2] is None and stated[3] is None)
|
||
else:
|
||
need(not defaults and all(x is None for x in stated) or "unknown-null" in SKIP)
|
||
digest = model_digest(v["model"]) if v["generator"] == GENERATOR else None
|
||
except (Bad, KeyError, TypeError):
|
||
raise Refuse("record_malformed")
|
||
check_signature(v, signer)
|
||
if (v["generator"] == GENERATOR) != (cls == "measured") and "class" not in SKIP: # §6.4 step 1
|
||
raise Refuse("distance_class_invalid")
|
||
wb = held.samples_of(work) # step 2
|
||
figures = list(stated)
|
||
if cls == "measured": # step 3
|
||
m = held.model(digest)
|
||
dset = [gen(m, v["prompt"], k, DEFAULT_STEPS, work["w"], work["h"]) for k in range(DEFAULT_SEEDS)]
|
||
dvec = [descriptors(b, work["w"], work["h"]) for b in dset]
|
||
for k, (s, d) in enumerate(defaults):
|
||
if (s != sha(dset[k]) or d != dvec[k]) and "defaults" not in SKIP:
|
||
raise Refuse("default_mismatch")
|
||
if wb is None:
|
||
raise Refuse("work_unheld")
|
||
wd = descriptors(wb, work["w"], work["h"])
|
||
if wd != work["d"] and "descriptors" not in SKIP:
|
||
raise Refuse("descriptor_mismatch")
|
||
figures = list(nearest(distance(wd, d) for d in dvec)[::-1]) + list(nearest(sample_distance(wb, b) for b in dset)[::-1])
|
||
elif cls == "estimate": # step 4
|
||
if wb is not None and descriptors(wb, work["w"], work["h"]) != work["d"] and "descriptors" not in SKIP:
|
||
raise Refuse("descriptor_mismatch")
|
||
figures[0:2] = nearest(distance(work["d"], d) for _, d in defaults)[::-1]
|
||
elif wb is not None and descriptors(wb, work["w"], work["h"]) != work["d"] and "descriptors" not in SKIP: # step 5
|
||
raise Refuse("descriptor_mismatch")
|
||
if figures != stated and "figures" not in SKIP: # step 6
|
||
raise Refuse("distance_mismatch")
|
||
out = {"kind": "wai-distance", "link": link.hex(), "class": cls}
|
||
out.update(zip(FIGURES, stated))
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------- §7 collections
|
||
|
||
COLLECTION = ["kind", "scope", "subject", "descriptors", "grid", "members", "coverage", "distances", "issued_at",
|
||
"signer_pubkey", "sig"]
|
||
SPREAD = ["pairs", "sum", "min", "median", "max", "width", "histogram"]
|
||
|
||
|
||
def spread(vs, width):
|
||
"""§7.2."""
|
||
ds = sorted(distance(vs[i], vs[j]) for i in range(len(vs)) for j in range(i + 1, len(vs)))
|
||
hist = [0] * 16
|
||
for d in ds:
|
||
hist[min(d // width, 15)] += 1
|
||
return {"pairs": len(ds), "sum": sum(ds), "min": ds[0], "median": ds[(len(ds) - 1) // 2], "max": ds[-1],
|
||
"width": width, "histogram": hist}
|
||
|
||
|
||
def verify_collection(text, case):
|
||
held = case["held"]
|
||
try:
|
||
v, signer, link = read_record(text, "wai-collection", COLLECTION)
|
||
need(v["scope"] in ("gallery", "room", "session", "archive") and is_term(v["subject"]))
|
||
need(v["descriptors"] == DESCRIPTOR_SET)
|
||
grid = read_grid(v["grid"])
|
||
need(isinstance(v["members"], list) and 2 <= len(v["members"]) <= 256)
|
||
members = [read_picture(x, True) for x in v["members"]]
|
||
c = exact(v["coverage"], ["occupied", "cells"])
|
||
occupied, n_cells = uint(c["occupied"]), uint(c["cells"])
|
||
s = exact(v["distances"], SPREAD)
|
||
for f in SPREAD[:5]:
|
||
uint(s[f])
|
||
uint(s["width"], 1, 6144)
|
||
need(isinstance(s["histogram"], list) and len(s["histogram"]) == 16)
|
||
for x in s["histogram"]:
|
||
uint(x)
|
||
except (Bad, KeyError, TypeError):
|
||
raise Refuse("record_malformed")
|
||
check_signature(v, signer)
|
||
vs = []
|
||
for p in members:
|
||
b = held.samples_of(p)
|
||
if b is None:
|
||
raise Refuse("member_unheld")
|
||
d = descriptors(b, p["w"], p["h"])
|
||
if d != p["d"] and "descriptors" not in SKIP:
|
||
raise Refuse("descriptor_mismatch")
|
||
vs.append(d)
|
||
if (occupied != len({niche(grid, d) for d in vs}) or n_cells != cells(grid)) and "coverage" not in SKIP:
|
||
raise Refuse("coverage_mismatch")
|
||
got = spread(vs, s["width"])
|
||
if got != s and "spread" not in SKIP:
|
||
raise Refuse("distances_mismatch")
|
||
return {"kind": "wai-collection", "link": link.hex(), "members": len(members), "occupied": occupied, "cells": n_cells,
|
||
"distances": s}
|
||
|
||
|
||
# ---------------------------------------------------------------- §9 rankings
|
||
|
||
RANKING = ["kind", "prior", "descriptors", "candidates", "scores", "order", "usage", "issued_at", "signer_pubkey", "sig"]
|
||
|
||
|
||
def has_term(vocabulary, t):
|
||
"""ai-usage §6.2 rule 4."""
|
||
if vocabulary == "aipref":
|
||
return t in AIPREF_LABELS
|
||
if vocabulary == "tdmrep":
|
||
return t == "tdm-reservation"
|
||
return t in CAWG_KEYS or (AU.cawg_key(t) and not t.startswith("cawg."))
|
||
|
||
|
||
def read_usage(x):
|
||
need(isinstance(x, list) and len(x) <= 3)
|
||
out = []
|
||
for e in x:
|
||
exact(e, ["vocabulary", "term"])
|
||
need(e["vocabulary"] in VOCABULARIES)
|
||
t = e["term"]
|
||
need(t is None or (isinstance(t, str) and 1 <= len(t.encode("utf-8")) <= 255 and has_term(e["vocabulary"], t)))
|
||
need(not out or out[-1][0] < e["vocabulary"])
|
||
out.append((e["vocabulary"], t))
|
||
return out
|
||
|
||
|
||
def order_of(scores):
|
||
"""§9.2."""
|
||
return sorted(range(len(scores)), key=lambda i: (-scores[i], i))
|
||
|
||
|
||
def verify_ranking(text, case):
|
||
held = case["held"]
|
||
try:
|
||
v, signer, link = read_record(text, "wai-ranking", RANKING)
|
||
digest = read_pin(v["prior"])
|
||
need(v["descriptors"] == DESCRIPTOR_SET)
|
||
need(isinstance(v["candidates"], list) and 1 <= len(v["candidates"]) <= 256)
|
||
cands = [read_picture(x, True) for x in v["candidates"]]
|
||
need(isinstance(v["scores"], list) and len(v["scores"]) == len(cands))
|
||
scores = [uint(x) for x in v["scores"]]
|
||
need(isinstance(v["order"], list) and len(v["order"]) == len(cands))
|
||
order = [uint(x) for x in v["order"]]
|
||
need(sorted(order) == list(range(len(cands))))
|
||
usage = read_usage(v["usage"])
|
||
except (Bad, KeyError, TypeError):
|
||
raise Refuse("record_malformed")
|
||
check_signature(v, signer)
|
||
prior = resolve(held, digest) # §9.4 step 1
|
||
recomputed = True
|
||
for p in cands: # step 2
|
||
b = held.samples_of(p)
|
||
if b is None:
|
||
recomputed = False
|
||
elif descriptors(b, p["w"], p["h"]) != p["d"] and "descriptors" not in SKIP:
|
||
raise Refuse("descriptor_mismatch")
|
||
want = [score(prior, p["d"]) for p in cands]
|
||
if (want != scores or order_of(want) != order) and "ranking" not in SKIP: # step 3
|
||
raise Refuse("ranking_mismatch")
|
||
st = prior["usage"] # step 4, §9.3
|
||
if st is not None and "usage-answered" not in SKIP:
|
||
for voc in st["prefs"]:
|
||
if not any(u[0] == voc for u in usage):
|
||
raise Refuse("prior_usage_unanswered")
|
||
uses = []
|
||
for voc, t in usage:
|
||
if t is None:
|
||
p = "outside"
|
||
elif st is None or AU.pref(st, voc, t) is None:
|
||
p = "absent"
|
||
else:
|
||
p = AU.pref(st, voc, t)
|
||
if p == "disallow" and "usage-allowed" not in SKIP:
|
||
raise Refuse("prior_usage_disallowed")
|
||
uses.append({"vocabulary": voc, "term": t, "preference": p})
|
||
return {"kind": "wai-ranking", "link": link.hex(), "prior": digest.hex(), "owner": prior["owner"].hex(),
|
||
"scope": prior["scope"], "order": order, "descriptors": "recomputed" if recomputed else "stated", "usage": uses}
|
||
|
||
|
||
# ---------------------------------------------------------------- §10 picks
|
||
|
||
PICK = ["kind", "act", "work", "from", "issued_at", "signer_pubkey", "sig"]
|
||
|
||
|
||
def read_from(x):
|
||
need(isinstance(x, dict) and "type" in x)
|
||
if x["type"] == "archive":
|
||
exact(x, ["type", "archive", "niche"])
|
||
return ("archive", hexb(x["archive"], 32), uint(x["niche"], 0, 4095))
|
||
if x["type"] == "set":
|
||
exact(x, ["type", "set"])
|
||
need(isinstance(x["set"], list) and 1 <= len(x["set"]) <= 256)
|
||
return ("set", [hexb(s, 32) for s in x["set"]])
|
||
need(x["type"] == "generation")
|
||
exact(x, ["type", "generator", "model", "prompt", "seed", "steps"])
|
||
need(is_term(x["generator"]) and is_term(x["model"]) and is_prompt(x["prompt"]))
|
||
seed = None if x["seed"] is None else uint(x["seed"], 0, (1 << 32) - 1)
|
||
steps = None if x["steps"] is None else uint(x["steps"], 1, 64)
|
||
digest = model_digest(x["model"]) if x["generator"] == GENERATOR else None
|
||
return ("generation", x["generator"], digest, x["prompt"], seed, steps)
|
||
|
||
|
||
def verify_pick(text, case, archives):
|
||
held, policy = case["held"], case["policy"]
|
||
try:
|
||
v, signer, link = read_record(text, "wai-pick", PICK, optional="objective")
|
||
need(v["act"] in ("generate", "select"))
|
||
work = read_picture(v["work"], False)
|
||
src = read_from(v["from"])
|
||
need((v["act"] == "generate") == (src[0] == "generation"))
|
||
objective = read_objective(v["objective"]) if "objective" in v else None
|
||
except (Bad, KeyError, TypeError):
|
||
raise Refuse("record_malformed")
|
||
check_signature(v, signer)
|
||
if objective is None and policy["objective"] and "disclosure" not in SKIP: # §10.4 step 1
|
||
raise Refuse("objective_undisclosed")
|
||
if objective is not None and objective[0] == "taste": # step 2
|
||
resolve(held, objective[1])
|
||
held.samples_of(work) # step 3
|
||
genome = None
|
||
if src[0] == "archive": # step 4
|
||
if src[1] not in archives:
|
||
raise Refuse("archive_unheld")
|
||
req, elites = archives[src[1]]
|
||
e = [x for x in elites if x["niche"] == src[2]]
|
||
if not e or ((e[0]["samples"] != work["samples"] or (req["w"], req["h"]) != (work["w"], work["h"])) and "in-archive" not in SKIP):
|
||
raise Refuse("pick_not_in_archive")
|
||
if req["objective"][0] == "taste" and (objective is None or objective[1] != req["objective"][1]) and "inherit" not in SKIP:
|
||
raise Refuse("objective_mismatch")
|
||
g = e[0]["genome"]
|
||
genome = {"seed": g[0], "steps": g[1], "mods": g[2], "prompt": derived(req["prompt"], req["modifiers"], g[2])}
|
||
check = "in-archive"
|
||
elif src[0] == "set": # step 5
|
||
if work["samples"] not in src[1] and "in-set" not in SKIP:
|
||
raise Refuse("pick_not_in_set")
|
||
check = "in-set"
|
||
elif src[1] == GENERATOR and src[4] is not None and src[5] is not None: # step 6
|
||
m = held.model(src[2])
|
||
if sha(gen(m, src[3], src[4], src[5], work["w"], work["h"])) != work["samples"] and "regenerate" not in SKIP:
|
||
raise Refuse("generation_mismatch")
|
||
genome = {"seed": src[4], "steps": src[5], "mods": 0, "prompt": src[3]}
|
||
check = "regenerated"
|
||
else:
|
||
check = "stated"
|
||
return {"kind": "wai-pick", "link": link.hex(), "act": v["act"],
|
||
"objective": "undisclosed" if objective is None else objective_report(objective), "check": check, "genome": genome}
|
||
|
||
|
||
# ---------------------------------------------------------------- §12 a case
|
||
|
||
def kind_of(text):
|
||
try:
|
||
v = strict_json(text)
|
||
return v.get("kind") if isinstance(v, dict) else None
|
||
except Bad:
|
||
return None
|
||
|
||
|
||
def verify_case(case):
|
||
archives, statements = {}, []
|
||
for i, text in enumerate(case["records"]):
|
||
k = kind_of(text)
|
||
try:
|
||
if k == "wai-archive":
|
||
s = verify_archive(text, case, archives)
|
||
elif k == "wai-distance":
|
||
s = verify_distance(text, case)
|
||
elif k == "wai-collection":
|
||
s = verify_collection(text, case)
|
||
elif k == "wai-ranking":
|
||
s = verify_ranking(text, case)
|
||
elif k == "wai-pick":
|
||
s = verify_pick(text, case, archives)
|
||
else:
|
||
raise Refuse("record_malformed")
|
||
except Refuse as r:
|
||
return {"outcome": "refuse", "reason": r.reason, "record": i}
|
||
statements.append(s)
|
||
return {"outcome": "accept", "statements": statements}
|
||
|
||
|
||
def source_bytes(m, s):
|
||
if "generate" in s:
|
||
g = s["generate"]
|
||
return gen(m, g["prompt"], g["seed"], g["steps"], g["width"], g["height"])
|
||
return bytes.fromhex(s["hex"])
|
||
|
||
|
||
def case_of(doc, c, model, model_sha):
|
||
held = Held({bytes.fromhex(d): model for d in c["held"]["models"] if bytes.fromhex(d) == model_sha},
|
||
{sha(p.encode("utf-8")): p.encode("utf-8") for p in c["held"]["priors"]},
|
||
{sha(b): b for b in (source_bytes(model, s) for s in c["held"]["samples"])})
|
||
return {"records": c["records"], "held": held,
|
||
"policy": {"replay": c["policy"]["replay"] == "required", "objective": c["policy"]["objective"] == "required"}}
|
||
|
||
|
||
def load_model(doc):
|
||
with open(os.path.join(WAI, doc["model"]["path"]), "rb") as f:
|
||
b = f.read()
|
||
return Model(json.loads(b)), sha(b)
|
||
|
||
|
||
def check(doc, quiet=False, only=None):
|
||
fails = []
|
||
model, model_sha = load_model(doc)
|
||
if model_sha.hex() != doc["model"]["sha256"]:
|
||
fails.append("the model file's SHA-256 is not the corpus's")
|
||
if only is None:
|
||
bad = [g for g in doc["generations"]
|
||
if sha(gen(model, g["prompt"], g["seed"], g["steps"], g["width"], g["height"])).hex() != g["sha256"]]
|
||
fails += ["generation %r seed %d: another picture" % (g["prompt"], g["seed"]) for g in bad]
|
||
if not quiet:
|
||
print(("PASS" if not bad else "FAIL") + " Appendix A: %d generations make their SHA-256" % len(doc["generations"]))
|
||
bad = [d for d in doc["descriptors"] if descriptors(source_bytes(model, d), d["width"], d["height"]) != d["descriptors"]]
|
||
fails += ["descriptor vector %s: another vector" % d["about"] for d in bad]
|
||
if not quiet:
|
||
print(("PASS" if not bad else "FAIL") + " §3: %d descriptor vectors recompute" % len(doc["descriptors"]))
|
||
for c in doc["cases"]:
|
||
if only is not None and c["name"] not in only:
|
||
continue
|
||
try:
|
||
got = verify_case(case_of(doc, c, model, model_sha))
|
||
except (KeyError, TypeError, ValueError, AssertionError) as e:
|
||
got = {"outcome": "error", "error": repr(e)}
|
||
want = {k: c[k] for k in ("outcome", "reason", "record", "statements") if k in c}
|
||
ok = got == want
|
||
if not ok:
|
||
fails.append("%s: got %s" % (c["name"], json.dumps(got)[:200]))
|
||
if not quiet:
|
||
print(("PASS" if ok else "FAIL") + " %s: %s%s" % (c["name"], c["outcome"], " " + c["reason"] if "reason" in c else ""))
|
||
return fails
|
||
|
||
|
||
# ---------------------------------------------------------------- the corpus's own checks
|
||
|
||
def reasons_of_extension():
|
||
with open(EXTENSION, encoding="utf-8") as f:
|
||
text = f.read()
|
||
at = text.index("\n## 13. Reasons")
|
||
end = text.index("\n## ", at + 1)
|
||
return [line.split("`")[1] for line in text[at:end].splitlines() if line.startswith("| `")]
|
||
|
||
|
||
REASONS = reasons_of_extension()
|
||
|
||
|
||
def resign_free(doc, case, i, f):
|
||
"""A copy of the corpus whose case `case` has record `i` edited by `f`, not re-signed."""
|
||
d = copy.deepcopy(doc)
|
||
c = next(x for x in d["cases"] if x["name"] == case)
|
||
v = json.loads(c["records"][i])
|
||
f(v)
|
||
c["records"][i] = json.dumps(v, separators=(",", ":"))
|
||
return d, case
|
||
|
||
|
||
def edit_case(doc, case, f):
|
||
d = copy.deepcopy(doc)
|
||
f(next(x for x in d["cases"] if x["name"] == case))
|
||
return d, case
|
||
|
||
|
||
def mutations(doc):
|
||
"""Copies of the corpus with one defect each, and the case each touches."""
|
||
def setp(path, x):
|
||
def f(v):
|
||
at = v
|
||
for p in path[:-1]:
|
||
at = at[p]
|
||
at[path[-1]] = x
|
||
return f
|
||
|
||
out = []
|
||
out.append(("an elite's seed changed (the signature now fails)", resign_free(doc, "archive-novelty", 0, setp(["elites", 0, "seed"], 1))))
|
||
out.append(("an archive's prompt changed", resign_free(doc, "archive-novelty", 0, setp(["prompt"], "a loud forest at noon"))))
|
||
out.append(("a distance's class changed", resign_free(doc, "distance-measured-pulled", 0, setp(["class"], "estimate"))))
|
||
out.append(("a ranking's order changed", resign_free(doc, "ranking-personal", 0, setp(["order", 0], 5))))
|
||
out.append(("a pick's objective dropped", resign_free(doc, "pick-from-steered-archive", 1, lambda v: v.pop("objective"))))
|
||
out.append(("the model not held", edit_case(doc, "archive-novelty", lambda c: c["held"].__setitem__("models", []))))
|
||
out.append(("the taste prior not held", edit_case(doc, "archive-taste", lambda c: c["held"].__setitem__("priors", []))))
|
||
out.append(("the policy switched to elite checks", edit_case(doc, "archive-elite-dropped", lambda c: c["policy"].__setitem__("replay", "elites"))))
|
||
out.append(("the policy stops requiring disclosure", edit_case(doc, "pick-undisclosed-required", lambda c: c["policy"].__setitem__("objective", "optional"))))
|
||
out.append(("a pulled work's samples replaced by a default's", edit_case(doc, "distance-measured-pulled",
|
||
lambda c: c["held"]["samples"][0]["generate"].__setitem__("seed", 3))))
|
||
out.append(("a collection member's samples dropped", edit_case(doc, "collection-defaults", lambda c: c["held"]["samples"].pop())))
|
||
out.append(("an accepted statement's figure changed", edit_case(doc, "distance-measured-default",
|
||
lambda c: c["statements"][0].__setitem__("sample_distance", 1))))
|
||
out.append(("a refusal's reason changed", edit_case(doc, "archive-niche-wrong", lambda c: c.__setitem__("reason", "descriptor_mismatch"))))
|
||
out.append(("a prior's text given with a trailing space", edit_case(doc, "ranking-personal",
|
||
lambda c: c["held"]["priors"].__setitem__(0, c["held"]["priors"][0] + " "))))
|
||
return out
|
||
|
||
|
||
CHECKS = {
|
||
"signature": "§2 the signature",
|
||
"coverage": "§5.4 step 2 and §7.3 the coverage figures",
|
||
"regenerate": "§5.4 step 5 and §10.4 step 6 regeneration",
|
||
"descriptors": "the descriptor recomputation",
|
||
"niche": "§5.4 step 5 the niche",
|
||
"score": "§5.4 step 5 the score",
|
||
"replay": "§5.4 step 6 the replay",
|
||
"class": "§6.4 step 1 the class",
|
||
"unknown-null": "§6.3 unknown is null",
|
||
"defaults": "§6.4 step 3 the default set",
|
||
"figures": "§6.4 step 6 the figures",
|
||
"spread": "§7.3 the distances",
|
||
"ranking": "§9.4 step 3 the order",
|
||
"usage-answered": "§9.3 every vocabulary answered",
|
||
"usage-allowed": "§9.3 no disallowed use",
|
||
"owner": "§8.4 the owner's signature",
|
||
"canonical-prior": "§8.1 the canonical file",
|
||
"disclosure": "§10.3 the disclosure policy",
|
||
"inherit": "§10.2 a steered archive's objective",
|
||
"in-archive": "§10.4 step 4 the elite",
|
||
"in-set": "§10.4 step 5 the set",
|
||
"sample-size": "the held samples' size",
|
||
}
|
||
|
||
|
||
def main(corpus_dir):
|
||
fails = []
|
||
st = ED._self_test()
|
||
print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
|
||
fails += ["ed25519 self-test"] if st else []
|
||
print(("PASS" if len(REASONS) == 27 else "FAIL") + " §13 read from the extension: %d reasons" % len(REASONS))
|
||
fails += [] if len(REASONS) == 27 else ["§13 does not read"]
|
||
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
||
doc = json.load(f)
|
||
fails += check(doc)
|
||
extra = sorted(set(os.listdir(corpus_dir)) - {"vectors.json", "README.md"})
|
||
print(("PASS" if not extra else "FAIL") + " no file but vectors.json and README.md")
|
||
fails += ["an extra file: %s" % n for n in extra]
|
||
given = {c["reason"] for c in doc["cases"] if c["outcome"] == "refuse"}
|
||
missing = [r for r in REASONS if r not in given]
|
||
unknown = sorted(given - set(REASONS))
|
||
print(("PASS" if not missing and not unknown else "FAIL") + " every reason of §13 refuses a case, and no other reason does")
|
||
fails += ["no case gives %s" % r for r in missing] + ["a reason §13 does not list: %s" % r for r in unknown]
|
||
for label_, (d, case) in mutations(doc):
|
||
caught = check(d, quiet=True, only={case})
|
||
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label_, " (%s)" % caught[0][:90] if caught else ""))
|
||
if not caught:
|
||
fails.append("a mutated copy verifies: %s" % label_)
|
||
for name, what in CHECKS.items():
|
||
SKIP.clear()
|
||
SKIP.add(name)
|
||
caught = check(doc, quiet=True, only={c["name"] for c in doc["cases"]})
|
||
SKIP.clear()
|
||
print(("PASS" if caught else "FAIL") + " a verifier without %s fails the corpus%s" % (what, " (%s)" % caught[0][:70] if caught else ""))
|
||
if not caught:
|
||
fails.append("the corpus does not catch a verifier without %s" % what)
|
||
for f in fails:
|
||
print("FAIL " + f)
|
||
print(("PASS" if not fails else "FAIL") + " exploration corpus: %d generations, %d descriptor vectors, %d cases"
|
||
% (len(doc["generations"]), len(doc["descriptors"]), len(doc["cases"])))
|
||
return 1 if fails else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) != 2:
|
||
print("usage: exploration_verify.py <live-conformance/exploration>", file=sys.stderr)
|
||
sys.exit(2)
|
||
sys.exit(main(sys.argv[1]))
|